blob: efd149c7f4b3065fb50fc628034a6e88f97db984 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
<?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();
}
}
|
