feat(output): fan out one stream to multiple outputs (PIPE-1446)

docs/configuration.md

@@ -175,11 +175,9 @@ For detailed configuration options for each generator, see the individual genera
 
 ### Output Configuration
 
-**Note:** Only a single output can be configured at a time.
-
 | YAML Path | Flag Name | Environment Variable | Default | Description |
 |-----------|-----------|---------------------|---------|-------------|
-| `output.type` | `--output-type` | `BLITZ_OUTPUT_TYPE` | `nop` | Output type. Valid values: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file` |
+| `output.type` | `--output-type` | `BLITZ_OUTPUT_TYPE` | `nop` | Output type. Valid values: `nop`, `stdout`, `tcp`, `udp`, `syslog`, `otlp-grpc`, `file`, `hec` |
 
 For detailed configuration options for each output, see the individual output documentation:
 
@@ -190,6 +188,32 @@ For detailed configuration options for each output, see the individual output do
 - [Syslog Output](https://github.com/observiq/blitz/blob/main/docs/output/syslog.md)
 - [OTLP gRPC Output](https://github.com/observiq/blitz/blob/main/docs/output/otlp-grpc.md)
 - [File Output](https://github.com/observiq/blitz/blob/main/docs/output/file.md)
+- [HEC Output](https://github.com/observiq/blitz/blob/main/docs/output/hec.md)
+
+#### Multiple Outputs (Fan-Out)
+
+To send one generated stream to several destinations at once, use the `outputs:`
+list instead of the singular `output:` block. Every generated record is fanned
+out to all listed outputs. Each entry takes the same fields as `output`, minus
+the `type` selector, which moves to the list item.
+
+```yaml
+outputs:
+  - type: stdout
+  - type: otlp-grpc
+    otlp-grpc:
+      host: localhost
+      port: 4317
+      insecure: true
+```
+
+Notes:
+
+- Set `output` or `outputs`, not both. A config with both is rejected, since it is ambiguous which one applies.
+- Writes are best-effort: one dead destination does not stop the others, and
+  each signal goes only to the outputs that support it.
+- Each list entry inherits any field it leaves unset (workers, and so on) from
+  the same defaults the singular `output` uses; fields you set on the entry win.
 
 ## Example Configurations
 

internal/config/config.go

@@ -3,6 +3,7 @@ package config
 
 import (
 	"fmt"
+	"reflect"
 	"strings"
 )
 
@@ -17,6 +18,10 @@ type Config struct {
 	Generators []Generator `yaml:"generators,omitempty" mapstructure:"generators,omitempty"`
 	// Output configuration
 	Output Output `yaml:"output,omitempty" mapstructure:"output,omitempty"`
+	// Outputs is the list of outputs for multi-output (fan-out) mode.
+	// If set, takes precedence over the singular Output field, and every
+	// generated record is fanned out to all of them.
+	Outputs []Output `yaml:"outputs,omitempty" mapstructure:"outputs,omitempty"`
 	// Metrics configuration (Prometheus scrape endpoint for self-metrics)
 	Metrics Metrics `yaml:"metrics,omitempty" mapstructure:"metrics,omitempty"`
 	// Environment configures the simulated datagen.Environment identities
@@ -38,7 +43,14 @@ func (c *Config) Validate() error {
 	if err := c.Generator.Validate(); err != nil {
 		return err
 	}
-	if err := c.Output.Validate(); err != nil {
+	if len(c.Outputs) > 0 {
+		// Validate the defaulted entries, since that is what gets built.
+		for i, o := range c.EffectiveOutputs() {
+			if err := o.Validate(); err != nil {
+				return fmt.Errorf("outputs[%d] validation failed: %w", i, err)
+			}
+		}
+	} else if err := c.Output.Validate(); err != nil {
 		return err
 	}
 	if err := c.Metrics.Validate(); err != nil {
@@ -66,6 +78,43 @@ func (c *Config) EffectiveGenerators() []Generator {
 	return expandGenerators([]Generator{c.Generator})
 }
 
+// EffectiveOutputs returns the list of outputs to use.
+// If Outputs is set, it takes precedence over the singular Output field.
+// The singular Output carries the defaults applied by the override system,
+// so each list entry inherits any field it leaves unset (workers, and so on)
+// from it, while user-set fields win.
+func (c *Config) EffectiveOutputs() []Output {
+	if len(c.Outputs) == 0 {
+		return []Output{c.Output}
+	}
+	outs := make([]Output, len(c.Outputs))
+	for i, o := range c.Outputs {
+		merged := o
+		fillZeroFields(reflect.ValueOf(&merged).Elem(), reflect.ValueOf(c.Output))
+		outs[i] = merged
+	}
+	return outs
+}
+
+// fillZeroFields sets each zero-valued field of dst to the corresponding
+// field of tmpl, recursing into nested structs. A non-zero field in dst is
+// left untouched, so an explicitly-set value always wins over the default.
+func fillZeroFields(dst, tmpl reflect.Value) {
+	for i := 0; i < dst.NumField(); i++ {
+		df := dst.Field(i)
+		if !df.CanSet() {
+			continue
+		}
+		if df.Kind() == reflect.Struct {
+			fillZeroFields(df, tmpl.Field(i))
+			continue
+		}
+		if df.IsZero() {
+			df.Set(tmpl.Field(i))
+		}
+	}
+}
+
 // expandGenerators expands comma-separated HostMetrics OS values.
 func expandGenerators(gens []Generator) []Generator {
 	var result []Generator

internal/config/output_exclusive.go

@@ -0,0 +1,29 @@
+package config
+
+import (
+	"fmt"
+
+	"github.com/spf13/viper"
+)
+
+// ValidateExclusiveOutputConfig rejects a config that sets both the singular
+// `output` and the plural `outputs`. The two together are ambiguous: `outputs`
+// wins for what runs, while `output` still seeds each entry's defaults, so a
+// value on `output` would leak silently into the list. Callers must pick one.
+//
+// It runs at load against the parsed config sources (`InConfig`), not
+// `IsSet`, so the defaults bound for `output.*` do not make it fire when only
+// `outputs` is written.
+func ValidateExclusiveOutputConfig(v *viper.Viper) error {
+	return rejectBothForms(v, "output", "outputs")
+}
+
+// rejectBothForms errors when both the singular and plural forms of a config
+// key are present in the parsed config. Shared by the output check here and,
+// later, the generator/generators check.
+func rejectBothForms(v *viper.Viper, singular, plural string) error {
+	if v.InConfig(singular) && v.InConfig(plural) {
+		return fmt.Errorf("set either %s or %s, not both", singular, plural)
+	}
+	return nil
+}

internal/config/output_exclusive_test.go

@@ -0,0 +1,37 @@
+package config
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
+)
+
+func viperFrom(t *testing.T, yaml string) *viper.Viper {
+	t.Helper()
+	v := viper.New()
+	v.SetConfigType("yaml")
+	require.NoError(t, v.ReadConfig(strings.NewReader(yaml)))
+	return v
+}
+
+func TestValidateExclusiveOutputConfig_BothSetFails(t *testing.T) {
+	v := viperFrom(t, "output:\n  type: stdout\noutputs:\n  - type: nop\n")
+	require.Error(t, ValidateExclusiveOutputConfig(v))
+}
+
+func TestValidateExclusiveOutputConfig_OnlyOutputsOK(t *testing.T) {
+	v := viperFrom(t, "outputs:\n  - type: nop\n")
+	require.NoError(t, ValidateExclusiveOutputConfig(v))
+}
+
+func TestValidateExclusiveOutputConfig_OnlyOutputOK(t *testing.T) {
+	v := viperFrom(t, "output:\n  type: stdout\n")
+	require.NoError(t, ValidateExclusiveOutputConfig(v))
+}
+
+func TestValidateExclusiveOutputConfig_NeitherOK(t *testing.T) {
+	v := viperFrom(t, "logging:\n  type: stdout\n")
+	require.NoError(t, ValidateExclusiveOutputConfig(v))
+}

internal/config/output_multi_test.go

@@ -0,0 +1,83 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEffectiveOutputs_SingleOutput(t *testing.T) {
+	cfg := &Config{Output: Output{Type: OutputTypeStdout}}
+
+	outs := cfg.EffectiveOutputs()
+	require.Len(t, outs, 1)
+	assert.Equal(t, OutputTypeStdout, outs[0].Type)
+}
+
+func TestEffectiveOutputs_MultiOutput(t *testing.T) {
+	cfg := &Config{
+		Outputs: []Output{
+			{Type: OutputTypeStdout},
+			{Type: OutputTypeNop},
+		},
+	}
+
+	outs := cfg.EffectiveOutputs()
+	require.Len(t, outs, 2)
+	assert.Equal(t, OutputTypeStdout, outs[0].Type)
+	assert.Equal(t, OutputTypeNop, outs[1].Type)
+}
+
+func TestEffectiveOutputs_MultiOverridesSingle(t *testing.T) {
+	cfg := &Config{
+		Output:  Output{Type: OutputTypeStdout},
+		Outputs: []Output{{Type: OutputTypeNop}, {Type: OutputTypeFile}},
+	}
+
+	outs := cfg.EffectiveOutputs()
+	require.Len(t, outs, 2, "Outputs field should take precedence")
+}
+
+func TestEffectiveOutputs_ListEntriesInheritDefaults(t *testing.T) {
+	// c.Output is the viper-defaulted template (workers filled). A list entry
+	// that omits workers must inherit that default; user-set fields win.
+	cfg := &Config{
+		Output: Output{TCP: TCPOutputConfig{Workers: 1}},
+		Outputs: []Output{
+			{Type: OutputTypeTCP, TCP: TCPOutputConfig{Host: "h", Port: 5000}},
+		},
+	}
+
+	outs := cfg.EffectiveOutputs()
+	require.Len(t, outs, 1)
+	assert.Equal(t, 1, outs[0].TCP.Workers, "unset workers should inherit the template default")
+	assert.Equal(t, "h", outs[0].TCP.Host, "user-set host must be preserved")
+	assert.Equal(t, 5000, outs[0].TCP.Port, "user-set port must be preserved")
+}
+
+func TestEffectiveOutputs_ListEntryUserValueOverridesDefault(t *testing.T) {
+	cfg := &Config{
+		Output: Output{TCP: TCPOutputConfig{Workers: 1}},
+		Outputs: []Output{
+			{Type: OutputTypeTCP, TCP: TCPOutputConfig{Host: "h", Port: 5000, Workers: 8}},
+		},
+	}
+
+	outs := cfg.EffectiveOutputs()
+	assert.Equal(t, 8, outs[0].TCP.Workers, "explicit workers must win over the template default")
+}
+
+func TestValidate_OutputsListValidatesEach(t *testing.T) {
+	cfg := &Config{
+		Logging: Logging{Type: LoggingTypeStdout},
+		Outputs: []Output{
+			{Type: OutputTypeNop},
+			{Type: OutputType("bogus")},
+		},
+	}
+
+	err := cfg.Validate()
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "outputs[1]")
+}

output/multi.go

@@ -0,0 +1,107 @@
+package output
+
+import (
+	"context"
+	"errors"
+
+	"github.com/observiq/blitz/telemetry"
+)
+
+// MultiOutput fans one generated stream out to several child outputs,
+// implementing Output plus MetricWriter and TraceWriter so callers treat it
+// like a single output. It records no telemetry itself; each child emits its
+// own output_type-tagged metrics, so there is no double-counting.
+type MultiOutput struct {
+	outputs []Output
+}
+
+// NewMultiOutput wraps the given outputs.
+func NewMultiOutput(outputs ...Output) *MultiOutput {
+	return &MultiOutput{outputs: outputs}
+}
+
+// WriteLog fans a log record to children that implement LogWriter and
+// declare Logs; others are skipped. Best-effort: a failing child does not
+// stop the others; the aggregated error is returned (nil if all succeed).
+func (m *MultiOutput) WriteLog(ctx context.Context, data LogRecord) error {
+	var errs []error
+	for _, o := range m.outputs {
+		w, ok := o.(LogWriter)
+		if !ok || !supports(o, telemetry.Logs) {
+			continue
+		}
+		if err := w.WriteLog(ctx, data); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// WriteMetric fans a metric record to children that implement MetricWriter
+// and declare Metrics; others are skipped.
+func (m *MultiOutput) WriteMetric(ctx context.Context, data MetricRecord) error {
+	var errs []error
+	for _, o := range m.outputs {
+		w, ok := o.(MetricWriter)
+		if !ok || !supports(o, telemetry.Metrics) {
+			continue
+		}
+		if err := w.WriteMetric(ctx, data); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// WriteTrace fans a trace record to children that implement TraceWriter and
+// declare Traces; others are skipped.
+func (m *MultiOutput) WriteTrace(ctx context.Context, data TraceRecord) error {
+	var errs []error
+	for _, o := range m.outputs {
+		w, ok := o.(TraceWriter)
+		if !ok || !supports(o, telemetry.Traces) {
+			continue
+		}
+		if err := w.WriteTrace(ctx, data); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// Stop stops every child, aggregating errors; one failure does not skip the rest.
+func (m *MultiOutput) Stop(ctx context.Context) error {
+	var errs []error
+	for _, o := range m.outputs {
+		if err := o.Stop(ctx); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// SupportedTelemetry returns the union of the children's supported types.
+func (m *MultiOutput) SupportedTelemetry() []telemetry.Type {
+	seen := make(map[telemetry.Type]struct{})
+	var union []telemetry.Type
+	for _, o := range m.outputs {
+		for _, t := range o.SupportedTelemetry() {
+			if _, dup := seen[t]; dup {
+				continue
+			}
+			seen[t] = struct{}{}
+			union = append(union, t)
+		}
+	}
+	return union
+}
+
+// supports reports whether the output declares the given telemetry type.
+func supports(o Output, t telemetry.Type) bool {
+	for _, s := range o.SupportedTelemetry() {
+		if s == t {
+			return true
+		}
+	}
+	return false
+}