enable timeout for http transports (#1275)

* enable timeout for http transports
Attempt to do our own discovery for some well-known PSR-18 clients, which allows
for configuring timeout (and in future certificates, keys etc).
This is not complete and a prototype for feedback, but I've updated an example to
show that it works for Guzzle and Symfony http clients

* improve client discovery

* refactor, test, add more implementations

* add timeout for logs and metrics

examples/metrics/exporters/otlp_http.php

@@ -14,7 +14,7 @@
 $clock = ClockFactory::getDefault();
 $reader = new ExportingReader(
     new MetricExporter(
-        PsrTransportFactory::discover()->create('http://collector:4318/v1/metrics', \OpenTelemetry\Contrib\Otlp\ContentTypes::JSON)
+        (new PsrTransportFactory())->create('http://collector:4318/v1/metrics', \OpenTelemetry\Contrib\Otlp\ContentTypes::JSON)
     )
 );
 

examples/sdk_builder.php

@@ -28,7 +28,7 @@
 
 $reader = new ExportingReader(
     new MetricExporter(
-        PsrTransportFactory::discover()->create('http://collector:4318/v1/metrics', 'application/x-protobuf')
+        (new PsrTransportFactory())->create('http://collector:4318/v1/metrics', 'application/x-protobuf')
     )
 );
 

examples/traces/exporters/otlp_http_from_factory.php

@@ -11,6 +11,7 @@
  */
 putenv('OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318');
 putenv('OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf');
+putenv('OTEL_EXPORTER_OTLP_TIMEOUT=1500'); //1.5s
 $factory = new \OpenTelemetry\SDK\Trace\TracerProviderFactory();
 $tracerProvider = $factory->create();
 

examples/traces/exporters/zipkin.php

@@ -11,7 +11,7 @@
 use OpenTelemetry\SDK\Trace\SpanProcessor\SimpleSpanProcessor;
 use OpenTelemetry\SDK\Trace\TracerProvider;
 
-$transport = PsrTransportFactory::discover()->create('http://zipkin:9411/api/v2/spans', 'application/json');
+$transport = (new PsrTransportFactory())->create('http://zipkin:9411/api/v2/spans', 'application/json');
 $zipkinExporter = new ZipkinExporter(
     $transport
 );

examples/traces/features/parent_span_example.php

@@ -17,7 +17,7 @@
 
 // zipkin exporter
 $zipkinExporter = new ZipkinExporter(
-    PsrTransportFactory::discover()->create('http://zipkin:9411/api/v2/spans', 'application/json')
+    (new PsrTransportFactory())->create('http://zipkin:9411/api/v2/spans', 'application/json')
 );
 
 $tracerProvider =  new TracerProvider(

examples/troubleshooting/logging_of_span_data.php

@@ -44,7 +44,7 @@
  */
 //$exporterEndpoint = 'http://example.com:9411/api/v2/spans';
 $exporter = new ZipkinExporter(
-    PsrTransportFactory::discover()->create($exporterEndpoint, 'application/json'),
+    (new PsrTransportFactory())->create($exporterEndpoint, 'application/json'),
 );
 /**
  * Decorate the Exporter

src/Contrib/Otlp/LogsExporterFactory.php

@@ -43,6 +43,7 @@ private function buildTransport(string $protocol): TransportInterface
 
         $headers = OtlpUtil::getHeaders(Signals::LOGS);
         $compression = $this->getCompression();
+        $timeout = $this->getTimeout();
 
         $factoryClass = Registry::transportFactory($protocol);
         $factory = $this->transportFactory ?: new $factoryClass();
@@ -52,6 +53,7 @@ private function buildTransport(string $protocol): TransportInterface
             Protocols::contentType($protocol),
             $headers,
             $compression,
+            $timeout,
         );
     }
 
@@ -76,4 +78,13 @@ private function getEndpoint(string $protocol): string
 
         return HttpEndpointResolver::create()->resolveToString($endpoint, Signals::LOGS);
     }
+
+    private function getTimeout(): float
+    {
+        $value = Configuration::has(Variables::OTEL_EXPORTER_OTLP_LOGS_TIMEOUT) ?
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_LOGS_TIMEOUT) :
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_TIMEOUT);
+
+        return $value/1000;
+    }
 }

src/Contrib/Otlp/MetricExporterFactory.php

@@ -44,12 +44,12 @@ private function buildTransport(string $protocol): TransportInterface
         /**
          * @todo (https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#periodic-exporting-metricreader)
          * - OTEL_METRIC_EXPORT_INTERVAL
-         * - OTEL_METRIC_EXPORT_TIMEOUT
          */
         $endpoint = $this->getEndpoint($protocol);
 
         $headers = OtlpUtil::getHeaders(Signals::METRICS);
         $compression = $this->getCompression();
+        $timeout = $this->getTimeout();
 
         $factoryClass = Registry::transportFactory($protocol);
         $factory = $this->transportFactory ?: new $factoryClass();
@@ -59,6 +59,7 @@ private function buildTransport(string $protocol): TransportInterface
             Protocols::contentType($protocol),
             $headers,
             $compression,
+            $timeout,
         );
     }
 
