feat(output): prometheus-scrape pull endpoint (PIPE-1488, BP-469)

cmd/blitz/build_output_test.go

@@ -7,6 +7,7 @@ import (
 	"go/ast"
 	"go/parser"
 	"go/token"
+	"net"
 	"path/filepath"
 	"strconv"
 	"testing"
@@ -33,6 +34,7 @@ func defaultOutput(t *testing.T, typ config.OutputType) config.Output {
 		config.OutputTypeHEC:                   "  hec:\n    host: 127.0.0.1\n    port: 9\n    token: t\n",
 		config.OutputTypeFlow:                  "  flow:\n    host: 127.0.0.1\n    port: 9\n",
 		config.OutputTypePrometheusRemoteWrite: "  prometheus-remote-write:\n    endpoint: http://127.0.0.1:9/api/v1/write\n",
+		config.OutputTypePrometheusScrape:      "  prometheus-scrape:\n    listenAddress: 127.0.0.1:0\n",
 	}
 	cfg, err := blitzconfig.Load([]byte(fmt.Sprintf("output:\n  type: %s\n%s", typ, required[typ])), blitzconfig.LoadOpts{})
 	require.NoError(t, err)
@@ -75,6 +77,10 @@ func outputTypes(t *testing.T) []config.OutputType {
 // with TLS enabled, so every case's options and TLS conversion run.
 func TestBuildOutput_EveryType(t *testing.T) {
 	missingCA := config.TLS{CertificateAuthority: []string{filepath.Join(t.TempDir(), "missing-ca.pem")}}
+	held, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = held.Close() })
+	busy := held.Addr().String()
 	cases := []struct {
 		name    string
 		typ     config.OutputType
@@ -114,6 +120,10 @@ func TestBuildOutput_EveryType(t *testing.T) {
 		{name: "prometheus-remote-write bad endpoint", typ: config.OutputTypePrometheusRemoteWrite, mutate: func(o *config.Output) {
 			o.PrometheusRemoteWrite.Endpoint = "ftp://127.0.0.1/write"
 		}, wantErr: "endpoint"},
+		{name: "prometheus-scrape", typ: config.OutputTypePrometheusScrape},
+		{name: "prometheus-scrape busy address", typ: config.OutputTypePrometheusScrape, mutate: func(o *config.Output) {
+			o.PrometheusScrape.ListenAddress = busy
+		}, wantErr: "listen"},
 	}
 	covered := map[config.OutputType]bool{}
 	for _, tc := range cases {
@@ -137,7 +147,7 @@ func TestBuildOutput_EveryType(t *testing.T) {
 			require.NoError(t, out.Stop(context.Background()))
 		})
 	}
-	_, err := buildOutput(zap.NewNop(), config.Output{Type: "bogus"}, embed.NopTelemetry())
+	_, err = buildOutput(zap.NewNop(), config.Output{Type: "bogus"}, embed.NopTelemetry())
 	require.ErrorContains(t, err, "invalid output type")
 }
 
