feat(generator): add per-process scraper to hostmetrics

docs/generator/hostmetrics.md

@@ -2,15 +2,15 @@
 
 **Class:** Producer (embed-eligible; see [docs/embed.md](../embed.md))
 
-The Host Metrics generator produces synthetic host-level metric data that mimics the OpenTelemetry Host Metrics receiver. It generates metrics for CPU, memory, disk, network, filesystem, load, paging, and processes.
+The Host Metrics generator produces synthetic host-level metric data that mimics the OpenTelemetry Host Metrics receiver. It generates metrics for CPU, memory, disk, network, filesystem, load, paging, processes, and per-process detail.
 
 ## Telemetry Type
 
 This generator produces **metrics** (not logs). It must be paired with an output that supports metrics, such as `otlp-grpc` or `stdout`.
 
 ## Scrapers
 
-The generator includes 8 scrapers, each producing metrics for a specific subsystem:
+The generator includes 9 scrapers, each producing metrics for a specific subsystem:
 
 | Scraper      | Metrics Produced                                           |
 |-------------|-----------------------------------------------------------|
@@ -22,6 +22,22 @@ The generator includes 8 scrapers, each producing metrics for a specific subsyst
 | `load`       | `system.cpu.load_average.1m`, `system.cpu.load_average.5m`, `system.cpu.load_average.15m` |
 | `paging`     | `system.paging.usage`, `system.paging.utilization`, `system.paging.operations`, `system.paging.faults` |
 | `processes`  | `system.processes.count`                                   |
+| `process`    | `process.memory.usage`, `process.memory.virtual`, `process.cpu.time`, `process.disk.io`, `process.threads`, `process.open_file_descriptors` |
+
+### The `process` scraper
+
+`processes` reports host-wide process *counts*; `process` reports metrics for
+*individual* processes. Each `process` record carries its own resource map with
+the process identity — `process.pid`, `process.parent_pid`,
+`process.executable.name`, `process.executable.path`, `process.command`,
+`process.command_line`, `process.command_args`, `process.owner`, and (Linux
+only) `process.cgroup`.
+
+Because process identity lives in resource attributes, this scraper is the one
+that produces genuinely high-cardinality output — useful for exercising
+reduction and normalization pipelines. The simulated process table varies by
+`generator.hostmetrics.os`, and always includes daemons whose resident memory
+sits under 1 MiB so memory-threshold filters have something to drop.
 
 ## Configuration
 

generator/hostmetrics/hostmetrics.go

@@ -273,5 +273,6 @@ func allScrapers() []Scraper {
 		&loadScraper{},
 		&pagingScraper{},
 		&processesScraper{},
+		&processScraper{},
 	}
 }

generator/hostmetrics/hostmetrics_test.go