@@ -84,6 +85,15 @@ private function getCompression(): string
             Configuration::getEnum(Variables::OTEL_EXPORTER_OTLP_COMPRESSION, self::DEFAULT_COMPRESSION);
     }
 
+    private function getTimeout(): float
+    {
+        $value = Configuration::has(Variables::OTEL_EXPORTER_OTLP_METRICS_TIMEOUT) ?
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_METRICS_TIMEOUT) :
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_TIMEOUT);
+
+        return $value/1000;
+    }
+
     private function getEndpoint(string $protocol): string
     {
         if (Configuration::has(Variables::OTEL_EXPORTER_OTLP_METRICS_ENDPOINT)) {

src/Contrib/Otlp/OtlpHttpTransportFactory.php

@@ -28,6 +28,7 @@ public function create(
             $compression = null;
         }
 
-        return PsrTransportFactory::discover()->create($endpoint, $contentType, $headers, $compression, $timeout, $retryDelay, $maxRetries, $cacert, $cert, $key);
+        return (new PsrTransportFactory())
+            ->create($endpoint, $contentType, $headers, $compression, $timeout, $retryDelay, $maxRetries, $cacert, $cert, $key);
     }
 }

src/Contrib/Otlp/SpanExporterFactory.php

@@ -21,7 +21,7 @@ class SpanExporterFactory implements SpanExporterFactoryInterface
 
     private const DEFAULT_COMPRESSION = 'none';
 
-    public function __construct(private ?TransportFactoryInterface $transportFactory = null)
+    public function __construct(private readonly ?TransportFactoryInterface $transportFactory = null)
     {
     }
 
@@ -46,11 +46,12 @@ private function buildTransport(): TransportInterface
         $endpoint = $this->getEndpoint($protocol);
         $headers = OtlpUtil::getHeaders(Signals::TRACE);
         $compression = $this->getCompression();
+        $timeout = $this->getTimeout();
 
         $factoryClass = Registry::transportFactory($protocol);
         $factory = $this->transportFactory ?: new $factoryClass();
 
-        return $factory->create($endpoint, $contentType, $headers, $compression);
+        return $factory->create($endpoint, $contentType, $headers, $compression, $timeout);
     }
 
     private function getProtocol(): string
@@ -81,4 +82,13 @@ private function getCompression(): string
             Configuration::getEnum(Variables::OTEL_EXPORTER_OTLP_TRACES_COMPRESSION) :
             Configuration::getEnum(Variables::OTEL_EXPORTER_OTLP_COMPRESSION, self::DEFAULT_COMPRESSION);
     }
+
+    private function getTimeout(): float
+    {
+        $value = Configuration::has(Variables::OTEL_EXPORTER_OTLP_TRACES_TIMEOUT) ?
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_TRACES_TIMEOUT) :
+            Configuration::getInt(Variables::OTEL_EXPORTER_OTLP_TIMEOUT);
+
+        return $value/1000;
+    }
 }

