98 lines
3.5 KiB
PHP
98 lines
3.5 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace OpenTelemetry\Contrib\Otlp;
|
|
|
|
use OpenTelemetry\API\Behavior\LogsMessagesTrait;
|
|
use Opentelemetry\Proto\Collector\Metrics\V1\ExportMetricsServiceResponse;
|
|
use OpenTelemetry\SDK\Common\Export\TransportInterface;
|
|
use OpenTelemetry\SDK\Metrics\AggregationTemporalitySelectorInterface;
|
|
use OpenTelemetry\SDK\Metrics\Data\Temporality;
|
|
use OpenTelemetry\SDK\Metrics\MetricMetadataInterface;
|
|
use OpenTelemetry\SDK\Metrics\PushMetricExporterInterface;
|
|
use RuntimeException;
|
|
use Throwable;
|
|
|
|
/**
|
|
* @see https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk_exporters/stdout.md#opentelemetry-metrics-exporter---standard-output
|
|
* @see https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/file-exporter.md#json-file-serialization
|
|
* @psalm-import-type SUPPORTED_CONTENT_TYPES from ProtobufSerializer
|
|
*/
|
|
final class MetricExporter implements PushMetricExporterInterface, AggregationTemporalitySelectorInterface
|
|
{
|
|
use LogsMessagesTrait;
|
|
|
|
private TransportInterface $transport;
|
|
private ProtobufSerializer $serializer;
|
|
/**
|
|
* @var string|Temporality|null
|
|
*/
|
|
private $temporality;
|
|
|
|
/**
|
|
* @param string|Temporality|null $temporality
|
|
*
|
|
* @psalm-param TransportInterface<SUPPORTED_CONTENT_TYPES> $transport
|
|
*/
|
|
public function __construct(TransportInterface $transport, $temporality = null)
|
|
{
|
|
if (!class_exists('\Google\Protobuf\Api')) {
|
|
throw new RuntimeException('No protobuf implementation found (ext-protobuf or google/protobuf)');
|
|
}
|
|
$this->transport = $transport;
|
|
$this->serializer = ProtobufSerializer::forTransport($transport);
|
|
$this->temporality = $temporality;
|
|
}
|
|
|
|
public function temporality(MetricMetadataInterface $metric)
|
|
{
|
|
return $this->temporality ?? $metric->temporality();
|
|
}
|
|
|
|
public function export(iterable $batch): bool
|
|
{
|
|
return $this->transport
|
|
->send($this->serializer->serialize((new MetricConverter($this->serializer))->convert($batch)))
|
|
->map(function (?string $payload): bool {
|
|
if ($payload === null) {
|
|
return true;
|
|
}
|
|
|
|
$serviceResponse = new ExportMetricsServiceResponse();
|
|
$this->serializer->hydrate($serviceResponse, $payload);
|
|
|
|
$partialSuccess = $serviceResponse->getPartialSuccess();
|
|
if ($partialSuccess !== null && $partialSuccess->getRejectedDataPoints()) {
|
|
self::logError('Export partial success', [
|
|
'rejected_data_points' => $partialSuccess->getRejectedDataPoints(),
|
|
'error_message' => $partialSuccess->getErrorMessage(),
|
|
]);
|
|
|
|
return false;
|
|
}
|
|
if ($partialSuccess !== null && $partialSuccess->getErrorMessage()) {
|
|
self::logWarning('Export success with warnings/suggestions', ['error_message' => $partialSuccess->getErrorMessage()]);
|
|
}
|
|
|
|
return true;
|
|
})
|
|
->catch(static function (Throwable $throwable): bool {
|
|
self::logError('Export failure', ['exception' => $throwable]);
|
|
|
|
return false;
|
|
})
|
|
->await();
|
|
}
|
|
|
|
public function shutdown(): bool
|
|
{
|
|
return $this->transport->shutdown();
|
|
}
|
|
|
|
public function forceFlush(): bool
|
|
{
|
|
return $this->transport->forceFlush();
|
|
}
|
|
}
|