From eea38d68a48229686c1f252832a5640508c17400 Mon Sep 17 00:00:00 2001 From: Dylan Myers Date: Thu, 6 Aug 2026 09:21:28 -0400 Subject: [PATCH] feat(o11y): per-batch output send spans + HEC ACK-poll span (PIPE-1066) Assisted-by: Claude Opus 4.8 --- output/adapter.go | 22 ++++++++++ output/adapter_test.go | 20 +++++++++ output/file/file.go | 27 +++++++++--- output/hec/ack.go | 13 ++++++ output/hec/hec.go | 10 +++++ output/otlp_grpc/otlp_grpc.go | 20 +++++++++ output/otlp_grpc/send_test.go | 83 +++++++++++++++++++++++++++++++++++ output/tcp/tcp.go | 25 ++++++++--- output/udp/udp.go | 25 ++++++++--- 9 files changed, 230 insertions(+), 15 deletions(-) create mode 100644 output/otlp_grpc/send_test.go diff --git a/output/adapter.go b/output/adapter.go index b769d6b..f6209fc 100644 --- a/output/adapter.go +++ b/output/adapter.go @@ -11,6 +11,28 @@ import ( // the consumer adapters create when TelemetrySettings.PerBatchSpans is on. const adapterTracerScope = "github.com/observiq/blitz/output" +// noopSendSpan is a non-recording span returned by StartSendSpan when per-batch +// spans are off, so callers can always defer span.End() without a nil check. +var noopSendSpan = trace.SpanFromContext(context.Background()) + +// StartSendSpan starts a gated per-batch send span from ctx, so an output's +// worker can trace the actual (async) send parented to the emit span the +// consumer adapter opened. When tel.PerBatchSpans is off it returns ctx +// unchanged and a non-recording span, so the caller always defers span.End(). +// +// The send runs in a worker goroutine after Write enqueued the batch, so the +// emit span has usually already ended by the time this fires. That is expected: +// OTel permits a child of an ended span, and the trace then reads as a brief +// enqueue (emit) with a later, longer send child, which is the correct picture +// of an asynchronous output. Callers carry the emit ctx through their internal +// channel to reach here. +func StartSendSpan(ctx context.Context, tel embed.TelemetrySettings, name string) (context.Context, trace.Span) { + if !tel.PerBatchSpans { + return ctx, noopSendSpan + } + return tel.Tracer(adapterTracerScope).Start(ctx, name) +} + // WriterAsLogConsumer wraps a Writer so it can be used in contexts that // expect an embed.LogConsumer. The adapter pushes each record in the // batch through Writer.Write in order, returning the first error it diff --git a/output/adapter_test.go b/output/adapter_test.go index 37699b3..fbefba6 100644 --- a/output/adapter_test.go +++ b/output/adapter_test.go @@ -8,10 +8,30 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/output" + "github.com/stretchr/testify/require" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" ) +// TestStartSendSpan_gated covers both branches of the shared output send-span +// helper: off yields a non-recording span, on records a named span. +func TestStartSendSpan_gated(t *testing.T) { + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + + off := embed.TelemetrySettings{TracerProvider: tp, PerBatchSpans: false} + _, span := output.StartSendSpan(context.Background(), off, "blitz.output.test.send") + span.End() + require.Empty(t, rec.Ended(), "no span expected when PerBatchSpans is off") + + on := embed.TelemetrySettings{TracerProvider: tp, PerBatchSpans: true} + _, span = output.StartSendSpan(context.Background(), on, "blitz.output.test.send") + span.End() + ended := rec.Ended() + require.Len(t, ended, 1) + require.Equal(t, "blitz.output.test.send", ended[0].Name()) +} + type recordingWriter struct { mu sync.Mutex records []output.LogRecord diff --git a/output/file/file.go b/output/file/file.go index 261be83..8b288ff 100644 --- a/output/file/file.go +++ b/output/file/file.go @@ -34,12 +34,20 @@ type RotationOptions struct { LocalTime bool } +// fileItem is one queued line plus the emit-span context it was written under, +// so the worker can parent its write span to the emit span. +type fileItem struct { + ctx context.Context + msg string +} + // File implements the Output interface for file writes type File struct { logger *zap.Logger + tel embed.TelemetrySettings path string workers int - dataChan chan string + dataChan chan fileItem ctx context.Context cancel context.CancelFunc workerManager *workermanager.WorkerManager @@ -77,9 +85,10 @@ func New(logger *zap.Logger, path string, workers int, rotation RotationOptions, f := &File{ logger: logger.Named("output-file"), + tel: tel, path: path, workers: workers, - dataChan: make(chan string, DefaultFileChannelSize), + dataChan: make(chan fileItem, DefaultFileChannelSize), ctx: ctx, cancel: cancel, writer: writer, @@ -117,7 +126,7 @@ func (f *File) ObserveBlitzOutputQueueSize(_ context.Context, observer metric.In // Write enqueues data for file workers. func (f *File) Write(ctx context.Context, data output.LogRecord) error { select { - case f.dataChan <- data.Message: + case f.dataChan <- fileItem{ctx: ctx, msg: data.Message}: f.metrics.BlitzOutputEntriesReceivedCounter.Add(ctx, 1, outputType, "logs") return nil case <-ctx.Done(): @@ -151,13 +160,21 @@ func (f *File) fileWorker(id int) { for { select { - case data, ok := <-f.dataChan: + case item, ok := <-f.dataChan: if !ok { f.logger.Info("File worker exiting - channel closed", zap.Int("worker_id", id)) return } - if err := f.writeData(data); err != nil { + // The write span covers the lumberjack write, which transparently + // absorbs any file rotation that fires during it. + _, span := output.StartSendSpan(item.ctx, f.tel, "blitz.output.file.write") + err := f.writeData(item.msg) + if err != nil { + span.RecordError(err) + } + span.End() + if err != nil { f.logger.Error("Failed to write file data", zap.Int("worker_id", id), zap.Error(err)) return } diff --git a/output/hec/ack.go b/output/hec/ack.go index 295d49d..8993524 100644 --- a/output/hec/ack.go +++ b/output/hec/ack.go @@ -9,6 +9,9 @@ import ( "time" "github.com/goccy/go-json" + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/output" + "go.opentelemetry.io/otel/attribute" "go.uber.org/zap" ) @@ -133,6 +136,7 @@ type ackPoller struct { maxRetries int resendCh chan resendItem // channel to send payloads back for resend metrics *hecMetrics + tel embed.TelemetrySettings done chan struct{} } @@ -148,6 +152,7 @@ func newACKPoller( maxRetries int, resendCh chan resendItem, metrics *hecMetrics, + tel embed.TelemetrySettings, ) *ackPoller { return &ackPoller{ logger: logger, @@ -161,6 +166,7 @@ func newACKPoller( maxRetries: maxRetries, resendCh: resendCh, metrics: metrics, + tel: tel, done: make(chan struct{}), } } @@ -191,6 +197,12 @@ func (p *ackPoller) poll() { return } + // The ACK poll is where HEC send latency really lives (indexing + // confirmation), so it gets its own span carrying the pending count. + _, span := output.StartSendSpan(ctx, p.tel, "blitz.output.hec.ack_poll") + span.SetAttributes(attribute.Int("blitz.ack.pending", len(ids))) + defer span.End() + p.metrics.recordACKPending(ctx, int64(len(ids))) // Query ACK status @@ -199,6 +211,7 @@ func (p *ackPoller) poll() { p.metrics.recordACKPollLatency(ctx, time.Since(startTime).Seconds()) if err != nil { + span.RecordError(err) p.logger.Error("Failed to query ACK status", zap.Error(err)) // Don't remove anything on query failure — will retry next cycle return diff --git a/output/hec/hec.go b/output/hec/hec.go index 844624f..cbed7e1 100644 --- a/output/hec/hec.go +++ b/output/hec/hec.go @@ -14,6 +14,7 @@ import ( "github.com/observiq/blitz/internal/config" "github.com/observiq/blitz/output" "github.com/observiq/blitz/telemetry" + "go.opentelemetry.io/otel/attribute" "go.uber.org/zap" ) @@ -268,6 +269,7 @@ func newWorker(id int, logger *zap.Logger, cfg Config, hostname string, dataChan cfg.maxRetries, w.resendCh, m, + cfg.tel, ) } @@ -363,11 +365,19 @@ func (w *worker) sendBatch(batch []output.LogRecord) { return } + // The batch POST covers many records from many emit spans, so it is a + // standalone operation span carrying the batch size. The indexing + // confirmation is traced separately in the ACK poller. + _, span := output.StartSendSpan(w.ctx, w.cfg.tel, "blitz.output.hec.send") + span.SetAttributes(attribute.Int("blitz.batch.size", len(batch))) + defer span.End() + startTime := time.Now() resp, err := w.postEvents(payload) latency := time.Since(startTime).Seconds() if err != nil { + span.RecordError(err) w.logger.Error("Failed to send HEC events", zap.Error(err), zap.Int("batch_size", len(batch))) w.metrics.recordSendError(ctx, "transport") return diff --git a/output/otlp_grpc/otlp_grpc.go b/output/otlp_grpc/otlp_grpc.go index 47610f0..8f277c2 100644 --- a/output/otlp_grpc/otlp_grpc.go +++ b/output/otlp_grpc/otlp_grpc.go @@ -12,6 +12,7 @@ import ( "github.com/observiq/blitz/internal/workermanager" "github.com/observiq/blitz/output" "github.com/observiq/blitz/telemetry" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1" collectormetrics "go.opentelemetry.io/proto/otlp/collector/metrics/v1" @@ -161,6 +162,7 @@ const outputType = "otlp-grpc" // OTLPGrpc implements the Output interface for OTLP gRPC connections type OTLPGrpc struct { logger *zap.Logger + tel embed.TelemetrySettings host string port string workers int @@ -234,6 +236,7 @@ func New(logger *zap.Logger, opts ...OTLPGrpcOption) (*OTLPGrpc, error) { otlp := &OTLPGrpc{ logger: logger.Named("output-otlp-grpc"), host: cfg.host, + tel: cfg.tel, port: cfg.port, workers: cfg.workers, insecure: cfg.insecure, @@ -535,6 +538,10 @@ func (o *OTLPGrpc) sendMetricBatch(client collectormetrics.MetricsServiceClient, return nil } + _, span := output.StartSendSpan(o.ctx, o.tel, "blitz.output.otlp.send") + span.SetAttributes(attribute.Int("blitz.batch.size", len(metrics)), attribute.String("blitz.signal", "metrics")) + defer span.End() + rm := buildMetricRequest(metrics, nil) request := &collectormetrics.ExportMetricsServiceRequest{ ResourceMetrics: []*metricspb.ResourceMetrics{rm}, @@ -547,6 +554,7 @@ func (o *OTLPGrpc) sendMetricBatch(client collectormetrics.MetricsServiceClient, startTime := time.Now() _, err := client.Export(ctx, request) if err != nil { + span.RecordError(err) o.metrics.BlitzOutputSendErrorsCounter.Add(context.Background(), 1, outputType, "metrics") return fmt.Errorf("failed to export metrics: %w", err) } @@ -566,6 +574,10 @@ func (o *OTLPGrpc) sendTraceBatch(client collectortrace.TraceServiceClient, batc return nil } + _, span := output.StartSendSpan(o.ctx, o.tel, "blitz.output.otlp.send") + span.SetAttributes(attribute.Int("blitz.batch.size", len(spans)), attribute.String("blitz.signal", "traces")) + defer span.End() + rs := buildTraceRequest(spans) request := &collectortrace.ExportTraceServiceRequest{ ResourceSpans: []*tracepb.ResourceSpans{rs}, @@ -672,6 +684,13 @@ func (o *OTLPGrpc) sendBatch(client collectorlogs.LogsServiceClient, batch *logB return nil } + // The batch send covers many records from many emit spans, so it is a + // standalone operation span carrying the batch size rather than a child of + // any single record's trace. + _, span := output.StartSendSpan(o.ctx, o.tel, "blitz.output.otlp.send") + span.SetAttributes(attribute.Int("blitz.batch.size", len(logs)), attribute.String("blitz.signal", "logs")) + defer span.End() + // Build OTLP request request := o.buildOTLPRequest(logs) @@ -683,6 +702,7 @@ func (o *OTLPGrpc) sendBatch(client collectorlogs.LogsServiceClient, batch *logB _, err := client.Export(ctx, request) if err != nil { + span.RecordError(err) o.recordSendError("export_error", err) return fmt.Errorf("failed to export logs: %w", err) } diff --git a/output/otlp_grpc/send_test.go b/output/otlp_grpc/send_test.go new file mode 100644 index 0000000..756f1c6 --- /dev/null +++ b/output/otlp_grpc/send_test.go @@ -0,0 +1,83 @@ +package otlpgrpc + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/observiq/blitz/embed" + "github.com/observiq/blitz/output" + "github.com/stretchr/testify/require" + collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1" + collectormetrics "go.opentelemetry.io/proto/otlp/collector/metrics/v1" + collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1" + logspb "go.opentelemetry.io/proto/otlp/logs/v1" + metricspb "go.opentelemetry.io/proto/otlp/metrics/v1" + tracepb "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/grpc" +) + +type mockLogsClient struct{ err error } + +func (m mockLogsClient) Export(context.Context, *collectorlogs.ExportLogsServiceRequest, ...grpc.CallOption) (*collectorlogs.ExportLogsServiceResponse, error) { + return &collectorlogs.ExportLogsServiceResponse{}, m.err +} + +type mockMetricsClient struct{ err error } + +func (m mockMetricsClient) Export(context.Context, *collectormetrics.ExportMetricsServiceRequest, ...grpc.CallOption) (*collectormetrics.ExportMetricsServiceResponse, error) { + return &collectormetrics.ExportMetricsServiceResponse{}, m.err +} + +type mockTraceClient struct{ err error } + +func (m mockTraceClient) Export(context.Context, *collectortrace.ExportTraceServiceRequest, ...grpc.CallOption) (*collectortrace.ExportTraceServiceResponse, error) { + return &collectortrace.ExportTraceServiceResponse{}, m.err +} + +// testOTLP builds a minimal OTLPGrpc for exercising the send methods directly, +// without standing up workers or a live collector. +func testOTLP(t *testing.T) *OTLPGrpc { + t.Helper() + m, err := output.NewMetrics(nil) + require.NoError(t, err) + return &OTLPGrpc{ + ctx: context.Background(), + tel: embed.TelemetrySettings{PerBatchSpans: true}, + metrics: m, + requestTimeout: time.Second, + batchTimeout: time.Second, + } +} + +// TestOTLPGrpc_sendBatchesEmitSpans exercises the three batch-send methods on +// both the success and error paths, covering the gated send span (including +// span.RecordError on failure) and the batch-size attribute. +func TestOTLPGrpc_sendBatchesEmitSpans(t *testing.T) { + o := testOTLP(t) + + lb := newLogBatch(10, time.Second) + lb.add(&logspb.LogRecord{}) + require.NoError(t, o.sendBatch(mockLogsClient{}, lb)) + + lbErr := newLogBatch(10, time.Second) + lbErr.add(&logspb.LogRecord{}) + require.Error(t, o.sendBatch(mockLogsClient{err: errors.New("boom")}, lbErr)) + + mb := newMetricBatch(10, time.Second) + mb.add(&metricspb.Metric{}) + require.NoError(t, o.sendMetricBatch(mockMetricsClient{}, mb)) + + mbErr := newMetricBatch(10, time.Second) + mbErr.add(&metricspb.Metric{}) + require.Error(t, o.sendMetricBatch(mockMetricsClient{err: errors.New("boom")}, mbErr)) + + tb := newTraceBatch(10, time.Second) + tb.add(&tracepb.Span{}) + require.NoError(t, o.sendTraceBatch(mockTraceClient{}, tb)) + + tbErr := newTraceBatch(10, time.Second) + tbErr.add(&tracepb.Span{}) + require.Error(t, o.sendTraceBatch(mockTraceClient{err: errors.New("boom")}, tbErr)) +} diff --git a/output/tcp/tcp.go b/output/tcp/tcp.go index 05a9998..28e0153 100644 --- a/output/tcp/tcp.go +++ b/output/tcp/tcp.go @@ -35,14 +35,22 @@ const ( // outputType is the output_type attribute value for TCP metrics. const outputType = "tcp" +// tcpItem is one queued message plus the emit-span context it was written +// under, so the worker can parent its send span to the emit span. +type tcpItem struct { + ctx context.Context + msg string +} + // TCP implements the Output interface for TCP connections type TCP struct { logger *zap.Logger + tel embed.TelemetrySettings host string port string workers int tlsConfig *tls.Config - dataChan chan string + dataChan chan tcpItem ctx context.Context cancel context.CancelFunc workerManager *workermanager.WorkerManager @@ -73,11 +81,12 @@ func New(logger *zap.Logger, host, port string, workers int, tlsConfig *tls.Conf tcp := &TCP{ logger: logger.Named("output-tcp"), + tel: tel, host: host, port: port, workers: workers, tlsConfig: tlsConfig, - dataChan: make(chan string, DefaultTCPChannelSize), + dataChan: make(chan tcpItem, DefaultTCPChannelSize), ctx: ctx, cancel: cancel, metrics: m, @@ -120,7 +129,7 @@ func (t *TCP) ObserveBlitzOutputQueueSize(_ context.Context, observer metric.Int // even if the data is not written to the channel. func (t *TCP) Write(ctx context.Context, data output.LogRecord) error { select { - case t.dataChan <- data.Message: + case t.dataChan <- tcpItem{ctx: ctx, msg: data.Message}: t.metrics.BlitzOutputEntriesReceivedCounter.Add(ctx, 1, outputType, "logs") return nil case <-ctx.Done(): @@ -173,13 +182,19 @@ func (t *TCP) tcpWorker(id int) { for { select { - case data, ok := <-t.dataChan: + case item, ok := <-t.dataChan: if !ok { t.logger.Info("TCP worker exiting - channel closed", zap.Int("worker_id", id)) return } - if err := t.sendData(conn, data); err != nil { + _, span := output.StartSendSpan(item.ctx, t.tel, "blitz.output.tcp.send") + err := t.sendData(conn, item.msg) + if err != nil { + span.RecordError(err) + } + span.End() + if err != nil { t.logger.Error("Failed to send TCP data", zap.Int("worker_id", id), zap.Error(err)) diff --git a/output/udp/udp.go b/output/udp/udp.go index 438192c..d8f3ae6 100644 --- a/output/udp/udp.go +++ b/output/udp/udp.go @@ -31,13 +31,21 @@ const ( // outputType is the output_type attribute value for UDP metrics. const outputType = "udp" +// udpItem is one queued message plus the emit-span context it was written +// under, so the worker can parent its send span to the emit span. +type udpItem struct { + ctx context.Context + msg string +} + // UDP implements the Output interface for UDP connections type UDP struct { logger *zap.Logger + tel embed.TelemetrySettings host string port string workers int - dataChan chan string + dataChan chan udpItem ctx context.Context cancel context.CancelFunc workerManager *workermanager.WorkerManager @@ -68,10 +76,11 @@ func New(logger *zap.Logger, host, port string, workers int, tel embed.Telemetry udp := &UDP{ logger: logger.Named("output-udp"), + tel: tel, host: host, port: port, workers: workers, - dataChan: make(chan string, DefaultUDPChannelSize), + dataChan: make(chan udpItem, DefaultUDPChannelSize), ctx: ctx, cancel: cancel, metrics: m, @@ -113,7 +122,7 @@ func (u *UDP) ObserveBlitzOutputQueueSize(_ context.Context, observer metric.Int // even if the data is not written to the channel. func (u *UDP) Write(ctx context.Context, data output.LogRecord) error { select { - case u.dataChan <- data.Message: + case u.dataChan <- udpItem{ctx: ctx, msg: data.Message}: u.metrics.BlitzOutputEntriesReceivedCounter.Add(ctx, 1, outputType, "logs") return nil case <-ctx.Done(): @@ -166,13 +175,19 @@ func (u *UDP) udpWorker(id int) { for { select { - case data, ok := <-u.dataChan: + case item, ok := <-u.dataChan: if !ok { u.logger.Info("UDP worker exiting - channel closed", zap.Int("worker_id", id)) return } - if err := u.sendData(conn, data); err != nil { + _, span := output.StartSendSpan(item.ctx, u.tel, "blitz.output.udp.send") + err := u.sendData(conn, item.msg) + if err != nil { + span.RecordError(err) + } + span.End() + if err != nil { u.logger.Error("Failed to send UDP data", zap.Int("worker_id", id), zap.Error(err))