src/Contrib/Zipkin/SpanExporterFactory.php

@@ -15,7 +15,8 @@ class SpanExporterFactory implements SpanExporterFactoryInterface
     public function create(): SpanExporterInterface
     {
         $endpoint = Configuration::getString(Variables::OTEL_EXPORTER_ZIPKIN_ENDPOINT);
-        $transport = PsrTransportFactory::discover()->create($endpoint, 'application/json');
+        $timeout = Configuration::getInt(Variables::OTEL_EXPORTER_ZIPKIN_TIMEOUT)/1000;
+        $transport = (new PsrTransportFactory())->create(endpoint: $endpoint, contentType: 'application/json', timeout: $timeout);
 
         return new Exporter($transport);
     }

src/SDK/Common/Configuration/Defaults.php

@@ -70,10 +70,10 @@ interface Defaults
     public const OTEL_EXPORTER_OTLP_METRICS_INSECURE = 'false';
     public const OTEL_EXPORTER_OTLP_LOGS_INSECURE = 'false';
     // Timeout (seconds)
-    public const OTEL_EXPORTER_OTLP_TIMEOUT = 10;
-    public const OTEL_EXPORTER_OTLP_TRACES_TIMEOUT = 10;
-    public const OTEL_EXPORTER_OTLP_METRICS_TIMEOUT = 10;
-    public const OTEL_EXPORTER_OTLP_LOGS_TIMEOUT = 10;
+    public const OTEL_EXPORTER_OTLP_TIMEOUT = 10000; //10s
+    public const OTEL_EXPORTER_OTLP_TRACES_TIMEOUT = 10000;
+    public const OTEL_EXPORTER_OTLP_METRICS_TIMEOUT = 10000;
+    public const OTEL_EXPORTER_OTLP_LOGS_TIMEOUT = 10000;
     // Protocol
     public const OTEL_EXPORTER_OTLP_PROTOCOL = 'http/protobuf';
     public const OTEL_EXPORTER_OTLP_TRACES_PROTOCOL = 'http/protobuf';
@@ -85,7 +85,7 @@ interface Defaults
      */
     public const OTEL_EXPORTER_ZIPKIN_ENDPOINT = 'http://localhost:9411/api/v2/spans';
     // Timeout (seconds)
