Merge pull request #43 from dapr/fix/tracing

Fix distributed tracing

Author: withinboredom

src/config.php

@@ -14,6 +14,7 @@
 use Dapr\Deserialization\DeserializationConfig;
 use Dapr\Deserialization\Deserializer;
 use Dapr\Deserialization\IDeserializer;
+use Dapr\DistributedTracing\ActiveTrace;
 use Dapr\PubSub\Publish;
 use Dapr\PubSub\Subscriptions;
 use Dapr\PubSub\Topic;
@@ -24,6 +25,7 @@
 use Dapr\State\IManageState;
 use Dapr\State\StateManager;
 use Dapr\State\TransactionalState;
+use DI\Container;
 use FastRoute\DataGenerator\GroupCountBased;
 use FastRoute\RouteCollector;
 use FastRoute\RouteParser\Std;
@@ -32,12 +34,14 @@
 use Monolog\Processor\PsrLogMessageProcessor;
 use Nyholm\Psr7\Factory\Psr17Factory;
 use Nyholm\Psr7Server\ServerRequestCreator;
+use Psr\Http\Message\RequestInterface;
 use Psr\Log\LoggerInterface;
 use Psr\Log\LogLevel;
 
 use function DI\autowire;
 use function DI\create;
 use function DI\env;
+use function DI\factory;
 use function DI\get;
 
 return [
@@ -63,6 +67,13 @@
     ),
 
     // SDK wiring
+    ActiveTrace::class             => factory(
+        fn(Container $container) => ActiveTrace::get_from_request(
+            $container->get(
+                RequestInterface::class
+            )
+        )
+    ),
     ActorConfig::class             => autowire()
         ->constructorParameter('actor_name_to_type', get('dapr.actors'))
         ->constructorParameter('idle_timeout', get('dapr.actors.idle_timeout'))

src/lib/App.php

@@ -8,6 +8,7 @@
 use Dapr\Actors\IActor;
 use Dapr\Actors\Reminder;
 use Dapr\Deserialization\InvokerParameterResolver;
+use Dapr\DistributedTracing\ActiveTrace;
 use Dapr\exceptions\Http\NotFound;
 use Dapr\PubSub\Subscriptions;
 use Dapr\Serialization\ISerializer;
@@ -176,7 +177,7 @@ public function start(?ServerRequestInterface $request = null): void
             )->withHeader('Content-Type', 'application/json');
             $this->logger->critical('Failed due to {exception}', ['exception' => $exception]);
         }