@@ -59,7 +59,7 @@ func TestNew(t *testing.T) {
 		require.NoError(t, err)
 		assert.NotNil(t, g)
 		assert.Equal(t, "test-host", g.hostname)
-		assert.Len(t, g.scrapers, 8)
+		assert.Len(t, g.scrapers, 9)
 	})
 
 	t.Run("auto hostname linux", func(t *testing.T) {
@@ -221,7 +221,7 @@ func TestScrapers(t *testing.T) {
 func TestBuildScrapers(t *testing.T) {
 	t.Run("empty returns all", func(t *testing.T) {
 		scrapers := buildScrapers(nil)
-		assert.Len(t, scrapers, 8)
+		assert.Len(t, scrapers, 9)
 	})
 
 	t.Run("specific names", func(t *testing.T) {

generator/hostmetrics/process.go

@@ -0,0 +1,181 @@
+package hostmetrics
+
+import (
+	"fmt"
+	"math/rand"
+	"strings"
+	"time"
+
+	"github.com/observiq/blitz/output"
+)
+
+// processScraper produces per-process metrics equivalent to the OpenTelemetry
+// hostmetrics receiver's `process` scraper. Unlike the system.* scrapers, each
+// record carries its own resource map: the process identity (pid, executable,
+// owner, cgroup, command line) lives in resource attributes, not datapoint
+// attributes, which is what makes these metrics high cardinality and therefore
+// the interesting case for reduction pipelines.
+type processScraper struct{}
+
+func (s *processScraper) Name() string { return "process" }
+
+// processTemplate describes a single simulated process. Memory is expressed as
+// a range so a scrape produces variation across cycles while keeping each
+// process in a plausible band.
+type processTemplate struct {
+	executable  string
+	path        string
+	cgroup      string
+	owner       string
+	args        []string
+	minMemoryKB int64
+	maxMemoryKB int64
+}
+
+// Linux process table. systemd-journald and cron sit well under 1 MiB so a
+// scrape always contains processes that memory-threshold filters can drop.
+var linuxProcesses = []processTemplate{
+	{"sshd", "/usr/sbin/sshd", "/system.slice/ssh.service", "root", []string{"-D", "-oCiphers=aes256-gcm@openssh.com"}, 512, 900},
+	{"nginx", "/usr/sbin/nginx", "/system.slice/nginx.service", "www-data", []string{"-g", "daemon off;"}, 65536, 131072},
+	{"postgres", "/usr/lib/postgresql/16/bin/postgres", "/system.slice/postgresql.service", "postgres", []string{"-D", "/var/lib/postgresql/16/main"}, 262144, 786432},
+	{"dockerd", "/usr/bin/dockerd", "/system.slice/docker.service", "root", []string{"-H", "fd://", "--containerd=/run/containerd/containerd.sock"}, 131072, 262144},
+	{"systemd-journald", "/lib/systemd/systemd-journald", "/system.slice/systemd-journald.service", "root", []string{}, 256, 800},
+	{"cron", "/usr/sbin/cron", "/system.slice/cron.service", "root", []string{"-f"}, 128, 700},
+	{"rsyslogd", "/usr/sbin/rsyslogd", "/system.slice/rsyslog.service", "syslog", []string{"-n", "-iNONE"}, 2048, 8192},
+	{"prometheus-node-exporter", "/usr/bin/prometheus-node-exporter", "/system.slice/prometheus-node-exporter.service", "prometheus", []string{"--collector.systemd"}, 16384, 32768},
+}
+
+// Windows process table. Paths and owners follow Windows conventions; cgroup is
+// left empty because it has no Windows equivalent.
+var windowsProcesses = []processTemplate{
+	{"sqlservr.exe", `C:\Program Files\Microsoft SQL Server\MSSQL16.MSSQLSERVER\MSSQL\Binn\sqlservr.exe`, "", `NT SERVICE\MSSQLSERVER`, []string{"-s", "MSSQLSERVER"}, 524288, 2097152},
+	{"w3wp.exe", `C:\Windows\System32\inetsrv\w3wp.exe`, "", `IIS APPPOOL\DefaultAppPool`, []string{"-ap", "DefaultAppPool"}, 131072, 393216},
+	{"MsMpEng.exe", `C:\Program Files\Windows Defender\MsMpEng.exe`, "", "LocalSystem", []string{}, 98304, 262144},
+	{"spoolsv.exe", `C:\Windows\System32\spoolsv.exe`, "", "LocalSystem", []string{}, 512, 900},
+	{"svchost.exe", `C:\Windows\System32\svchost.exe`, "", `NT AUTHORITY\NETWORK SERVICE`, []string{"-k", "netsvcs", "-p"}, 8192, 40960},
+	{"wininit.exe", `C:\Windows\System32\wininit.exe`, "", "LocalSystem", []string{}, 256, 800},
+}
+
+func (s *processScraper) Scrape(r *rand.Rand, _ string, resource map[string]string) []output.MetricRecord {
+	now := time.Now()
+
+	templates := linuxProcesses
+	if resource["os.type"] == "windows" {
+		templates = windowsProcesses
+	}
+
+	var records []output.MetricRecord
+	for _, tmpl := range templates {
+		pid := int64(r.Intn(30000) + 100) // #nosec G404
+		res := processResource(resource, tmpl, pid)
+
+		memKB := tmpl.minMemoryKB
+		if span := tmpl.maxMemoryKB - tmpl.minMemoryKB; span > 0 {
+			memKB += r.Int63n(span) // #nosec G404
+		}
+		rss := memKB * 1024
+		// Virtual memory runs 2-4x resident for a typical daemon.
+		virtual := rss * int64(2+r.Intn(3)) // #nosec G404
+
+		records = append(records,
+			output.MetricRecord{
+				Name: "process.memory.usage", Description: "The amount of physical memory in use",
+				Unit: "By", Type: output.MetricTypeGauge,
+				IntValue: int64Ptr(rss),
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{},
+					Resource:   res,
+				},
+			},
+			output.MetricRecord{
+				Name: "process.memory.virtual", Description: "Virtual memory size",
+				Unit: "By", Type: output.MetricTypeGauge,
+				IntValue: int64Ptr(virtual),
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{},
+					Resource:   res,
+				},
+			},
+			output.MetricRecord{
+				Name: "process.threads", Description: "Process threads count",
+				Unit: "{thread}", Type: output.MetricTypeGauge,
+				IntValue: int64Ptr(int64(r.Intn(64) + 1)), // #nosec G404
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{},
+					Resource:   res,
+				},
+			},
+			output.MetricRecord{
+				Name: "process.open_file_descriptors", Description: "Number of file descriptors in use by the process",
+				Unit: "{count}", Type: output.MetricTypeGauge,
+				IntValue: int64Ptr(int64(r.Intn(512) + 3)), // #nosec G404
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{},
+					Resource:   res,
+				},
+			},
+		)
+
+		for _, state := range []string{"user", "system"} {
+			records = append(records, output.MetricRecord{
+				Name: "process.cpu.time", Description: "Total CPU seconds broken down by different states",
+				Unit: "s", Type: output.MetricTypeSum,
+				DoubleValue: float64Ptr(r.Float64() * 1000), // #nosec G404
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{"state": state},
+					Resource:   res,
+				},
+			})
+		}
+
+		for _, direction := range []string{"read", "write"} {
+			records = append(records, output.MetricRecord{
+				Name: "process.disk.io", Description: "Disk bytes transferred",
+				Unit: "By", Type: output.MetricTypeSum,
+				IntValue: int64Ptr(int64(r.Intn(1 << 30))), // #nosec G404
+				Metadata: output.MetricPointMetadata{
+					Timestamp:  now,
+					Attributes: map[string]string{"direction": direction},
+					Resource:   res,
+				},
+			})
+		}
+	}
+
+	return records
+}
+
+// processResource copies the shared host resource and layers the per-process
+// identity attributes on top, so every process gets its own resource map and
+// no scrape mutates the caller's map. process.cgroup is omitted when the
+// template has none (Windows).
+func processResource(base map[string]string, tmpl processTemplate, pid int64) map[string]string {
+	res := make(map[string]string, len(base)+8)
+	for k, v := range base {
+		res[k] = v
+	}
+
+	commandLine := tmpl.path
+	if len(tmpl.args) > 0 {
+		commandLine = tmpl.path + " " + strings.Join(tmpl.args, " ")
+	}
+
+	res["process.pid"] = fmt.Sprintf("%d", pid)
+	res["process.parent_pid"] = "1"
+	res["process.executable.name"] = tmpl.executable
+	res["process.executable.path"] = tmpl.path
+	res["process.command"] = tmpl.path
+	res["process.command_line"] = commandLine
+	res["process.command_args"] = strings.Join(tmpl.args, " ")
+	res["process.owner"] = tmpl.owner
+	if tmpl.cgroup != "" {
+		res["process.cgroup"] = tmpl.cgroup
+	}
+
+	return res
+}

generator/hostmetrics/process_test.go

@@ -0,0 +1,122 @@
+package hostmetrics
+
+import (
+	"math/rand"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProcessScraperName(t *testing.T) {
+	s := &processScraper{}
+	assert.Equal(t, "process", s.Name())
+}
+
+// Process identity must live in resource attributes, not datapoint attributes —
+// that placement is what reduction pipelines key off.
+func TestProcessScraperResourceAttributes(t *testing.T) {
+	s := &processScraper{}
+	r := rand.New(rand.NewSource(7)) // #nosec G404
+	base := map[string]string{"host.name": "test", "os.type": "linux"}
+
+	records := s.Scrape(r, "test-host", base)
+	require.NotEmpty(t, records)
+
+	for _, rec := range records {
+		res := rec.Metadata.Resource
+		assert.Equal(t, "test", res["host.name"], "base resource should be carried through")
+		for _, key := range []string{
+			"process.pid",
+			"process.parent_pid",
+			"process.executable.name",
+			"process.executable.path",
+			"process.command",
+			"process.command_line",
+			"process.owner",
+		} {
+			assert.NotEmpty(t, res[key], "%s: resource should carry %s", rec.Name, key)
+		}
+	}
+
+	// The shared base map must not be mutated by the scrape.
+	assert.Len(t, base, 2, "scrape must not write process attributes into the shared resource map")
+}
+
+// Every simulated process needs a distinct resource map; sharing one would
+// collapse the cardinality the scraper exists to produce.
+func TestProcessScraperDistinctResourcePerProcess(t *testing.T) {
+	s := &processScraper{}
+	r := rand.New(rand.NewSource(11)) // #nosec G404
+
+	records := s.Scrape(r, "test-host", map[string]string{"host.name": "test", "os.type": "linux"})
+	require.NotEmpty(t, records)
+
+	executables := map[string]struct{}{}
+	for _, rec := range records {
+		executables[rec.Metadata.Resource["process.executable.name"]] = struct{}{}
+	}
+	assert.Len(t, executables, len(linuxProcesses), "each template should appear as its own process")
+}
+
+// The blueprint use case filters processes under 1 MiB resident, so a scrape has
+// to contain some.
+func TestProcessScraperEmitsSubMiBProcesses(t *testing.T) {
+	s := &processScraper{}
+	r := rand.New(rand.NewSource(3)) // #nosec G404
+
+	records := s.Scrape(r, "test-host", map[string]string{"host.name": "test", "os.type": "linux"})
+
+	var small, large int
+	for _, rec := range records {
+		if rec.Name != "process.memory.usage" {
+			continue
+		}
+		require.NotNil(t, rec.IntValue)
+		if *rec.IntValue < 1048576 {
+			small++
+		} else {
+			large++
+		}
+	}
+	assert.Positive(t, small, "expected at least one process under 1 MiB resident")
+	assert.Positive(t, large, "expected at least one process over 1 MiB resident")
+}
+
+func TestProcessScraperWindows(t *testing.T) {
+	s := &processScraper{}
+	r := rand.New(rand.NewSource(19)) // #nosec G404
+
+	records := s.Scrape(r, "test-host", map[string]string{"host.name": "test", "os.type": "windows"})
+	require.NotEmpty(t, records)
+
+	for _, rec := range records {
+		res := rec.Metadata.Resource
+		assert.True(t, strings.HasSuffix(res["process.executable.name"], ".exe"),
+			"windows processes should use .exe names, got %q", res["process.executable.name"])
+		assert.NotContains(t, res, "process.cgroup", "cgroup has no Windows equivalent")
+	}
+}
+
+func TestProcessResourceCommandLine(t *testing.T) {
+	t.Run("with args", func(t *testing.T) {
+		tmpl := processTemplate{
+			executable: "sshd", path: "/usr/sbin/sshd", cgroup: "/system.slice/ssh.service",
+			owner: "root", args: []string{"-D", "-oPort=22"},
+		}
+		res := processResource(map[string]string{"host.name": "test"}, tmpl, 1842)
+		assert.Equal(t, "/usr/sbin/sshd -D -oPort=22", res["process.command_line"])
+		assert.Equal(t, "-D -oPort=22", res["process.command_args"])
+		assert.Equal(t, "1842", res["process.pid"])
+		assert.Equal(t, "/system.slice/ssh.service", res["process.cgroup"])
+	})
+
+	t.Run("without args", func(t *testing.T) {
+		tmpl := processTemplate{executable: "cron", path: "/usr/sbin/cron", owner: "root"}
+		res := processResource(map[string]string{}, tmpl, 42)
+		assert.Equal(t, "/usr/sbin/cron", res["process.command_line"])
+		assert.Empty(t, res["process.command_args"])
+		assert.NotContains(t, res, "process.cgroup")
+	})
+}

internal/config/generator_hostmetrics.go

@@ -36,7 +36,7 @@ type HostMetricsGeneratorConfig struct {
 
 // ValidScrapers is the list of valid scraper names.
 var ValidScrapers = []string{
-	"cpu", "memory", "disk", "network", "filesystem", "load", "paging", "processes",
+	"cpu", "memory", "disk", "network", "filesystem", "load", "paging", "processes", "process",
 }
 
 // Validate validates the host metrics generator configuration