feat(o11y): per-batch output send spans + HEC ACK-poll span (PIPE-1066)

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

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

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
 			}

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

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

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)
 	}

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))
+}