From e96c48122425553ce5f7df5c905252ef66917d89 Mon Sep 17 00:00:00 2001 From: Chelsea Wright Date: Thu, 13 Aug 2026 14:34:01 +0200 Subject: [PATCH] feat(generator): add per-process scraper to hostmetrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The existing `processes` scraper reports host-wide counts (system.processes.count) only, so nothing in blitz emitted per-process telemetry. The new `process` scraper emits process.memory.usage, process.memory.virtual, process.cpu.time, process.disk.io, process.threads and process.open_file_descriptors. Each record carries its own resource map with the process identity: pid, parent_pid, executable name and path, command, command_line, command_args, owner and — on Linux — cgroup. That resource-level identity is what makes process metrics high cardinality, and therefore what reduction and normalization pipelines act on; placing it in datapoint attributes instead would not exercise the same code paths downstream. The simulated process table varies by the configured OS (Linux daemons vs Windows services), and always contains daemons under 1 MiB resident so memory-threshold filters have something to drop. --- docs/generator/hostmetrics.md | 20 ++- generator/hostmetrics/hostmetrics.go | 1 + generator/hostmetrics/hostmetrics_test.go | 4 +- generator/hostmetrics/process.go | 181 ++++++++++++++++++++++ generator/hostmetrics/process_test.go | 122 +++++++++++++++ internal/config/generator_hostmetrics.go | 2 +- 6 files changed, 325 insertions(+), 5 deletions(-) create mode 100644 generator/hostmetrics/process.go create mode 100644 generator/hostmetrics/process_test.go diff --git a/docs/generator/hostmetrics.md b/docs/generator/hostmetrics.md index 552b343..2224586 100644 --- a/docs/generator/hostmetrics.md +++ b/docs/generator/hostmetrics.md @@ -2,7 +2,7 @@ **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 @@ -10,7 +10,7 @@ This generator produces **metrics** (not logs). It must be paired with an output ## 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 diff --git a/generator/hostmetrics/hostmetrics.go b/generator/hostmetrics/hostmetrics.go index 0cd072c..61fb51e 100644 --- a/generator/hostmetrics/hostmetrics.go +++ b/generator/hostmetrics/hostmetrics.go @@ -273,5 +273,6 @@ func allScrapers() []Scraper { &loadScraper{}, &pagingScraper{}, &processesScraper{}, + &processScraper{}, } } diff --git a/generator/hostmetrics/hostmetrics_test.go b/generator/hostmetrics/hostmetrics_test.go index b0ca4a3..f7e1943 100644 --- a/generator/hostmetrics/hostmetrics_test.go +++ b/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) { diff --git a/generator/hostmetrics/process.go b/generator/hostmetrics/process.go new file mode 100644 index 0000000..89e77eb --- /dev/null +++ b/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 +} diff --git a/generator/hostmetrics/process_test.go b/generator/hostmetrics/process_test.go new file mode 100644 index 0000000..77ff4d0 --- /dev/null +++ b/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") + }) +} diff --git a/internal/config/generator_hostmetrics.go b/internal/config/generator_hostmetrics.go index 5f3f2bc..20198f4 100644 --- a/internal/config/generator_hostmetrics.go +++ b/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