@@ -146,7 +156,7 @@ func TestBuildOutput_EveryType(t *testing.T) {
 // The Prometheus outputs keep running without their self-metrics when those
 // cannot be built, so they are the exception (their own tests pin that).
 func TestBuildOutput_ConstructorErrors(t *testing.T) {
-	bestEffortMetrics := map[config.OutputType]bool{config.OutputTypePrometheusRemoteWrite: true}
+	bestEffortMetrics := map[config.OutputType]bool{config.OutputTypePrometheusRemoteWrite: true, config.OutputTypePrometheusScrape: true}
 	tel := embed.TelemetrySettings{MeterProvider: failingMeterProvider{}}
 	for _, typ := range outputTypes(t) {
 		out, err := buildOutput(zap.NewNop(), defaultOutput(t, typ), tel)

cmd/blitz/main.go

@@ -37,6 +37,7 @@ import (
 	"github.com/observiq/blitz/output/nop"
 	otlpgrpc "github.com/observiq/blitz/output/otlp_grpc"
 	promrw "github.com/observiq/blitz/output/promrw"
+	"github.com/observiq/blitz/output/promscrape"
 	stdoutout "github.com/observiq/blitz/output/stdout"
 	syslogout "github.com/observiq/blitz/output/syslog"
 	"github.com/observiq/blitz/output/tcp"
@@ -612,6 +613,13 @@ func buildOutput(logger *zap.Logger, out config.Output, tel embed.TelemetrySetti
 			logger.Error("Failed to create prometheus-remote-write output", zap.Error(err))
 			return nil, err
 		}
+	case config.OutputTypePrometheusScrape:
+		ps := out.PrometheusScrape
+		inst, err = promscrape.New(ps.ListenAddress, ps.MetricsPath, ps.EmitTimestamps, ps.MetricExpiration, tel, logger)
+		if err != nil {
+			logger.Error("Failed to create prometheus-scrape output", zap.Error(err))
+			return nil, err
+		}
 	default:
 		logger.Error("Invalid output type", zap.String("type", string(out.Type)))
 		return nil, fmt.Errorf("invalid output type: %s", out.Type)

cmd/blitz/prometheus_outputs_test.go

@@ -40,3 +40,10 @@ func TestRunWritesToPrometheusRemoteWrite(t *testing.T) {
 	require.NoError(t, runCountedHostMetrics(t, fmt.Sprintf("output:\n  type: prometheus-remote-write\n  prometheus-remote-write:\n    endpoint: %s\n", srv.URL)))
 	require.NotZero(t, posts.Load(), "stopping the output flushes the counted points")
 }
+
+func TestRunReportsAPrometheusScrapeListenError(t *testing.T) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = l.Close() })
+	require.ErrorContains(t, runCountedHostMetrics(t, fmt.Sprintf("output:\n  type: prometheus-scrape\n  prometheus-scrape:\n    listenAddress: %s\n", l.Addr())), l.Addr().String())
+}

cmd/blitz/prometheus_scrape_unix_test.go

@@ -0,0 +1,83 @@
+//go:build !windows
+
+package main
+
+// The scrape test stops run by sending the process SIGTERM, and syscall.Kill
+// does not exist on Windows, so it builds on the other platforms only.
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestRunServesPrometheusScrape scrapes a running CLI at its configured
+// address and metrics path, so the output block must reach the endpoint.
+func TestRunServesPrometheusScrape(t *testing.T) {
+	addrs := freeAddrs(t, 2)
+	addr, metricsPort := addrs[0], addrs[1].(*net.TCPAddr).Port
+	yaml := fmt.Sprintf("generator:\n  type: hostmetrics\n  hostmetrics:\n    rate: 10ms\n"+
+		"output:\n  type: prometheus-scrape\n  prometheus-scrape:\n    listenAddress: %s\n    metricsPath: /custom\n"+
+		"metrics:\n  port: %d\n", addr, metricsPort)
+	path := filepath.Join(t.TempDir(), "blitz.yaml")
+	require.NoError(t, os.WriteFile(path, []byte(yaml), 0o600))
+	origFile := configFile
+	t.Cleanup(func() { configFile = origFile })
+	configFile = path
+
+	done := make(chan error, 1)
+	go func() { done <- run(nil, nil) }()
+	var body string
+	var exited bool
+	var runErr error
+	require.Eventually(t, func() bool {
+		select {
+		case runErr = <-done:
+			exited = true
+			return true
+		default:
+		}
+		resp, err := http.Get("http://" + addr.String() + "/custom")
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		b, _ := io.ReadAll(resp.Body)
+		body = string(b)
+		return resp.StatusCode == http.StatusOK && strings.Contains(body, "system_")
+	}, 10*time.Second, 20*time.Millisecond, "scrape the configured address and path")
+	require.False(t, exited, "run exited before serving the scrape: %v", runErr)
+
+	// run stops on SIGTERM, which its own handler catches.
+	require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGTERM))
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("run did not stop on SIGTERM")
+	}
+}
+
+// freeAddrs returns n distinct loopback addresses with ports free at call
+// time. Every listener stays open until all n are picked, so the kernel cannot
+// hand the same port out twice.
+func freeAddrs(t *testing.T, n int) []net.Addr {
+	t.Helper()
+	addrs := make([]net.Addr, 0, n)
+	for range n {
+		l, err := net.Listen("tcp", "127.0.0.1:0")
+		require.NoError(t, err)
+		defer func() { _ = l.Close() }()
+		addrs = append(addrs, l.Addr())
+	}
+	return addrs
+}

docker/README.md

@@ -65,6 +65,7 @@ docker compose -f docker/docker-compose.telemetry-generator.yml up
 | `BLITZ_PII_WORKERS` | `1` | Number of workers for PII generator |
 | `PROM_RW_ENDPOINT` | `http://bdot-collector:19291/api/v1/write` | Remote-write receiver URL for `blitz-hostmetrics-promrw` |
 | `PROM_RW_VERSION` | `1.0` | Remote-write protocol version (`1.0` or `2.0`) for `blitz-hostmetrics-promrw` |
