feat(datagen): environment foundations for simulated host identity (PIPE-1036)

embed/host.go

@@ -1,6 +1,10 @@
 package embed
 
-import "go.uber.org/zap"
+import (
+	"go.uber.org/zap"
+
+	"github.com/observiq/blitz/internal/datagen"
+)
 
 // Host is the bundle of consumers and ambient resources a host process
 // supplies to an embedded blitz runner.
@@ -31,6 +35,21 @@ type Host struct {
 	// treat their own reference as frozen once they hand the Host off.
 	// See cloneResource in this package.
 	Resource map[string]string
+
+	// Environment is the simulated datagen.Environment that generators draw
+	// their host identities (host.name / os.type) from (PIPE-1036). Nil means
+	// generators fall back to the running host's os.Hostname().
+	//
+	// Read-only for the lifetime of a single Runner.Start: workers only read
+	// it, so they share the pointer without synchronization. It is not
+	// deep-copied the way Resource is; copying the whole identity graph would
+	// be costly and buys nothing, since the Environment is never mutated in
+	// place. Reconfiguration swaps it by rebuilding the runner with a fresh
+	// Host (Stop, New, Start), not by mutating the live value, so a caller
+	// replaces Environments across a rebuild rather than underneath running
+	// workers. Callers must treat their reference as read-only once they hand
+	// the Host off.
+	Environment *datagen.Environment
 }
 
 // cloneResource returns a defensive copy of m. Runner.Start uses it so

generator/resource/resource.go

