Send each request under the trace context stored on it

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
      *