86 lines
3.0 KiB
PHP
86 lines
3.0 KiB
PHP
|
<?php
|
||
|
|
|
||
|
|
declare(strict_types=1);
|
||
|
|
|
||
|
|
namespace OpenTelemetry\Contrib\Otlp;
|
||
|
|
|
||
|
|
use OpenTelemetry\API\Behavior\LogsMessagesTrait;
|
||
|
|
use Opentelemetry\Proto\Collector\Logs\V1\ExportLogsServiceResponse;
|
||
|
|
use OpenTelemetry\SDK\Common\Export\TransportInterface;
|
||
|
|
use OpenTelemetry\SDK\Common\Future\CancellationInterface;
|
||
|
|
use OpenTelemetry\SDK\Common\Future\FutureInterface;
|
||
|
|
use OpenTelemetry\SDK\Logs\LogRecordExporterInterface;
|
||
|
|
use OpenTelemetry\SDK\Logs\ReadableLogRecord;
|
||
|
|
use RuntimeException;
|
||
|
|
use Throwable;
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @psalm-import-type SUPPORTED_CONTENT_TYPES from ProtobufSerializer
|
||
|
|
*/
|
||
|
|
class LogsExporter implements LogRecordExporterInterface
|
||
|
|
{
|
||
|
|
use LogsMessagesTrait;
|
||
|
|
|
||
|
|
private TransportInterface $transport;
|
||
|
|
private ProtobufSerializer $serializer;
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @psalm-param TransportInterface<SUPPORTED_CONTENT_TYPES> $transport
|
||
|
|
*/
|
||
|
|
public function __construct(TransportInterface $transport)
|
||
|
|
{
|
||
|
|
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);
|
||
|
|
}
|
||
|
|
|
||
|
|
/**
|
||
|
|
* @param iterable<ReadableLogRecord> $batch
|
||
|
|
*/
|
||
|
|
public function export(iterable $batch, ?CancellationInterface $cancellation = null): FutureInterface
|
||
|
|
{
|
||
|
|
return $this->transport
|
||
|
|
->send($this->serializer->serialize((new LogsConverter($this->serializer))->convert($batch)), $cancellation)
|
||
|
|
->map(function (?string $payload): bool {
|
||
|
|
if ($payload === null) {
|
||
|
|
return true;
|
||
|
|
}
|
||
|
|
|
||
|
|
$serviceResponse = new ExportLogsServiceResponse();
|
||
|
|
$this->serializer->hydrate($serviceResponse, $payload);
|
||
|
|
|
||
|
|
$partialSuccess = $serviceResponse->getPartialSuccess();
|
||
|
|
if ($partialSuccess !== null && $partialSuccess->getRejectedLogRecords()) {
|
||
|
|
self::logError('Export partial success', [
|
||
|
|
'rejected_logs' => $partialSuccess->getRejectedLogRecords(),
|
||
|
|
'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;
|
||
|
|
});
|
||
|
|
}
|
||
|
|
|
||
|
|
public function forceFlush(?CancellationInterface $cancellation = null): bool
|
||
|
|
{
|
||
|
|
return $this->transport->forceFlush($cancellation);
|
||
|
|
}
|
||
|
|
|
||
|
|
public function shutdown(?CancellationInterface $cancellation = null): bool
|
||
|
|
{
|
||
|
|
return $this->transport->shutdown($cancellation);
|
||
|
|
}
|
||
|
|
}
|