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