Send each request under the trace context stored on it (#172)

The Guzzle auto-instrumentation strips traceparent/tracestate off the
outgoing request and re-injects them from the active context. Since the
pool sends every request in a chunk under the worker's own span, all of
them reached their destination sharing that one parent, and the trace
context recorded when the row was created never made it onto the wire.

Activating the row's own context around the send makes the client span
parent to the trace that created the request insurance instead.

Author: LauJosefsen

src/RequestInsurance/AsyncRequests/RequestPool.php

@@ -7,10 +7,12 @@
 use GuzzleHttp\Pool;
 use GuzzleHttp\Client;
 use GuzzleHttp\TransferStats;
+use OpenTelemetry\Context\Context;
 use Illuminate\Support\Facades\Config;
 use GuzzleHttp\Promise\PromiseInterface;
 use Illuminate\Database\Eloquent\Collection;
 use Cego\RequestInsurance\Models\RequestInsurance;
+use OpenTelemetry\API\Trace\Propagation\TraceContextPropagator;
 
 class RequestPool
 {
@@ -76,13 +78,44 @@ protected function requestProvider(Client $client, Collection $requestInsurances
      */
     private function convertRequestToPromise(Client $client, RequestInsurance $requestInsurance): PromiseInterface
     {
-        return $client->requestAsync(mb_strtoupper($requestInsurance->method), $requestInsurance->url, [
-            'headers'     => array_merge($requestInsurance->getHeadersCastToArray(), ['User-Agent' => sprintf('RequestInsurance %s', Config::get('app.name', 'unknown'))]),
+        $headers = $requestInsurance->getHeadersCastToArray();
+
+        $send = fn () => $client->requestAsync(mb_strtoupper($requestInsurance->method), $requestInsurance->url, [
+            'headers'     => array_merge($headers, ['User-Agent' => sprintf('RequestInsurance %s', Config::get('app.name', 'unknown'))]),
             'body'        => $requestInsurance->payload,
             'timeout'     => $requestInsurance->getEffectiveTimeout(),
             'on_stats'    => fn (TransferStats $stats) => $requestInsurance->setTimings($stats),
             'http_errors' => false,
         ]);
+
+        return $this->withTraceContextOf($headers, $send);
+    }
+
+    /**
+     * Runs the given callback with the trace context stored on the request insurance as the active context
+     *
+     * Auto-instrumentation strips the stored traceparent off the outgoing request and re-injects
+     * one built from the active context. Without this, every request in a chunk is sent under the
+     * worker's own span instead of under the trace that created the row.
+     *
+     * @param array<string, string> $headers
+     * @param callable(): PromiseInterface $callback
+     *
+     * @return PromiseInterface
+     */
+    private function withTraceContextOf(array $headers, callable $callback): PromiseInterface
+    {
+        if ( ! class_exists(TraceContextPropagator::class)) {
+            return $callback();
+        }
+
+        $scope = Context::storage()->attach(TraceContextPropagator::getInstance()->extract($headers));
+
+        try {
+            return $callback();
+        } finally {
+            $scope->detach();
+        }
     }
 
     /**

tests/Unit/RequestPoolTest.php

@@ -5,9 +5,12 @@
 use Tests\TestCase;
 use GuzzleHttp\Client;
 use GuzzleHttp\Psr7\Response;
+use OpenTelemetry\API\Trace\Span;
+use OpenTelemetry\Context\Context;
 use GuzzleHttp\Promise\FulfilledPromise;
 use GuzzleHttp\Promise\PromiseInterface;
 use Illuminate\Database\Eloquent\Collection;
+use OpenTelemetry\API\Trace\SpanContextInterface;
 use Cego\RequestInsurance\Models\RequestInsurance;
 use Cego\RequestInsurance\AsyncRequests\RequestPool;
 
@@ -28,6 +31,68 @@ public function test_it_uppercases_a_mixed_case_method(): void
         $this->assertMethodGivenToClient('DeLeTe', 'DELETE');
     }
 
+    public function test_it_sends_a_request_under_the_trace_context_stored_on_it(): void
+    {
+        // Arrange
+        $traceId = '0af7651916cd43dd8448eb211c80319c';
+        $spanId = 'b7ad6b7169203331';
+
+        $requestInsurance = RequestInsurance::getBuilder()
+            ->url('https://test.lupinsdev.dk')
+            ->method('POST')
+            ->headers(['traceparent' => sprintf('00-%s-%s-01', $traceId, $spanId)])
+            ->create();
+
+        // Act
+        $spanContext = $this->captureActiveSpanContextWhileSending($requestInsurance);
+
+        // Assert
+        $this->assertSame($traceId, $spanContext->getTraceId());
+        $this->assertSame($spanId, $spanContext->getSpanId());
+    }
+
+    public function test_it_leaves_the_active_context_alone_for_a_request_without_a_trace_context(): void
+    {
+        // Arrange
+        $requestInsurance = RequestInsurance::getBuilder()
+            ->url('https://test.lupinsdev.dk')
+            ->method('POST')
+            ->create();
+
+        // Act
+        $spanContext = $this->captureActiveSpanContextWhileSending($requestInsurance);
+
+        // Assert
+        $this->assertFalse($spanContext->isValid());
+    }
+
+    /**
+     * Returns the span context that was active while the pool handed the request to the Guzzle client
+     *
+     * @param RequestInsurance $requestInsurance
+     *
+     * @return SpanContextInterface
+     */
+    protected function captureActiveSpanContextWhileSending(RequestInsurance $requestInsurance): SpanContextInterface
+    {
+        $client = new class () extends Client {
+            public ?SpanContextInterface $spanContext = null;
+
+            public function requestAsync(string $method, $uri = '', array $options = []): PromiseInterface
+            {
+                $this->spanContext = Span::fromContext(Context::getCurrent())->getContext();
+
+                return new FulfilledPromise(new Response(200));
+            }
+        };
+
+        (new RequestPool($client, new Collection([$requestInsurance])))->getResponses();
+
+        $this->assertNotNull($client->spanContext, 'the client was never handed a request');
+
+        return $client->spanContext;
+    }
+
     /**
      * Asserts which HTTP method the pool hands to the Guzzle client for a stored method
      *