feat(prommap): shared Prometheus metric-mapping package (PIPE-1486, BP-471)

cmd/blitz/main.go

@@ -167,6 +167,7 @@ func run(cmd *cobra.Command, args []string) error {
 	// Emit Warn-level banners for any deprecated generator types
 	// configured by the user. Fires once per startup, not per record.
 	config.LogGeneratorDeprecations(logger, cfg)
+	config.LogRemovedSettings(logger, cfg)
 
 	if err := setupMetrics(ctx, cfg, logger); err != nil {
 		logger.Error("Failed to setup metrics", zap.Error(err))

config/loader_test.go

@@ -168,7 +168,6 @@ func TestLoadModules_HostMetricsRequiresMetricConsumer(t *testing.T) {
 generator:
   type: hostmetrics
   hostmetrics:
-    workers: 1
     rate: 1s
     os: linux
 output:

docker/docker-compose.telemetry-generator.yml

@@ -188,7 +188,6 @@ services:
     <<: *blitz-common
     environment:
       BLITZ_GENERATOR_TYPE: hostmetrics
-      BLITZ_GENERATOR_HOSTMETRICS_WORKERS: ${BLITZ_WORKERS:-1}
       BLITZ_GENERATOR_HOSTMETRICS_RATE: ${BLITZ_RATE:-1s}
       BLITZ_GENERATOR_HOSTMETRICS_OS: linux
       BLITZ_OUTPUT_TYPE: otlp-grpc
@@ -200,7 +199,6 @@ services:
     <<: *blitz-common
     environment:
       BLITZ_GENERATOR_TYPE: hostmetrics
-      BLITZ_GENERATOR_HOSTMETRICS_WORKERS: ${BLITZ_WORKERS:-1}
       BLITZ_GENERATOR_HOSTMETRICS_RATE: ${BLITZ_RATE:-1s}
       BLITZ_GENERATOR_HOSTMETRICS_OS: windows
       BLITZ_OUTPUT_TYPE: otlp-grpc

docs/generator/hostmetrics.md

@@ -28,19 +28,21 @@ The generator includes 8 scrapers, each producing metrics for a specific subsyst
 | YAML Path                          | Flag Name                          | Environment Variable                    | Default   | Description                                                          |
 |------------------------------------|------------------------------------|-----------------------------------------|-----------|----------------------------------------------------------------------|
 | `generator.type`                   | `--generator-type`                 | `BLITZ_GENERATOR_TYPE`                  | `nop`     | Generator type. Set to `hostmetrics` to use this generator.          |
-| `generator.hostmetrics.workers`    | `--generator-hostmetrics-workers`  | `BLITZ_GENERATOR_HOSTMETRICS_WORKERS`   | `1`       | Number of worker goroutines.                                         |
 | `generator.hostmetrics.rate`       | `--generator-hostmetrics-rate`     | `BLITZ_GENERATOR_HOSTMETRICS_RATE`      | `1s`      | Scrape interval for host metrics.                                    |
 | `generator.hostmetrics.os`         | `--generator-hostmetrics-os`       | `BLITZ_GENERATOR_HOSTMETRICS_OS`        | `linux`   | Simulated operating system. One of: `linux`, `windows`.              |
 | `generator.hostmetrics.hostname`   | `--generator-hostmetrics-hostname` | `BLITZ_GENERATOR_HOSTMETRICS_HOSTNAME`  | (random)  | Simulated hostname. If empty, a random hostname is generated.        |
 | `generator.hostmetrics.scrapers`   | `--generator-hostmetrics-scrapers` | `BLITZ_GENERATOR_HOSTMETRICS_SCRAPERS`  | (all)     | Scrapers to enable. If empty, all scrapers are enabled.              |
 
+`generator.hostmetrics.workers` has been removed. One simulated host runs one worker, so use `rate` for more
+frequent writes and add `generators:` entries for more hosts. A configured value is ignored with a warning, and is
+expected to fail config validation as of v0.25.0.
+
 ## Example Configuration
 
 ```yaml
 generator:
   type: hostmetrics
   hostmetrics:
-    workers: 1
     rate: 1s
     os: linux
     scrapers:

embed/record.go

@@ -45,13 +45,13 @@ type LogRecordMetadata struct {
 type MetricType string
 
 const (
-	// MetricTypeGauge represents a gauge metric.
+	// MetricTypeGauge is a point-in-time value that can rise or fall.
 	MetricTypeGauge MetricType = "gauge"
-	// MetricTypeSum represents a sum metric.
+	// MetricTypeSum is a non-monotonic sum (OTel Sum monotonic=false; Prometheus gauge).
 	MetricTypeSum MetricType = "sum"
-	// MetricTypeCounter represents a counter metric.
+	// MetricTypeCounter is a monotonic cumulative count (OTel Sum monotonic=true; Prometheus counter, _total).
 	MetricTypeCounter MetricType = "counter"
-	// MetricTypeHistogram represents a histogram metric.
+	// MetricTypeHistogram is a bucketed distribution (Prometheus _bucket/_sum/_count).
 	MetricTypeHistogram MetricType = "histogram"
 )
 

generator/hostmetrics/hostmetrics.go

@@ -24,9 +24,13 @@ const generatorType = "hostmetrics"
 type Config struct {
 	// Logger is the zap logger used for diagnostic output. Required.
 	Logger *zap.Logger
-	// Workers is the number of worker goroutines. Required, >= 1.
+	// Workers is ignored: one simulated host runs one worker, and Rate is the
+	// load knob. Parallel workers only emitted duplicate series for the same
+	// host.
+	//
+	// Deprecated: ignored; kept so existing embed callers still compile.
 	Workers int
-	// Rate is the scrape interval per worker. Required, > 0.
+	// Rate is the scrape interval. Required, > 0.
 	Rate time.Duration
 	// OS is the simulated operating system ("linux" or "windows"). Ignored
 	// when Identity is set (the identity's own OS is used instead).
@@ -65,7 +69,6 @@ type Generator struct {
 	embed.ProducerMarker
 
 	logger   *zap.Logger
-	workers  int
 	rate     time.Duration
 	osType   string
 	hostname string
@@ -93,9 +96,6 @@ func New(cfg Config) (*Generator, error) {
 	if cfg.Consumer == nil {
 		return nil, fmt.Errorf("MetricConsumer cannot be nil")
 	}
-	if cfg.Workers < 1 {
-		return nil, fmt.Errorf("workers must be 1 or greater, got %d", cfg.Workers)
-	}
 	if cfg.Rate <= 0 {
 		return nil, fmt.Errorf("rate must be greater than 0, got %s", cfg.Rate)
 	}
@@ -116,7 +116,6 @@ func New(cfg Config) (*Generator, error) {
 
 	return &Generator{
 		logger:   cfg.Logger.Named("generator-hostmetrics"),
-		workers:  cfg.Workers,
 		rate:     cfg.Rate,
 		osType:   sys.OSInfo.Type.SemconvOSType(),
 		hostname: sys.Hostname,
@@ -169,27 +168,25 @@ func (g *Generator) SetCountTracker(tracker *count.Tracker) {
 	g.tracker = tracker
 }
 
-// Start launches the worker goroutines.
+// Start launches the single worker: one simulated host emits one sample per
+// series per rate interval.
 func (g *Generator) Start(_ context.Context) error {
 	g.logger.Info("Starting host metrics generator",
-		zap.Int("workers", g.workers),
 		zap.Duration("rate", g.rate),
 		zap.String("os.type", g.osType),
 		zap.String("hostname", g.hostname),
 		zap.Int("scrapers", len(g.scrapers)),
 	)
 
-	g.metrics.BlitzGeneratorActiveWorkersGauge.Record(context.Background(), int64(g.workers), generatorType)
+	g.metrics.BlitzGeneratorActiveWorkersGauge.Record(context.Background(), 1, generatorType)
 
-	for i := range g.workers {
-		g.wg.Add(1)
-		go g.worker(i)
-	}
+	g.wg.Add(1)
+	go g.worker(0)
 
 	return nil
 }
 
-// Stop signals workers to drain and waits for them to exit.
+// Stop signals the worker to drain and waits for it to exit.
 func (g *Generator) Stop(ctx context.Context) error {
 	g.logger.Info("Stopping host metrics generator")
 
@@ -253,9 +250,8 @@ func (g *Generator) scrape(r *rand.Rand) {
 
 	// The host-identity resource is fixed for this generator's lifetime, so it
 	// is built once (StaticResources in New) and shared read-only across every
-	// scrape and worker. Scrapers only attach it to the MetricRecords they
-	// return — they never mutate it — so handing out the zero-allocation shared
-	// map is safe under concurrent workers.
+	// scrape. Scrapers only attach it to the MetricRecords they return — they
+	// never mutate it — so handing out the zero-allocation shared map is safe.
 	res := g.static.Record()
 
 	for _, scraper := range g.scrapers {

generator/hostmetrics/hostmetrics_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/observiq/blitz/telemetry"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -101,14 +103,45 @@ func TestNew(t *testing.T) {
 		require.Error(t, err)
 	})
 
-	t.Run("invalid workers", func(t *testing.T) {
+	t.Run("workers unset is valid", func(t *testing.T) {
 		cfg := baseCfg(t, &mockMetricConsumer{})
 		cfg.Workers = 0
 		_, err := New(cfg)
-		require.Error(t, err)
+		require.NoError(t, err)
 	})
 }
 
+// TestWorkersIgnored asserts one simulated host runs exactly one worker no
+// matter what Workers says: parallel workers would emit duplicate series for
+// the same host, and rate is the load knob.
+func TestWorkersIgnored(t *testing.T) {
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+
+	cfg := baseCfg(t, &mockMetricConsumer{})
+	cfg.Workers = 4
+	cfg.Telemetry = embed.TelemetrySettings{MeterProvider: mp}
+	g, err := New(cfg)
+	require.NoError(t, err)
+	require.NoError(t, g.Start(context.Background()))
+	t.Cleanup(func() { _ = g.Stop(context.Background()) })
+
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(context.Background(), &rm))
+	var active int64 = -1
+	for _, sm := range rm.ScopeMetrics {
+		for _, md := range sm.Metrics {
+			if md.Name != "blitz.generator.active_workers" {
+				continue
+			}
+			if gauge, ok := md.Data.(metricdata.Gauge[int64]); ok && len(gauge.DataPoints) > 0 {
+				active = gauge.DataPoints[0].Value
+			}
+		}
+	}
+	require.Equal(t, int64(1), active)
+}
+
 // TestNewProjectsIdentityResource confirms that when a resolved datagen
 // identity is supplied, the generator's static resource carries the full
 // host.* / os.* / deployment.* projection (os.type as the semconv value, so

internal/config/generator_hostmetrics.go

@@ -9,7 +9,11 @@ import (
 
 // HostMetricsGeneratorConfig contains configuration for host metrics generator
 type HostMetricsGeneratorConfig struct {
-	// Workers is the number of worker goroutines for host metrics generation
+	// Workers is removed and ignored: one simulated host runs one worker, and
+	// Rate is the load knob. Kept only so a still-configured value can be
+	// detected and warned about (see LogRemovedSettings).
+	//
+	// Deprecated: ignored; expected to fail validation as of v0.25.0.
 	Workers int `yaml:"workers,omitempty" mapstructure:"workers,omitempty"`
 	// Rate is the scrape interval for host metrics
 	Rate time.Duration `yaml:"rate,omitempty" mapstructure:"rate,omitempty"`
@@ -45,10 +49,6 @@ var ValidScrapers = []string{
 
 // Validate validates the host metrics generator configuration
 func (c *HostMetricsGeneratorConfig) Validate() error {
-	if c.Workers < 1 {
-		return fmt.Errorf("hostmetrics generator workers must be 1 or greater, got %d", c.Workers)
-	}
-
 	if c.Rate <= 0 {
 		return fmt.Errorf("hostmetrics generator rate must be positive, got %v", c.Rate)
 	}

internal/config/generator_hostmetrics_test.go

@@ -39,13 +39,11 @@ func TestHostMetricsGeneratorConfig_Validate(t *testing.T) {
 			},
 		},
 		{
-			name: "invalid workers",
+			// workers was removed (rate is the load knob); unset is valid.
+			name: "workers unset",
 			config: HostMetricsGeneratorConfig{
-				Workers: 0,
-				Rate:    time.Second,
+				Rate: time.Second,
 			},
-			wantErr: true,
-			errMsg:  "workers must be 1 or greater",
 		},
 		{
 			name: "invalid rate",

internal/config/migrate.go

@@ -78,3 +78,32 @@ func LogGeneratorDeprecations(logger *zap.Logger, cfg *Config) {
 		}
 	}
 }
+
+// LogRemovedSettings emits a Warn once per startup for every configured
+// setting that has been removed. A removed setting is ignored rather than
+// rejected during a deprecation window, so existing configs keep loading.
+//
+// Currently emits warnings for:
+//   - generator.hostmetrics.workers: one simulated host runs one worker, and
+//     rate is the load knob. Parallel workers only duplicated the same host's
+//     series.
+//
+// TODO: hostmetrics `workers` was removed (rate is the load knob). Expected in
+// v0.25.0: turn this warning into a config validation error and drop the
+// deprecated --generator-hostmetrics-workers flag.
+func LogRemovedSettings(logger *zap.Logger, cfg *Config) {
+	if logger == nil || cfg == nil {
+		return
+	}
+	for _, g := range cfg.EffectiveGenerators() {
+		if g.Type == GeneratorTypeHostMetrics && g.HostMetrics.Workers != 0 {
+			logger.Warn(HostMetricsWorkersRemoved)
+		}
+	}
+}
+
+// HostMetricsWorkersRemoved is the warning for the removed
+// generator.hostmetrics.workers setting, shared by the startup log and the
+// deprecated CLI flag.
+const HostMetricsWorkersRemoved = "`generator.hostmetrics.workers` is no longer supported and is ignored; " +
+	"use `rate` for more frequent writes. Setting it is expected to fail config validation as of v0.25.0."

internal/config/migrate_test.go

@@ -221,3 +221,62 @@ func TestLogGeneratorDeprecations_NilSafe(t *testing.T) {
 		config.LogGeneratorDeprecations(nil, &config.Config{})
 	})
 }
+
+// TestLogRemovedSettings_HostMetricsWorkers asserts a Warn fires when the
+// removed generator.hostmetrics.workers setting is still configured, and that
+// it points to rate and the expected v0.25.0 validation failure.
+func TestLogRemovedSettings_HostMetricsWorkers(t *testing.T) {
+	core, recorded := observer.New(zap.WarnLevel)
+	logger := zap.New(core)
+
+	cfg := &config.Config{
+		Generator: config.Generator{
+			Type:        config.GeneratorTypeHostMetrics,
+			HostMetrics: config.HostMetricsGeneratorConfig{Workers: 4, Rate: time.Second},
+		},
+	}
+
+	config.LogRemovedSettings(logger, cfg)
+
+	entries := recorded.FilterMessageSnippet("generator.hostmetrics.workers").All()
+	require.Len(t, entries, 1)
+	assert.Contains(t, entries[0].Message, "ignored")
+	assert.Contains(t, entries[0].Message, "rate")
+	assert.Contains(t, entries[0].Message, "v0.25.0")
+}
+
+// TestLogRemovedSettings_GeneratorsList covers hostmetrics entries in the
+// generators: list, one Warn per offending entry.
+func TestLogRemovedSettings_GeneratorsList(t *testing.T) {
+	core, recorded := observer.New(zap.WarnLevel)
+	logger := zap.New(core)
+
+	cfg := &config.Config{
+		Generators: []config.Generator{
+			{Type: config.GeneratorTypeHostMetrics, HostMetrics: config.HostMetricsGeneratorConfig{Workers: 2, Rate: time.Second}},
+			{Type: config.GeneratorTypeHostMetrics, HostMetrics: config.HostMetricsGeneratorConfig{Rate: time.Second}},
+		},
+	}
+
+	config.LogRemovedSettings(logger, cfg)
+
+	assert.Len(t, recorded.FilterMessageSnippet("generator.hostmetrics.workers").All(), 1)
+}
+
+// TestLogRemovedSettings_NotSetNoWarn confirms no warning when workers is unset.
+func TestLogRemovedSettings_NotSetNoWarn(t *testing.T) {
+	core, recorded := observer.New(zap.WarnLevel)
+	logger := zap.New(core)
+
+	cfg := &config.Config{
+		Generator: config.Generator{
+			Type:        config.GeneratorTypeHostMetrics,
+			HostMetrics: config.HostMetricsGeneratorConfig{Rate: time.Second},
+		},
+	}
+
+	config.LogRemovedSettings(logger, cfg)
+
+	assert.Empty(t, recorded.All())
+	assert.NotPanics(t, func() { config.LogRemovedSettings(nil, nil) })
+}

internal/config/override.go

@@ -21,6 +21,9 @@ type Override struct {
 	Usage string
 	// Default is the default value for the override
 	Default any
+	// Deprecated, when set, hides the flag and prints this message when it
+	// is used, while keeping it bound for a deprecation window.
+	Deprecated string
 }
 
 // NewOverride creates a new override
@@ -37,6 +40,11 @@ func NewOverride(field, usage string, def any) *Override {
 // Bind binds the override to the viper instance
 func (o *Override) Bind(flags *pflag.FlagSet) error {
 	flag := o.createFlag(flags)
+	if o.Deprecated != "" {
+		if err := flags.MarkDeprecated(o.Flag, o.Deprecated); err != nil {
+			return err
+		}
+	}
 	if err := viper.BindPFlag(o.Field, flag); err != nil {
 		return err
 	}
@@ -277,7 +285,15 @@ func DefaultOverrides() []*Override {
 		NewOverride("generator.filegen.cache-ttl", "file cache time-to-live (0 = never expire)", time.Duration(0)),
 		NewOverride("generator.okta.workers", "number of Okta generator workers", 1),
 		NewOverride("generator.okta.rate", "rate at which Okta logs are generated per worker", 1*time.Second),
-		NewOverride("generator.hostmetrics.workers", "number of host metrics generator workers", 1),
+		// Removed setting kept bound for the deprecation window; see LogRemovedSettings.
+		&Override{
+			Field:      "generator.hostmetrics.workers",
+			Flag:       createFlagName("generator.hostmetrics.workers"),
+			Env:        createEnvName("generator.hostmetrics.workers"),
+			Usage:      "removed: ignored, use --generator-hostmetrics-rate",
+			Default:    0,
+			Deprecated: HostMetricsWorkersRemoved,
+		},
 		NewOverride("generator.hostmetrics.rate", "scrape interval for host metrics generation", 1*time.Second),
 		NewOverride("generator.hostmetrics.os", "simulated operating system. One of: linux|windows", "linux"),
 		NewOverride("generator.hostmetrics.hostname", "simulated hostname (empty = random)", ""),