Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions docs/generator/hostmetrics.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
|-------------|-----------------------------------------------------------|
Expand All @@ -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

Expand Down
1 change: 1 addition & 0 deletions generator/hostmetrics/hostmetrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -273,5 +273,6 @@ func allScrapers() []Scraper {
&loadScraper{},
&pagingScraper{},
&processesScraper{},
&processScraper{},
}
}
4 changes: 2 additions & 2 deletions generator/hostmetrics/hostmetrics_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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) {
Expand Down
181 changes: 181 additions & 0 deletions generator/hostmetrics/process.go
Original file line number Diff line number Diff line change
@@ -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
}
122 changes: 122 additions & 0 deletions generator/hostmetrics/process_test.go
Original file line number Diff line number Diff line change
@@ -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")
})
}
2 changes: 1 addition & 1 deletion internal/config/generator_hostmetrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading