Optimize kafka reporter (#6075)

hyperf · Aug 23, 2023 · 2fbef18 · 2fbef18
1 parent bdc7deb
commit 2fbef18
Show file tree

Hide file tree

Showing 6 changed files with 139 additions and 77 deletions.
diff --git a/src/Adapter/HttpClientFactory.php b/src/Adapter/HttpClientFactory.php
@@ -11,41 +11,9 @@
  */
 namespace Hyperf\Tracer\Adapter;
 
-use Hyperf\Guzzle\ClientFactory as GuzzleClientFactory;
-use RuntimeException;
-use Zipkin\Reporters\Http\ClientFactory;
-
-class HttpClientFactory implements ClientFactory
+/**
+ * @deprecated v3.0, will remove in v3.1, use \Hyperf\Tracer\Adapter\Reporter\HttpClientFactory instead.
+ */
+class HttpClientFactory extends Reporter\HttpClientFactory
 {
-    public function __construct(private GuzzleClientFactory $guzzleClientFactory)
-    {
-    }
-
-    public function build(array $options): callable
-    {
-        return function (string $payload) use ($options): void {
-            $url = $options['endpoint_url'];
-            unset($options['endpoint_url']);
-            $client = $this->guzzleClientFactory->create($options);
-            $additionalHeaders = $options['headers'] ?? [];
-            $requiredHeaders = [
-                'Content-Type' => 'application/json',
-                'Content-Length' => strlen($payload),
-                'b3' => '0',
-            ];
-            $headers = array_merge($additionalHeaders, $requiredHeaders);
-            $response = $client->post($url, [
-                'body' => $payload,
-                'headers' => $headers,
-                // If 'no_aspect' option is true, then the HttpClientAspect will not modify the client options.
-                'no_aspect' => true,
-            ]);
-            $statusCode = $response->getStatusCode();
-            if ($statusCode !== 202) {
-                throw new RuntimeException(
-                    sprintf('Reporting of spans failed, status code %d', $statusCode)
-                );
-            }
-        };
-    }
 }
diff --git a/src/Adapter/Reporter/HttpClientFactory.php b/src/Adapter/Reporter/HttpClientFactory.php
@@ -0,0 +1,51 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Tracer\Adapter\Reporter;
+
+use Hyperf\Guzzle\ClientFactory as GuzzleClientFactory;
+use RuntimeException;
+use Zipkin\Reporters\Http\ClientFactory;
+
+class HttpClientFactory implements ClientFactory
+{
+    public function __construct(private GuzzleClientFactory $guzzleClientFactory)
+    {
+    }
+
+    public function build(array $options): callable
+    {
+        return function (string $payload) use ($options): void {
+            $url = $options['endpoint_url'];
+            unset($options['endpoint_url']);
+            $client = $this->guzzleClientFactory->create($options);
+            $additionalHeaders = $options['headers'] ?? [];
+            $requiredHeaders = [
+                'Content-Type' => 'application/json',
+                'Content-Length' => strlen($payload),
+                'b3' => '0',
+            ];
+            $headers = array_merge($additionalHeaders, $requiredHeaders);
+            $response = $client->post($url, [
+                'body' => $payload,
+                'headers' => $headers,
+                // If 'no_aspect' option is true, then the HttpClientAspect will not modify the client options.
+                'no_aspect' => true,
+            ]);
+            $statusCode = $response->getStatusCode();
+            if ($statusCode !== 202) {
+                throw new RuntimeException(
+                    sprintf('Reporting of spans failed, status code %d', $statusCode)
+                );
+            }
+        };
+    }
+}
diff --git a/src/Adapter/Reporter/Kafka.php b/src/Adapter/Reporter/Kafka.php
@@ -11,8 +11,6 @@
  */
 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;
@@ -21,72 +19,50 @@
 use Zipkin\Reporters\JsonV2Serializer;
 use Zipkin\Reporters\SpanSerializer;
 
+use function count;
+use function json_last_error;
 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,
+        private array $options,
+        private KafkaClientFactory $clientFactory,
         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)) {
+        if (count($spans) === 0) {
             return;
         }
 
-        try {
-            $this->producer->send(
-                $this->topic,
-                $this->serializer->serialize($spans),
-                uniqid('', true)
+        $payload = $this->serializer->serialize($spans);
+
+        if (! $payload) {
+            $this->logger->error(
+                sprintf('failed to encode spans with code %d', json_last_error())
             );
-        } catch (Throwable $e) {
-            $this->logger->error(sprintf('failed to report spans: %s', $e->getMessage()));
+            return;
         }
-    }
 
-    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();
+        $client = $this->clientFactory->build($this->options);
 
-        $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']);
+        try {
+            $client($payload);
+        } catch (Throwable $e) {
+            $this->logger->error(sprintf('failed to report spans: %s', $e->getMessage()));
         }
-
-        return new Producer($config);
     }
 }
diff --git a/src/Adapter/Reporter/KafkaClientFactory.php b/src/Adapter/Reporter/KafkaClientFactory.php
@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+namespace Hyperf\Tracer\Adapter\Reporter;
+
+use longlang\phpkafka\Producer\Producer;
+use longlang\phpkafka\Producer\ProducerConfig;
+
+class KafkaClientFactory
+{
+    private ?Producer $producer = null;
+
+    public function build(array $options): callable
+    {
+        $this->producer ??= $this->createProducer($options);
+
+        return function (string $payload) use ($options): void {
+            $this->producer->send(
+                $options['topic'] ?? 'zipkin',
+                $payload,
+                uniqid('', true)
+            );
+        };
+    }
+
+    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);
+    }
+}
diff --git a/src/Adapter/Reporter/ReporterFactory.php b/src/Adapter/Reporter/ReporterFactory.php
@@ -18,11 +18,20 @@
 
 class ReporterFactory
 {
+    public function __construct(
+        private HttpClientFactory $httpClientFactory,
+    ) {
+    }
+
     public function make(array $option = []): Reporter
     {
         $class = $option['class'] ?? '';
         $constructor = $option['constructor'] ?? [];
 
+        if ($class === \Zipkin\Reporters\Http::class) {
+            $option['constructor']['requesterFactory'] = $this->httpClientFactory;
+        }
+
         if (! class_exists($class)) {
             throw new RuntimeException(sprintf('Class %s is not exists.', $class));
         }

diff --git a/tests/TracerFactoryTest.php b/tests/TracerFactoryTest.php
@@ -133,7 +133,7 @@ public function testJaegerFactory()
     protected function getContainer($config)
     {
         $container = Mockery::mock(Container::class);
-        $client = Mockery::mock(\Hyperf\Tracer\Adapter\HttpClientFactory::class);
+        $client = Mockery::mock(\Hyperf\Tracer\Adapter\Reporter\HttpClientFactory::class);
         $reporter = Mockery::mock(\Hyperf\Tracer\Adapter\Reporter\ReporterFactory::class);
         $reporter->shouldReceive('make')
             ->andReturn(new \Zipkin\Reporters\Http([], $client));