+| `PROM_SCRAPE_PORT` | `9464` | Host port that publishes `blitz-hostmetrics-promscrape`'s `/metrics` endpoint |
 
 ### Examples
 
@@ -110,6 +111,7 @@ docker compose -f docker/docker-compose.telemetry-generator.yml up -d
 | `blitz-hostmetrics-linux` | Host Metrics | Synthetic host metrics (CPU, memory, disk, etc.) for Linux |
 | `blitz-hostmetrics-windows` | Host Metrics | Synthetic host metrics (CPU, memory, disk, etc.) for Windows |
 | `blitz-hostmetrics-promrw` | Host Metrics | Synthetic host metrics sent via the Prometheus remote-write output (needs a receiver at `PROM_RW_ENDPOINT`; starts only with `--profile prometheus-remote-write`) |
+| `blitz-hostmetrics-promscrape` | Host Metrics | Synthetic host metrics served on a Prometheus scrape (pull) endpoint at host port `PROM_SCRAPE_PORT`; starts only with `--profile prometheus-scrape` |
 | `blitz-traces` | Traces | Synthetic distributed traces (HTTP + DB spans) |
 | `blitz-fix` | FIX | FIX protocol messages (4.2 / 4.4 / 5.0 SP2) across 10 asset categories — NewOrderSingle, ExecutionReport, cancel/replace/status |
 | `blitz-flow` | Network Flow | Synthetic NetFlow/IPFIX/sFlow records exported over UDP |

docker/docker-compose.telemetry-generator.yml

@@ -19,6 +19,8 @@
 #   - PROM_RW_ENDPOINT: remote-write receiver URL for blitz-hostmetrics-promrw
 #     (default: http://bdot-collector:19291/api/v1/write)
 #   - PROM_RW_VERSION: remote-write protocol version, 1.0 or 2.0 (default: 1.0)
+#   - PROM_SCRAPE_PORT: host port for blitz-hostmetrics-promscrape's /metrics
+#     (default: 9464)
 
 x-blitz-common: &blitz-common
   image: ghcr.io/observiq/blitz:latest
