feat(o11y): per-instance metrics via injectable MeterProvider, drop package globals (PIPE-1066)

Makefile

@@ -69,15 +69,15 @@ generate-o11y-check: generate-o11y
 
 generate-o11y:
 	@echo "Discovering metric registries..."
-	@for manifest in $$(find . -name "registry_manifest.yaml" -path "*/monitoring/registry_manifest.yaml" | sort); do \
+	@set -e; for manifest in $$(find . -name "registry_manifest.yaml" -path "*/monitoring/registry_manifest.yaml" | sort); do \
 		registry_path=$$(dirname $$(dirname $$manifest) | sed 's|^\./||'); \
 		depth=$$(echo "$$registry_path" | tr '/' '\n' | wc -l | tr -d ' '); \
 		rel_path=""; \
 		for i in $$(seq 1 $$depth); do rel_path="../$$rel_path"; done; \
 		echo "Generating metrics for $$registry_path..."; \
-		docker run --rm -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry check --registry=./monitoring; \
-		docker run --rm -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry generate --registry=./monitoring --templates=$${rel_path}weaver/templates --config=$${rel_path}weaver-go.yaml go .; \
-		docker run --rm -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry generate --registry=./monitoring --templates=$${rel_path}weaver/templates --config=$${rel_path}weaver-markdown.yaml markdown .; \
+		docker run --rm --user $$(id -u):$$(id -g) -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry check --registry=./monitoring; \
+		docker run --rm --user $$(id -u):$$(id -g) -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry generate --registry=./monitoring --templates=$${rel_path}weaver/templates --config=$${rel_path}weaver-go.yaml go .; \
+		docker run --rm --user $$(id -u):$$(id -g) -v ${PWD}:/workspace -w /workspace/$$registry_path $(WEAVER_IMAGE) registry generate --registry=./monitoring --templates=$${rel_path}weaver/templates --config=$${rel_path}weaver-markdown.yaml markdown .; \
 		echo "✓ Generated $$registry_path/monitoring.go and $$registry_path/monitoring.md"; \
 	done
 	@go fmt ./...

cmd/blitz/main.go

@@ -15,6 +15,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/observiq/blitz/embed"
 	"github.com/observiq/blitz/generator/count"
 	gennop "github.com/observiq/blitz/generator/nop"
 	"github.com/observiq/blitz/generator/winevt"