-        $this->emitter->emit($response);
+        $this->emitter->emit(ActiveTrace::decorate_response($this->container->get(ActiveTrace::class), $response));
     }
 
     /**

src/lib/DaprClient.php

@@ -3,6 +3,7 @@
 namespace Dapr;
 
 use Dapr\Deserialization\IDeserializer;
+use Dapr\DistributedTracing\ActiveTrace;
 use Dapr\exceptions\DaprException;
 use JetBrains\PhpStorm\Pure;
 use Psr\Log\LoggerInterface;
@@ -13,15 +14,14 @@
  */
 class DaprClient
 {
-    private static array $trace;
-    private static bool $added_trace = false;
     // temp hack to allow custom headers
     private static DaprClient $client;
     public array $extra_headers = [];
 
     public function __construct(
         protected LoggerInterface $logger,
         protected IDeserializer $deserializer,
+        protected ActiveTrace|null $trace,
         protected string $port
     ) {
         self::$client = $this;
@@ -63,7 +63,6 @@ public function get(string $url, ?array $params = null): DaprResponse
             $return->headers,
             fn($carry, $item) => str_starts_with($item, 'etag:') ? str_replace('etag: ', '', $item) : $carry
         );
-        self::detect_trace_from_response($return);
 
         $this->logger->debug('Got response: {response}', ['response' => $return]);
 
@@ -106,48 +105,12 @@ public function get(string $url, ?array $params = null): DaprResponse
 
     private function get_headers(): array
     {
-        return array_merge(["Accept: application/json"], self::detect_trace(), $this->extra_headers);
-    }
-
-    private function detect_trace(): array
-    {
-        if (isset(self::$trace)) {
-            return self::$trace;
-        }
-
-        $existing_headers = getallheaders();
-        self::$trace      = isset($existing_headers['Traceparent']) ? ['Traceparent: '.$existing_headers['Traceparent']] : [];
-
-        if ( ! self::$added_trace && isset($existing_headers['Traceparent'])) {
-            header('Traceparent: '.$existing_headers['Traceparent']);
-            self::$added_trace = true;
-        }
-
-        return self::$trace;
-    }
-
-    /**
-     * @param DaprResponse $response
-     */
-    private function detect_trace_from_response(DaprResponse $response): void
-    {
-        if (isset(self::$trace)) {
-            return;
-        }
+        $trace = $this->trace ? [
+            'tracestate: '.$this->trace->trace_state,
+            'traceparent: '.$this->trace->trace_parent,
+        ] : [];
 
-        $header = array_filter(
-            $response->headers,
-            function ($ii) {
-                return str_starts_with($ii, 'Traceparent:');
-            }
-        );
-        if ( ! empty($header)) {
-            self::$trace = $header;
-            if ( ! self::$added_trace) {
-                header('Traceparent: '.$header[0]);
-                self::$added_trace = true;
-            }
-        }
+        return array_merge(["Accept: application/json"], $trace, $this->extra_headers);
     }
 
     /**
@@ -180,7 +143,6 @@ public function post(string $url, mixed $data, ?array $params = null): DaprRespo
         $response->data    = json_decode($response->data, true);
         $response->code    = curl_getinfo($curl, CURLINFO_HTTP_CODE);
         $response->headers = explode("\r\n", curl_getinfo($curl, CURLINFO_HEADER_OUT));
-        self::detect_trace_from_response($response);
         $this->logger->debug('Got response: {r}', ['r' => $response]);
 
         if ($this->deserializer->is_exception($response->data)) {
@@ -226,7 +188,6 @@ public function delete(string $url, ?array $params = []): DaprResponse
         $response->data    = json_decode(curl_exec($curl), true);
         $response->code    = curl_getinfo($curl, CURLINFO_HTTP_CODE);
         $response->headers = explode("\r\n", curl_getinfo($curl, CURLINFO_HEADER_OUT));
-        self::detect_trace_from_response($response);
 
         $this->logger->debug('Got response: {r}', ['r' => $response]);
 

src/lib/DistributedTracing/ActiveTrace.php

@@ -0,0 +1,60 @@
+<?php
+
+namespace Dapr\DistributedTracing;
+
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+
+/**
+ * Class ActiveTrace
+ * @package Dapr\DistributedTracing
+ */
+class ActiveTrace
+{
+    /**
+     * ActiveTrace constructor.
+     *
+     * @param string $trace_parent
+     * @param string|null $trace_state
+     */
+    private function __construct(public string $trace_parent, public string|null $trace_state)
+    {
+    }
+
+    /**
+     * @param ActiveTrace|null $trace
+     * @param ResponseInterface $response
+     *
+     * @return ResponseInterface
+     */
+    public static function decorate_response(ActiveTrace|null $trace, ResponseInterface $response)
+    {
+        if ($trace === null) {
+            return $response;
+        }
+        if ($trace->trace_state) {
+            return $response
+                ->withHeader('traceparent', $trace->trace_parent)
+                ->withHeader('tracestate', $trace->trace_state);
+        }
+
+        return $response->withHeader('traceparent', $trace->trace_parent);
+    }
+
+    /**
+     * @param RequestInterface $request
+     *
+     * @return ActiveTrace|null
+     */
+    public static function get_from_request(RequestInterface $request): ActiveTrace|null
+    {
+        if ($request->hasHeader('traceparent')) {
+            return new ActiveTrace(
+                $request->getHeader('traceparent')[0],
+                $request->hasHeader('tracestate') ? $request->getHeader('tracestate')[0] : null
+            );
+        }
+
+        return null;
+    }
+}