Add OTLP log sink - #541
Open
jeremyudis wants to merge 2 commits into
Open
Conversation
Adds a standalone OtlpWriter (LogStreamWriter) that exports log messages to an OpenTelemetry collector over OTLP/gRPC or OTLP/HTTP (JSON or protobuf), alongside the existing Kafka/S3/Pulsar writers. - OtlpWriter: batch and committable write paths, exponential backoff on retryable failures, request splitting by max size, OTLP partial-success handling per spec. - OtlpGrpcExporter / OtlpHttpExporter behind a shared OtlpExporter transport interface; OtlpChannelManager shares one gRPC channel per endpoint, mirroring KafkaProducerManager. - OtlpLogRecordConverter maps LogMessage -> OTLP LogRecord/Resource, including k8s pod metadata as semantic-convention attributes and optional Docker JSON / CRI container log unwrapping. - New WriterType.OTLP config schema (OtlpWriterConfig, OtlpFormatConfig, ContainerLogFormat) plus config parsing and DefaultLogMonitor wiring. - Also adds the KafkaMessageFormat/OtlpFormatConfig schema and config parsing for an OTLP envelope mode on the Kafka writer (wired up to KafkaWriter in a follow-up commit). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JuL9Hi8NSsf2sAhdJf1p4N
The KafkaWriterConfig.messageFormat/otlpFormatConfig schema and parsing already existed but nothing in KafkaWriter or CommittableKafkaWriter ever used them, so kafka.messageFormat=otlp_json/otlp_proto silently wrote raw bytes instead of the OTLP envelope. - KafkaWriter#initOtlpMessageFormat builds an OtlpMessageFormatter when messageFormat != RAW; #serializeValue routes both the batch (messageCollation) and committable (CommittableKafkaWriter) write paths through it. DefaultLogMonitor now calls this after constructing either writer variant. - The injected "messageCRC" Kafka header is now computed from the actual record value (the OTLP envelope, when configured) instead of always using the pre-transform raw-bytes checksum, so a consumer recomputing CRC32 over what it received from Kafka gets a match. RAW format is unaffected (record value and raw bytes are identical). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JuL9Hi8NSsf2sAhdJf1p4N
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
OtlpWriterLogStreamWriterthat exports log messages directly to an OpenTelemetry collector over OTLP/gRPC or OTLP/HTTP (JSON or protobuf), with retry/backoff, request-size-based batch splitting, and optional Docker JSON / CRI container-log unwrapping for Kubernetes pod logs.KafkaWriterConfig.messageFormat/otlpFormatConfigschema sokafka.messageFormat=otlp_json/otlp_protoactually serializes Kafka record values as single-record OTLP envelopes instead of silently writing raw bytes.messageCRCKafka header to reflect the actual record value (the OTLP envelope, when configured) rather than always the pre-transform raw bytes, so downstream consumers recomputing CRC32 over what they read from Kafka get a match.RAWformat (the default) is unaffected.Test plan
mvn test -pl singer -Dtest=TestOtlpConfigParsing,com.pinterest.singer.writer.otlp.*(44 tests) — config parsing, container log parsers, gRPC exporter against a real in-process gRPC server, HTTP exporter against a realHttpServer, log record conversion, writer batching/retry/backoff/partial-success semantics.mvn test -pl singer -Dtest=TestKafkaWriter,TestCommittableKafkaWriter(20 tests) — both Kafka write paths route through the OTLP formatter when configured and are unchanged for theRAWdefault;messageCRCheader now matches the actual record value.mvn clean package -pl singer -am -DskipTests— full module build.DefaultLogStreamProcessorTest/TestMemoryEfficientLogStreamProcessorfailures seen in a fullmvn testrun are present onmastertoo (unrelated to this change) by reverting these commits locally and re-running.Co-Authored-By: Claude Sonnet 5 noreply@anthropic.com
🤖 Generated with Claude Code
https://claude.ai/code/session_01JuL9Hi8NSsf2sAhdJf1p4N