@@ -262,6 +264,22 @@ services:
       BLITZ_OUTPUT_PROMETHEUS_REMOTE_WRITE_ENDPOINT: ${PROM_RW_ENDPOINT:-http://bdot-collector:19291/api/v1/write}
       BLITZ_OUTPUT_PROMETHEUS_REMOTE_WRITE_VERSION: ${PROM_RW_VERSION:-1.0}
 
+  # Host metrics served on a Prometheus scrape (pull) endpoint. Point a scraper
+  # (the collector prometheusreceiver, or a Prometheus server) at
+  # blitz-hostmetrics-promscrape:9464/metrics to pull the exposition. Starts
+  # only with --profile prometheus-scrape, since it publishes a host port.
+  blitz-hostmetrics-promscrape:
+    <<: *blitz-common
+    profiles: ["prometheus-scrape"]
+    environment:
+      BLITZ_GENERATOR_TYPE: hostmetrics
+      BLITZ_GENERATOR_HOSTMETRICS_RATE: ${BLITZ_RATE:-1s}
+      BLITZ_GENERATOR_HOSTMETRICS_OS: linux
+      BLITZ_OUTPUT_TYPE: prometheus-scrape
+      BLITZ_OUTPUT_PROMETHEUS_SCRAPE_LISTENADDRESS: 0.0.0.0:9464
+    ports:
+      - "${PROM_SCRAPE_PORT:-9464}:9464"
+
 networks:
   telemetry-net:
     driver: bridge

docs/configuration.md

@@ -197,7 +197,7 @@ For detailed configuration options for each generator, see the individual genera
 
 | YAML Path | Flag Name | Environment Variable | Default | Description |
 |-----------|-----------|---------------------|---------|-------------|
-| `output.type` | `--output-type` | `BLITZ_OUTPUT_TYPE` | `nop` | Output type. Valid values: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file`, `hec`, `flow`, `prometheus-remote-write` |
+| `output.type` | `--output-type` | `BLITZ_OUTPUT_TYPE` | `nop` | Output type. Valid values: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file`, `hec`, `flow`, `prometheus-remote-write`, `prometheus-scrape` |
 
 For detailed configuration options for each output, see the individual output documentation:
 
@@ -211,6 +211,7 @@ For detailed configuration options for each output, see the individual output do
 - [HEC Output](https://github.com/observiq/blitz/blob/main/docs/output/hec.md)
 - [Flow Output](https://github.com/observiq/blitz/blob/main/docs/output/flow.md)
 - [Prometheus Remote-Write Output](https://github.com/observiq/blitz/blob/main/docs/output/prometheus-remote-write.md)
+- [Prometheus Scrape Output](https://github.com/observiq/blitz/blob/main/docs/output/prometheus-scrape.md)
 
 #### Multiple Outputs (Fan-Out)
 
@@ -476,7 +477,7 @@ Duration values (like `generator.json.rate`, `generator.winevt.rate`, or `genera
 - `logging.type` - Must be one of: `stdout`, `file`
 - `logging.file.path` - Required when `logging.type` is `file`
 - `generator.type` - Must be one of: `nop`, `json`, `winevt`, `palo-alto`, `apache-common`, `apache-combined`, `apache-error`, `nginx`, `postgres`, `kubernetes`, `fix`, `flow`
-- `output.type` - Must be one of: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file`, `hec`, `flow`, `prometheus-remote-write`
+- `output.type` - Must be one of: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file`, `hec`, `flow`, `prometheus-remote-write`, `prometheus-scrape`
 
 ## Error Handling
 

docs/output/prometheus-scrape.md

@@ -0,0 +1,64 @@
+# Prometheus Scrape Output
+
+The prometheus-scrape output is a metrics-only pull endpoint. It hosts an HTTP `/metrics` endpoint in the Prometheus exposition format, so a Prometheus server or the collector `prometheusreceiver` can scrape blitz directly. Unlike the push-based [remote-write output](prometheus-remote-write.md), the scraper controls timing: blitz holds the current value of each series and serves a snapshot on every scrape.
+
+## Metric Mapping
+
+The endpoint serves what the OpenTelemetry Collector's `prometheusexporter` (v0.161.0, default settings) would serve for the same points, built with the same Prometheus client library:
+
+- Gauge and non-monotonic Sum become a gauge. Counter becomes a counter. Histogram becomes cumulative `_bucket` series with `le` labels, including `+Inf` from the point's count, plus `_sum` and `_count`. A histogram needs one more bucket count than bounds; `WriteMetric` rejects any other point.
+- Names follow the exporter's default naming with suffixes: the unit becomes a suffix (`s` to `_seconds`, `By` to `_bytes`), counters end in `_total`, and gauges with unit `1` end in `_ratio`, so `system.memory.usage` (unit `By`) becomes `system_memory_usage_bytes`.
+- Attribute names are sanitized to the Prometheus grammar, and empty values are kept. A point with two attributes that sanitize to one label name is not served, and the conversion error is logged.
+- Every series carries `otel_scope_name`, `otel_scope_version`, and `otel_scope_schema_url`, empty because blitz sends an empty instrumentation scope.
+- Every metric has a `# HELP` line, empty when the point has no description. The first description and type seen for a name are kept; a later point of another type under the same name is not served.
+
+The endpoint keeps the latest point per series (name, type, attributes, and resource). A point older than the stored one is ignored. A series not updated within `metricExpiration` is dropped, so a host that stops reporting stops being exposed.
+
+`job` comes from `service.namespace`/`service.name` and `instance` from `service.instance.id`, as in the exporter. When `service.name` is absent, `job` falls back to `telemetry.source`, and when `service.instance.id` is absent, `instance` falls back to `host.name`, which blitz generators set instead; this is the one difference from the exporter, and it keeps each simulated host's series distinct. As in the exporter, a present but empty `service.name` or `service.instance.id` gives an empty label with no fallback. A label is left off when its source is missing. Each target with both a `job` and an `instance` gets one `target_info` series (`# HELP target_info Target metadata`) carrying its other resource attributes; keys that sanitize to one label are joined with `;`. As in the exporter, a resource key that is no valid label name drops every `target_info` from that scrape.
+
+By default no explicit timestamp is written, and the scraper stamps each sample at scrape time. Set `emitTimestamps: true` to append each point's millisecond timestamp, as the exporter's `send_timestamps` does. The response format and compression follow the scraper's `Accept` and `Accept-Encoding` headers.
+
+## Configuration
+
+| YAML Path                                   | Flag                                          | Environment Variable                              | Default        | Description                                                               |
+|---------------------------------------------|-----------------------------------------------|---------------------------------------------------|----------------|---------------------------------------------------------------------------|
+| `output.type`                               | `--output-type`                               | `BLITZ_OUTPUT_TYPE`                               | `nop`          | Set to `prometheus-scrape` to use this output.                            |
+| `output.prometheus-scrape.listenAddress`    | `--output-prometheus-scrape-listenaddress`    | `BLITZ_OUTPUT_PROMETHEUS_SCRAPE_LISTENADDRESS`    | `0.0.0.0:9464` | Host:port the metrics endpoint binds to.                                  |
+| `output.prometheus-scrape.metricsPath`      | `--output-prometheus-scrape-metricspath`      | `BLITZ_OUTPUT_PROMETHEUS_SCRAPE_METRICSPATH`      | `/metrics`     | URL path the exposition is served on.                                     |
+| `output.prometheus-scrape.emitTimestamps`   | `--output-prometheus-scrape-emittimestamps`   | `BLITZ_OUTPUT_PROMETHEUS_SCRAPE_EMITTIMESTAMPS`   | `false`        | Append per-sample millisecond timestamps.                                 |
+| `output.prometheus-scrape.metricExpiration` | `--output-prometheus-scrape-metricexpiration` | `BLITZ_OUTPUT_PROMETHEUS_SCRAPE_METRICEXPIRATION` | `5m`           | Drop a series not updated within this duration. `0` keeps series forever. |
+
+The endpoint has no authentication or TLS. By default it listens on every
+interface, since Prometheus usually scrapes from another host. Set
+`listenAddress` to a loopback address such as `127.0.0.1:9464` to keep it
+local.
+
+## Example Configuration
+
+```yaml
+output:
+  type: prometheus-scrape
+  prometheus-scrape:
+    listenAddress: 0.0.0.0:9464
+    metricsPath: /metrics
+```
+
+Point a scraper at it, for example a Prometheus scrape config:
+
+```yaml
+scrape_configs:
+  - job_name: blitz
+    static_configs:
+      - targets: ["blitz-host:9464"]
+```
+
+## Self-Telemetry
+
+blitz records these instruments about the output's own operation and exports them through its telemetry pipeline. They are not the metrics the output exposes to scrapers.
+
+- **`blitz.output.prometheus_scrape.scrapes`** (Counter): scrape requests served.
+- **`blitz.output.prometheus_scrape.series_exposed`** (Gauge): series exposed at the last scrape, each histogram bucket, `_sum`, and `_count` counted (not counting `target_info`).
+- **`blitz.output.prometheus_scrape.exposition_bytes`** (Histogram): bytes sent per scrape, compressed when the scraper accepts gzip.
+- **`blitz.output.prometheus_scrape.scrape_latency`** (Histogram, ms): latency of serving a scrape request.
+
+The shared `blitz.output.entries_received` counter also records each point the output receives, tagged `output_type=prometheus-scrape`.

internal/config/output.go

@@ -28,6 +28,8 @@ const (
 	OutputTypeFlow OutputType = "flow"
 	// OutputTypePrometheusRemoteWrite represents Prometheus remote-write output
 	OutputTypePrometheusRemoteWrite OutputType = "prometheus-remote-write"
+	// OutputTypePrometheusScrape represents Prometheus scrape (pull) output
+	OutputTypePrometheusScrape OutputType = "prometheus-scrape"
 )
 
 // Output contains configuration for output destinations
@@ -52,6 +54,8 @@ type Output struct {
 	Flow FlowOutputConfig `yaml:"flow,omitempty" mapstructure:"flow,omitempty"`
 	// PrometheusRemoteWrite contains Prometheus remote-write output configuration
 	PrometheusRemoteWrite PrometheusRemoteWriteOutputConfig `yaml:"prometheus-remote-write,omitempty" mapstructure:"prometheus-remote-write,omitempty"`
+	// PrometheusScrape contains Prometheus scrape output configuration
+	PrometheusScrape PrometheusScrapeOutputConfig `yaml:"prometheus-scrape,omitempty" mapstructure:"prometheus-scrape,omitempty"`
 }
 
 // Validate validates the output configuration
@@ -100,8 +104,12 @@ func (o *Output) Validate() error {
 		if err := o.PrometheusRemoteWrite.Validate(); err != nil {
 			return fmt.Errorf("prometheus-remote-write output validation failed: %w", err)
 		}
+	case OutputTypePrometheusScrape:
+		if err := o.PrometheusScrape.Validate(); err != nil {
+			return fmt.Errorf("prometheus-scrape output validation failed: %w", err)
+		}
 	default:
-		return fmt.Errorf("invalid output type: %s, must be one of: nop, stdout, tcp, udp, syslog, otlp-grpc, file, hec, flow, prometheus-remote-write", o.Type)
+		return fmt.Errorf("invalid output type: %s, must be one of: nop, stdout, tcp, udp, syslog, otlp-grpc, file, hec, flow, prometheus-remote-write, prometheus-scrape", o.Type)
 	}
 
 	return nil