@@ -51,12 +51,67 @@ func Hostname() string {
 //	resource.Default("apache", "apache.format", "common")
 //	// → {"host.name": "<host>", "telemetry.source": "apache", "apache.format": "common"}
 func Default(source string, extras ...string) map[string]string {
+	return WithHost(Hostname(), source, extras...)
+}
+
+// WithHost returns a Resource map like Default but with an explicit host.name,
+// for generators whose hostname comes from a resolved datagen SystemIdentity
+// (PIPE-1036) rather than the process's os.Hostname(). extras follow the same
+// key/value convention as Default.
+func WithHost(hostname, source string, extras ...string) map[string]string {
 	r := map[string]string{
-		"host.name":        Hostname(),
+		"host.name":        hostname,
 		"telemetry.source": source,
 	}
 	for i := 0; i+1 < len(extras); i += 2 {
 		r[extras[i]] = extras[i+1]
 	}
 	return r
 }
+
+// StaticResources is an immutable set of resource attributes that stay constant
+// for a generator worker's lifetime: the host-identity projection (host.name,
+// os.type, ...) plus per-generator constants (telemetry.source, format flavor,
+// version). Build it once at construction and reuse it for every record the
+// worker emits (PIPE-1036).
+//
+// The model is: Static + Dynamic (per record) = Record. Static carries the
+// fields that never change for the worker; Record merges in the few that do.
+type StaticResources struct {
+	attrs map[string]string
+}
+
+// NewStaticResources builds a StaticResources from a base attribute set. The
+// map is copied, so a caller that retains and later mutates attrs does not
+// affect the constructed value.
+func NewStaticResources(attrs map[string]string) *StaticResources {
+	cp := make(map[string]string, len(attrs))
+	for k, v := range attrs {
+		cp[k] = v
+	}
+	return &StaticResources{attrs: cp}
+}
+
+// Record returns the resource attributes for a single emitted record: the
+// static set merged with the given dynamic key/value pairs (same even-length
+// convention as Default's extras).
+//
+// When no dynamic pairs are supplied — the common case, since most generators
+// vary nothing per record — Record returns the shared static map with no
+// allocation. Callers MUST treat that returned map as read-only; mutating it
+// corrupts every other record and races concurrent workers. When dynamic pairs
+// are supplied, Record returns a fresh merged map that is safe to mutate and
+// leaves the static set untouched.
+func (s *StaticResources) Record(dynamicKV ...string) map[string]string {
+	if len(dynamicKV) < 2 {
+		return s.attrs
+	}
+	out := make(map[string]string, len(s.attrs)+len(dynamicKV)/2)
+	for k, v := range s.attrs {
+		out[k] = v
+	}
+	for i := 0; i+1 < len(dynamicKV); i += 2 {
+		out[dynamicKV[i]] = dynamicKV[i+1]
+	}
+	return out
+}

generator/resource/resource_static_test.go

@@ -0,0 +1,76 @@
+package resource
+
+import (
+	"reflect"
+	"testing"
+)
+
+func mapPtr(m map[string]string) uintptr { return reflect.ValueOf(m).Pointer() }
+
+func TestStaticResourcesConstructorCopies(t *testing.T) {
+	base := map[string]string{"host.name": "thor-web-01", "telemetry.source": "apache"}
+	s := NewStaticResources(base)
+
+	// Mutating the input after construction must not leak into the static set.
+	base["host.name"] = "mutated"
+	base["injected"] = "x"
+
+	got := s.Record()
+	if got["host.name"] != "thor-web-01" {
+		t.Errorf("host.name = %q, want thor-web-01 (constructor must copy)", got["host.name"])
+	}
+	if _, ok := got["injected"]; ok {
+		t.Error("post-construction input mutation leaked into the static set")
+	}
+}
+
+func TestStaticResourcesRecordNoDynamicIsSharedAndReadOnly(t *testing.T) {
+	s := NewStaticResources(map[string]string{"host.name": "thor-web-01", "telemetry.source": "apache"})
+
+	a := s.Record()
+	b := s.Record()
+
+	// Zero-allocation path: repeated no-dynamic calls return the SAME map.
+	if mapPtr(a) != mapPtr(b) {
+		t.Error("Record() with no dynamic pairs should return the shared static map, not a fresh copy")
+	}
+	if a["telemetry.source"] != "apache" {
+		t.Errorf("telemetry.source = %q, want apache", a["telemetry.source"])
+	}
+}
+
+func TestStaticResourcesRecordWithDynamicMergesWithoutMutatingStatic(t *testing.T) {
+	s := NewStaticResources(map[string]string{"host.name": "thor-web-01", "telemetry.source": "wel"})
+
+	rec := s.Record("wel.channel", "Security", "wel.role", "dc")
+
+	// Merged map carries both static and dynamic.
+	if rec["host.name"] != "thor-web-01" || rec["wel.channel"] != "Security" || rec["wel.role"] != "dc" {
+		t.Errorf("merged record missing expected keys: %#v", rec)
+	}
+	// It must be a distinct map from the shared static one.
+	if mapPtr(rec) == mapPtr(s.Record()) {
+		t.Error("Record(dynamic...) must return a fresh map, not the shared static map")
+	}
+	// The static set must be untouched by the merge.
+	if _, ok := s.Record()["wel.channel"]; ok {
+		t.Error("dynamic pair leaked into the static set")
+	}
+}
+
+func TestStaticResourcesRecordOddArgsIgnoresTrailing(t *testing.T) {
+	s := NewStaticResources(map[string]string{"telemetry.source": "json"})
+
+	// A single unpaired arg is treated as "no complete dynamic pair": shared static.
+	if mapPtr(s.Record("dangling")) != mapPtr(s.Record()) {
+		t.Error("a single unpaired dynamic arg should yield the shared static map")
+	}
+	// An odd count keeps complete pairs and drops the trailing unpaired key.
+	rec := s.Record("json.type", "pii", "dangling")
+	if rec["json.type"] != "pii" {
+		t.Errorf("json.type = %q, want pii", rec["json.type"])
+	}
+	if _, ok := rec["dangling"]; ok {
+		t.Error("trailing unpaired key should be dropped")
+	}
+}

generator/resource/resource_withhost_test.go

@@ -0,0 +1,36 @@
+package resource
+
+import "testing"
+
+func TestWithHost(t *testing.T) {
+	r := WithHost("web-01", "apache", "apache.format", "common")
+	if r["host.name"] != "web-01" {
+		t.Errorf(`host.name = %q, want "web-01"`, r["host.name"])
+	}
+	if r["telemetry.source"] != "apache" {
+		t.Errorf(`telemetry.source = %q, want "apache"`, r["telemetry.source"])
+	}
+	if r["apache.format"] != "common" {
+		t.Errorf(`apache.format = %q, want "common"`, r["apache.format"])
+	}
+}
+
+func TestWithHost_OddExtrasIgnoresDangling(t *testing.T) {
+	r := WithHost("h", "src", "onlykey")
+	if _, ok := r["onlykey"]; ok {
+		t.Error("dangling extra key should be ignored")
+	}
+	if r["host.name"] != "h" {
+		t.Errorf("host.name = %q, want h", r["host.name"])
+	}
+}
+
+func TestDefaultUsesProcessHostname(t *testing.T) {
+	r := Default("apache")
+	if r["host.name"] != Hostname() {
+		t.Errorf("Default host.name = %q, want process hostname %q", r["host.name"], Hostname())
+	}
+	if r["telemetry.source"] != "apache" {
+		t.Errorf("telemetry.source = %q, want apache", r["telemetry.source"])
+	}
+}

internal/config/config.go

@@ -19,6 +19,9 @@ type Config struct {
 	Output Output `yaml:"output,omitempty" mapstructure:"output,omitempty"`
 	// Metrics configuration
 	Metrics Metrics `yaml:"metrics,omitempty" mapstructure:"metrics,omitempty"`
+	// Environment configures the simulated datagen.Environment identities
+	// that generators draw their host.name/OS from (PIPE-1036).
+	Environment EnvironmentConfig `yaml:"environment,omitempty" mapstructure:"environment,omitempty"`
 	// OnFinish controls behavior when finite generation completes.
 	// One of: "exit" (default), "idle"
 	OnFinish string `yaml:"onFinish,omitempty" mapstructure:"onFinish,omitempty"`
@@ -38,6 +41,9 @@ func (c *Config) Validate() error {
 	if err := c.Metrics.Validate(); err != nil {
 		return err
 	}
+	if err := c.Environment.Validate(); err != nil {
+		return err
+	}
 	if c.OnFinish != "" && c.OnFinish != "exit" && c.OnFinish != "idle" {
 		return fmt.Errorf("onFinish must be one of: exit, idle, got %q", c.OnFinish)
 	}

internal/config/environment.go

@@ -0,0 +1,124 @@
+// Package config contains the top level configuration structures and logic
+package config
+
+import (
+	"fmt"
+
+	"github.com/observiq/blitz/internal/datagen"
+	"go.uber.org/zap"
+)
+
+// EnvironmentConfig configures the simulated datagen.Environment that
+// generators draw their host identities from. The block is optional; an
+// omitted environment yields a randomized default Environment (PIPE-1036).
+type EnvironmentConfig struct {
+	// DomainName is the AD/DNS domain for the environment. Empty = datagen default.
+	DomainName string `yaml:"domain_name,omitempty" mapstructure:"domain_name,omitempty"`
+	// SeedConfig controls per-identity-type determinism.
+	SeedConfig EnvironmentSeedConfig `yaml:"seed_config,omitempty" mapstructure:"seed_config,omitempty"`
+	// Counts controls how many of each identity type are generated.
+	Counts EnvironmentCounts `yaml:"counts,omitempty" mapstructure:"counts,omitempty"`
+}
+
+// EnvironmentSeedConfig mirrors datagen.SeedConfig as optional YAML keys. An
+// omitted (nil) field randomizes that identity type; an explicit value —
+// including 0 — is a deterministic seed, per the datagen seed contract.
+type EnvironmentSeedConfig struct {
+	Shared         *int64 `yaml:"shared,omitempty" mapstructure:"shared,omitempty"`
+	Systems        *int64 `yaml:"systems,omitempty" mapstructure:"systems,omitempty"`
+	Users          *int64 `yaml:"users,omitempty" mapstructure:"users,omitempty"`
+	Groups         *int64 `yaml:"groups,omitempty" mapstructure:"groups,omitempty"`
+	Services       *int64 `yaml:"services,omitempty" mapstructure:"services,omitempty"`
+	Applications   *int64 `yaml:"applications,omitempty" mapstructure:"applications,omitempty"`
+	Networks       *int64 `yaml:"networks,omitempty" mapstructure:"networks,omitempty"`
+	Domains        *int64 `yaml:"domains,omitempty" mapstructure:"domains,omitempty"`
+	StorageSystems *int64 `yaml:"storage_systems,omitempty" mapstructure:"storage_systems,omitempty"`
+	NetworkSystems *int64 `yaml:"network_systems,omitempty" mapstructure:"network_systems,omitempty"`
+}
+
+// EnvironmentCounts mirrors the count fields of datagen.EnvironmentOpts. A zero
+// (omitted) count uses the datagen package default for that type.
+type EnvironmentCounts struct {
+	Systems        int `yaml:"systems,omitempty" mapstructure:"systems,omitempty"`
+	Users          int `yaml:"users,omitempty" mapstructure:"users,omitempty"`
+	Groups         int `yaml:"groups,omitempty" mapstructure:"groups,omitempty"`
+	Networks       int `yaml:"networks,omitempty" mapstructure:"networks,omitempty"`
+	StorageSystems int `yaml:"storage_systems,omitempty" mapstructure:"storage_systems,omitempty"`
+	NetworkSystems int `yaml:"network_systems,omitempty" mapstructure:"network_systems,omitempty"`
+	DomainAdmins   int `yaml:"domain_admins,omitempty" mapstructure:"domain_admins,omitempty"`
+}
+
+// Validate checks the environment configuration: counts must not be negative.
+func (e EnvironmentConfig) Validate() error {
+	counts := map[string]int{
+		"systems":         e.Counts.Systems,
+		"users":           e.Counts.Users,
+		"groups":          e.Counts.Groups,
+		"networks":        e.Counts.Networks,
+		"storage_systems": e.Counts.StorageSystems,
+		"network_systems": e.Counts.NetworkSystems,
+		"domain_admins":   e.Counts.DomainAdmins,
+	}
+	for name, v := range counts {
+		if v < 0 {
+			return fmt.Errorf("environment count %q must not be negative, got %d", name, v)
+		}
+	}
+	return nil
+}
+
+// Build resolves the configured Environment, hydrating a datagen.SeedConfig
+// (omitted seeds randomize), initializing it (which logs the effective seeds),
+// and composing the Environment. A nil logger is treated as a no-op logger.
+func (e EnvironmentConfig) Build(logger *zap.Logger) (*datagen.Environment, error) {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+
+	seeds := datagen.NewSeedConfig()
+	sc := e.SeedConfig
+	if sc.Shared != nil {
+		seeds.Shared = *sc.Shared
+	}
+	if sc.Systems != nil {
+		seeds.Systems = *sc.Systems
+	}
+	if sc.Users != nil {
+		seeds.Users = *sc.Users
+	}
+	if sc.Groups != nil {
+		seeds.Groups = *sc.Groups
+	}
+	if sc.Services != nil {
+		seeds.Services = *sc.Services
+	}
+	if sc.Applications != nil {
+		seeds.Applications = *sc.Applications
+	}
+	if sc.Networks != nil {
+		seeds.Networks = *sc.Networks
+	}
+	if sc.Domains != nil {
+		seeds.Domains = *sc.Domains
+	}
+	if sc.StorageSystems != nil {
+		seeds.StorageSystems = *sc.StorageSystems
+	}
+	if sc.NetworkSystems != nil {
+		seeds.NetworkSystems = *sc.NetworkSystems
+	}
+	seeds.Init(logger)
+
+	opts := &datagen.EnvironmentOpts{
+		DomainName:         e.DomainName,
+		SystemCount:        e.Counts.Systems,
+		UserCount:          e.Counts.Users,
+		GroupCount:         e.Counts.Groups,
+		NetworkCount:       e.Counts.Networks,
+		StorageSystemCount: e.Counts.StorageSystems,
+		NetworkSystemCount: e.Counts.NetworkSystems,
+		DomainAdminsCount:  e.Counts.DomainAdmins,
+		Logger:             logger,
+	}
+	return datagen.GenerateEnvironment(seeds, opts)
+}