Added Kafka reporter for zipkin

src/tracer/composer.json

@@ -28,6 +28,7 @@
     },
     "suggest": {
         "hyperf/event": "Required to use DbQueryExecutedListener.",
+        "longlang/phpkafka": "Required (^1.2) to use Kafka Producer.",
         "jonahgeorge/jaeger-client-php": "Required (^0.6) to use jaeger tracing."
     },
     "autoload": {

src/tracer/publish/opentracing.php

@@ -32,9 +32,34 @@
                 'ipv6' => null,
                 'port' => 9501,
             ],
-            'options' => [
-                'endpoint_url' => env('ZIPKIN_ENDPOINT_URL', 'http://localhost:9411/api/v2/spans'),
-                'timeout' => env('ZIPKIN_TIMEOUT', 1),
+            'reporter' => env('ZIPKIN_REPORTER', 'http'), // kafka, http
+            'reporters' => [
+                // options for http reporter
+                'http' => [
+                    'class' => \Zipkin\Reporters\Http::class,
+                    'constructor' => [
+                        'options' => [
+                            'endpoint_url' => env('ZIPKIN_ENDPOINT_URL', 'http://localhost:9411/api/v2/spans'),
+                            'timeout' => env('ZIPKIN_TIMEOUT', 1),
+                        ],
+                    ],
+                ],
+                // options for kafka reporter
+                'kafka' => [
+                    'class' => \Hyperf\Tracer\Adapter\Reporter\Kafka::class,
+                    'constructor' => [
+                        'options' => [
+                            'topic' => env('ZIPKIN_KAFKA_TOPIC', 'zipkin'),
+                            'bootstrap_servers' => env('ZIPKIN_KAFKA_BOOTSTRAP_SERVERS', '127.0.0.1:9092'),
+                            'acks' => (int) env('ZIPKIN_KAFKA_ACKS', -1),
+                            'connect_timeout' => (int) env('ZIPKIN_KAFKA_CONNECT_TIMEOUT', 1),
+                            'send_timeout' => (int) env('ZIPKIN_KAFKA_SEND_TIMEOUT', 1),
+                        ],
+                    ],
+                ],
+                'noop' => [
+                    'class' => \Zipkin\Reporters\Noop::class,
+                ],
             ],
             'sampler' => BinarySampler::createAsAlwaysSample(),
         ],

src/tracer/src/Adapter/Reporter/Kafka.php

@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Tracer\Adapter\Reporter;
+
+use longlang\phpkafka\Producer\Producer;
+use longlang\phpkafka\Producer\ProducerConfig;
+use Psr\Log\LoggerInterface;
+use Psr\Log\NullLogger;
+use Throwable;
+use Zipkin\Recording\Span;
+use Zipkin\Reporter;
+use Zipkin\Reporters\JsonV2Serializer;
+use Zipkin\Reporters\SpanSerializer;
+
+use function sprintf;
+
+class Kafka implements Reporter
+{
+    private Producer $producer;
+
+    private string $topic;
+
+    private LoggerInterface $logger;
+
+    private SpanSerializer $serializer;
+
+    public function __construct(
+        array $options = [],
+        Producer $producer = null,
+        LoggerInterface $logger = null,
+        SpanSerializer $serializer = null
+    ) {
+        $this->topic = $options['topic'] ?? 'zipkin';
+        $this->serializer = $serializer ?? new JsonV2Serializer();
+        $this->logger = $logger ?? new NullLogger();
+        $this->producer = $producer ?? $this->createProducer($options);
+    }
+
+    /**
+     * @param array|Span[] $spans
+     */
+    public function report(array $spans): void
+    {
+        if (empty($spans)) {
+            return;
+        }
+
+        try {
+            $this->producer->send(
+                $this->topic,
+                $this->serializer->serialize($spans),
+                uniqid('', true)
+            );
+        } catch (Throwable $e) {
+            $this->logger->error(sprintf('failed to report spans: %s', $e->getMessage()));
+        }
+    }
+
+    private function createProducer(array $options): Producer
+    {
+        $options = array_replace([
+            'bootstrap_servers' => '127.0.0.1:9092',
+            'acks' => -1,
+            'connect_timeout' => 1,
+            'send_timeout' => 1,
+        ], $options);
+        $config = new ProducerConfig();
+
+        $config->setBootstrapServer($options['bootstrap_servers']);
+        $config->setUpdateBrokers(true);
+        if (is_int($options['acks'])) {
+            $config->setAcks($options['acks']);
+        }
+        if (is_float($options['connect_timeout'])) {
+            $config->setConnectTimeout($options['connect_timeout']);
+        }
+        if (is_float($options['send_timeout'])) {
+            $config->setSendTimeout($options['send_timeout']);
+        }
+
+        return new Producer($config);
+    }
+}