@@ -142,6 +143,11 @@ 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}
+
 	// Configure output first
 	var outputInstance output.Output
 	switch cfg.Output.Type {
@@ -175,6 +181,7 @@ func run(cmd *cobra.Command, args []string) error {
 			strconv.Itoa(cfg.Output.TCP.Port),
 			cfg.Output.TCP.Workers,
 			tlsConfig,
+			tel,
 		)
 		if err != nil {
 			logger.Error("Failed to create TCP output", zap.Error(err))
@@ -186,6 +193,7 @@ func run(cmd *cobra.Command, args []string) error {
 			cfg.Output.UDP.Host,
 			strconv.Itoa(cfg.Output.UDP.Port),
 			cfg.Output.UDP.Workers,
+			tel,
 		)
 		if err != nil {
 			logger.Error("Failed to create UDP output", zap.Error(err))
@@ -214,6 +222,7 @@ func run(cmd *cobra.Command, args []string) error {
 			MsgID:            cfg.Output.Syslog.MsgID,
 			MaxDatagramBytes: cfg.Output.Syslog.MaxDatagramBytes,
 			TLSConfig:        tlsConfig,
+			Telemetry:        tel,
 		}
 		outputInstance, err = syslogout.New(logger, sysCfg)
 		if err != nil {
@@ -240,6 +249,7 @@ func run(cmd *cobra.Command, args []string) error {
 		}
 		// Set insecure flag
 		opts = append(opts, otlpgrpc.WithInsecure(cfg.Output.OTLPGrpc.Insecure))
+		opts = append(opts, otlpgrpc.WithTelemetry(tel))
 		// If TLS is enabled and not insecure, set up TLS
 		if cfg.Output.OTLPGrpc.EnableTLS && !cfg.Output.OTLPGrpc.Insecure {
 			var tlsConfig *tls.Config
@@ -268,6 +278,7 @@ func run(cmd *cobra.Command, args []string) error {
 			cfg.Output.File.Path,
 			cfg.Output.File.Workers,
 			rot,
+			tel,
 		)
 		if err != nil {
 			logger.Error("Failed to create File output", zap.Error(err))
@@ -290,6 +301,7 @@ func run(cmd *cobra.Command, args []string) error {
 			hecout.WithSourceType(cfg.Output.HEC.SourceType),
 			hecout.WithIndex(cfg.Output.HEC.Index),
 			hecout.WithEnableTLS(cfg.Output.HEC.EnableTLS),
+			hecout.WithTelemetry(tel),
 		}
 		if cfg.Output.HEC.EnableTLS {
 			var tlsConfig *tls.Config
@@ -316,7 +328,7 @@ func run(cmd *cobra.Command, args []string) error {
 	var tracker *count.Tracker
 
 	for _, genCfg := range effectiveGens {
-		gen, genErr := createGenerator(logger, genCfg, outputInstance)
+		gen, genErr := createGenerator(logger, genCfg, outputInstance, tel)
 		if genErr != nil {
 			logger.Error("Failed to create generator",
 				zap.String("type", string(genCfg.Type)),
@@ -395,7 +407,7 @@ shutdown:
 	return nil
 }
 
-func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Output) (any, error) {
+func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Output, tel embed.TelemetrySettings) (any, error) {
 	// Standalone-CLI-only generator types that dispatch.ForEmbed does not
 	// construct (winevt is deprecated for embed; nop yields no records).
 	// All other generators delegate to dispatch.ForEmbed so the
@@ -404,7 +416,7 @@ func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Out
 	case config.GeneratorTypeNop:
 		return gennop.New(logger)
 	case config.GeneratorTypeWinevt:
-		return winevt.New(logger, genCfg.Winevt.Workers, genCfg.Winevt.Rate)
+		return winevt.New(logger, genCfg.Winevt.Workers, genCfg.Winevt.Rate, tel)
 	}
 
 	// All remaining (embed-eligible) types go through the canonical
@@ -423,7 +435,7 @@ func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Out
 	if tw, ok := out.(output.TraceWriter); ok {
 		consumers.TraceConsumer = output.WriterAsTraceConsumer(tw)
 	}
-	mod, err := dispatch.ForEmbed(logger, genCfg, consumers, nil)
+	mod, err := dispatch.ForEmbed(logger, genCfg, consumers, nil, tel)
 	if err != nil {
 		return nil, err
 	}

config/loader.go

@@ -122,6 +122,11 @@ type EmbedOpts struct {
 	// the CLI's BLITZ_* env-var path; blitz never reads os.Environ()
 	// itself in embedded mode.
 	EnvOverrides map[string]string
+
+	// Telemetry carries the OTel providers blitz routes its own
+	// self-telemetry through, forwarded to every constructed generator. The
+	// zero value falls back to the process-global providers.
+	Telemetry embed.TelemetrySettings
 }
 
 // LoadModules parses blitz YAML bytes, constructs the corresponding
@@ -156,7 +161,7 @@ func LoadModules(yamlBytes []byte, opts EmbedOpts) ([]embed.ProducerModule, erro
 	}
 	modules := make([]embed.ProducerModule, 0, len(gens))
 	for i, gen := range gens {
-		mod, err := dispatch.ForEmbed(logger, gen, consumers, opts.FileGenLibrary)
+		mod, err := dispatch.ForEmbed(logger, gen, consumers, opts.FileGenLibrary, opts.Telemetry)
 		if err != nil {
 			return nil, fmt.Errorf("generator[%d] type=%q: %w", i, gen.Type, err)
 		}

embed/host.go

@@ -1,6 +1,10 @@
 package embed
 
-import "go.uber.org/zap"
+import (
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/zap"
+)
 
 // Host is the bundle of consumers and ambient resources a host process
 // supplies to an embedded blitz runner.
@@ -31,6 +35,21 @@ type Host struct {
 	// treat their own reference as frozen once they hand the Host off.
 	// See cloneResource in this package.
 	Resource map[string]string
+
+	// MeterProvider is the OTel MeterProvider blitz routes its own internal
+	// metrics through (generator and output self-telemetry). Nil falls back
+	// to the process-global provider, matching standalone behavior.
+	MeterProvider metric.MeterProvider
+
+	// TracerProvider is the OTel TracerProvider blitz routes its own internal
+	// spans through. Nil falls back to the process-global provider. Reserved
+	// for the self-tracing phase; blitz emits no internal spans yet.
+	TracerProvider trace.TracerProvider
+
+	// PerBatchSpans enables the higher-volume per-emit-cycle spans once
+	// self-tracing lands. Off by default; the coarse spans do not depend on
+	// it.
+	PerBatchSpans bool
 }
 
 // cloneResource returns a defensive copy of m. Runner.Start uses it so

embed/integration_test.go

@@ -67,7 +67,7 @@ func TestEmbed_ApacheRecordsFlowToMemoryConsumer(t *testing.T) {
 	logger := zaptest.NewLogger(t)
 	consumer := &memoryLogConsumer{}
 
-	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer)
+	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer, embed.NopTelemetry())
 	require.NoError(t, err)
 
 	runner, err := embed.New(embed.Config{
@@ -232,7 +232,7 @@ func TestEmbed_HostMetricsPointsFlowToMemoryConsumer(t *testing.T) {
 func TestEmbed_RunnerRejectsDoubleStart(t *testing.T) {
 	logger := zaptest.NewLogger(t)
 	consumer := &memoryLogConsumer{}
-	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer)
+	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer, embed.NopTelemetry())
 	require.NoError(t, err)
 
 	runner, err := embed.New(embed.Config{Modules: []embed.ProducerModule{gen}})

embed/new_race_test.go

@@ -24,7 +24,7 @@ func TestRunner_ConcurrentStartIsSerialized(t *testing.T) {
 	logger := zaptest.NewLogger(t)
 	consumer := &memoryLogConsumer{}
 
-	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer)
+	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer, embed.NopTelemetry())
 	require.NoError(t, err)
 
 	runner, err := embed.New(embed.Config{Modules: []embed.ProducerModule{gen}})
@@ -72,7 +72,7 @@ func TestRunner_ConcurrentStopIsIdempotent(t *testing.T) {
 	logger := zaptest.NewLogger(t)
 	consumer := &memoryLogConsumer{}
 
-	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer)
+	gen, err := apache.New(logger, 1, 10*time.Millisecond, consumer, embed.NopTelemetry())
 	require.NoError(t, err)
 
 	runner, err := embed.New(embed.Config{Modules: []embed.ProducerModule{gen}})

generator/apache/apache.go

@@ -46,11 +46,12 @@ type ApacheLogGenerator struct {
 	wg       sync.WaitGroup
 	stopCh   chan struct{}
 	tracker  *count.Tracker
+	metrics  *generator.Metrics
 }
 
 // New creates a new Apache log generator. The consumer receives each
 // generated record as a size-1 batch via ConsumeLogs.
-func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.LogConsumer) (*ApacheLogGenerator, error) {
+func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.LogConsumer, tel embed.TelemetrySettings) (*ApacheLogGenerator, error) {
 	if logger == nil {
 		return nil, fmt.Errorf("logger cannot be nil")
 	}
@@ -63,11 +64,17 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log
 		return nil, fmt.Errorf("consumer cannot be nil")
 	}
 
+	metrics, err := generator.NewMetrics(tel.MeterProvider)
+	if err != nil {
+		return nil, fmt.Errorf("build generator metrics: %w", err)
+	}
+
 	return &ApacheLogGenerator{
 		logger:   logger,
 		workers:  workers,
 		rate:     rate,
 		consumer: consumer,
+		metrics:  metrics,
 		stopCh:   make(chan struct{}),
 	}, nil
 }
@@ -83,7 +90,7 @@ func (g *ApacheLogGenerator) Start(_ context.Context) error {
 		zap.Duration("rate", g.rate))
 
 	// Record initial active workers count
-	generator.BlitzGeneratorActiveWorkersGauge.Record(context.Background(), int64(g.workers), componentName)
+	g.metrics.BlitzGeneratorActiveWorkersGauge.Record(context.Background(), int64(g.workers), componentName)
 
 	for i := 0; i < g.workers; i++ {
 		g.wg.Add(1)
@@ -99,7 +106,7 @@ func (g *ApacheLogGenerator) Stop(ctx context.Context) error {
 	g.logger.Info("Stopping Apache log generator")
 
 	// Record zero active workers
-	generator.BlitzGeneratorActiveWorkersGauge.Record(ctx, 0, componentName)
+	g.metrics.BlitzGeneratorActiveWorkersGauge.Record(ctx, 0, componentName)
 
 	close(g.stopCh)
 
@@ -181,7 +188,7 @@ func (g *ApacheLogGenerator) generateAndWriteLog(_ int) error {
 	}
 
 	// Record logs generated counter
-	generator.BlitzGeneratorEntriesCounter.Add(context.Background(), 1, componentName)
+	g.metrics.BlitzGeneratorEntriesCounter.Add(context.Background(), 1, componentName)
 
 	// Push as a size-1 batch with timeout
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -393,7 +400,7 @@ func parseApacheCLF(line string) (map[string]any, error) {
 
 // recordWriteError records metrics for write errors
 func (g *ApacheLogGenerator) recordWriteError(errorType string, _ error) {
-	generator.BlitzGeneratorWriteErrorsCounter.Add(context.Background(), 1, componentName,
+	g.metrics.BlitzGeneratorWriteErrorsCounter.Add(context.Background(), 1, componentName,
 		metric.WithAttributeSet(attribute.NewSet(attribute.String("error_type", errorType))),
 	)
 }