diff --git a/cmd/blitz/main.go b/cmd/blitz/main.go index ad5de87..85a352b 100644 --- a/cmd/blitz/main.go +++ b/cmd/blitz/main.go @@ -20,6 +20,7 @@ import ( "github.com/observiq/blitz/generator/winevt" "github.com/observiq/blitz/internal/build" "github.com/observiq/blitz/internal/config" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/internal/dispatch" "github.com/observiq/blitz/internal/logging" "github.com/observiq/blitz/internal/service" @@ -310,13 +311,22 @@ func run(cmd *cobra.Command, args []string) error { return fmt.Errorf("invalid output type: %s", cfg.Output.Type) } + // Build the simulated identity environment once, up front, so every + // generator resolves its host identity from the same fleet (PIPE-1036). A + // live-reconfigure path (deferred) would rebuild this and swap it in. + env, err := cfg.Environment.Build(logger) + if err != nil { + logger.Error("Failed to build simulated environment", zap.Error(err)) + return err + } + // Configure generators effectiveGens := cfg.EffectiveGenerators() var generators []any var tracker *count.Tracker for _, genCfg := range effectiveGens { - gen, genErr := createGenerator(logger, genCfg, outputInstance) + gen, genErr := createGenerator(logger, genCfg, outputInstance, env) if genErr != nil { logger.Error("Failed to create generator", zap.String("type", string(genCfg.Type)), @@ -395,7 +405,7 @@ shutdown: return nil } -func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Output) (any, error) { +func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Output, env *datagen.Environment) (any, error) { // Standalone-CLI-only generator types that dispatch.ForEmbed does not // construct (winevt is deprecated for embed; nop yields no records). // All other generators delegate to dispatch.ForEmbed so the @@ -423,7 +433,7 @@ func createGenerator(logger *zap.Logger, genCfg config.Generator, out output.Out if tw, ok := out.(output.TraceWriter); ok { consumers.TraceConsumer = output.WriterAsTraceConsumer(tw) } - mod, err := dispatch.ForEmbed(logger, genCfg, consumers, nil) + mod, err := dispatch.ForEmbed(logger, genCfg, consumers, nil, env) if err != nil { return nil, err } diff --git a/config/loader.go b/config/loader.go index 96d35ea..067614f 100644 --- a/config/loader.go +++ b/config/loader.go @@ -149,6 +149,14 @@ func LoadModules(yamlBytes []byte, opts EmbedOpts) ([]embed.ProducerModule, erro return nil, err } gens := cfg.EffectiveGenerators() + + // Resolve the simulated identity environment once for the whole config, so + // every generator draws its host identity from the same fleet (PIPE-1036). + env, err := cfg.Environment.Build(logger) + if err != nil { + return nil, fmt.Errorf("build environment: %w", err) + } + consumers := dispatch.EmbedConsumers{ LogConsumer: opts.LogConsumer, MetricConsumer: opts.MetricConsumer, @@ -156,7 +164,7 @@ func LoadModules(yamlBytes []byte, opts EmbedOpts) ([]embed.ProducerModule, erro } modules := make([]embed.ProducerModule, 0, len(gens)) for i, gen := range gens { - mod, err := dispatch.ForEmbed(logger, gen, consumers, opts.FileGenLibrary) + mod, err := dispatch.ForEmbed(logger, gen, consumers, opts.FileGenLibrary, env) if err != nil { return nil, fmt.Errorf("generator[%d] type=%q: %w", i, gen.Type, err) } diff --git a/docs/embed.md b/docs/embed.md index 1230d32..3c4e796 100644 --- a/docs/embed.md +++ b/docs/embed.md @@ -162,9 +162,14 @@ Consumer errors are best-effort: blitz logs the error, increments a `consumer_er Every blitz record carries a per-record `Metadata.Resource` map describing the entity that emitted it (host, module, format, version). The three signal types follow a parallel shape: -- `LogRecord.Metadata.Resource` — `map[string]string` -- `MetricPoint.Metadata.Resource` — `map[string]string` -- `Span.Metadata.Resource` — `map[string]string` +- `LogRecord.Metadata.Resource` — `map[string]any` +- `MetricPoint.Metadata.Resource` — `map[string]any` +- `Span.Metadata.Resource` — `map[string]any` + +Resource values are `any` so an attribute can be a scalar (`host.name`) or a list +(`host.ip`, `host.mac` are `[]string`, serialized as OTLP array values). Only the +OTLP metrics builder and the stdout JSON output serialize Resource today; other +outputs drop it. Similarly for `Metadata.Attributes`: @@ -188,6 +193,13 @@ Every shipped Producer populates at least: - `host.name` — the hostname the record semantically describes (defaults to `os.Hostname()`, falls back to `blitz`). - `telemetry.source` — the module identifier (`apache`, `nginx`, `paloalto`, `fix`, `wel`, …). +When a simulated identity environment is configured, every Producer additionally +projects the resolved host's identity onto each record: `host.id`, `host.arch`, +the `os.*` set (`os.type`, `os.name`, `os.version`, `os.build_id`, +`os.description`), `host.ip`/`host.mac` (from the host's interfaces), and +`deployment.environment.name`. See [environment.md](environment.md) for how the +environment is configured and how each generator is mapped to a simulated host. + Some Producers populate additional dimensions: | Source | Extra Resource keys | @@ -197,7 +209,7 @@ Some Producers populate additional dimensions: | `kubernetes` | `kubernetes.format` — currently `cri-o` (only supported format today) | | `filegen` | `filegen.source` — the file / package / glob the line came from | | `wel` | `wel.channel`, `wel.computer`, `wel.domain`, `wel.role` | -| `fix` (when it lands) | `fix.version` — `FIX.4.2` / `FIX.4.4` / `FIX.5.0SP2` | +| `fix` | `fix.version` — `FIX.4.2` / `FIX.4.4` / `FIX.5.0SP2` | Generators MUST NOT carry secret or per-deployment-specific values they don't already legitimately know — that remains the host's concern via `embed.Host.Resource`. @@ -220,6 +232,24 @@ Metadata: embed.LogRecordMetadata{ `resource.Default(source, extras...)` returns a fresh map per call (so consumers can mutate safely without affecting subsequent emissions) and memoizes `os.Hostname()` once per process. +To describe a **simulated** host rather than the running process, build the +resource once at worker construction from a resolved `datagen.SystemIdentity`: + +```go +// At construction — projects host.* / os.* / deployment.* from the identity, +// stamps telemetry.source, and appends any per-generator constants: +g.static = resource.FromIdentity(identity, "my-module", "my-module.format", formatName) + +// Per record — the shared static set, plus any per-record dynamic pairs: +Resource: g.static.Record("my-module.channel", channel) +``` + +`FromIdentity(nil, source, extras...)` falls back to the process hostname, so a +generator has one uniform construction path whether or not an environment is +wired. `Record()` with no dynamic pairs returns a shared, read-only map +(zero-allocation) that callers MUST NOT mutate; with dynamic pairs it returns a +fresh merged map. See [environment.md](environment.md). + ## Configuration Two supported paths: diff --git a/docs/environment.md b/docs/environment.md new file mode 100644 index 0000000..202efc7 --- /dev/null +++ b/docs/environment.md @@ -0,0 +1,123 @@ +# Simulated Identity Environment + +Blitz can generate telemetry that describes a coherent, simulated fleet of hosts +rather than the single machine the process runs on. A `datagen.Environment` is a +cross-referenced graph of identities — a domain, networks, users, groups, +systems, and storage/network appliances — built deterministically from seeds. +When an environment is configured, every generator resolves a simulated host +from it and stamps that host's identity onto each emitted record. + +This page covers the user-facing surface: how to configure the environment, how +generators are mapped to simulated hosts, and which resource attributes get +projected. For the low-level identity model (the identity hierarchy, seed +contract, hostname pools, appliance taxonomy), see +[datagen.md](datagen.md). + +## Configuration + +The environment is configured under the top-level `environment:` key. The whole +block is optional — an omitted `environment` yields a randomized default +environment, so records still carry a coherent simulated host without any +configuration. + +```yaml +environment: + # AD/DNS domain for the fleet. Default: blitz.local + domain_name: corp.example.com + + # Per-identity-type determinism. Each field is optional; an omitted field + # randomizes that identity type, while an explicit value — including 0 — is a + # deterministic seed. Changing one seed only re-randomizes that slice of the + # output. + seed_config: + shared: 42 # base seed mixed into every type unless that type is set + systems: 100 # host identities (OS, hostname, interfaces, specs) + users: 101 + groups: 102 + services: 103 + applications: 104 + networks: 105 + domains: 106 + storage_systems: 107 + network_systems: 108 + + # How many of each identity type to generate. A zero (omitted) count uses the + # datagen default shown below. + counts: + systems: 20 # machines (default 20) + users: 50 # default 50 + groups: 10 # default 10 + networks: 4 # subnets (default 4) + storage_systems: 2 # storage arrays (default 2) + network_systems: 4 # network devices (default 4) + domain_admins: 0 # exact Domain Admins membership; 0 = user-count-scaled default +``` + +Counts must not be negative; the config is rejected at load time otherwise. + +## How generators map to hosts + +Each generator resolves exactly one host from the environment, keyed by the +generator's component name (`hostmetrics`, `traces`, `apache`, `nginx`, `wel`, +…). The mapping is deterministic — the same component always resolves to the +same host for a given environment — so a component attributes every record it +emits to one consistent machine, and distinct components spread across the +fleet. + +When no environment is available, generators fall back to the running process's +`os.Hostname()` (or `blitz` if that fails), exactly as before — so nothing about +the default output shape changes when the environment is absent. + +> Finer per-worker granularity — one distinct host per worker within a single +> generator — is a planned opt-in. Today the granularity is one host per +> generator component. + +## Projected resource attributes + +With an environment configured, every record's `Metadata.Resource` carries the +resolved host's identity, following OpenTelemetry semantic conventions: + +| Attribute | Source | +|-------------------------------|------------------------------------------------------------| +| `host.name` | the system's hostname | +| `host.id` | OS-appropriate machine id (machine-id / GUID / UUID) | +| `host.arch` | CPU architecture (semconv `host.arch` value) | +| `os.type` | semconv value — macOS is reported as `darwin` | +| `os.name` | e.g. `Ubuntu`, `Microsoft Windows Server 2022`, `macOS` | +| `os.version` | e.g. `22.04.5`, `10.0.20348.2762`, `14.6.1` | +| `os.build_id` | kernel release / Windows build number / macOS build | +| `os.description` | e.g. `Ubuntu 22.04.5 LTS` | +| `host.ip` | `[]string` of the host's interface IPv4 + IPv6 addresses | +| `host.mac` | `[]string` of the host's interface MAC addresses | +| `deployment.environment.name` | deployment tier — `production` / `staging` / `test` / `development` | +| `telemetry.source` | the generating module (`apache`, `nginx`, …) | + +Empty identity fields are omitted rather than emitted blank. Per-generator +constants (`apache.format`, `wel.channel`, `json.type`, …) are layered on top of +this set — see [embed.md](embed.md#resource-attributes). + +`host.image.*` (VM/OS image provenance) is a reserved framework hook. It is not +emitted today; a future cloud-identity source will populate it. + +## OS → hostname convention + +Hostnames are drawn from mythology pools, chosen by OS and role so a hostname +hints at what the machine is: + +| Pool | Mythology | Convention | +|----------|-----------|-----------------------------------| +| Norse | Norse | Linux servers | +| Roman | Roman | Windows servers and workstations | +| Greek | Greek | Domain Controllers | +| Egyptian | Egyptian | Network appliances / routers | +| Celtic | Celtic | macOS / developer workstations | + +Windows and Domain Controller hostnames render in the uppercase NetBIOS style; +Linux and macOS hosts use the lowercase style. + +## Determinism + +Given the same `seed_config` (and a fixed time anchor for time-dependent fields +like certificate validity windows), the generated environment is reproducible +run to run. Omit the seeds for a fresh randomized fleet each run; pin them for +reproducible fixtures, demos, and snapshot tests. diff --git a/generator/apache/apache.go b/generator/apache/apache.go index a52f6fc..7aac646 100644 --- a/generator/apache/apache.go +++ b/generator/apache/apache.go @@ -14,6 +14,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/internal/generator/security" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" @@ -43,6 +44,7 @@ type ApacheLogGenerator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -68,6 +70,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, componentName, "apache.format", "common"), stopCh: make(chan struct{}), }, nil } @@ -75,6 +78,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *ApacheLogGenerator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *ApacheLogGenerator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName, "apache.format", "common") +} + // Start launches the worker goroutines that push generated records to // the configured consumer. Start returns once workers are running. func (g *ApacheLogGenerator) Start(_ context.Context) error { @@ -174,7 +185,7 @@ func (g *ApacheLogGenerator) generateAndWriteLog(_ int) error { } // Format log data as Apache CLF - logRecord, err := formatAsApacheCLF(logData) + logRecord, err := formatAsApacheCLF(logData, g.static) if err != nil { g.recordWriteError("unknown", err) return fmt.Errorf("format log as Apache CLF: %w", err) @@ -302,7 +313,7 @@ func generateStatusAndSeverity(r *rand.Rand) (int, string) { // formatAsApacheCLF converts apacheLogData to Apache Common Log Format // Format: remotehost rfc931 authuser [date] "request" status bytes // Example: 127.0.0.1 - - [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326 -func formatAsApacheCLF(data *apacheLogData) (embed.LogRecord, error) { +func formatAsApacheCLF(data *apacheLogData, static *resource.StaticResources) (embed.LogRecord, error) { // Format timestamp as [dd/MMM/yyyy:HH:mm:ss -TZ] // Use local timezone offset loc := time.Now().Location() @@ -337,7 +348,7 @@ func formatAsApacheCLF(data *apacheLogData) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: data.timestamp, Severity: data.severity, - Resource: resource.Default(componentName, "apache.format", "common"), + Resource: static.Record(), }, }, nil } diff --git a/generator/apache/apache_test.go b/generator/apache/apache_test.go index 675b7a6..997ab36 100644 --- a/generator/apache/apache_test.go +++ b/generator/apache/apache_test.go @@ -11,6 +11,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -396,3 +397,18 @@ func BenchmarkApacheGenerator(b *testing.B) { defer cancel() _ = generator.Stop(ctx) } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockConsumer()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/apache_combined/apache_combined.go b/generator/apache_combined/apache_combined.go index 0deb344..be05597 100644 --- a/generator/apache_combined/apache_combined.go +++ b/generator/apache_combined/apache_combined.go @@ -45,6 +45,7 @@ type ApacheCombinedLogGenerator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -70,6 +71,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, "apache", "apache.format", "combined"), stopCh: make(chan struct{}), }, nil } @@ -77,6 +79,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *ApacheCombinedLogGenerator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *ApacheCombinedLogGenerator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, "apache", "apache.format", "combined") +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *ApacheCombinedLogGenerator) Start(_ context.Context) error { @@ -171,7 +181,7 @@ func (g *ApacheCombinedLogGenerator) generateAndWriteLog(_ int) error { } // Format log data as Apache Combined Log Format - logRecord, err := formatAsApacheCombined(logData) + logRecord, err := formatAsApacheCombined(logData, g.static) if err != nil { g.recordWriteError("unknown", err) return fmt.Errorf("format log as Apache Combined: %w", err) @@ -269,7 +279,7 @@ func generateReferer(r *rand.Rand) string { // formatAsApacheCombined converts apacheCombinedLogData to Apache Combined Log Format // Format: remotehost rfc931 authuser [date] "request" status bytes "referer" "user-agent" // Example: 127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326 "http://www.example.com/start.html" "Mozilla/4.08 [en] (Win98; I ;Nav)" -func formatAsApacheCombined(data *apacheCombinedLogData) (embed.LogRecord, error) { +func formatAsApacheCombined(data *apacheCombinedLogData, static *resource.StaticResources) (embed.LogRecord, error) { // Format timestamp as [dd/MMM/yyyy:HH:mm:ss -TZ] // Use local timezone offset loc := time.Now().Location() @@ -337,7 +347,7 @@ func formatAsApacheCombined(data *apacheCombinedLogData) (embed.LogRecord, error Metadata: embed.LogRecordMetadata{ Timestamp: data.timestamp, Severity: data.severity, - Resource: resource.Default("apache", "apache.format", "combined"), + Resource: static.Record(), }, }, nil } diff --git a/generator/apache_combined/apache_combined_test.go b/generator/apache_combined/apache_combined_test.go index 34936e9..ecb7f98 100644 --- a/generator/apache_combined/apache_combined_test.go +++ b/generator/apache_combined/apache_combined_test.go @@ -11,6 +11,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -387,3 +388,18 @@ func BenchmarkApacheCombinedGenerator(b *testing.B) { defer cancel() _ = generator.Stop(ctx) } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockConsumer()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/apache_error/apache_error.go b/generator/apache_error/apache_error.go index 9150324..55eaaf3 100644 --- a/generator/apache_error/apache_error.go +++ b/generator/apache_error/apache_error.go @@ -42,6 +42,7 @@ type ApacheErrorLogGenerator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -67,6 +68,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, "apache", "apache.format", "error"), stopCh: make(chan struct{}), }, nil } @@ -74,6 +76,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *ApacheErrorLogGenerator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *ApacheErrorLogGenerator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, "apache", "apache.format", "error") +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *ApacheErrorLogGenerator) Start(_ context.Context) error { @@ -168,7 +178,7 @@ func (g *ApacheErrorLogGenerator) generateAndWriteLog(_ int) error { } // Format log data as Apache Error Log Format - logRecord, err := formatAsApacheError(logData) + logRecord, err := formatAsApacheError(logData, g.static) if err != nil { g.recordWriteError("unknown", err) return fmt.Errorf("format log as Apache Error: %w", err) @@ -354,7 +364,7 @@ func generateErrorMessage(r *rand.Rand, level string) string { // formatAsApacheError converts apacheErrorLogData to Apache Error Log Format // Format: [timestamp] [level] [pid:tid] [client] message // Example: [Wed Oct 11 14:32:52 2000] [error] [client 127.0.0.1] client denied by server configuration: /export/home/live/ap/htdocs/test -func formatAsApacheError(data *apacheErrorLogData) (embed.LogRecord, error) { +func formatAsApacheError(data *apacheErrorLogData, static *resource.StaticResources) (embed.LogRecord, error) { // Format timestamp as [Day Mon DD HH:MM:SS YYYY] timestampStr := data.timestamp.Format("[Mon Jan 02 15:04:05 2006]") @@ -443,7 +453,7 @@ func formatAsApacheError(data *apacheErrorLogData) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: data.timestamp, Severity: data.severity, - Resource: resource.Default("apache", "apache.format", "error"), + Resource: static.Record(), }, }, nil } diff --git a/generator/apache_error/apache_error_test.go b/generator/apache_error/apache_error_test.go index 5f1bd97..6fee419 100644 --- a/generator/apache_error/apache_error_test.go +++ b/generator/apache_error/apache_error_test.go @@ -10,6 +10,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -389,3 +390,18 @@ func BenchmarkApacheErrorGenerator(b *testing.B) { defer cancel() _ = generator.Stop(ctx) } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockConsumer()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/filegen/filegen.go b/generator/filegen/filegen.go index 932701a..a91e5fe 100644 --- a/generator/filegen/filegen.go +++ b/generator/filegen/filegen.go @@ -20,6 +20,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/internal/generator/ctime" "github.com/observiq/blitz/telemetry" "go.uber.org/zap" @@ -97,6 +98,7 @@ type FileLogGenerator struct { rate time.Duration source string // file path or directory path or glob pattern consumer embed.LogConsumer + static *resource.StaticResources dataLibrary fs.FS // optional; nil falls back to ./data_library on disk for "package:" / bare-name sources stopCh chan struct{} tracker *count.Tracker @@ -150,6 +152,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, source string, cac rate: rate, source: source, consumer: consumer, + static: resource.FromIdentity(nil, "filegen", "filegen.source", source), dataLibrary: dataLibrary, stopCh: make(chan struct{}), cache: cache, @@ -159,6 +162,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, source string, cac // Name returns the module identifier. func (g *FileLogGenerator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *FileLogGenerator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, "filegen", "filegen.source", g.source) +} + // Start starts the File log generator and launches workers that push // records to the configured consumer. func (g *FileLogGenerator) Start(_ context.Context) error { @@ -525,7 +536,7 @@ func (g *FileLogGenerator) readAndWriteFile(filename string) error { Message: processedLine, Metadata: embed.LogRecordMetadata{ Timestamp: time.Now(), - Resource: resource.Default("filegen", "filegen.source", g.source), + Resource: g.static.Record(), }, }}) cancel() diff --git a/generator/filegen/filegen_test.go b/generator/filegen/filegen_test.go index f88ef71..9029c1a 100644 --- a/generator/filegen/filegen_test.go +++ b/generator/filegen/filegen_test.go @@ -12,6 +12,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -700,3 +701,24 @@ func TestFileLogGenerator_CountLimited(t *testing.T) { writes := consumer.getWrites() assert.Equal(t, 5, len(writes), "Expected exactly 5 logs with count tracker") } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + + tmpfile, err := os.CreateTemp("", "filegen-identity-*.log") + require.NoError(t, err) + defer os.Remove(tmpfile.Name()) + tmpfile.Close() + + gen, err := New(logger, 1, 100*time.Millisecond, tmpfile.Name(), true, 0, newMockConsumer(), nil) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/fix/fix.go b/generator/fix/fix.go index cb9204f..13e3e27 100644 --- a/generator/fix/fix.go +++ b/generator/fix/fix.go @@ -36,6 +36,7 @@ import ( "github.com/observiq/blitz/generator/fix/catalog/v44/app" "github.com/observiq/blitz/generator/fix/state" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" // Bring in per-category and per-version registrations. _ "github.com/observiq/blitz/generator/fix/catalog/v42" @@ -96,6 +97,7 @@ type Generator struct { logger *zap.Logger cfg Config consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} @@ -134,6 +136,7 @@ func New(logger *zap.Logger, cfg Config, consumer embed.LogConsumer) (*Generator logger: logger, cfg: cfg, consumer: consumer, + static: resource.FromIdentity(nil, componentName, "fix.version", cfg.Version.String()), stopCh: make(chan struct{}), }, nil } @@ -141,6 +144,14 @@ func New(logger *zap.Logger, cfg Config, consumer embed.LogConsumer) (*Generator // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName, "fix.version", g.cfg.Version.String()) +} + // Start launches the worker goroutines. func (g *Generator) Start(_ context.Context) error { g.logger.Info("Starting FIX generator", @@ -207,9 +218,7 @@ func (g *Generator) runWorker(workerIdx int) { Message: string(msg), Metadata: embed.LogRecordMetadata{ Severity: "INFO", - Resource: resource.Default(componentName, - "fix.version", g.cfg.Version.String(), - ), + Resource: g.static.Record(), }, } if err := g.consumer.ConsumeLogs(ctx, []embed.LogRecord{rec}); err != nil { diff --git a/generator/fix/fix_test.go b/generator/fix/fix_test.go index 6429283..fd6ad61 100644 --- a/generator/fix/fix_test.go +++ b/generator/fix/fix_test.go @@ -13,6 +13,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/fix/catalog" + "github.com/observiq/blitz/internal/datagen" ) // captureConsumer buffers every record passed to ConsumeLogs. @@ -245,3 +246,17 @@ func TestGeneratorSatisfiesProducerModule(t *testing.T) { require.NoError(t, err) var _ embed.ProducerModule = g } + +func TestSetHostIdentity(t *testing.T) { + g, err := New(zap.NewNop(), DefaultConfig(), &captureConsumer{}) + require.NoError(t, err) + + g.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", g.static.Record()["host.name"]) + + g.SetHostIdentity(nil) + assert.NotEmpty(t, g.static.Record()["host.name"]) +} diff --git a/generator/hostmetrics/hostmetrics.go b/generator/hostmetrics/hostmetrics.go index 0cd072c..41f5e46 100644 --- a/generator/hostmetrics/hostmetrics.go +++ b/generator/hostmetrics/hostmetrics.go @@ -28,11 +28,18 @@ type Config struct { Workers int // Rate is the scrape interval per worker. Required, > 0. Rate time.Duration - // OS is the simulated operating system ("linux" or "windows"). + // OS is the simulated operating system ("linux" or "windows"). Ignored + // when Identity is set (the identity's own OS is used instead). OS string // Hostname is the simulated hostname. If empty, a random hostname is - // generated per the OS style. + // generated per the OS style. Ignored when Identity is set. Hostname string + // Identity, when non-nil, is the resolved simulated host this generator's + // metrics describe (PIPE-1036). Its full host.* / os.* / deployment.* + // projection becomes the static resource on every emitted point. When nil, + // a minimal identity is synthesized from OS + Hostname, preserving the + // standalone-CLI behavior. + Identity *datagen.SystemIdentity // ScraperNames restricts emission to a named subset. Empty = all. ScraperNames []string // Consumer receives every scraped batch. Required. @@ -56,8 +63,9 @@ type Generator struct { logger *zap.Logger workers int rate time.Duration - os string + osType string hostname string + static *resource.StaticResources scrapers []Scraper consumer embed.MetricConsumer seed int64 @@ -87,12 +95,43 @@ func New(cfg Config) (*Generator, error) { return nil, fmt.Errorf("rate must be greater than 0, got %s", cfg.Rate) } + // Resolve the simulated host: an explicit Environment identity when + // supplied, otherwise a minimal identity synthesized from the OS/Hostname + // knobs. Either way the resource projection is built once here and reused + // for the generator's lifetime. + sys := cfg.Identity + if sys == nil { + sys = syntheticIdentity(cfg) + } + + return &Generator{ + logger: cfg.Logger.Named("generator-hostmetrics"), + workers: cfg.Workers, + rate: cfg.Rate, + osType: sys.OSInfo.Type.SemconvOSType(), + hostname: sys.Hostname, + static: resource.FromIdentity(sys, generatorType), + scrapers: buildScrapers(cfg.ScraperNames), + consumer: cfg.Consumer, + seed: cfg.Seed, + stopCh: make(chan struct{}), + }, nil +} + +// syntheticIdentity builds a minimal host identity from the generator's OS and +// Hostname knobs, used when no simulated Environment identity is wired +// (cfg.Identity == nil). The hostname is generated deterministically from Seed +// in the OS-appropriate style when cfg.Hostname is empty, preserving the prior +// standalone-CLI behavior. The style is derived from the datagen.OSType so an +// empty or non-windows OS renders a Linux-style host. +func syntheticIdentity(cfg Config) *datagen.SystemIdentity { + osType := datagen.OSType(cfg.OS) hostname := cfg.Hostname if hostname == "" { // Hostname-only RNG; intentionally seeded once at construction // since hostname is fixed for the lifetime of the generator. style := datagen.StyleLinux - if cfg.OS == "windows" { + if osType == datagen.OSWindows { style = datagen.StyleWindows } seed := cfg.Seed @@ -105,18 +144,10 @@ func New(cfg Config) (*Generator, error) { datagen.AllMythologyNames, ) } - - return &Generator{ - logger: cfg.Logger.Named("generator-hostmetrics"), - workers: cfg.Workers, - rate: cfg.Rate, - os: cfg.OS, - hostname: hostname, - scrapers: buildScrapers(cfg.ScraperNames), - consumer: cfg.Consumer, - seed: cfg.Seed, - stopCh: make(chan struct{}), - }, nil + return &datagen.SystemIdentity{ + Hostname: hostname, + OSInfo: datagen.OSInfo{Type: osType}, + } } // Name returns the module identifier for ProducerModule. @@ -132,7 +163,7 @@ func (g *Generator) Start(_ context.Context) error { g.logger.Info("Starting host metrics generator", zap.Int("workers", g.workers), zap.Duration("rate", g.rate), - zap.String("os", g.os), + zap.String("os.type", g.osType), zap.String("hostname", g.hostname), zap.Int("scrapers", len(g.scrapers)), ) @@ -209,17 +240,12 @@ func (g *Generator) scrape(r *rand.Rand) { ctx := context.Background() - // Build a fresh resource map per scrape. Future distributed-blitz - // simulation may derive resource from the simulated host's - // Environment record at scrape time — keep the allocation local so - // no scrape-to-scrape mutation can bleed state. resource.Default - // supplies telemetry.source + host.name (real process hostname); we - // override host.name with the datagen-generated simulated hostname - // because hostmetrics describes a simulated machine, not the host - // blitz is running on. - res := resource.Default(generatorType) - res["host.name"] = g.hostname - res["os.type"] = g.os + // The host-identity resource is fixed for this generator's lifetime, so it + // is built once (StaticResources in New) and shared read-only across every + // scrape and worker. Scrapers only attach it to the MetricRecords they + // return — they never mutate it — so handing out the zero-allocation shared + // map is safe under concurrent workers. + res := g.static.Record() for _, scraper := range g.scrapers { points := scraper.Scrape(r, g.hostname, res) diff --git a/generator/hostmetrics/hostmetrics_test.go b/generator/hostmetrics/hostmetrics_test.go index 8109541..8e17034 100644 --- a/generator/hostmetrics/hostmetrics_test.go +++ b/generator/hostmetrics/hostmetrics_test.go @@ -9,6 +9,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -108,6 +109,66 @@ func TestNew(t *testing.T) { }) } +// TestNewProjectsIdentityResource confirms that when a resolved datagen +// identity is supplied, the generator's static resource carries the full +// host.* / os.* / deployment.* projection (os.type as the semconv value, so +// macOS becomes darwin) rather than just host.name + os raw string. +func TestNewProjectsIdentityResource(t *testing.T) { + cfg := baseCfg(t, &mockMetricConsumer{}) + cfg.Hostname = "" + cfg.OS = "" + cfg.Identity = &datagen.SystemIdentity{ + Hostname: "THOR-01", + HostID: "abc123", + Arch: datagen.ArchAMD64, + Tier: datagen.TierProd, + OSInfo: datagen.OSInfo{Type: datagen.OSMacOS, Name: "macOS", Version: "14.6.1"}, + } + + g, err := New(cfg) + require.NoError(t, err) + + res := g.static.Record() + assert.Equal(t, "THOR-01", res["host.name"]) + assert.Equal(t, "abc123", res["host.id"]) + assert.Equal(t, "darwin", res["os.type"]) + assert.Equal(t, "macOS", res["os.name"]) + assert.Equal(t, "production", res["deployment.environment.name"]) + assert.Equal(t, "hostmetrics", res["telemetry.source"]) + assert.Equal(t, "THOR-01", g.hostname) +} + +// TestSyntheticIdentityOSTypeProjection confirms the no-Identity path builds a +// synthetic identity from the OS knob: an empty OS omits os.type entirely +// (rather than emitting an empty string), and a set OS projects through +// SemconvOSType. +func TestSyntheticIdentityOSTypeProjection(t *testing.T) { + cfg := baseCfg(t, &mockMetricConsumer{}) + cfg.OS = "" + + g, err := New(cfg) + require.NoError(t, err) + _, ok := g.static.Record()["os.type"] + assert.False(t, ok, "empty OS must not emit an empty os.type") + + cfg.OS = "windows" + g2, err := New(cfg) + require.NoError(t, err) + assert.Equal(t, "windows", g2.static.Record()["os.type"]) +} + +// TestSyntheticIdentityRandomSeed exercises the randomize branch of the +// synthetic-identity hostname generation (Seed < 0 → wall-clock seed). +func TestSyntheticIdentityRandomSeed(t *testing.T) { + cfg := baseCfg(t, &mockMetricConsumer{}) + cfg.Hostname = "" + cfg.Seed = -1 + + g, err := New(cfg) + require.NoError(t, err) + assert.NotEmpty(t, g.hostname) +} + func TestNameAndSupportedTelemetry(t *testing.T) { g, err := New(baseCfg(t, &mockMetricConsumer{})) require.NoError(t, err) diff --git a/generator/json/json.go b/generator/json/json.go index f54549b..5ec7ca3 100644 --- a/generator/json/json.go +++ b/generator/json/json.go @@ -13,6 +13,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/internal/generator/logtypes" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" @@ -81,6 +82,7 @@ type JSONLogGenerator struct { rate time.Duration logType string consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -117,6 +119,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, logType string, co rate: rate, logType: logType, consumer: consumer, + static: resource.FromIdentity(nil, componentName), stopCh: make(chan struct{}), }, nil } @@ -124,6 +127,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, logType string, co // Name returns the module identifier. func (g *JSONLogGenerator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *JSONLogGenerator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *JSONLogGenerator) Start(_ context.Context) error { @@ -239,7 +250,7 @@ func (g *JSONLogGenerator) generateAndWriteLog(_ int) error { } // Format log data as JSON - logRecord, err := formatAsJSON(logData) + logRecord, err := formatAsJSON(logData, g.static) if err != nil { g.recordWriteError("unknown", err) return fmt.Errorf("format log as JSON: %w", err) @@ -266,7 +277,7 @@ func (g *JSONLogGenerator) generateAndWriteLog(_ int) error { } // formatAsJSON converts LogData to a JSON-formatted LogRecord -func formatAsJSON(data logtypes.LogData) (embed.LogRecord, error) { +func formatAsJSON(data logtypes.LogData, static *resource.StaticResources) (embed.LogRecord, error) { var jsonData any var timestamp time.Time var severity string @@ -320,7 +331,7 @@ func formatAsJSON(data logtypes.LogData) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: timestamp, Severity: severity, - Resource: resource.Default(componentName, "json.type", jsonType), + Resource: static.Record("json.type", jsonType), }, }, nil } diff --git a/generator/json/json_test.go b/generator/json/json_test.go index 1a5b5c4..b85d640 100644 --- a/generator/json/json_test.go +++ b/generator/json/json_test.go @@ -10,6 +10,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/internal/generator/logtypes" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -469,3 +470,18 @@ func BenchmarkGenerateDefaultLog(b *testing.B) { _ = err } } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, "default", newMockWriter()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/kubernetes/kubernetes.go b/generator/kubernetes/kubernetes.go index 80bd1ad..a5d1d32 100644 --- a/generator/kubernetes/kubernetes.go +++ b/generator/kubernetes/kubernetes.go @@ -14,6 +14,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -57,6 +58,7 @@ type Generator struct { rate time.Duration format ContainerLogFormat consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -91,6 +93,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, format string, con rate: rate, format: logFormat, consumer: consumer, + static: resource.FromIdentity(nil, componentName, "kubernetes.format", formatCRIO), stopCh: make(chan struct{}), }, nil } @@ -98,6 +101,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, format string, con // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName, "kubernetes.format", formatCRIO) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *Generator) Start(_ context.Context) error { @@ -198,7 +209,7 @@ func (g *Generator) generateAndWriteLog(_ int) error { Metadata: embed.LogRecordMetadata{ Timestamp: timestamp, Severity: g.extractSeverity(appLog), - Resource: resource.Default(componentName, "kubernetes.format", formatCRIO), + Resource: g.static.Record(), }, } diff --git a/generator/kubernetes/kubernetes_test.go b/generator/kubernetes/kubernetes_test.go index f24aa3d..97d8a8d 100644 --- a/generator/kubernetes/kubernetes_test.go +++ b/generator/kubernetes/kubernetes_test.go @@ -8,6 +8,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -100,3 +101,18 @@ func TestGenerator_CountLimited(t *testing.T) { writes := writer.getWrites() assert.Equal(t, 5, len(writes), "Expected exactly 5 logs with count tracker") } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 50*time.Millisecond, "cri-o", newMockWriter()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/nginx/nginx.go b/generator/nginx/nginx.go index 58226a2..2a6b8ef 100644 --- a/generator/nginx/nginx.go +++ b/generator/nginx/nginx.go @@ -65,6 +65,7 @@ type Generator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -94,6 +95,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, componentName), stopCh: make(chan struct{}), }, nil } @@ -101,6 +103,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *Generator) Start(_ context.Context) error { @@ -193,7 +203,7 @@ func (g *Generator) generateAndWriteLog(_ int) error { return fmt.Errorf("generate NGINX log data: %w", err) } - logRecord, err := formatAsNginxCombined(logData) + logRecord, err := formatAsNginxCombined(logData, g.static) if err != nil { g.recordWriteError(errorTypeUnknown, err) return fmt.Errorf("format log as NGINX Combined: %w", err) @@ -297,7 +307,7 @@ func generateReferer(r *rand.Rand) string { // formatAsNginxCombined converts nginxLogData to NGINX Combined Log Format // Format: $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent" // Example: 127.0.0.1 - - [25/Dec/2023:10:15:30 -0800] "GET /index.html HTTP/1.1" 200 2326 "https://www.example.com/" "Mozilla/5.0..." -func formatAsNginxCombined(data *nginxLogData) (embed.LogRecord, error) { +func formatAsNginxCombined(data *nginxLogData, static *resource.StaticResources) (embed.LogRecord, error) { loc := time.Now().Location() localTime := data.timestamp.In(loc) _, offset := localTime.Zone() @@ -347,7 +357,7 @@ func formatAsNginxCombined(data *nginxLogData) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: data.timestamp, Severity: data.severity, - Resource: resource.Default(componentName), + Resource: static.Record(), }, }, nil } diff --git a/generator/nginx/nginx_test.go b/generator/nginx/nginx_test.go index 107af4f..ebba819 100644 --- a/generator/nginx/nginx_test.go +++ b/generator/nginx/nginx_test.go @@ -11,11 +11,25 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" ) +// TestSetHostIdentity confirms the setter projects the given host onto the +// static resource, and that a nil identity keeps the process-hostname fallback. +func TestSetHostIdentity(t *testing.T) { + g, err := New(zaptest.NewLogger(t), 1, time.Second, newMockWriter()) + require.NoError(t, err) + + g.SetHostIdentity(&datagen.SystemIdentity{Hostname: "IDENTITY-HOST", OSInfo: datagen.OSInfo{Type: datagen.OSLinux}}) + assert.Equal(t, "IDENTITY-HOST", g.static.Record()["host.name"]) + + g.SetHostIdentity(nil) + assert.NotEmpty(t, g.static.Record()["host.name"]) +} + // Compile-time assertion: the migrated generator satisfies embed.ProducerModule. var _ embed.ProducerModule = (*Generator)(nil) diff --git a/generator/okta/okta.go b/generator/okta/okta.go index f313a84..580f800 100644 --- a/generator/okta/okta.go +++ b/generator/okta/okta.go @@ -13,6 +13,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -34,6 +35,7 @@ type Generator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -207,6 +209,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, componentName), stopCh: make(chan struct{}), }, nil } @@ -214,6 +217,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *Generator) Start(_ context.Context) error { @@ -433,7 +444,7 @@ func (g *Generator) generateOktaLog(r *rand.Rand) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: now, Severity: event.severity, - Resource: resource.Default(componentName), + Resource: g.static.Record(), }, }, nil } diff --git a/generator/okta/okta_test.go b/generator/okta/okta_test.go index 532f27e..3c4761b 100644 --- a/generator/okta/okta_test.go +++ b/generator/okta/okta_test.go @@ -11,6 +11,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -418,3 +419,18 @@ func TestGenerateRequestID(t *testing.T) { assert.Len(t, id, 20) assert.Regexp(t, `^[A-Za-z0-9]+$`, id) } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockWriter()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/paloalto/paloalto.go b/generator/paloalto/paloalto.go index 37a1f98..54e596b 100644 --- a/generator/paloalto/paloalto.go +++ b/generator/paloalto/paloalto.go @@ -33,6 +33,7 @@ type Generator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} @@ -57,6 +58,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, componentName), stopCh: make(chan struct{}), }, nil } @@ -64,6 +66,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *Generator) Start(_ context.Context) error { @@ -156,7 +166,7 @@ func (g *Generator) generateAndWrite(_ int) error { Message: line, Metadata: embed.LogRecordMetadata{ Severity: "INFO", - Resource: resource.Default(componentName), + Resource: g.static.Record(), }, } diff --git a/generator/paloalto/paloalto_test.go b/generator/paloalto/paloalto_test.go index 584beb4..909dcff 100644 --- a/generator/paloalto/paloalto_test.go +++ b/generator/paloalto/paloalto_test.go @@ -10,6 +10,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -383,3 +384,18 @@ func TestGenerator_CountLimited(t *testing.T) { writes := writer.getWrites() assert.Equal(t, 5, len(writes), "Expected exactly 5 logs with count tracker") } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockWriter()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/postgres/postgres.go b/generator/postgres/postgres.go index 9eba378..23d2445 100644 --- a/generator/postgres/postgres.go +++ b/generator/postgres/postgres.go @@ -13,6 +13,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/count" "github.com/observiq/blitz/generator/resource" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -61,6 +62,7 @@ type Generator struct { workers int rate time.Duration consumer embed.LogConsumer + static *resource.StaticResources wg sync.WaitGroup stopCh chan struct{} tracker *count.Tracker @@ -246,6 +248,7 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log workers: workers, rate: rate, consumer: consumer, + static: resource.FromIdentity(nil, componentName), stopCh: make(chan struct{}), }, nil } @@ -253,6 +256,14 @@ func New(logger *zap.Logger, workers int, rate time.Duration, consumer embed.Log // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName) +} + // Start launches the worker goroutines that push generated records to // the configured consumer. func (g *Generator) Start(_ context.Context) error { @@ -344,7 +355,7 @@ func (g *Generator) generateAndWriteLog(_ int) error { return fmt.Errorf("generate PostgreSQL log data: %w", err) } - logRecord, err := formatAsPostgres(logData) + logRecord, err := formatAsPostgres(logData, g.static) if err != nil { g.recordWriteError(errorTypeUnknown, err) return fmt.Errorf("format log as PostgreSQL: %w", err) @@ -404,7 +415,7 @@ func generateRandomIP(r *rand.Rand) string { // formatAsPostgres converts postgresLogData to PostgreSQL log format // Format: %t [%p]: user=%u,db=%d,app=%a,client=%h : // Example: 2024-01-15 10:23:45.123 UTC [12345]: user=postgres,db=mydb,app=psql,client=127.0.0.1 LOG: statement: SELECT * FROM users; -func formatAsPostgres(data *postgresLogData) (embed.LogRecord, error) { +func formatAsPostgres(data *postgresLogData, static *resource.StaticResources) (embed.LogRecord, error) { // Format timestamp as PostgreSQL does: YYYY-MM-DD HH:MM:SS.mmm UTC timestampStr := data.timestamp.UTC().Format("2006-01-02 15:04:05.000 MST") @@ -497,7 +508,7 @@ func formatAsPostgres(data *postgresLogData) (embed.LogRecord, error) { Metadata: embed.LogRecordMetadata{ Timestamp: data.timestamp, Severity: data.severity, - Resource: resource.Default(componentName), + Resource: static.Record(), }, }, nil } diff --git a/generator/postgres/postgres_test.go b/generator/postgres/postgres_test.go index b2c80ae..cd4b464 100644 --- a/generator/postgres/postgres_test.go +++ b/generator/postgres/postgres_test.go @@ -11,6 +11,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap/zaptest" @@ -337,3 +338,18 @@ func BenchmarkPostgresGenerator(b *testing.B) { defer cancel() _ = generator.Stop(ctx) } + +func TestSetHostIdentity(t *testing.T) { + logger := zaptest.NewLogger(t) + gen, err := New(logger, 1, 100*time.Millisecond, newMockWriter()) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + assert.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + assert.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/generator/resource/identity.go b/generator/resource/identity.go new file mode 100644 index 0000000..2ff97f2 --- /dev/null +++ b/generator/resource/identity.go @@ -0,0 +1,77 @@ +package resource + +import ( + "github.com/observiq/blitz/internal/datagen" +) + +// FromIdentity builds a StaticResources for a generator worker from a resolved +// datagen host identity. It projects the identity's OpenTelemetry host.* / os.* / +// deployment.* resource attributes, stamps telemetry.source = source, and appends +// any per-generator constants in extras (same even-length key/value convention as +// Default). Empty identity fields are omitted rather than emitted blank. +// +// A nil sys means no simulated environment is wired: FromIdentity falls back to +// the running host's name, matching Default(source, extras...), so every +// generator has a single uniform construction path regardless of whether an +// Environment is present. +// +// host.image.* is deliberately never emitted: SystemIdentity.Image is an unwired +// framework hook for a future CloudIdentity source (PIPE-1036). +func FromIdentity(sys *datagen.SystemIdentity, source string, extras ...string) *StaticResources { + if sys == nil { + return NewStaticResources(WithHost(Hostname(), source, extras...)) + } + attrs := projectIdentity(sys) + attrs["telemetry.source"] = source + for i := 0; i+1 < len(extras); i += 2 { + attrs[extras[i]] = extras[i+1] + } + return NewStaticResources(attrs) +} + +// projectIdentity maps a host identity to its OpenTelemetry resource attributes: +// host.name / host.id / host.arch, the os.* set (os.type carrying the semconv +// value, so macOS becomes darwin), host.ip[] / host.mac[] gathered from the +// identity's interfaces, and deployment.environment.name. Empty fields are +// omitted. It does not set telemetry.source (a per-generator constant) and never +// emits host.image.* (an unwired framework hook). +func projectIdentity(sys *datagen.SystemIdentity) map[string]any { + attrs := make(map[string]any, 12) + putNonEmpty(attrs, "host.name", sys.Hostname) + putNonEmpty(attrs, "host.id", sys.HostID) + putNonEmpty(attrs, "host.arch", string(sys.Arch)) + putNonEmpty(attrs, "os.type", sys.OSInfo.Type.SemconvOSType()) + putNonEmpty(attrs, "os.name", sys.OSInfo.Name) + putNonEmpty(attrs, "os.version", sys.OSInfo.Version) + putNonEmpty(attrs, "os.build_id", sys.OSInfo.BuildID) + putNonEmpty(attrs, "os.description", sys.OSInfo.Description) + putNonEmpty(attrs, "deployment.environment.name", string(sys.Tier)) + + var ips, macs []string + for _, iface := range sys.Interfaces { + if iface.IPv4 != "" { + ips = append(ips, iface.IPv4) + } + if iface.IPv6 != "" { + ips = append(ips, iface.IPv6) + } + if iface.MACAddress != "" { + macs = append(macs, iface.MACAddress) + } + } + if len(ips) > 0 { + attrs["host.ip"] = ips + } + if len(macs) > 0 { + attrs["host.mac"] = macs + } + return attrs +} + +// putNonEmpty sets m[key] = val only when val is non-empty, so blank identity +// fields never produce empty-string resource attributes. +func putNonEmpty(m map[string]any, key, val string) { + if val != "" { + m[key] = val + } +} diff --git a/generator/resource/identity_test.go b/generator/resource/identity_test.go new file mode 100644 index 0000000..c6d082f --- /dev/null +++ b/generator/resource/identity_test.go @@ -0,0 +1,145 @@ +package resource + +import ( + "os" + "reflect" + "testing" + + "github.com/observiq/blitz/internal/datagen" +) + +// fullIdentity is a completely-populated host identity for projection tests. +func fullIdentity() *datagen.SystemIdentity { + return &datagen.SystemIdentity{ + Hostname: "THOR-WEB-01", + HostID: "6b3a2f1e-1122-3344-5566-778899aabbcc", + Arch: datagen.ArchAMD64, + Tier: datagen.TierProd, + OSInfo: datagen.OSInfo{ + Type: datagen.OSWindows, + Name: "Microsoft Windows Server 2022", + Version: "10.0.20348.2762", + BuildID: "20348", + Description: "Microsoft Windows [Version 10.0.20348.2762]", + }, + Interfaces: []datagen.NetworkInterface{ + {Name: "Ethernet0", IPv4: "10.10.1.20", IPv6: "fe80::1", MACAddress: "00:1a:2b:3c:4d:5e"}, + }, + } +} + +func TestFromIdentityFullProjection(t *testing.T) { + rec := FromIdentity(fullIdentity(), "wel").Record() + + want := map[string]string{ + "host.name": "THOR-WEB-01", + "host.id": "6b3a2f1e-1122-3344-5566-778899aabbcc", + "host.arch": "amd64", + "os.type": "windows", + "os.name": "Microsoft Windows Server 2022", + "os.version": "10.0.20348.2762", + "os.build_id": "20348", + "os.description": "Microsoft Windows [Version 10.0.20348.2762]", + "deployment.environment.name": "production", + "telemetry.source": "wel", + } + for k, v := range want { + if rec[k] != v { + t.Errorf("%s = %v, want %q", k, rec[k], v) + } + } +} + +func TestFromIdentityMacOSTypeIsDarwin(t *testing.T) { + sys := &datagen.SystemIdentity{ + Hostname: "brigid-mbp", + OSInfo: datagen.OSInfo{Type: datagen.OSMacOS, Name: "macOS", Version: "14.6.1"}, + } + rec := FromIdentity(sys, "json").Record() + if rec["os.type"] != "darwin" { + t.Errorf("os.type = %v, want darwin (semconv value for macOS)", rec["os.type"]) + } +} + +func TestFromIdentityOmitsEmptyFields(t *testing.T) { + // A bare identity: only a hostname, everything else zero-valued. + sys := &datagen.SystemIdentity{Hostname: "sparse-01"} + rec := FromIdentity(sys, "apache").Record() + + if rec["host.name"] != "sparse-01" { + t.Errorf("host.name = %v, want sparse-01", rec["host.name"]) + } + // No empty os.type / os.name / host.id / host.arch / deployment.* should be present. + for _, k := range []string{"host.id", "host.arch", "os.type", "os.name", "os.version", "os.build_id", "os.description", "deployment.environment.name", "host.ip", "host.mac"} { + if _, ok := rec[k]; ok { + t.Errorf("expected %q to be omitted for a sparse identity, got %v", k, rec[k]) + } + } +} + +func TestFromIdentityHostIPAndMACArrays(t *testing.T) { + sys := &datagen.SystemIdentity{ + Hostname: "multi-nic", + Interfaces: []datagen.NetworkInterface{ + {IPv4: "10.10.1.20", IPv6: "fe80::1", MACAddress: "00:1a:2b:3c:4d:5e"}, + {IPv4: "10.10.2.30", IPv6: "", MACAddress: "00:1a:2b:3c:4d:5f"}, + }, + } + rec := FromIdentity(sys, "paloalto").Record() + + ips, ok := rec["host.ip"].([]string) + if !ok { + t.Fatalf("host.ip = %T, want []string (OTLP ArrayValue via PIPE-1253)", rec["host.ip"]) + } + wantIPs := []string{"10.10.1.20", "fe80::1", "10.10.2.30"} + if !reflect.DeepEqual(ips, wantIPs) { + t.Errorf("host.ip = %v, want %v", ips, wantIPs) + } + + macs, ok := rec["host.mac"].([]string) + if !ok { + t.Fatalf("host.mac = %T, want []string", rec["host.mac"]) + } + wantMACs := []string{"00:1a:2b:3c:4d:5e", "00:1a:2b:3c:4d:5f"} + if !reflect.DeepEqual(macs, wantMACs) { + t.Errorf("host.mac = %v, want %v", macs, wantMACs) + } +} + +func TestFromIdentityNilFallsBackToHostname(t *testing.T) { + rec := FromIdentity(nil, "nginx").Record() + + h, _ := os.Hostname() + if h == "" { + h = "blitz" + } + if rec["host.name"] != h { + t.Errorf("host.name = %v, want process hostname %q on nil identity", rec["host.name"], h) + } + if rec["telemetry.source"] != "nginx" { + t.Errorf("telemetry.source = %v, want nginx", rec["telemetry.source"]) + } + // No simulated os.* attributes when there is no identity. + if _, ok := rec["os.type"]; ok { + t.Errorf("os.type should be absent on nil identity, got %v", rec["os.type"]) + } +} + +func TestFromIdentityDoesNotEmitHostImage(t *testing.T) { + sys := fullIdentity() + sys.Image = &datagen.HostImage{ID: "ami-0abc", Name: "win-2022", Version: "20240115"} + rec := FromIdentity(sys, "wel").Record() + + for _, k := range []string{"host.image.id", "host.image.name", "host.image.version"} { + if _, ok := rec[k]; ok { + t.Errorf("%q must not be emitted (unwired framework hook), got %v", k, rec[k]) + } + } +} + +func TestFromIdentityExtrasApplied(t *testing.T) { + rec := FromIdentity(fullIdentity(), "wel", "wel.role", "dc").Record() + if rec["wel.role"] != "dc" { + t.Errorf("extras not applied: wel.role = %v, want dc", rec["wel.role"]) + } +} diff --git a/generator/traces/traces.go b/generator/traces/traces.go index c9957df..67c56bb 100644 --- a/generator/traces/traces.go +++ b/generator/traces/traces.go @@ -46,8 +46,15 @@ type Config struct { // a deterministic Linux-style hostname is generated from Seed via // datagen.GenerateHostname, matching the hostmetrics convention so // records from both signals attribute consistently to the same - // simulated machine when configured with the same Seed. + // simulated machine when configured with the same Seed. Ignored when + // Identity is set. Hostname string + // Identity, when non-nil, is the resolved simulated host these traces + // describe (PIPE-1036). Its full host.* / os.* / deployment.* projection + // becomes the static resource on every span. When nil, a minimal + // Linux-style identity is synthesized from Hostname, preserving the prior + // standalone-CLI behavior. + Identity *datagen.SystemIdentity // Consumer receives each emitted span individually as it "completes" // (its EndTime is reached on wall-clock). Required. Consumer embed.TraceConsumer @@ -90,6 +97,7 @@ type Generator struct { workers int rate time.Duration hostname string + static *resource.StaticResources consumer embed.TraceConsumer seed int64 @@ -118,6 +126,33 @@ func New(cfg Config) (*Generator, error) { return nil, fmt.Errorf("rate must be greater than 0, got %s", cfg.Rate) } + // Resolve the simulated host: an explicit Environment identity when + // supplied, otherwise a minimal Linux-style identity synthesized from the + // Hostname knob. The resource projection is built once here and reused for + // every span. + sys := cfg.Identity + if sys == nil { + sys = syntheticIdentity(cfg) + } + + return &Generator{ + logger: cfg.Logger.Named("generator-traces"), + workers: cfg.Workers, + rate: cfg.Rate, + hostname: sys.Hostname, + static: resource.FromIdentity(sys, generatorType), + consumer: cfg.Consumer, + seed: cfg.Seed, + stopCh: make(chan struct{}), + }, nil +} + +// syntheticIdentity builds a minimal Linux-style host identity from the Hostname +// knob, used when no simulated Environment identity is wired (cfg.Identity == +// nil). When cfg.Hostname is empty a hostname is generated deterministically +// from Seed, matching the hostmetrics convention so both signals attribute to +// the same simulated machine under the same Seed. +func syntheticIdentity(cfg Config) *datagen.SystemIdentity { hostname := cfg.Hostname if hostname == "" { // Hostname-only RNG; seeded once at construction since the @@ -132,16 +167,7 @@ func New(cfg Config) (*Generator, error) { datagen.AllMythologyNames, ) } - - return &Generator{ - logger: cfg.Logger.Named("generator-traces"), - workers: cfg.Workers, - rate: cfg.Rate, - hostname: hostname, - consumer: cfg.Consumer, - seed: cfg.Seed, - stopCh: make(chan struct{}), - }, nil + return &datagen.SystemIdentity{Hostname: hostname} } // Name returns the module identifier for ProducerModule. @@ -289,8 +315,10 @@ func (g *Generator) startTrace(r *mathrand.Rand) { } } - res := resource.Default(generatorType) - res["host.name"] = g.hostname + // The static host-identity resource is shared read-only; each span takes a + // defensive clone (cloneResource) so per-span mutations can't bleed across + // spans or into the shared set. + res := g.static.Record() traceID := generateTraceID() now := time.Now() diff --git a/generator/traces/traces_test.go b/generator/traces/traces_test.go index 7445b9c..f0a4e08 100644 --- a/generator/traces/traces_test.go +++ b/generator/traces/traces_test.go @@ -8,6 +8,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/count" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -63,6 +64,45 @@ func baseCfg(t *testing.T, cons embed.TraceConsumer) Config { } } +// TestNewProjectsIdentityResource confirms that when a resolved datagen +// identity is supplied, the generator's static resource carries the full +// host.* / os.* / deployment.* projection, so spans describe the simulated +// host rather than just its name. +func TestNewProjectsIdentityResource(t *testing.T) { + cfg := baseCfg(t, &mockTraceConsumer{}) + cfg.Identity = &datagen.SystemIdentity{ + Hostname: "odin-api-01", + HostID: "def456", + Arch: datagen.ArchARM64, + Tier: datagen.TierStaging, + OSInfo: datagen.OSInfo{Type: datagen.OSLinux, Name: "Ubuntu", Version: "22.04.5"}, + } + + g, err := New(cfg) + require.NoError(t, err) + + res := g.static.Record() + assert.Equal(t, "odin-api-01", res["host.name"]) + assert.Equal(t, "def456", res["host.id"]) + assert.Equal(t, "arm64", res["host.arch"]) + assert.Equal(t, "linux", res["os.type"]) + assert.Equal(t, "staging", res["deployment.environment.name"]) + assert.Equal(t, "traces", res["telemetry.source"]) + assert.Equal(t, "odin-api-01", g.hostname) +} + +// TestSyntheticIdentityRandomSeed exercises the randomize branch of the +// synthetic-identity hostname generation (Seed < 0 → wall-clock seed). +func TestSyntheticIdentityRandomSeed(t *testing.T) { + cfg := baseCfg(t, &mockTraceConsumer{}) + cfg.Hostname = "" + cfg.Seed = -1 + + g, err := New(cfg) + require.NoError(t, err) + assert.NotEmpty(t, g.hostname) +} + func TestNew(t *testing.T) { t.Run("valid", func(t *testing.T) { g, err := New(baseCfg(t, &mockTraceConsumer{})) diff --git a/generator/wel/wel.go b/generator/wel/wel.go index 6fc3ab9..e13c777 100644 --- a/generator/wel/wel.go +++ b/generator/wel/wel.go @@ -13,6 +13,7 @@ import ( "github.com/observiq/blitz/generator" "github.com/observiq/blitz/generator/resource" "github.com/observiq/blitz/generator/wel/catalog" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -37,6 +38,7 @@ type Generator struct { role catalog.MachineRole channels []string consumer embed.LogConsumer + static *resource.StaticResources registry *catalog.Registry state *catalog.StateTracker @@ -125,6 +127,11 @@ func New(cfg Config) (*Generator, error) { role: cfg.Role, channels: channels, consumer: cfg.Consumer, + static: resource.FromIdentity(nil, componentName, + "wel.computer", cfg.Computer, + "wel.domain", cfg.Domain, + "wel.role", string(cfg.Role), + ), registry: reg, state: state, opts: opts, @@ -135,6 +142,18 @@ func New(cfg Config) (*Generator, error) { // Name returns the module identifier. func (g *Generator) Name() string { return componentName } +// SetHostIdentity sets the simulated host whose identity every emitted record +// carries (PIPE-1036). A nil identity keeps the process-hostname fallback. Must +// be called before Start; the resource it builds is read concurrently by +// workers thereafter. +func (g *Generator) SetHostIdentity(id *datagen.SystemIdentity) { + g.static = resource.FromIdentity(id, componentName, + "wel.computer", g.computer, + "wel.domain", g.domain, + "wel.role", string(g.role), + ) +} + // Start launches the worker goroutines that yield generated records // to the configured consumer. Start returns once workers are running. func (g *Generator) Start(_ context.Context) error { @@ -239,12 +258,7 @@ func (g *Generator) generateAndWrite(rng *rand.Rand) error { Message: xml, Metadata: embed.LogRecordMetadata{ Severity: record.LevelName, - Resource: resource.Default(componentName, - "wel.channel", record.Channel, - "wel.computer", g.computer, - "wel.domain", g.domain, - "wel.role", string(g.role), - ), + Resource: g.static.Record("wel.channel", record.Channel), }, } diff --git a/generator/wel/wel_test.go b/generator/wel/wel_test.go index 113472c..d666968 100644 --- a/generator/wel/wel_test.go +++ b/generator/wel/wel_test.go @@ -9,6 +9,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/generator/wel/catalog" + "github.com/observiq/blitz/internal/datagen" "github.com/observiq/blitz/telemetry" "github.com/stretchr/testify/require" "go.uber.org/zap" @@ -177,3 +178,23 @@ func TestGeneratorSupportedTelemetry(t *testing.T) { t.Errorf("expected Logs telemetry type, got %v", types[0]) } } + +func TestSetHostIdentity(t *testing.T) { + gen, err := New(Config{ + Logger: zap.NewNop(), + Workers: 1, + Rate: time.Second, + Usernames: []string{"test"}, + Consumer: &mockConsumer{}, + }) + require.NoError(t, err) + + gen.SetHostIdentity(&datagen.SystemIdentity{ + Hostname: "IDENTITY-HOST", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux}, + }) + require.Equal(t, "IDENTITY-HOST", gen.static.Record()["host.name"]) + + gen.SetHostIdentity(nil) + require.NotEmpty(t, gen.static.Record()["host.name"]) +} diff --git a/internal/datagen/arch_test.go b/internal/datagen/arch_test.go new file mode 100644 index 0000000..a000215 --- /dev/null +++ b/internal/datagen/arch_test.go @@ -0,0 +1,33 @@ +package datagen + +import "testing" + +func TestParseArch(t *testing.T) { + valid := map[string]Arch{ + "amd64": ArchAMD64, + "arm32": ArchARM32, + "arm64": ArchARM64, + "ia64": ArchIA64, + "ppc32": ArchPPC32, + "ppc64": ArchPPC64, + "s390x": ArchS390X, + "x86": ArchX86, + " AMD64 ": ArchAMD64, // trimmed + lowercased + } + for in, want := range valid { + got, err := ParseArch(in) + if err != nil { + t.Errorf("ParseArch(%q): unexpected error: %v", in, err) + continue + } + if got != want { + t.Errorf("ParseArch(%q) = %q, want %q", in, got, want) + } + } + + for _, in := range []string{"", "sparc", "riscv64", "mips"} { + if _, err := ParseArch(in); err == nil { + t.Errorf("ParseArch(%q): want error, got nil", in) + } + } +} diff --git a/internal/datagen/environment.go b/internal/datagen/environment.go index ddcf5c6..6741420 100644 --- a/internal/datagen/environment.go +++ b/internal/datagen/environment.go @@ -293,6 +293,15 @@ func generateSystems(systemSeed, servicesSeed, applicationsSeed int64, count int } weights := []float64{0.10, 0.40, 0.30, 0.15, 0.05} + // Prod-tier baseline OS release per family, chosen once so every prod host + // of a family is pinned to the same conservative release (real fleets keep + // prod uniform). Non-prod hosts roll newer and vary per host. + prodBaseline := map[OSType]OSInfo{ + OSLinux: osInfoForTier(r, OSLinux, true), + OSWindows: osInfoForTier(r, OSWindows, true), + OSMacOS: osInfoForTier(r, OSMacOS, true), + } + systems := make([]*SystemIdentity, count) for i := 0; i < count; i++ { // Pick OS/role using weighted selection @@ -302,16 +311,25 @@ func generateSystems(systemSeed, servicesSeed, applicationsSeed int64, count int return nil, err } + // Assign a deployment tier and cluster the OS release by it: prod hosts + // share the pinned family baseline; non-prod hosts roll newer and vary. + sys.Tier = weightedSelect(r, DeploymentTiers, deploymentTierWeights) + if sys.Tier == TierProd { + sys.OSInfo = prodBaseline[spec.os] + } else { + sys.OSInfo = osInfoForTier(r, spec.os, false) + } + // Services and applications use their own RNGs so the seeds in // SeedConfig actually drive what's generated, per identity type. - sys.Services = GenerateServicesForSystem(rServices, sys.OS, sys.Role, sys.Hostname) - sys.Applications = GenerateApplicationsForSystem(rApplications, sys.OS, sys.Role, sys.Hostname, logger) + sys.Services = GenerateServicesForSystem(rServices, sys.OSInfo.Type, sys.Role, sys.Hostname) + sys.Applications = GenerateApplicationsForSystem(rApplications, sys.OSInfo.Type, sys.Role, sys.Hostname, logger) // Assign network interface if len(networks) > 0 { net := pickNetworkForRole(r, sys.Role, networks) iface := NetworkInterface{ - Name: interfaceName(sys.OS), + Name: interfaceName(sys.OSInfo.Type), IPv4: RandomIPInCIDR(r, net.CIDR), IPv6: RandomIPv6(r), MACAddress: RandomMAC(r), diff --git a/internal/datagen/osrelease.go b/internal/datagen/osrelease.go new file mode 100644 index 0000000..d4e7629 --- /dev/null +++ b/internal/datagen/osrelease.go @@ -0,0 +1,177 @@ +package datagen + +import ( + "fmt" + "math/rand" + "strings" +) + +// DeploymentTier is a host's deployment environment, emitted as the OTel +// deployment.environment.name resource attribute. The set is hardcoded for now; +// making the tier vocabulary and fleet distribution user-configurable is a +// future project (see stub-dynamic-deployment-tiers / round 16). +type DeploymentTier string + +const ( + TierProd DeploymentTier = "production" + TierStaging DeploymentTier = "staging" + TierTest DeploymentTier = "test" + TierDev DeploymentTier = "development" +) + +// DeploymentTiers is the hardcoded tier set; deploymentTierWeights is the +// default prod-heavy fleet distribution (prod 55%, the rest 15% each). +var ( + DeploymentTiers = []DeploymentTier{TierProd, TierStaging, TierTest, TierDev} + deploymentTierWeights = []float64{0.55, 0.15, 0.15, 0.15} +) + +// OSInfo is the OpenTelemetry os.* projection source for a system: the fields +// map 1:1 to os.type / os.name / os.version / os.build_id / os.description. The +// values within one OSInfo are internally consistent (a real name/version/ +// build/description that actually go together), sourced from authentic release +// data (see the pools below), not synthesized piecemeal. +type OSInfo struct { + Type OSType // os.type + Name string // os.name e.g. "Ubuntu", "Microsoft Windows Server 2022", "macOS" + Version string // os.version e.g. "22.04.5", "10.0.20348.2762", "14.6.1" + BuildID string // os.build_id e.g. "5.15.0-91-generic", "20348", "23G80" + Description string // os.description e.g. "Ubuntu 22.04.5 LTS" +} + +// linuxRelease is one authentic Linux distro release. kernel becomes +// os.build_id; pretty is the os-release PRETTY_NAME (os.description). +type linuxRelease struct{ name, version, kernel, pretty string } + +// windowsRelease is one Windows product. build is the CurrentBuildNumber +// (os.build_id); ubrs are real Update Build Revisions from the product's update +// history — one is chosen per generation to form the full version + ver-string. +type windowsRelease struct { + name, build string + ubrs []int +} + +// macRelease is one authentic macOS point release (version + ProductBuildVersion). +type macRelease struct{ version, build string } + +// Authentic release data. Linux: os-release NAME / VERSION_ID / kernel / +// PRETTY_NAME. Windows: product name / build number / real UBRs from MS update +// history. macOS: verified version→build pairs (Apple/Wikipedia). +var ( + linuxReleases = []linuxRelease{ + {"Ubuntu", "22.04.5", "5.15.0-91-generic", "Ubuntu 22.04.5 LTS"}, + {"Debian GNU/Linux", "12", "6.1.0-18-amd64", "Debian GNU/Linux 12 (bookworm)"}, + {"Debian GNU/Linux", "11", "5.10.0-27-amd64", "Debian GNU/Linux 11 (bullseye)"}, + {"Red Hat Enterprise Linux", "9.3", "5.14.0-362.el9.x86_64", "Red Hat Enterprise Linux 9.3 (Plow)"}, + {"Fedora Linux", "39", "6.6.9-200.fc39.x86_64", "Fedora Linux 39 (Server Edition)"}, + } + + windowsReleases = []windowsRelease{ + {"Microsoft Windows Server 2022", "20348", []int{2227, 2322, 2340, 2762, 4405, 4529, 4893, 5020, 5139, 5386}}, + {"Microsoft Windows Server 2019", "17763", []int{8276, 8389, 8511, 8647, 8755, 8880, 9020}}, + {"Microsoft Windows Server 2016", "14393", []int{2724, 8330, 8783, 8868, 8957, 9062, 9140}}, + {"Microsoft Windows 11 Pro", "22631", []int{6199, 6276, 6345, 6491, 6649, 6783, 6936, 7079, 7219, 7376}}, + {"Microsoft Windows 10 Pro", "19045", []int{6575, 6691, 6809, 6937, 7058, 7184, 7291, 7417, 7548}}, + } + + macReleases = []macRelease{ + {"14.2.1", "23C71"}, {"14.3.1", "23D60"}, {"14.4", "23E214"}, {"14.4.1", "23E224"}, + {"14.5", "23F79"}, {"14.6", "23G80"}, {"14.6.1", "23G93"}, {"14.7", "23H124"}, + {"14.7.1", "23H222"}, {"14.7.2", "23H311"}, + } +) + +// GenerateOSInfo returns a coherent OSInfo for the given OS type, drawn from the +// authentic release pools. Windows picks one real UBR per selection. +// Deterministic for a given (r, os). +func GenerateOSInfo(r *rand.Rand, os OSType) OSInfo { + switch os { + case OSWindows: + rel := windowsReleases[r.Intn(len(windowsReleases))] // #nosec G404 + return buildWindowsOSInfo(rel, rel.ubrs[r.Intn(len(rel.ubrs))]) // #nosec G404 + case OSMacOS: + return buildMacOSInfo(macReleases[r.Intn(len(macReleases))]) // #nosec G404 + default: + return buildLinuxOSInfo(linuxReleases[r.Intn(len(linuxReleases))]) // #nosec G404 + } +} + +// osInfoForTier picks a release biased by patch currency: the pinned baseline +// (older=true) draws from the conservative/older end of an ordered patch +// timeline, non-prod (older=false) from the newer end. Linux has no +// cross-distro age ordering, so it draws any release regardless. +func osInfoForTier(r *rand.Rand, os OSType, older bool) OSInfo { + switch os { + case OSWindows: + rel := windowsReleases[r.Intn(len(windowsReleases))] // #nosec G404 + return buildWindowsOSInfo(rel, rel.ubrs[pickHalfIndex(r, len(rel.ubrs), older)]) + case OSMacOS: + return buildMacOSInfo(macReleases[pickHalfIndex(r, len(macReleases), older)]) + default: + return buildLinuxOSInfo(linuxReleases[r.Intn(len(linuxReleases))]) // #nosec G404 + } +} + +// pickHalfIndex returns an index into an n-element, oldest-first pool: the older +// half [0, n/2) when older is true, else the newer half [n/2, n). +func pickHalfIndex(r *rand.Rand, n int, older bool) int { + half := n / 2 + if older { + if half == 0 { + return 0 + } + return r.Intn(half) // #nosec G404 + } + return half + r.Intn(n-half) // #nosec G404 +} + +func buildWindowsOSInfo(rel windowsRelease, ubr int) OSInfo { + version := fmt.Sprintf("10.0.%s.%d", rel.build, ubr) + return OSInfo{ + Type: OSWindows, + Name: rel.name, + Version: version, + BuildID: rel.build, + Description: fmt.Sprintf("Microsoft Windows [Version %s]", version), + } +} + +func buildMacOSInfo(rel macRelease) OSInfo { + return OSInfo{ + Type: OSMacOS, + Name: "macOS", + Version: rel.version, + BuildID: rel.build, + Description: fmt.Sprintf("macOS %s (%s)", rel.version, rel.build), + } +} + +func buildLinuxOSInfo(rel linuxRelease) OSInfo { + return OSInfo{ + Type: OSLinux, + Name: rel.name, + Version: rel.version, + BuildID: rel.kernel, + Description: rel.pretty, + } +} + +// GenerateHostID returns an OS-appropriate host.id: a /etc/machine-id-style +// 32-char lowercase hex string on Linux, a registry MachineGuid-style GUID on +// Windows, and an uppercase IOPlatformUUID on macOS. +func GenerateHostID(r *rand.Rand, os OSType) string { + h := randomHex(r, 16) // 32 lowercase hex chars + switch os { + case OSWindows: + return formatUUID(h) + case OSMacOS: + return strings.ToUpper(formatUUID(h)) + default: + return h + } +} + +// formatUUID inserts UUID dashes (8-4-4-4-12) into a 32-char hex string. +func formatUUID(hex32 string) string { + return hex32[0:8] + "-" + hex32[8:12] + "-" + hex32[12:16] + "-" + hex32[16:20] + "-" + hex32[20:32] +} diff --git a/internal/datagen/osrelease_test.go b/internal/datagen/osrelease_test.go new file mode 100644 index 0000000..498e2ad --- /dev/null +++ b/internal/datagen/osrelease_test.go @@ -0,0 +1,100 @@ +package datagen + +import ( + "math/rand" + "regexp" + "strings" + "testing" +) + +func TestGenerateOSInfo_Coherent(t *testing.T) { + for _, os := range []OSType{OSLinux, OSWindows, OSMacOS} { + r := rand.New(rand.NewSource(1)) // #nosec G404 + info := GenerateOSInfo(r, os) + if info.Type != os { + t.Errorf("%s: Type = %q, want %q", os, info.Type, os) + } + if info.Name == "" || info.Version == "" || info.BuildID == "" || info.Description == "" { + t.Errorf("%s: incomplete OSInfo: %+v", os, info) + } + } +} + +func TestGenerateOSInfo_Deterministic(t *testing.T) { + a := GenerateOSInfo(rand.New(rand.NewSource(7)), OSWindows) // #nosec G404 + b := GenerateOSInfo(rand.New(rand.NewSource(7)), OSWindows) // #nosec G404 + if a != b { + t.Errorf("GenerateOSInfo not deterministic: %+v vs %+v", a, b) + } +} + +func TestGenerateOSInfo_WindowsUBR(t *testing.T) { + // Windows os.description is the ver-string carrying a real UBR, and + // os.build_id is the bare build number. + info := GenerateOSInfo(rand.New(rand.NewSource(3)), OSWindows) // #nosec G404 + if !strings.HasPrefix(info.Description, "Microsoft Windows [Version 10.0.") { + t.Errorf("Windows description = %q, want ver-string form", info.Description) + } + if !regexp.MustCompile(`^\d+$`).MatchString(info.BuildID) { + t.Errorf("Windows build_id = %q, want bare build number", info.BuildID) + } + // The build number appears in the version quad. + if !strings.Contains(info.Version, info.BuildID) { + t.Errorf("Windows version %q should contain build_id %q", info.Version, info.BuildID) + } +} + +func TestGenerateOSInfo_MacOS(t *testing.T) { + info := GenerateOSInfo(rand.New(rand.NewSource(2)), OSMacOS) // #nosec G404 + if info.Name != "macOS" { + t.Errorf("macOS name = %q, want macOS", info.Name) + } + // Description is "macOS ()". + want := "macOS " + info.Version + " (" + info.BuildID + ")" + if info.Description != want { + t.Errorf("macOS description = %q, want %q", info.Description, want) + } +} + +func TestPickHalfIndex(t *testing.T) { + r := rand.New(rand.NewSource(1)) // #nosec G404 + for i := 0; i < 30; i++ { + if idx := pickHalfIndex(r, 10, true); idx < 0 || idx >= 5 { + t.Fatalf("older index %d out of [0,5)", idx) + } + if idx := pickHalfIndex(r, 10, false); idx < 5 || idx >= 10 { + t.Fatalf("newer index %d out of [5,10)", idx) + } + } + // Single-element pool: the guard avoids Intn(0); both halves resolve to 0. + if pickHalfIndex(r, 1, true) != 0 { + t.Error("n=1 older should be 0") + } + if pickHalfIndex(r, 1, false) != 0 { + t.Error("n=1 newer should be 0") + } +} + +func TestGenerateHostID(t *testing.T) { + linuxRE := regexp.MustCompile(`^[0-9a-f]{32}$`) + uuidLowerRE := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`) + uuidUpperRE := regexp.MustCompile(`^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$`) + + linux := GenerateHostID(rand.New(rand.NewSource(1)), OSLinux) // #nosec G404 + if !linuxRE.MatchString(linux) { + t.Errorf("Linux host.id = %q, want 32-char lowercase hex", linux) + } + win := GenerateHostID(rand.New(rand.NewSource(1)), OSWindows) // #nosec G404 + if !uuidLowerRE.MatchString(win) { + t.Errorf("Windows host.id = %q, want GUID", win) + } + mac := GenerateHostID(rand.New(rand.NewSource(1)), OSMacOS) // #nosec G404 + if !uuidUpperRE.MatchString(mac) { + t.Errorf("macOS host.id = %q, want uppercase UUID", mac) + } + + // Deterministic. + if GenerateHostID(rand.New(rand.NewSource(9)), OSLinux) != GenerateHostID(rand.New(rand.NewSource(9)), OSLinux) { // #nosec G404 + t.Error("GenerateHostID not deterministic") + } +} diff --git a/internal/datagen/systems.go b/internal/datagen/systems.go index 05089dd..e73e095 100644 --- a/internal/datagen/systems.go +++ b/internal/datagen/systems.go @@ -19,12 +19,31 @@ const ( // Arch represents a CPU architecture. type Arch string +// Arch values are the OpenTelemetry semconv host.arch value set. Random system +// generation uses the common ones (amd64, arm64); the rest are selectable via +// explicit configuration (e.g. an s390x mainframe or ppc64 host). const ( ArchAMD64 Arch = "amd64" + ArchARM32 Arch = "arm32" ArchARM64 Arch = "arm64" + ArchIA64 Arch = "ia64" + ArchPPC32 Arch = "ppc32" + ArchPPC64 Arch = "ppc64" + ArchS390X Arch = "s390x" ArchX86 Arch = "x86" ) +// ParseArch maps a user-supplied CPU architecture string to an Arch, accepting +// the OpenTelemetry semconv host.arch value set. Unknown values return an error. +func ParseArch(s string) (Arch, error) { + switch a := Arch(strings.ToLower(strings.TrimSpace(s))); a { + case ArchAMD64, ArchARM32, ArchARM64, ArchIA64, ArchPPC32, ArchPPC64, ArchS390X, ArchX86: + return a, nil + default: + return "", fmt.Errorf("datagen: unsupported architecture %q", s) + } +} + // SystemRole represents a machine's role in the environment. type SystemRole string @@ -35,41 +54,17 @@ const ( RoleRouter SystemRole = "router" ) -// OS version pools. -var ( - LinuxVersions = NewPool( - "5.15.0-91-generic", // Ubuntu 22.04 - "6.1.0-18-amd64", // Debian 12 - "5.14.0-362.el9", // RHEL 9 - "6.6.9-200.fc39", // Fedora 39 - "5.10.0-27-amd64", // Debian 11 - ) - - WindowsVersions = NewPool( - "10.0.20348", // Server 2022 - "10.0.17763", // Server 2019 - "10.0.14393", // Server 2016 - "10.0.22631", // Windows 11 23H2 - "10.0.19045", // Windows 10 22H2 - ) - - MacOSVersions = NewPool( - "14.2.1", // Sonoma - "13.6.3", // Ventura - "12.7.2", // Monterey - ) -) - // SystemIdentity represents a machine in the simulated environment. type SystemIdentity struct { - Hostname string - FQDN string // hostname + domain - OS OSType - OSVersion string - Arch Arch - Role SystemRole - Domain string // back-reference to DomainIdentity.Name - OUPath string // "OU=Servers,DC=contoso,DC=com" + Hostname string + FQDN string // hostname + domain + OSInfo OSInfo // os.type / os.name / os.version / os.build_id / os.description + HostID string // host.id (OS-appropriate machine identifier) + Arch Arch + Role SystemRole + Tier DeploymentTier // deployment.environment.name + Domain string // back-reference to DomainIdentity.Name + OUPath string // "OU=Servers,DC=contoso,DC=com" // Hardware CPUCores int @@ -82,11 +77,26 @@ type SystemIdentity struct { // TLS cert issued by the domain CA Cert *CertInfo + // Image is the host's OS/VM image provenance (host.image.*). It is a + // framework hook for a future CloudIdentity source and is nil today; core + // system generation never populates it. A nil Image means the resource + // projection emits no host.image.* attributes. + Image *HostImage + // Sub-identities (populated by environment generation) Services []*ServiceIdentity Applications []*ApplicationIdentity } +// HostImage is the OpenTelemetry host.image.* projection source: the VM image +// or OS install a host was instantiated from. It is not populated by core +// system generation — a future CloudIdentity source will set it on cloud hosts. +type HostImage struct { + ID string // host.image.id + Name string // host.image.name + Version string // host.image.version +} + // NetworkInterface represents a NIC bound to a network subnet. type NetworkInterface struct { Name string // "eth0", "Ethernet0" @@ -139,16 +149,9 @@ func GenerateSystemIdentity(r *rand.Rand, os OSType, role SystemRole, domain *Do fqdn = strings.ToLower(hostname) + "." + domain.Name } - // Pick OS version - var osVersion string - switch os { - case OSLinux: - osVersion = LinuxVersions.Random(r) - case OSWindows: - osVersion = WindowsVersions.Random(r) - case OSMacOS: - osVersion = MacOSVersions.Random(r) - } + // Coherent OS release + machine id, from authentic release data. + osInfo := GenerateOSInfo(r, os) + hostID := GenerateHostID(r, os) // Pick arch arch := ArchAMD64 @@ -166,18 +169,18 @@ func GenerateSystemIdentity(r *rand.Rand, os OSType, role SystemRole, domain *Do cert := generateCertInfo(r, fqdn, hostname, domain.CA) return &SystemIdentity{ - Hostname: hostname, - FQDN: fqdn, - OS: os, - OSVersion: osVersion, - Arch: arch, - Role: role, - Domain: domain.Name, - OUPath: ouPath, - CPUCores: cpu, - MemoryMB: mem, - DiskGB: disk, - Cert: cert, + Hostname: hostname, + FQDN: fqdn, + OSInfo: osInfo, + HostID: hostID, + Arch: arch, + Role: role, + Domain: domain.Name, + OUPath: ouPath, + CPUCores: cpu, + MemoryMB: mem, + DiskGB: disk, + Cert: cert, }, nil } diff --git a/internal/datagen/systems_test.go b/internal/datagen/systems_test.go index 4a6d851..2825eef 100644 --- a/internal/datagen/systems_test.go +++ b/internal/datagen/systems_test.go @@ -34,15 +34,15 @@ func TestSystemRoles(t *testing.T) { } } -func TestVersionPools(t *testing.T) { - if LinuxVersions.Len() < 3 { - t.Errorf("LinuxVersions has %d items, want at least 3", LinuxVersions.Len()) +func TestReleasePools(t *testing.T) { + if len(linuxReleases) < 3 { + t.Errorf("linuxReleases has %d items, want at least 3", len(linuxReleases)) } - if WindowsVersions.Len() < 3 { - t.Errorf("WindowsVersions has %d items, want at least 3", WindowsVersions.Len()) + if len(windowsReleases) < 3 { + t.Errorf("windowsReleases has %d items, want at least 3", len(windowsReleases)) } - if MacOSVersions.Len() < 3 { - t.Errorf("MacOSVersions has %d items, want at least 3", MacOSVersions.Len()) + if len(macReleases) < 3 { + t.Errorf("macReleases has %d items, want at least 3", len(macReleases)) } } @@ -55,8 +55,8 @@ func TestGenerateSystemIdentity(t *testing.T) { if err != nil { t.Fatalf("GenerateSystemIdentity: %v", err) } - if sys.OS != OSLinux { - t.Errorf("expected OS %q, got %q", OSLinux, sys.OS) + if sys.OSInfo.Type != OSLinux { + t.Errorf("expected OS %q, got %q", OSLinux, sys.OSInfo.Type) } if sys.Role != RoleServer { t.Errorf("expected Role %q, got %q", RoleServer, sys.Role) @@ -80,8 +80,8 @@ func TestGenerateSystemIdentity(t *testing.T) { if err != nil { t.Fatalf("GenerateSystemIdentity: %v", err) } - if sys.OS != OSWindows { - t.Errorf("expected OS %q, got %q", OSWindows, sys.OS) + if sys.OSInfo.Type != OSWindows { + t.Errorf("expected OS %q, got %q", OSWindows, sys.OSInfo.Type) } if sys.Hostname != strings.ToUpper(sys.Hostname) { t.Errorf("windows hostname %q should be uppercase", sys.Hostname) @@ -131,6 +131,28 @@ func TestGenerateSystemIdentity(t *testing.T) { }) } +func TestHostImageFrameworkHook(t *testing.T) { + // The Image hook exists so a future CloudIdentity source can populate + // host.image.* on a system, but it is unwired today: GenerateSystemIdentity + // must leave it nil so the projection emits no host.image.* attributes. + r := rand.New(rand.NewSource(7)) + domain := GenerateDomainIdentity(7, "contoso.com", time.Now()) + + sys, err := GenerateSystemIdentity(r, OSLinux, RoleServer, domain, NorseNames) + if err != nil { + t.Fatalf("GenerateSystemIdentity: %v", err) + } + if sys.Image != nil { + t.Errorf("expected Image nil (unwired framework hook), got %+v", sys.Image) + } + + // HostImage carries the OTel host.image.* semconv fields. + img := &HostImage{ID: "ami-0abc", Name: "ubuntu-22.04", Version: "20240115"} + if img.ID != "ami-0abc" || img.Name != "ubuntu-22.04" || img.Version != "20240115" { + t.Errorf("HostImage fields did not round-trip: %+v", img) + } +} + func TestGenerateSystemIdentityErrorsOnNilInputs(t *testing.T) { r := rand.New(rand.NewSource(1)) domain := GenerateDomainIdentity(1, "", time.Now()) diff --git a/internal/datagen/tiers_test.go b/internal/datagen/tiers_test.go new file mode 100644 index 0000000..c1b2c46 --- /dev/null +++ b/internal/datagen/tiers_test.go @@ -0,0 +1,52 @@ +package datagen + +import ( + "testing" + "time" + + "go.uber.org/zap" +) + +func TestGenerateSystems_Tiers(t *testing.T) { + seeds := NewSeedConfig() + seeds.Shared = 42 + seeds.Init(zap.NewNop()) + + env, err := GenerateEnvironment(seeds, &EnvironmentOpts{ + SystemCount: 300, + Now: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + }) + if err != nil { + t.Fatalf("GenerateEnvironment: %v", err) + } + + counts := map[DeploymentTier]int{} + prodWindows := map[OSInfo]struct{}{} + for _, s := range env.Systems { + if s.Tier == "" { + t.Fatal("system has no deployment tier assigned") + } + counts[s.Tier]++ + if s.Tier == TierProd && s.OSInfo.Type == OSWindows { + prodWindows[s.OSInfo] = struct{}{} + } + } + + // All four tiers appear in a fleet of 300. + for _, tr := range DeploymentTiers { + if counts[tr] == 0 { + t.Errorf("tier %q never assigned", tr) + } + } + // Prod is the plurality (55% vs 15%). + if counts[TierProd] <= counts[TierStaging] { + t.Errorf("prod (%d) should dominate staging (%d)", counts[TierProd], counts[TierStaging]) + } + // Prod Windows hosts are pinned to a single OS release (uniform fleet). + if len(prodWindows) > 1 { + t.Errorf("prod Windows hosts should share one release, got %d distinct", len(prodWindows)) + } + if len(prodWindows) == 0 { + t.Error("expected at least one prod Windows host in a fleet of 300") + } +} diff --git a/internal/dispatch/embed.go b/internal/dispatch/embed.go index 557cc1d..6d2ef16 100644 --- a/internal/dispatch/embed.go +++ b/internal/dispatch/embed.go @@ -27,6 +27,7 @@ import ( "github.com/observiq/blitz/generator/wel" welcatalog "github.com/observiq/blitz/generator/wel/catalog" "github.com/observiq/blitz/internal/config" + "github.com/observiq/blitz/internal/datagen" "go.uber.org/zap" ) @@ -71,11 +72,16 @@ func (c EmbedConsumers) requireTrace(typ config.GeneratorType) error { // the snapshot shipped in the blitz module, or nil to fall back to // reading ./data_library/ from the process cwd. // +// env is the simulated identity environment (PIPE-1036). When non-nil, the +// generator's host identity is resolved from it (see hostIdentity) so emitted +// records carry the simulated host's host.* / os.* / deployment.* attributes; +// when nil, generators fall back to the running process's hostname. +// // Returns an error when the configured generator type requires a // consumer that is nil in `consumers` (e.g. hostmetrics without a // MetricConsumer, traces without a TraceConsumer), and for generator // types that are not embed-eligible at all (nop, winevt — see PIPE-1032). -func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsumers, fileGenLibrary fs.FS) (embed.ProducerModule, error) { +func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsumers, fileGenLibrary fs.FS, env *datagen.Environment) (embed.ProducerModule, error) { if logger == nil { return nil, fmt.Errorf("logger cannot be nil") } @@ -84,52 +90,62 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return jsongen.New(logger, genCfg.JSON.Workers, genCfg.JSON.Rate, genCfg.JSON.Type, consumers.LogConsumer) + mod, err := jsongen.New(logger, genCfg.JSON.Workers, genCfg.JSON.Rate, genCfg.JSON.Type, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypePaloAlto: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return paloalto.New(logger, genCfg.PaloAlto.Workers, genCfg.PaloAlto.Rate, consumers.LogConsumer) + mod, err := paloalto.New(logger, genCfg.PaloAlto.Workers, genCfg.PaloAlto.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeApache: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return apachegen.New(logger, genCfg.Apache.Workers, genCfg.Apache.Rate, consumers.LogConsumer) + mod, err := apachegen.New(logger, genCfg.Apache.Workers, genCfg.Apache.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeApacheCombined: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return apachecombinedgen.New(logger, genCfg.ApacheCombined.Workers, genCfg.ApacheCombined.Rate, consumers.LogConsumer) + mod, err := apachecombinedgen.New(logger, genCfg.ApacheCombined.Workers, genCfg.ApacheCombined.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeApacheError: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return apacheerrorgen.New(logger, genCfg.ApacheError.Workers, genCfg.ApacheError.Rate, consumers.LogConsumer) + mod, err := apacheerrorgen.New(logger, genCfg.ApacheError.Workers, genCfg.ApacheError.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeNginx: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return nginx.New(logger, genCfg.Nginx.Workers, genCfg.Nginx.Rate, consumers.LogConsumer) + mod, err := nginx.New(logger, genCfg.Nginx.Workers, genCfg.Nginx.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypePostgres: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return postgres.New(logger, genCfg.Postgres.Workers, genCfg.Postgres.Rate, consumers.LogConsumer) + mod, err := postgres.New(logger, genCfg.Postgres.Workers, genCfg.Postgres.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeKubernetes: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return kubernetes.New(logger, genCfg.Kubernetes.Workers, genCfg.Kubernetes.Rate, genCfg.Kubernetes.Format, consumers.LogConsumer) + mod, err := kubernetes.New(logger, genCfg.Kubernetes.Workers, genCfg.Kubernetes.Rate, genCfg.Kubernetes.Format, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeFile: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return filegen.New(logger, genCfg.Filegen.Workers, genCfg.Filegen.Rate, genCfg.Filegen.Source, genCfg.Filegen.CacheEnabled, genCfg.Filegen.CacheTTL, consumers.LogConsumer, fileGenLibrary) + mod, err := filegen.New(logger, genCfg.Filegen.Workers, genCfg.Filegen.Rate, genCfg.Filegen.Source, genCfg.Filegen.CacheEnabled, genCfg.Filegen.CacheTTL, consumers.LogConsumer, fileGenLibrary) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeOkta: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return okta.New(logger, genCfg.Okta.Workers, genCfg.Okta.Rate, consumers.LogConsumer) + mod, err := okta.New(logger, genCfg.Okta.Workers, genCfg.Okta.Rate, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeWel: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err @@ -138,7 +154,7 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum if role == "" { role = welcatalog.RoleMember } - return wel.New(wel.Config{ + mod, err := wel.New(wel.Config{ Logger: logger, Workers: genCfg.Wel.Workers, Rate: genCfg.Wel.Rate, @@ -148,11 +164,13 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum Channels: genCfg.Wel.Channels, Consumer: consumers.LogConsumer, }) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeFIX: if err := consumers.requireLog(genCfg.Type); err != nil { return nil, err } - return newFIX(logger, genCfg.FIX, consumers.LogConsumer) + mod, err := newFIX(logger, genCfg.FIX, consumers.LogConsumer) + return applyHostIdentity(mod, err, env, genCfg.Type) case config.GeneratorTypeHostMetrics: if err := consumers.requireMetric(genCfg.Type); err != nil { return nil, err @@ -166,6 +184,7 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum ScraperNames: genCfg.HostMetrics.Scrapers, Consumer: consumers.MetricConsumer, Seed: yamlSeedDefault(genCfg.HostMetrics.Seed), + Identity: hostIdentity(env, genCfg.Type), }) case config.GeneratorTypeTraces: if err := consumers.requireTrace(genCfg.Type); err != nil { @@ -178,6 +197,7 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum Hostname: genCfg.Traces.Hostname, Consumer: consumers.TraceConsumer, Seed: yamlSeedDefault(genCfg.Traces.Seed), + Identity: hostIdentity(env, genCfg.Type), }) case config.GeneratorTypeNop: return nil, fmt.Errorf("generator type %q does not produce records; not embed-eligible", genCfg.Type) @@ -188,6 +208,44 @@ func ForEmbed(logger *zap.Logger, genCfg config.Generator, consumers EmbedConsum } } +// hostIdentitySetter is implemented by the log generators, whose positional +// constructors take identity after construction (via SetHostIdentity) rather +// than as a constructor argument. Metric- and trace-yielding generators take +// their identity as a Config field instead, so they do not implement this. +type hostIdentitySetter interface { + SetHostIdentity(*datagen.SystemIdentity) +} + +// applyHostIdentity resolves the component's simulated host from env and applies +// it to a just-constructed log generator, then returns it for the ForEmbed case +// to hand back. It is a no-op when construction failed (err != nil) or when the +// module does not accept a post-construction identity. +func applyHostIdentity(mod embed.ProducerModule, err error, env *datagen.Environment, component config.GeneratorType) (embed.ProducerModule, error) { + if err != nil { + return nil, err + } + if setter, ok := mod.(hostIdentitySetter); ok { + setter.SetHostIdentity(hostIdentity(env, component)) + } + return mod, nil +} + +// hostIdentity resolves the simulated host a generator component's records +// describe: the environment's deterministic SystemForKey selection keyed by the +// generator type, so the same component always maps to the same host and +// distinct components spread across the fleet. Returns nil when no environment +// is configured, leaving the generator on its process-hostname fallback. +// +// Keying by component gives one host per generator (the default granularity). +// Finer per-worker granularity — one host per worker — is a future opt-in that +// would key SystemForKey by component plus worker index. +func hostIdentity(env *datagen.Environment, component config.GeneratorType) *datagen.SystemIdentity { + if env == nil { + return nil + } + return env.SystemForKey(string(component)) +} + // yamlSeedDefault translates a YAML-loaded Seed value into the // generator-Config Seed value, applying the "stochastic by default" // architectural intent for YAML users. YAML zero-value (omitted `seed:` diff --git a/internal/dispatch/embed_test.go b/internal/dispatch/embed_test.go index fa27c70..b893fdc 100644 --- a/internal/dispatch/embed_test.go +++ b/internal/dispatch/embed_test.go @@ -2,6 +2,7 @@ package dispatch import ( "context" + "sync" "testing" "time" @@ -11,6 +12,7 @@ import ( "github.com/observiq/blitz/embed" "github.com/observiq/blitz/internal/config" + "github.com/observiq/blitz/internal/datagen" ) type noopConsumer struct{} @@ -31,6 +33,143 @@ func logsOnly() EmbedConsumers { return EmbedConsumers{LogConsumer: noopConsumer{}} } +// capturingMetricConsumer records emitted points so a test can assert on the +// resource attributes the generator attached. +type capturingMetricConsumer struct { + mu sync.Mutex + points []embed.MetricPoint +} + +func (c *capturingMetricConsumer) ConsumeMetrics(_ context.Context, batch []embed.MetricPoint) error { + c.mu.Lock() + defer c.mu.Unlock() + c.points = append(c.points, batch...) + return nil +} + +func (c *capturingMetricConsumer) snapshot() []embed.MetricPoint { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]embed.MetricPoint, len(c.points)) + copy(out, c.points) + return out +} + +// capturingLogConsumer records emitted log records for resource assertions. +type capturingLogConsumer struct { + mu sync.Mutex + records []embed.LogRecord +} + +func (c *capturingLogConsumer) ConsumeLogs(_ context.Context, batch []embed.LogRecord) error { + c.mu.Lock() + defer c.mu.Unlock() + c.records = append(c.records, batch...) + return nil +} + +func (c *capturingLogConsumer) snapshot() []embed.LogRecord { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]embed.LogRecord, len(c.records)) + copy(out, c.records) + return out +} + +// TestForEmbedLogGeneratorWiresEnvironmentIdentity proves the setter path: a log +// generator built through ForEmbed with an environment has its host identity +// applied (via SetHostIdentity), so emitted records carry the simulated host. +func TestForEmbedLogGeneratorWiresEnvironmentIdentity(t *testing.T) { + env := &datagen.Environment{ + Systems: []*datagen.SystemIdentity{{ + Hostname: "PANTHEON-LOG-01", + OSInfo: datagen.OSInfo{Type: datagen.OSLinux, Name: "Ubuntu"}, + }}, + } + cons := &capturingLogConsumer{} + cfg := config.Generator{ + Type: config.GeneratorTypeNginx, + Nginx: config.NginxGeneratorConfig{Workers: 1, Rate: 20 * time.Millisecond}, + } + + mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{LogConsumer: cons}, nil, env) + require.NoError(t, err) + require.NoError(t, mod.Start(context.Background())) + require.Eventually(t, func() bool { return len(cons.snapshot()) > 0 }, 2*time.Second, 10*time.Millisecond) + require.NoError(t, mod.Stop(context.Background())) + + recs := cons.snapshot() + require.NotEmpty(t, recs) + assert.Equal(t, "PANTHEON-LOG-01", recs[0].Metadata.Resource["host.name"]) + assert.Equal(t, "nginx", recs[0].Metadata.Resource["telemetry.source"]) +} + +// TestForEmbedPropagatesConstructorError confirms applyHostIdentity forwards a +// constructor error (here nginx.New rejecting Workers=0) rather than trying to +// apply an identity to a nil module. +func TestForEmbedPropagatesConstructorError(t *testing.T) { + cfg := config.Generator{ + Type: config.GeneratorTypeNginx, + Nginx: config.NginxGeneratorConfig{Workers: 0, Rate: time.Second}, + } + mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{LogConsumer: noopConsumer{}}, nil, nil) + require.Error(t, err) + require.Nil(t, mod) +} + +// TestHostIdentityResolvesFromEnvironment covers the component-keyed identity +// resolution: a nil environment yields nil (process-hostname fallback), and a +// populated environment returns a deterministic SystemForKey selection. +func TestHostIdentityResolvesFromEnvironment(t *testing.T) { + assert.Nil(t, hostIdentity(nil, config.GeneratorTypeHostMetrics)) + + env := &datagen.Environment{ + Systems: []*datagen.SystemIdentity{ + {Hostname: "PANTHEON-01", OSInfo: datagen.OSInfo{Type: datagen.OSLinux}}, + }, + } + got := hostIdentity(env, config.GeneratorTypeHostMetrics) + require.NotNil(t, got) + assert.Equal(t, "PANTHEON-01", got.Hostname) +} + +// TestForEmbedHostMetricsWiresEnvironmentIdentity proves the full wiring: a +// hostmetrics module built through ForEmbed with an environment emits points +// carrying the resolved simulated host's identity attributes. +func TestForEmbedHostMetricsWiresEnvironmentIdentity(t *testing.T) { + env := &datagen.Environment{ + Systems: []*datagen.SystemIdentity{{ + Hostname: "PANTHEON-01", + HostID: "id-1", + Arch: datagen.ArchAMD64, + Tier: datagen.TierProd, + OSInfo: datagen.OSInfo{Type: datagen.OSLinux, Name: "Ubuntu", Version: "22.04.5"}, + }}, + } + cons := &capturingMetricConsumer{} + cfg := config.Generator{ + Type: config.GeneratorTypeHostMetrics, + HostMetrics: config.HostMetricsGeneratorConfig{ + Workers: 1, + Rate: 20 * time.Millisecond, + Scrapers: []string{"cpu"}, + }, + } + + mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{MetricConsumer: cons}, nil, env) + require.NoError(t, err) + require.NoError(t, mod.Start(context.Background())) + require.Eventually(t, func() bool { return len(cons.snapshot()) > 0 }, 2*time.Second, 10*time.Millisecond) + require.NoError(t, mod.Stop(context.Background())) + + pts := cons.snapshot() + require.NotEmpty(t, pts) + res := pts[0].Metadata.Resource + assert.Equal(t, "PANTHEON-01", res["host.name"]) + assert.Equal(t, "linux", res["os.type"]) + assert.Equal(t, "production", res["deployment.environment.name"]) +} + func TestForEmbedWelReturnsProducerModule(t *testing.T) { cfg := config.Generator{ Type: config.GeneratorTypeWel, @@ -40,7 +179,7 @@ func TestForEmbedWelReturnsProducerModule(t *testing.T) { Role: "member", }, } - mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.NoError(t, err) require.NotNil(t, mod) assert.Equal(t, "wel", mod.Name()) @@ -54,14 +193,14 @@ func TestForEmbedWelDefaultsEmptyRole(t *testing.T) { Rate: 50 * time.Millisecond, }, } - mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.NoError(t, err) require.NotNil(t, mod) } func TestForEmbedWinevtRejectionMentionsWel(t *testing.T) { cfg := config.Generator{Type: config.GeneratorTypeWinevt} - _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "DEPRECATED") assert.Contains(t, err.Error(), "`wel` generator") @@ -77,7 +216,7 @@ func TestForEmbedFIXReturnsProducerModule(t *testing.T) { Version: "4.4", }, } - mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + mod, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.NoError(t, err) require.NotNil(t, mod) assert.Equal(t, "fix", mod.Name()) @@ -92,7 +231,7 @@ func TestForEmbedFIXRejectsUnknownVersion(t *testing.T) { Version: "4.3", }, } - _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "unknown version") } @@ -106,18 +245,18 @@ func TestForEmbedFIXRejectsUnknownCategory(t *testing.T) { EnabledCategories: []string{"crypto"}, }, } - _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil) + _, err := ForEmbed(zap.NewNop(), cfg, logsOnly(), nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "unknown asset category") } func TestForEmbedRejectsNilLogger(t *testing.T) { - _, err := ForEmbed(nil, config.Generator{Type: config.GeneratorTypeFIX}, logsOnly(), nil) + _, err := ForEmbed(nil, config.Generator{Type: config.GeneratorTypeFIX}, logsOnly(), nil, nil) require.Error(t, err) } func TestForEmbedRejectsMissingLogConsumerForLogType(t *testing.T) { - _, err := ForEmbed(zap.NewNop(), config.Generator{Type: config.GeneratorTypeFIX}, EmbedConsumers{}, nil) + _, err := ForEmbed(zap.NewNop(), config.Generator{Type: config.GeneratorTypeFIX}, EmbedConsumers{}, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "LogConsumer") } @@ -133,7 +272,7 @@ func TestForEmbedHostMetricsReturnsProducerModule(t *testing.T) { Hostname: "test-host", }, } - mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{MetricConsumer: noopMetricConsumer{}}, nil) + mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{MetricConsumer: noopMetricConsumer{}}, nil, nil) require.NoError(t, err) require.NotNil(t, mod) assert.Equal(t, "hostmetrics", mod.Name()) @@ -148,7 +287,7 @@ func TestForEmbedHostMetricsRejectsMissingMetricConsumer(t *testing.T) { OS: "linux", }, } - _, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{}, nil) + _, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{}, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "MetricConsumer") } @@ -162,7 +301,7 @@ func TestForEmbedTracesReturnsProducerModule(t *testing.T) { Rate: 50 * time.Millisecond, }, } - mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{TraceConsumer: noopTraceConsumer{}}, nil) + mod, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{TraceConsumer: noopTraceConsumer{}}, nil, nil) require.NoError(t, err) require.NotNil(t, mod) assert.Equal(t, "traces", mod.Name()) @@ -176,7 +315,7 @@ func TestForEmbedTracesRejectsMissingTraceConsumer(t *testing.T) { Rate: time.Second, }, } - _, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{}, nil) + _, err := ForEmbed(zap.NewNop(), cfg, EmbedConsumers{}, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "TraceConsumer") }