feat(o11y): self-tracing via runtime spans, adapter emit spans, and OTLP trace export (PIPE-1066)

cmd/blitz/main.go

@@ -25,6 +25,7 @@ import (
 	"github.com/observiq/blitz/internal/logging"
 	"github.com/observiq/blitz/internal/service"
 	"github.com/observiq/blitz/internal/telemetry/metrics"
+	"github.com/observiq/blitz/internal/telemetry/traces"
 	"github.com/observiq/blitz/output"
 	fileout "github.com/observiq/blitz/output/file"
 	hecout "github.com/observiq/blitz/output/hec"
@@ -143,10 +144,27 @@ func run(cmd *cobra.Command, args []string) error {
 		cancel()
 	}()
 
-	// Blitz routes its own self-telemetry through this bundle. Standalone
-	// leaves the providers nil so they fall back to the process-global
-	// provider configured by setupMetrics (Prometheus).
-	tel := embed.TelemetrySettings{Logger: logger}
+	// Blitz routes its own self-telemetry through this bundle. Metrics leave
+	// the provider nil so they fall back to the process-global provider
+	// configured by setupMetrics (Prometheus). Trace export is opt-in via the
+	// telemetry.traces config: when an OTLP endpoint is set, spans export
+	// there; otherwise the nil TracerProvider means spans are created but
+	// dropped by the global no-op provider.
+	tel := embed.TelemetrySettings{
+		Logger:        logger,
+		PerBatchSpans: cfg.Telemetry.Traces.PerBatchSpans,
+	}
+	if cfg.Telemetry.Traces.OTLPEndpoint != "" {
+		otlpTraces, terr := traces.NewOTLP(ctx, cfg.Telemetry.Traces.OTLPEndpoint, cfg.Telemetry.Traces.Insecure)
+		if terr != nil {
+			logger.Error("Failed to enable self-telemetry trace export", zap.Error(terr))
+			return terr
+		}
+		defer func() { _ = otlpTraces.Shutdown(context.Background()) }()
+		tel.TracerProvider = otlpTraces.Provider()
+		logger.Info("self-telemetry trace export enabled",
+			zap.String("endpoint", cfg.Telemetry.Traces.OTLPEndpoint))
+	}
 
 	// Configure output first
 	var outputInstance output.Output
@@ -355,7 +373,7 @@ func run(cmd *cobra.Command, args []string) error {
 	// Set up SIGUSR1 restart signal handler
 	setupRestartSignal(ctx, logger, tracker)
 
-	svc, err := service.New(logger, generators, outputInstance)
+	svc, err := service.New(logger, generators, outputInstance, tel)
 	if err != nil {
 		logger.Error("Failed to create service", zap.Error(err))
 		return err
@@ -427,13 +445,13 @@ func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Out
 	// when an output doesn't support a signal the configured generator
 	// needs.
 	consumers := dispatch.EmbedConsumers{
-		LogConsumer: output.WriterAsLogConsumer(out),
+		LogConsumer: output.WriterAsLogConsumer(out, tel),
 	}
 	if mw, ok := out.(output.MetricWriter); ok {
-		consumers.MetricConsumer = output.WriterAsMetricConsumer(mw)
+		consumers.MetricConsumer = output.WriterAsMetricConsumer(mw, tel)
 	}
 	if tw, ok := out.(output.TraceWriter); ok {
-		consumers.TraceConsumer = output.WriterAsTraceConsumer(tw)
+		consumers.TraceConsumer = output.WriterAsTraceConsumer(tw, tel)
 	}
 	mod, err := dispatch.ForEmbed(logger, genCfg, consumers, nil, tel)
 	if err != nil {

embed/new.go

@@ -54,7 +54,7 @@ func (r *runner) Start(ctx context.Context, host Host) error {
 	for i, m := range r.cfg.Modules {
 		rtModules[i] = m
 	}
-	rt := runtime.New(logger, rtModules)
+	rt := runtime.New(logger, rtModules, host.TracerProvider)
 	if err := rt.Start(ctx); err != nil {
 		return err
 	}

embed/telemetry.go

@@ -1,6 +1,7 @@
 package embed
 
 import (
+	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/metric"
 	metricnoop "go.opentelemetry.io/otel/metric/noop"
 	"go.opentelemetry.io/otel/trace"
@@ -35,6 +36,17 @@ type TelemetrySettings struct {
 	PerBatchSpans bool
 }
 
+// Tracer returns a tracer for the given instrumentation scope from the bundle's
+// TracerProvider. A nil TracerProvider falls back to the process global, so the
+// result is always safe to use.
+func (t TelemetrySettings) Tracer(scope string) trace.Tracer {
+	tp := t.TracerProvider
+	if tp == nil {
+		tp = otel.GetTracerProvider()
+	}
+	return tp.Tracer(scope)
+}
+
 // NopTelemetry returns a TelemetrySettings wired to no-op providers and a nop
 // logger. Use it where a caller has no telemetry to route, most commonly in
 // tests and in construction paths that record nothing. It is distinct from a

embed/telemetry_test.go

@@ -1,9 +1,12 @@
 package embed
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 )
 
 func TestNopTelemetry(t *testing.T) {
@@ -28,3 +31,25 @@ func TestTelemetrySettings_zeroValueFieldsAreNil(t *testing.T) {
 	require.Nil(t, tel.MeterProvider)
 	require.Nil(t, tel.TracerProvider)
 }
+
+func TestTelemetrySettings_Tracer_usesProvidedProvider(t *testing.T) {
+	exporter := tracetest.NewInMemoryExporter()
+	tp := sdktrace.NewTracerProvider(sdktrace.WithSyncer(exporter))
+	tel := TelemetrySettings{TracerProvider: tp}
+
+	_, span := tel.Tracer("test").Start(context.Background(), "op")
+	span.End()
+
+	spans := exporter.GetSpans()
+	require.Len(t, spans, 1)
+	require.Equal(t, "op", spans[0].Name)
+}
+
+func TestTelemetrySettings_Tracer_nilFallsBackToGlobal(t *testing.T) {
+	var tel TelemetrySettings
+
+	require.NotPanics(t, func() {
+		_, span := tel.Tracer("test").Start(context.Background(), "op")
+		span.End()
+	})
+}

go.mod

@@ -20,6 +20,8 @@ require (
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.44.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.66.0
 	go.opentelemetry.io/otel/metric v1.44.0
 	go.opentelemetry.io/otel/sdk v1.44.0
@@ -42,6 +44,7 @@ require (
 	github.com/anthropics/anthropic-sdk-go v1.19.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/ccojocar/zxcvbn-go v1.0.4 // indirect
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect

go.sum

@@ -18,6 +18,8 @@ github.com/ccojocar/zxcvbn-go v1.0.4 h1:FWnCIRMXPj43ukfX000kvBZvV6raSxakYr1nzyNr
 github.com/ccojocar/zxcvbn-go v1.0.4/go.mod h1:3GxGX+rHmueTUMvm5ium7irpyjmm7ikxYFOSJB21Das=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
+github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
@@ -159,6 +161,10 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6h
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q=
 go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
 go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 h1:qazEJlUOQzhCpzQpFETGby7EdqjI1wsd0W+6Gg1SCTU=
+go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0/go.mod h1:fOD2Yefuxixkx3ahVNf0O/PERb6r4OlbxfATVnYvzCo=
 go.opentelemetry.io/otel/exporters/prometheus v0.66.0 h1:vkrK8PAznv2NKt2r+kdu252ccGzkEqLc2aSXbQIALYQ=
 go.opentelemetry.io/otel/exporters/prometheus v0.66.0/go.mod h1:V/UB6D3vMF/UBOL5igAsAYnk1nG/bzYYTzvsB16cy7o=
 go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=

internal/config/config.go

@@ -17,8 +17,11 @@ type Config struct {
 	Generators []Generator `yaml:"generators,omitempty" mapstructure:"generators,omitempty"`
 	// Output configuration
 	Output Output `yaml:"output,omitempty" mapstructure:"output,omitempty"`
-	// Metrics configuration
+	// Metrics configuration (Prometheus scrape endpoint for self-metrics)
 	Metrics Metrics `yaml:"metrics,omitempty" mapstructure:"metrics,omitempty"`
+	// Telemetry configures export of blitz's own self-telemetry (self-traces,
+	// and later self-logs) via OTLP.
+	Telemetry Telemetry `yaml:"telemetry,omitempty" mapstructure:"telemetry,omitempty"`
 	// OnFinish controls behavior when finite generation completes.
 	// One of: "exit" (default), "idle"
 	OnFinish string `yaml:"onFinish,omitempty" mapstructure:"onFinish,omitempty"`
@@ -38,6 +41,9 @@ func (c *Config) Validate() error {
 	if err := c.Metrics.Validate(); err != nil {
 		return err
 	}
+	if err := c.Telemetry.Validate(); err != nil {
+		return err
+	}
 	if c.OnFinish != "" && c.OnFinish != "exit" && c.OnFinish != "idle" {
 		return fmt.Errorf("onFinish must be one of: exit, idle, got %q", c.OnFinish)
 	}

internal/config/override.go

@@ -338,6 +338,9 @@ func DefaultOverrides() []*Override {
 		NewOverride("output.hec.source", "HEC event source metadata", DefaultHECSource),
 		NewOverride("output.hec.sourceType", "HEC event sourcetype metadata", DefaultHECSourceType),
 		NewOverride("output.hec.index", "HEC target index (empty = token default)", ""),
+		NewOverride("telemetry.traces.otlpEndpoint", "OTLP gRPC endpoint (host:port) for exporting blitz's own spans (empty = disabled)", ""),
+		NewOverride("telemetry.traces.insecure", "send blitz's own spans over plaintext gRPC (no TLS)", false),
+		NewOverride("telemetry.traces.perBatchSpans", "enable higher-volume per-emit-cycle spans (off by default)", false),
 	}
 
 	overrides = append(overrides, tcpTLSOverrides()...)

internal/config/override_test.go

@@ -140,6 +140,9 @@ func getTestOverrideFlagsArgs() []string {
 		"--output-hec-tls-min-version", "1.3",
 		"--output-stdout-flushinterval", "50ms",
 		"--metrics-port", "8080",
+		"--telemetry-traces-otlpendpoint", "traces.example:4317",
+		"--telemetry-traces-insecure", "true",
+		"--telemetry-traces-perbatchspans", "true",
 	}
 }
 
@@ -272,6 +275,9 @@ func getTestOverrideEnvs() map[string]string {
 		"BLITZ_OUTPUT_HEC_TLS_MIN_VERSION":          "1.2",
 		"BLITZ_OUTPUT_STDOUT_FLUSHINTERVAL":         "75ms",
 		"BLITZ_METRICS_PORT":                        "9100",
+		"BLITZ_TELEMETRY_TRACES_OTLPENDPOINT":       "traces.env.example:4317",
+		"BLITZ_TELEMETRY_TRACES_INSECURE":           "true",
+		"BLITZ_TELEMETRY_TRACES_PERBATCHSPANS":      "true",
 	}
 }
 
@@ -677,6 +683,13 @@ func TestOverrideFlags(t *testing.T) {
 		Metrics: Metrics{
 			Port: 8080,
 		},
+		Telemetry: Telemetry{
+			Traces: TracesTelemetry{
+				OTLPEndpoint:  "traces.example:4317",
+				Insecure:      true,
+				PerBatchSpans: true,
+			},
+		},
 	}
 	require.Equal(t, expectedCfg, cfg)
 }
@@ -886,6 +899,13 @@ func TestOverrideEnvs(t *testing.T) {
 		Metrics: Metrics{
 			Port: 9100,
 		},
+		Telemetry: Telemetry{
+			Traces: TracesTelemetry{
+				OTLPEndpoint:  "traces.env.example:4317",
+				Insecure:      true,
+				PerBatchSpans: true,
+			},
+		},
 	}
 	require.Equal(t, expectedCfg, cfg)
 }

internal/config/telemetry.go

@@ -0,0 +1,34 @@
+package config
+
+// Telemetry configures export of blitz's OWN self-telemetry (its internal
+// logs, metrics, and traces). This is distinct from the data blitz generates.
+// The existing `metrics` block still governs the Prometheus scrape endpoint
+// for self-metrics; this block adds OTLP export for self-traces (and, in a
+// later phase, self-logs).
+type Telemetry struct {
+	// Traces configures OTLP export of blitz's internal spans.
+	Traces TracesTelemetry `yaml:"traces,omitempty" mapstructure:"traces,omitempty"`
+}
+
+// TracesTelemetry configures OTLP gRPC export of blitz's internal spans.
+type TracesTelemetry struct {
+	// OTLPEndpoint is the OTLP gRPC endpoint (host:port). Empty disables trace
+	// export: spans are still created but routed to a no-op provider.
+	OTLPEndpoint string `yaml:"otlpEndpoint,omitempty" mapstructure:"otlpEndpoint,omitempty"`
+
+	// Insecure sends spans over plaintext gRPC (no TLS). Defaults to false.
+	Insecure bool `yaml:"insecure,omitempty" mapstructure:"insecure,omitempty"`
+
+	// PerBatchSpans enables the higher-volume per-emit-cycle spans. Off by
+	// default; the coarse session and generator-lifecycle spans do not depend
+	// on it.
+	PerBatchSpans bool `yaml:"perBatchSpans,omitempty" mapstructure:"perBatchSpans,omitempty"`
+}
+
+// Validate validates the telemetry configuration. Export is off by default and
+// all fields are optional, so there is nothing to reject today; the method
+// exists to match the config-block Validate convention and to host future
+// checks (e.g. endpoint format).
+func (t Telemetry) Validate() error {
+	return nil
+}

internal/runtime/runtime.go

@@ -4,9 +4,16 @@ import (
 	"context"
 	"fmt"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 )
 
+// tracerScope is the instrumentation scope for the runtime's self-telemetry
+// spans.
+const tracerScope = "github.com/observiq/blitz/internal/runtime"
+
 // Module is the narrow lifecycle interface Runtime operates on. The
 // embed.ProducerModule type can be wrapped to satisfy this contract —
 // Runtime stays decoupled from the embed package to avoid an import
@@ -31,40 +38,68 @@ type Module interface {
 // host-level concerns. CLI adds signal handling, YAML loading, and
 // output wiring; embed adds host-supplied consumers and resource
 // attributes.
+//
+// Runtime emits blitz's session-level self-telemetry: a root "blitz.session"
+// span covering Start to Stop, with a child "blitz.generator.run" span per
+// module (bounded by that module's lifetime). These spans are decoupled from
+// the embed package; the caller passes a raw trace.TracerProvider.
 type Runtime struct {
 	logger  *zap.Logger
 	modules []Module
+	tracer  trace.Tracer
+
+	// sessionSpan and moduleSpans hold the open self-telemetry spans between
+	// Start and Stop. Start and Stop are called once each and never
+	// concurrently, so no synchronization is needed. moduleSpans is
+	// index-aligned with modules.
+	sessionSpan trace.Span
+	moduleSpans []trace.Span
 }
 
-// New returns a Runtime configured with the given logger and modules.
-func New(logger *zap.Logger, modules []Module) *Runtime {
+// New returns a Runtime configured with the given logger, modules, and tracer
+// provider. A nil tracerProvider falls back to the process global, so span
+// emission is always safe.
+func New(logger *zap.Logger, modules []Module, tracerProvider trace.TracerProvider) *Runtime {
 	if logger == nil {
 		logger = zap.NewNop()
 	}
+	if tracerProvider == nil {
+		tracerProvider = otel.GetTracerProvider()
+	}
 	return &Runtime{
 		logger:  logger,
 		modules: modules,
+		tracer:  tracerProvider.Tracer(tracerScope),
 	}
 }
 
 // Start begins every configured module. If any module's Start returns
 // an error, Start stops the modules already started (in reverse order)
 // and returns the failure.
 func (r *Runtime) Start(ctx context.Context) error {
+	ctx, r.sessionSpan = r.tracer.Start(ctx, "blitz.session")
+	r.moduleSpans = make([]trace.Span, 0, len(r.modules))
+
 	started := make([]Module, 0, len(r.modules))
 	for _, m := range r.modules {
-		if err := m.Start(ctx); err != nil {
+		mctx, mspan := r.tracer.Start(ctx, "blitz.generator.run",
+			trace.WithAttributes(attribute.String("blitz.generator.name", m.Name())))
+		if err := m.Start(mctx); err != nil {
+			mspan.End()
 			// Roll back: stop modules already started, in reverse order.
 			for i := len(started) - 1; i >= 0; i-- {
 				if stopErr := started[i].Stop(ctx); stopErr != nil {
 					r.logger.Warn("module stop failed during start rollback",
 						zap.String("module", started[i].Name()),
 						zap.Error(stopErr))
 				}
+				r.moduleSpans[i].End()
 			}
+			r.sessionSpan.End()
 			return fmt.Errorf("start module %s: %w", m.Name(), err)
 		}
 		started = append(started, m)
+		r.moduleSpans = append(r.moduleSpans, mspan)
 	}
 	return nil
 }
@@ -83,6 +118,12 @@ func (r *Runtime) Stop(ctx context.Context) error {
 				firstErr = fmt.Errorf("stop module %s: %w", m.Name(), err)
 			}
 		}
+		if i < len(r.moduleSpans) && r.moduleSpans[i] != nil {
+			r.moduleSpans[i].End()
+		}
+	}
+	if r.sessionSpan != nil {
+		r.sessionSpan.End()
 	}
 	return firstErr
 }