Add Memcached Reporter

composer.json

@@ -22,6 +22,7 @@
     "require": {
         "php": "^7.3 || ^8.0",
         "ext-curl": "*",
+        "ext-memcached": "*",
         "psr/http-message": "~1.0",
         "psr/log": "^1.0"
     },

src/Zipkin/Reporters/Aggregation/MemcachedClient.php

@@ -0,0 +1,111 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Zipkin\Reporters\Aggregation;
+
+use Memcached;
+use Exception;
+
+class MemcachedClient
+{
+    const GET_EXTENDED = Memcached::GET_EXTENDED;
+
+    /**
+     * @var Memcached
+     */
+    private $client;
+
+    /**
+     * @var string
+     */
+    private $server;
+
+    /**
+     * @var int
+     */
+    private $port;
+
+    /**
+     * @param string       $server
+     * @param int          $port
+     * @param bool         $enableCompression
+     */
+    public function __construct(
+        string $server = '127.0.0.1',
+        int $port = 11211,
+        bool $enableCompression = true
+    ) {
+        $this->server = $server;
+        $this->port = $port;
+
+        $this->client = new Memcached();
+        $this->client->setOption(Memcached::OPT_COMPRESSION, $enableCompression);
+        $this->client->addServer($this->server, $this->port);
+    }
+
+    /**
+     * Check connection
+     *
+     * @return bool
+     */
+    public function ping(): bool
+    {
+        if (false === @fsockopen($this->server, $this->port)) {
+            throw new Exception(
+                "Unable to connect to memcached server {$this->server}:{$this->port}"
+            );
+        }
+
+        return true;
+    }
+
+    /**
+     * Set an item
+     *
+     * @param string $key
+     * @param mixed  $value
+     * @param int    $expiration
+     */
+    public function set($key, $value, $expiration = 0): bool
+    {
+        return $this->client->set($key, $value, $expiration);
+    }
+
+    /**
+     * Get item by key.
+     *
+     * @param string $key
+     * @param mixed  $cacheCallback
+     * @param int    $flags
+     *
+     * @return mixed
+     */
+    public function get($key, $cacheCallback = null, $flags = null)
+    {
+        return $this->client->get($key, $cacheCallback, $flags);
+    }
+
+    /**
+     * Compare and swap an item.
+     *
+     * @param float  $casToken
+     * @param string $key
+     * @param mixed  $value
+     * @param int    $expiration
+     */
+    public function cas($casToken, $key, $value, $expiration = 0): bool
+    {
+        return $this->client->cas($casToken, $key, $value, $expiration);
+    }
+
+    /**
+     * Quit all connections.
+     *
+     * @return bool
+     */
+    public function quit(): bool
+    {
+        return $this->client->quit();
+    }
+}

src/Zipkin/Reporters/Http.php

@@ -73,7 +73,7 @@ public function __construct(
     }
 
     /**
-     * @param ReadbackSpan[] $spans
+     * @param ReadbackSpan[]|array $spans
      * @return void
      */
     public function report(array $spans): void
@@ -82,7 +82,8 @@ public function report(array $spans): void
             return;
         }
 