src/tracer/src/Adapter/Reporter/ReporterFactory.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Tracer\Adapter\Reporter;
+
+use RuntimeException;
+use Zipkin\Reporter;
+
+use function Hyperf\Support\make;
+
+class ReporterFactory
+{
+    public function make(array $option = []): Reporter
+    {
+        $class = $option['class'] ?? '';
+        $constructor = $option['constructor'] ?? [];
+
+        if (! class_exists($class)) {
+            throw new RuntimeException(sprintf('Class %s is not exists.', $class));
+        }
+
+        if (! is_a($class, Reporter::class, true)) {
+            throw new RuntimeException('Unsupported reporter.');
+        }
+
+        return make($class, $constructor);
+    }
+}

src/tracer/src/Adapter/ZipkinTracerFactory.php

@@ -12,9 +12,9 @@
 namespace Hyperf\Tracer\Adapter;
 
 use Hyperf\Contract\ConfigInterface;
+use Hyperf\Tracer\Adapter\Reporter\ReporterFactory;
 use Hyperf\Tracer\Contract\NamedFactoryInterface;
 use Zipkin\Endpoint;
-use Zipkin\Reporters\Http;
 use Zipkin\Samplers\BinarySampler;
 use Zipkin\TracingBuilder;
 use ZipkinOpenTracing\Tracer;
@@ -25,16 +25,16 @@ class ZipkinTracerFactory implements NamedFactoryInterface
 
     private string $name = '';
 
-    public function __construct(private ConfigInterface $config, private HttpClientFactory $clientFactory)
+    public function __construct(private ConfigInterface $config, private ReporterFactory $reportFactory)
     {
     }
 
     public function make(string $name): \OpenTracing\Tracer
     {
         $this->name = $name;
-        [$app, $options, $sampler] = $this->parseConfig();
+        [$app, $sampler, $reporterOption] = $this->parseConfig();
         $endpoint = Endpoint::create($app['name'], $app['ipv4'], $app['ipv6'], $app['port']);
-        $reporter = new Http($options, $this->clientFactory);
+        $reporter = $this->reportFactory->make($reporterOption);
         $tracing = TracingBuilder::create()
             ->havingLocalEndpoint($endpoint)
             ->havingSampler($sampler)
@@ -46,17 +46,21 @@ public function make(string $name): \OpenTracing\Tracer
     private function parseConfig(): array
     {
         // @TODO Detect the ipv4, ipv6, port from server object or system info automatically.
+        $reporter = (string) $this->getConfig('reporter', 'http');
         return [
             $this->getConfig('app', [
                 'name' => 'skeleton',
                 'ipv4' => '127.0.0.1',
                 'ipv6' => null,
                 'port' => 9501,
             ]),
-            $this->getConfig('options', [
-                'timeout' => 1,
-            ]),
             $this->getConfig('sampler', BinarySampler::createAsAlwaysSample()),
+            $this->getConfig('reporters.' . $reporter, [
+                'class' => \Zipkin\Reporters\Http::class,
+                'constructor' => [
+                    'options' => $this->getConfig('options', []),
+                ],
+            ]),
         ];
     }
 

src/tracer/tests/TracerFactoryTest.php

@@ -134,10 +134,13 @@ protected function getContainer($config)
     {
         $container = Mockery::mock(Container::class);
         $client = Mockery::mock(\Hyperf\Tracer\Adapter\HttpClientFactory::class);
+        $reporter = Mockery::mock(\Hyperf\Tracer\Adapter\Reporter\ReporterFactory::class);
+        $reporter->shouldReceive('make')
+            ->andReturn(new \Zipkin\Reporters\Http([], $client));
 
         $container->shouldReceive('get')
             ->with(\Hyperf\Tracer\Adapter\ZipkinTracerFactory::class)
-            ->andReturn(new \Hyperf\Tracer\Adapter\ZipkinTracerFactory($config, $client));
+            ->andReturn(new \Hyperf\Tracer\Adapter\ZipkinTracerFactory($config, $reporter));
         $container->shouldReceive('get')
             ->with(\Hyperf\Tracer\Adapter\JaegerTracerFactory::class)
             ->andReturn(new \Hyperf\Tracer\Adapter\JaegerTracerFactory($config));