feat(generators): emit simulated host identity on every record (PIPE-1036)

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
 	}

config/loader.go

@@ -149,14 +149,22 @@ 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,
 		TraceConsumer:  opts.TraceConsumer,
 	}
 	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)
 		}

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:

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.

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,13 +70,22 @@ 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
 }
 
 // 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
 }

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"])
+}