-        $payload = $this->serializer->serialize($spans);
+        $payload = (is_object($spans[0])) ? $this->serializer->serialize($spans) : json_encode($spans);
+
         if ($payload === false) {
             $this->logger->error(
                 \sprintf('failed to encode spans with code %d', \json_last_error())
@@ -91,6 +92,7 @@ public function report(array $spans): void
         }
 
         $client = $this->clientFactory->build($this->options);
+
         try {
             $client($payload);
         } catch (RuntimeException $e) {

src/Zipkin/Reporters/Memcached.php

@@ -0,0 +1,176 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Zipkin\Reporters;
+
+use Zipkin\Reporter;
+use Psr\Log\NullLogger;
+use Psr\Log\LoggerInterface;
+use Zipkin\Reporters\SpanSerializer;
+use Zipkin\Reporters\JsonV2Serializer;
+use Zipkin\Reporters\Aggregation\MemcachedClient;
+use Exception;
+
+final class Memcached implements Reporter
+{
+    public const DEFAULT_OPTIONS = [
+        'cache_key' => 'zipkin_traces',
+    ];
+
+    /**
+     * @var array
+     */
+    private $options;
+
+    /**
+     * @var MemcachedClient
+     */
+    private $memcachedClient;
+
+    /**
+     * @var LoggerInterface
+     */
+    private $logger;
+
+    /**
+     * @var SpanSerializer
+     */
+    private $serializer;
+
+    /**
+     * @param array           $options
+     * @param MemcachedClient $memcachedClient
+     * @param LoggerInterface $logger
+     * @param SpanSerializer  $serializer
+     */
+    public function __construct(
+        array $options = [],
+        MemcachedClient $memcachedClient = null,
+        LoggerInterface $logger = null,
+        SpanSerializer $serializer = null
+    ) {
+        $this->options = \array_merge(self::DEFAULT_OPTIONS, $options);
+        $this->memcachedClient = $memcachedClient ?? new MemcachedClient();
+        $this->logger = $logger ?? new NullLogger();
+        $this->serializer = $serializer ?? new JsonV2Serializer();
+    }
+
+    /**
+     * @param  array  $spans
+     */
+    public function report(array $spans): void
+    {
+        try {
+            $this->memcachedClient->ping();
+
+            // Fetch stored spans
+            $result = $this->memcachedClient->get(
+                $this->options['cache_key'],
+                null,
+                MemcachedClient::GET_EXTENDED
+            );
+
+            $payload = $this->serializer->serialize($spans);
+
+            if ($payload === false) {
+                $this->logger->error(
+                    \sprintf('failed to encode spans with code %d', \json_last_error())
+                );
+            }
+
+            // Store spans if there aren't any previous spans
+            if (empty($result)) {
+                $this->memcachedClient->set($this->options['cache_key'], $payload);
+                $this->memcachedClient->quit();
+                return;
+            }
+
+            $status = false;
+
+            // Merge the new spans with the stored spans only if
+            // the item not updated by a different concurrent proceess
+            while (!$status) {
+                $result['value'] = array_merge(
+                    json_decode($result['value'], true),
+                    json_decode($payload, true)
+                );
+
+                $status = $this->memcachedClient->cas(
+                    $result['cas'],
+                    $this->options['cache_key'],
+                    json_encode($result['value'])
+                );
+
+                if (!$status) {
+                    $result = $this->memcachedClient->get(
+                        $this->options['cache_key'],
+                        null,
+                        MemcachedClient::GET_EXTENDED
+                    );
+                }
+            }
+
+            $this->memcachedClient->quit();
+        } catch (Exception $e) {
+            $this->logger->error(
+                \sprintf('Error while calling memcached server: %s', $e->getMessage())
+            );
+        }
+
+        return;
+    }
+
+    /**
+     * @return array
+     */
+    public function flush(): array
+    {
+        try {
+            $this->memcachedClient->ping();
+
+            // Fetch stored spans
+            $result = $this->memcachedClient->get(
+                $this->options['cache_key'],
+                null,
+                MemcachedClient::GET_EXTENDED
+            );
+
+            if (empty($result)) {
+                $this->memcachedClient->quit();
+
+                return [];
+            }
+
+            $status = false;
+
+            // Return stored spans and set the key value as empty only if
+            // the item not updated by a different concurrent proceess
+            while (!$status) {
+                $status = $this->memcachedClient->cas(
+                    $result['cas'],
+                    $this->options['cache_key'],
+                    json_encode([])
+                );
+
+                if (!$status) {
+                    $result = $this->memcachedClient->get(
+                        $this->options['cache_key'],
+                        null,
+                        MemcachedClient::GET_EXTENDED
+                    );
+                }
+            }
+
+            $this->memcachedClient->quit();
+
+            return json_decode($result['value'], true);
+        } catch (Exception $e) {
+            $this->logger->error(
+                \sprintf('Error while calling memcached server: %s', $e->getMessage())
+            );
+        }
+
+        return [];
+    }
+}