-    public const OTEL_EXPORTER_ZIPKIN_TIMEOUT = 10;
+    public const OTEL_EXPORTER_ZIPKIN_TIMEOUT = 10000; //10s
     /**
      * Prometheus Exporter
      * @see https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/configuration/sdk-environment-variables.md#prometheus-exporter

src/SDK/Common/Export/Http/PsrTransportFactory.php

@@ -10,16 +10,17 @@
 use Http\Discovery\Psr18ClientDiscovery;
 use InvalidArgumentException;
 use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery;
 use Psr\Http\Client\ClientInterface;
 use Psr\Http\Message\RequestFactoryInterface;
 use Psr\Http\Message\StreamFactoryInterface;
 
 final class PsrTransportFactory implements TransportFactoryInterface
 {
     public function __construct(
-        private readonly ClientInterface $client,
-        private readonly RequestFactoryInterface $requestFactory,
-        private readonly StreamFactoryInterface $streamFactory,
+        private ?ClientInterface $client = null,
+        private ?RequestFactoryInterface $requestFactory = null,
+        private ?StreamFactoryInterface $streamFactory = null,
     ) {
     }
 
@@ -43,6 +44,12 @@ public function create(
         }
         assert(!empty($endpoint));
 
+        $this->client ??= Discovery::find([
+            'timeout' => $timeout,
+        ]);
+        $this->requestFactory ??= Psr17FactoryDiscovery::findRequestFactory();
+        $this->streamFactory ??= Psr17FactoryDiscovery::findStreamFactory();
+
         return new PsrTransport(
             $this->client,
             $this->requestFactory,
@@ -56,6 +63,9 @@ public function create(
         );
     }
 
+    /**
+     * @deprecated
+     */
     public static function discover(): self
     {
         return new self(

src/SDK/Common/Http/Psr/Client/Discovery.php

@@ -0,0 +1,73 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\SDK\Common\Http\Psr\Client;
+
+use Generator;
+use Http\Discovery\Psr18ClientDiscovery;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery\Buzz;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery\CurlClient;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery\DiscoveryInterface;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery\Guzzle;
+use OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery\Symfony;
+use Psr\Http\Client\ClientInterface;
+
+class Discovery
+{
+    private static ?array $discoverers;
+
+    /**
+     * @var list<class-string<DiscoveryInterface>>
+     */
+    private const DEFAULTS = [
+        Guzzle::class,
+        Symfony::class,
+        Buzz::class,
+        CurlClient::class,
+    ];
+
+    /**
+     * Attempt discovery of a configurable psr-18 http client, falling back to Psr18ClientDiscovery.
+     */
+    public static function find(array $options = []): ClientInterface
+    {
+        $options = array_filter($options);
+
+        foreach (self::discoverers() as $clientDiscovery) {
+            /** @var DiscoveryInterface $clientDiscovery */
+            if ($clientDiscovery->available()) {
+                return $clientDiscovery->create($options);
+            }
+        }
+
+        return Psr18ClientDiscovery::find();
+    }
+
+    /**
+     * @internal
+     */
+    public static function setDiscoverers(array $discoverers): void
+    {
+        self::$discoverers = $discoverers;
+    }
+
+    /**
+     * @internal
+     */
+    public static function reset(): void
+    {
+        self::$discoverers = null;
+    }
+
+    private static function discoverers(): Generator
+    {
+        foreach (self::$discoverers ?? self::DEFAULTS as $discoverer) {
+            if (is_string($discoverer)) {
+                yield new $discoverer();
+            } else {
+                yield $discoverer;
+            }
+        }
+    }
+}

src/SDK/Common/Http/Psr/Client/Discovery/Buzz.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery;
+
+use Buzz\Client\FileGetContents;
+use Http\Discovery\Psr17FactoryDiscovery;
+use Psr\Http\Client\ClientInterface;
+
+class Buzz implements DiscoveryInterface
+{
+    /**
+     * @phan-suppress PhanUndeclaredClassReference
+     */
+    public function available(): bool
+    {
+        return class_exists(FileGetContents::class);
+    }
+
+    /**
+     * @phan-suppress PhanUndeclaredClassReference,PhanTypeMismatchReturn,PhanUndeclaredClassMethod
+     */
+    public function create(mixed $options): ClientInterface
+    {
+        return new FileGetContents(Psr17FactoryDiscovery::findResponseFactory(), $options);
+    }
+}

src/SDK/Common/Http/Psr/Client/Discovery/CurlClient.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery;
+
+use Http\Client\Curl\Client;
+use Psr\Http\Client\ClientInterface;
+
+class CurlClient implements DiscoveryInterface
+{
+    /**
+     * @phan-suppress PhanUndeclaredClassReference
+     */
+    public function available(): bool
+    {
+        return extension_loaded('curl') && class_exists(Client::class);
+    }
+
+    /**
+     * @phan-suppress PhanUndeclaredClassReference,PhanTypeMismatchReturn,PhanUndeclaredClassMethod
+     * @psalm-suppress UndefinedClass,InvalidReturnType,InvalidReturnStatement
+     */
+    public function create(mixed $options): ClientInterface
+    {
+        $options = [
+            \CURLOPT_TIMEOUT => $options['timeout'] ?? null,
+        ];
+
+        /** @phpstan-ignore-next-line  */
+        return new Client(options: array_filter($options));
+    }
+}

src/SDK/Common/Http/Psr/Client/Discovery/DiscoveryInterface.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OpenTelemetry\SDK\Common\Http\Psr\Client\Discovery;
+
+use Psr\Http\Client\ClientInterface;
+
+interface DiscoveryInterface
+{
+    public function available(): bool;
+    public function create(mixed $options): ClientInterface;
+}