diff --git a/CHANGELOG.md b/CHANGELOG.md index a5182671a2..f171b5426e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## main / (unreleased) +* [CHANGE] eventrecorder: All outputs now require a configured `name`, used instead of paths, URLs, brokers, topics, or other destination configuration in metric label values. Webhook URLs and Kafka configuration are also omitted from logs; file paths remain available in file-output error logs. + ## 0.34.0 / 2026-08-16 * [CHANGE] notify: The `reason` label on `alertmanager_notifications_failed_total` now distinguishes `authError` (HTTP 401/403) and `rateLimited` (HTTP 429) from the generic `clientError`. Dashboards/alerts matching `reason="clientError"` for these codes must be updated. #5332 diff --git a/config/config_test.go b/config/config_test.go index 2a37784de3..8a4c3500cb 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -53,7 +53,8 @@ receivers: - name: default event_recorder: webhook_outputs: - - url: https://stream-id.ingest.cloudflare.com + - name: pipelines + url: https://stream-id.ingest.cloudflare.com batch: true http_config: authorization: @@ -61,6 +62,7 @@ event_recorder: `) require.NoError(t, err) require.Len(t, cfg.EventRecorder.WebhookOutputs, 1) + require.Equal(t, "pipelines", cfg.EventRecorder.WebhookOutputs[0].Name) require.True(t, cfg.EventRecorder.WebhookOutputs[0].Batch) resolveFilepaths("/etc/alertmanager", cfg) diff --git a/docs/configuration.md b/docs/configuration.md index be9a73f841..7d0d542d30 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2146,7 +2146,14 @@ Event recording is configured under the top-level `event_recorder` key. Outputs are grouped by type, one list per destination kind (mirroring the way receivers group their integrations). Every recorded event is sent to -every output across all lists. +every output across all lists. Every output requires a name, which is used +with its type as the output identifier in metrics and logs (for example, +`webhook:primary`). Destination configuration such as paths, URLs, brokers, +and topics is not included in metric labels. URLs, brokers, and topics are +also omitted from logs; file paths remain in file-output error logs for +troubleshooting. Names must be unique within each output type, no longer than +128 characters, and contain only letters, digits, hyphens, underscores, and +periods. ```yaml # JSONL file outputs. @@ -2173,6 +2180,9 @@ when the parent directory observes a rename/remove/create on the target path (for compatibility with `logrotate` and similar tools). ```yaml +# Name used to identify this output in metrics and logs. +name: + # Path to the JSONL output file. Will be created if it does not exist. path: ``` @@ -2188,6 +2198,9 @@ duplicate events after ambiguous failures. With multiple workers, requests may complete out of order; set `workers: 1` when request ordering matters. ```yaml +# Name used to identify this output in metrics and logs. +name: + # URL to POST events to. url: @@ -2230,7 +2243,8 @@ as a batched webhook output: ```yaml event_recorder: webhook_outputs: - - url: https://.ingest.cloudflare.com + - name: pipelines + url: https://.ingest.cloudflare.com batch: true http_config: # The token must have the "Workers Pipeline Send" permission when @@ -2255,6 +2269,9 @@ The target topic must already exist (or the brokers must be configured to auto-create topics); Alertmanager does not create it. ```yaml +# Name used to identify this output in metrics and logs. +name: + # Seed broker list (host:port). At least one entry is required. brokers: [ - ... ] @@ -2302,4 +2319,7 @@ driver (Docker, Kubernetes, etc.) captures stdout automatically. > distinct formats on the same stream that may complicate downstream > log parsing. -This output type takes no additional configuration fields. +```yaml +# Name used to identify this output in metrics and logs. +name: +``` diff --git a/eventrecorder/config.go b/eventrecorder/config.go index 622976c797..c665ae33bc 100644 --- a/eventrecorder/config.go +++ b/eventrecorder/config.go @@ -13,6 +13,12 @@ package eventrecorder +import ( + "fmt" +) + +const maxOutputNameLength = 128 + // Config configures the event recorder feature. // // Outputs are grouped by type, one list per destination kind, mirroring @@ -26,6 +32,75 @@ type Config struct { StdoutOutputs []StdoutOutputConfig `yaml:"stdout_outputs,omitempty" json:"stdout_outputs,omitempty"` } +// UnmarshalYAML implements the yaml.Unmarshaler interface, validating that +// each output identifier is unique. +func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { + type plain Config + if err := unmarshal((*plain)(c)); err != nil { + return err + } + return c.validate() +} + +func (c Config) validate() error { + seen := make(map[string]struct{}, c.totalOutputs()) + add := func(kind, name string) error { + id, err := outputIdentifier(kind, name) + if err != nil { + return err + } + if _, ok := seen[id]; ok { + return fmt.Errorf("event_recorder output name %q is duplicated for type %s", name, kind) + } + seen[id] = struct{}{} + return nil + } + for _, out := range c.FileOutputs { + if err := add("file", out.Name); err != nil { + return err + } + } + for _, out := range c.WebhookOutputs { + if err := add("webhook", out.Name); err != nil { + return err + } + } + for _, out := range c.KafkaOutputs { + if err := add("kafka", out.Name); err != nil { + return err + } + } + for _, out := range c.StdoutOutputs { + if err := add("stdout", out.Name); err != nil { + return err + } + } + return nil +} + +func outputIdentifier(kind, name string) (string, error) { + if name == "" { + return "", fmt.Errorf("event_recorder %s output requires a name", kind) + } + if len(name) > maxOutputNameLength { + return "", fmt.Errorf("event_recorder %s output name must not exceed %d characters", kind, maxOutputNameLength) + } + for _, r := range name { + if (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') && r != '-' && r != '_' && r != '.' { + return "", fmt.Errorf("event_recorder %s output name must contain only letters, digits, hyphens, underscores, and periods", kind) + } + } + return kind + ":" + name, nil +} + +func safeOutputIdentifier(kind, name string) string { + id, err := outputIdentifier(kind, name) + if err != nil { + return kind + ":" + } + return id +} + // totalOutputs returns the number of configured outputs across all // destination kinds. func (c Config) totalOutputs() int { diff --git a/eventrecorder/file.go b/eventrecorder/file.go index d29572de66..82ba41838f 100644 --- a/eventrecorder/file.go +++ b/eventrecorder/file.go @@ -29,6 +29,8 @@ import ( // FileOutputConfig configures a JSONL file event recorder output. type FileOutputConfig struct { + // Name identifies this output in metrics and logs. + Name string `yaml:"name" json:"name"` // Path is the JSONL file to append events to. Created if absent. Path string `yaml:"path" json:"path"` } @@ -38,6 +40,9 @@ type FileOutputConfig struct { func (c *FileOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { type plain FileOutputConfig if err := unmarshal((*plain)(c)); err != nil { + return errors.New("invalid event_recorder file output configuration") + } + if _, err := outputIdentifier("file", c.Name); err != nil { return err } if c.Path == "" { @@ -48,7 +53,7 @@ func (c *FileOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { // equal reports whether two file output configs are semantically equal. func (c FileOutputConfig) equal(o FileOutputConfig) bool { - return c.Path == o.Path + return c.Name == o.Name && c.Path == o.Path } // FileOutput writes pre-serialized JSON event bytes to a JSONL file. @@ -56,6 +61,7 @@ func (c FileOutputConfig) equal(o FileOutputConfig) bool { // logrotate). type FileOutput struct { path string + name string mu sync.Mutex f *os.File closed bool @@ -66,20 +72,28 @@ type FileOutput struct { // Name returns a stable identifier for this output. func (fo *FileOutput) Name() string { - return fmt.Sprintf("file:%s", fo.path) + return fo.name } -// NewFileOutput creates a new file-based event recorder output at the given -// path. The file is watched with fsnotify so that external log +// NewFileOutput creates a new file-based event recorder output. The file is +// watched with fsnotify so that external log // rotation tools (e.g., logrotate) trigger an immediate reopen. -func NewFileOutput(path string, logger *slog.Logger) (*FileOutput, error) { - f, err := openAppend(path) +func NewFileOutput(cfg FileOutputConfig, logger *slog.Logger) (*FileOutput, error) { + name, err := outputIdentifier("file", cfg.Name) + if err != nil { + return nil, err + } + if cfg.Path == "" { + return nil, errors.New("file output requires a path") + } + f, err := openAppend(cfg.Path) if err != nil { return nil, err } fo := &FileOutput{ - path: path, + path: cfg.Path, + name: name, f: f, logger: logger, done: make(chan struct{}), @@ -111,7 +125,7 @@ func (fo *FileOutput) reopen() { } f, err := openAppend(fo.path) if err != nil { - fo.logger.Error("Failed to reopen event recorder file", "path", fo.path, "err", err) + fo.logger.Error("Failed to reopen event recorder file", "output", fo.name, "path", fo.path, "err", err) return } fo.f = f @@ -163,7 +177,7 @@ func (fo *FileOutput) watchLoop(ready chan<- error) { if !ok { return } - fo.logger.Error("fsnotify error on event recorder directory", "err", err) + fo.logger.Error("fsnotify error on event recorder directory", "output", fo.name, "path", fo.path, "err", err) case <-fo.done: return } diff --git a/eventrecorder/file_test.go b/eventrecorder/file_test.go index b8130fe6e1..26fa567816 100644 --- a/eventrecorder/file_test.go +++ b/eventrecorder/file_test.go @@ -29,11 +29,11 @@ func TestFileOutput_SendEvent(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "events.jsonl") - fo, err := NewFileOutput(path, slog.Default()) + fo, err := NewFileOutput(FileOutputConfig{Name: "primary", Path: path}, slog.Default()) require.NoError(t, err) defer fo.Close() - require.Equal(t, "file:"+path, fo.Name()) + require.Equal(t, "file:primary", fo.Name()) n1, err := fo.SendEvent(sampleEvent()) require.NoError(t, err) @@ -57,7 +57,7 @@ func TestFileOutput_ReopenAfterRename(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "events.jsonl") - fo, err := NewFileOutput(path, slog.Default()) + fo, err := NewFileOutput(FileOutputConfig{Name: "rotate", Path: path}, slog.Default()) require.NoError(t, err) defer fo.Close() @@ -93,7 +93,7 @@ func TestFileOutput_ReopenAfterRemove(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "events.jsonl") - fo, err := NewFileOutput(path, slog.Default()) + fo, err := NewFileOutput(FileOutputConfig{Name: "remove", Path: path}, slog.Default()) require.NoError(t, err) defer fo.Close() @@ -120,7 +120,7 @@ func TestFileOutput_Close(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "events.jsonl") - fo, err := NewFileOutput(path, slog.Default()) + fo, err := NewFileOutput(FileOutputConfig{Name: "close", Path: path}, slog.Default()) require.NoError(t, err) _, err = fo.SendEvent(sampleEvent()) @@ -133,7 +133,7 @@ func TestFileOutput_Close(t *testing.T) { } func TestFileOutput_InvalidPath(t *testing.T) { - _, err := NewFileOutput("/nonexistent/dir/events.jsonl", slog.Default()) + _, err := NewFileOutput(FileOutputConfig{Name: "invalid", Path: "/nonexistent/dir/events.jsonl"}, slog.Default()) require.Error(t, err) } @@ -148,14 +148,20 @@ func TestFileOutputConfig_UnmarshalYAML(t *testing.T) { }{ { name: "valid", - yaml: "path: /tmp/events.jsonl\n", + yaml: "name: primary\npath: /tmp/events.jsonl\n", check: func(t *testing.T, c FileOutputConfig) { + require.Equal(t, "primary", c.Name) require.Equal(t, "/tmp/events.jsonl", c.Path) }, }, { name: "missing path", - yaml: "{}\n", + yaml: "name: primary\n", + wantErr: true, + }, + { + name: "missing name", + yaml: "path: /tmp/events.jsonl\n", wantErr: true, }, } @@ -176,10 +182,13 @@ func TestFileOutputConfig_UnmarshalYAML(t *testing.T) { } func TestEventRecorderConfigEqual_File(t *testing.T) { - a := Config{FileOutputs: []FileOutputConfig{{Path: "/tmp/events.jsonl"}}} - b := Config{FileOutputs: []FileOutputConfig{{Path: "/tmp/events.jsonl"}}} + a := Config{FileOutputs: []FileOutputConfig{{Name: "primary", Path: "/tmp/events.jsonl"}}} + b := Config{FileOutputs: []FileOutputConfig{{Name: "primary", Path: "/tmp/events.jsonl"}}} require.True(t, configEqual(a, b)) b.FileOutputs[0].Path = "/tmp/other.jsonl" require.False(t, configEqual(a, b)) + b.FileOutputs[0].Path = a.FileOutputs[0].Path + b.FileOutputs[0].Name = "secondary" + require.False(t, configEqual(a, b)) } diff --git a/eventrecorder/kafka.go b/eventrecorder/kafka.go index d60f7082b7..30a5da2784 100644 --- a/eventrecorder/kafka.go +++ b/eventrecorder/kafka.go @@ -36,6 +36,8 @@ const defaultKafkaBufferSize = 1024 // KafkaOutputConfig configures a Kafka event recorder output. type KafkaOutputConfig struct { + // Name identifies this output in metrics and logs. + Name string `yaml:"name" json:"name"` // Brokers is the list of Kafka seed brokers in host:port form. Brokers []string `yaml:"brokers" json:"brokers"` // Topic is the Kafka topic to produce events to. @@ -67,12 +69,13 @@ type KafkaOutputConfig struct { func (c *KafkaOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { type plain KafkaOutputConfig if err := unmarshal((*plain)(c)); err != nil { + return errors.New("invalid event_recorder kafka output configuration") + } + if _, err := outputIdentifier("kafka", c.Name); err != nil { return err } if err := c.clientOptions().Validate(); err != nil { - // The shared validator's messages already say "kafka: ..."; we - // prefix with the event_recorder context for user clarity. - return fmt.Errorf("event_recorder %w", err) + return errors.New("event_recorder kafka output has invalid client configuration") } if c.Topic == "" { return errors.New("event_recorder kafka output requires a topic") @@ -81,7 +84,7 @@ func (c *KafkaOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { c.Format = kafka.FormatJSON } if err := kafka.ValidateFormat(c.Format); err != nil { - return fmt.Errorf("event_recorder %w", err) + return errors.New("event_recorder kafka output has an invalid format") } return nil } @@ -103,6 +106,9 @@ func (c KafkaOutputConfig) clientOptions() kafka.ClientOptions { // equal. Broker lists are compared order-independently because // reordering brokers in YAML is semantically a no-op. func (c KafkaOutputConfig) equal(o KafkaOutputConfig) bool { + if c.Name != o.Name { + return false + } if !kafka.BrokerListsEqual(c.Brokers, o.Brokers) { return false } @@ -142,7 +148,7 @@ type KafkaOutput struct { topic string instance string // used as the message key format kafka.Format - name string // "kafka:/" + name string // "kafka:" logger *slog.Logger drops prometheus.Counter produceErrs *prometheus.CounterVec @@ -163,6 +169,10 @@ func NewKafkaOutput( produceErrors *prometheus.CounterVec, logger *slog.Logger, ) (*KafkaOutput, error) { + name, err := outputIdentifier("kafka", cfg.Name) + if err != nil { + return nil, err + } if cfg.Topic == "" { return nil, errors.New("kafka output requires a topic") } @@ -176,7 +186,9 @@ func NewKafkaOutput( // Shared validation + franz-go option construction lives in the // kafka package so a future Kafka receiver can reuse it. - kopts, err := kafka.BuildOpts(cfg.clientOptions(), logger) + // franz-go logs can include broker and topic configuration. Keep its + // internal logger disabled and emit safe, output-scoped logs here. + kopts, err := kafka.BuildOpts(cfg.clientOptions(), nil) if err != nil { return nil, err } @@ -191,8 +203,6 @@ func NewKafkaOutput( bufferSize = defaultKafkaBufferSize } - name := fmt.Sprintf("kafka:%s/%s", kafka.BrokerList(cfg.Brokers), cfg.Topic) - ko := &KafkaOutput{ client: client, topic: cfg.Topic, @@ -210,7 +220,7 @@ func NewKafkaOutput( // Best-effort connectivity check runs in the background so that // alertmanager startup (and event_recorder hot reload) is never // blocked by an unreachable broker. - kafka.PingInBackground(client, logger) + ko.pingInBackground() ko.wg.Add(1) go ko.dispatch() @@ -221,6 +231,16 @@ func NewKafkaOutput( // Name returns the stable identifier for this output. func (ko *KafkaOutput) Name() string { return ko.name } +func (ko *KafkaOutput) pingInBackground() { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), kafka.DefaultPingTimeout) + defer cancel() + if err := ko.client.Ping(ctx); err != nil { + ko.logger.Warn("Kafka event recorder output could not reach brokers at startup; will retry in background", "output", ko.name) + } + }() +} + // SendEvent serializes the event in the configured format (JSON or // protobuf) and queues it for asynchronous delivery. It returns the // serialized size (for the bytes-written metric). @@ -324,7 +344,7 @@ func (ko *KafkaOutput) produce(value []byte) { ko.logger.Warn("Kafka producer buffer full, dropping event", "output", ko.name) default: ko.produceErrs.WithLabelValues(ko.name, string(kafka.ClassifyError(err))).Inc() - ko.logger.Warn("Kafka event recorder produce failed", "output", ko.name, "err", err) + ko.logger.Warn("Kafka event recorder produce failed", "output", ko.name, "error_type", kafka.ClassifyError(err)) } }) } @@ -357,7 +377,7 @@ func (ko *KafkaOutput) Close() error { defer cancel() if err := ko.client.Flush(ctx); err != nil { ko.logger.Warn("Kafka event recorder flush did not complete within budget; remaining records will be dropped", - "output", ko.name, "err", err) + "output", ko.name) } ko.client.Close() return nil diff --git a/eventrecorder/kafka_test.go b/eventrecorder/kafka_test.go index 9fac564542..79fae41cd5 100644 --- a/eventrecorder/kafka_test.go +++ b/eventrecorder/kafka_test.go @@ -14,9 +14,11 @@ package eventrecorder import ( + "bytes" "context" "errors" "log/slog" + "sync" "testing" "time" @@ -34,6 +36,29 @@ import ( "github.com/prometheus/alertmanager/kafka" ) +type synchronizedBuffer struct { + mu sync.Mutex + b bytes.Buffer +} + +func (b *synchronizedBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.b.Write(p) +} + +func (b *synchronizedBuffer) Len() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.b.Len() +} + +func (b *synchronizedBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.b.String() +} + // --- helpers. func testOutputDrops() *prometheus.CounterVec { @@ -130,6 +155,7 @@ func TestKafkaOutput_SendEvent_JSON(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "json", Brokers: brokers, Topic: topic, Format: kafka.FormatJSON, @@ -164,6 +190,7 @@ func TestKafkaOutput_SendEvent_Protobuf(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "protobuf", Brokers: brokers, Topic: topic, Format: kafka.FormatProtobuf, @@ -200,6 +227,7 @@ func TestKafkaOutput_KeyIsInstance(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "key", Brokers: brokers, Topic: topic, Format: kafka.FormatJSON, @@ -232,6 +260,7 @@ func TestKafkaOutput_DropsOnFullBuffer(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "drops", Brokers: brokers, Topic: topic, Format: kafka.FormatJSON, @@ -273,6 +302,7 @@ func TestKafkaOutput_SendAfterClose(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "after-close", Brokers: brokers, Topic: topic, Format: kafka.FormatJSON, @@ -299,6 +329,7 @@ func TestKafkaOutput_CloseFlushesQueue(t *testing.T) { ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "flush", Brokers: brokers, Topic: topic, Format: kafka.FormatJSON, @@ -326,9 +357,12 @@ func TestKafkaOutput_ContinuesOnInitialPingFailure(t *testing.T) { // must succeed and Name() must be well-formed. Importantly, the // constructor must NOT block on the ping timeout — that runs in // the background. + var logs synchronizedBuffer + logger := slog.New(slog.NewTextHandler(&logs, nil)) start := time.Now() ko, err := NewKafkaOutput( KafkaOutputConfig{ + Name: "unreachable", Brokers: []string{"127.0.0.1:1"}, Topic: "no-broker", Format: kafka.FormatJSON, @@ -336,12 +370,12 @@ func TestKafkaOutput_ContinuesOnInitialPingFailure(t *testing.T) { "test-host", testOutputDrops(), testKafkaProduceErrors(), - slog.Default(), + logger, ) constructDur := time.Since(start) require.NoError(t, err) require.NotNil(t, ko) - require.Equal(t, "kafka:127.0.0.1:1/no-broker", ko.Name()) + require.Equal(t, "kafka:unreachable", ko.Name()) // Construction must return well before the ping timeout (5s). // 1s is generous for CI but still 5x faster than the timeout. @@ -355,6 +389,12 @@ func TestKafkaOutput_ContinuesOnInitialPingFailure(t *testing.T) { require.NoError(t, ko.Close()) require.Less(t, time.Since(closeStart), 2*time.Second, "Close must abort the in-flight ping") + require.Eventually(t, func() bool { + return logs.Len() > 0 + }, time.Second, 10*time.Millisecond) + require.Contains(t, logs.String(), "kafka:unreachable") + require.NotContains(t, logs.String(), "127.0.0.1:1") + require.NotContains(t, logs.String(), "no-broker") } func TestKafkaOutput_RejectsBadConfig(t *testing.T) { @@ -364,11 +404,12 @@ func TestKafkaOutput_RejectsBadConfig(t *testing.T) { }{ { name: "no brokers", - cfg: KafkaOutputConfig{Topic: "t", Format: kafka.FormatJSON}, + cfg: KafkaOutputConfig{Name: "test", Topic: "t", Format: kafka.FormatJSON}, }, { name: "empty broker entry", cfg: KafkaOutputConfig{ + Name: "test", Brokers: []string{"127.0.0.1:9092", ""}, Topic: "t", Format: kafka.FormatJSON, @@ -376,15 +417,20 @@ func TestKafkaOutput_RejectsBadConfig(t *testing.T) { }, { name: "no topic", - cfg: KafkaOutputConfig{Brokers: []string{"127.0.0.1:9092"}, Format: kafka.FormatJSON}, + cfg: KafkaOutputConfig{Name: "test", Brokers: []string{"127.0.0.1:9092"}, Format: kafka.FormatJSON}, + }, + { + name: "no name", + cfg: KafkaOutputConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "t", Format: kafka.FormatJSON}, }, { name: "bad format", - cfg: KafkaOutputConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "t", Format: "yaml"}, + cfg: KafkaOutputConfig{Name: "test", Brokers: []string{"127.0.0.1:9092"}, Topic: "t", Format: "yaml"}, }, { name: "bad acks", cfg: KafkaOutputConfig{ + Name: "test", Brokers: []string{"127.0.0.1:9092"}, Topic: "t", Format: kafka.FormatJSON, Acks: "majority", }, @@ -392,6 +438,7 @@ func TestKafkaOutput_RejectsBadConfig(t *testing.T) { { name: "bad compression", cfg: KafkaOutputConfig{ + Name: "test", Brokers: []string{"127.0.0.1:9092"}, Topic: "t", Format: kafka.FormatJSON, Compression: "deflate", }, @@ -405,28 +452,17 @@ func TestKafkaOutput_RejectsBadConfig(t *testing.T) { } } -func TestKafkaOutput_NameIsStable(t *testing.T) { - // The Name() format ("kafka:/") is composed - // here in eventrecorder; broker-list sorting is handled by the - // shared kafka package (and tested there). This test pins the - // composition formula so reordering brokers in YAML doesn't change - // the Prometheus label value. - const topic = "topic" - a := "kafka:" + kafka.BrokerList([]string{"b:9092", "a:9092"}) + "/" + topic - b := "kafka:" + kafka.BrokerList([]string{"a:9092", "b:9092"}) + "/" + topic - require.Equal(t, a, b) - require.Equal(t, "kafka:a:9092,b:9092/topic", a) -} - // --- config tests. func TestEventRecorderConfigEqual_KafkaBrokerOrder(t *testing.T) { a := Config{KafkaOutputs: []KafkaOutputConfig{{ + Name: "test", Brokers: []string{"b:9092", "a:9092"}, Topic: "t", Format: kafka.FormatJSON, }}} b := Config{KafkaOutputs: []KafkaOutputConfig{{ + Name: "test", Brokers: []string{"a:9092", "b:9092"}, Topic: "t", Format: kafka.FormatJSON, @@ -439,6 +475,10 @@ func TestEventRecorderConfigEqual_KafkaBrokerOrder(t *testing.T) { b.KafkaOutputs[0].Topic = "t" b.KafkaOutputs[0].Format = kafka.FormatProtobuf require.False(t, configEqual(a, b), "differing formats must compare unequal") + + b.KafkaOutputs[0].Format = kafka.FormatJSON + b.KafkaOutputs[0].Name = "other" + require.False(t, configEqual(a, b), "differing names must compare unequal") } func TestKafkaOutputConfig_UnmarshalYAML(t *testing.T) { @@ -451,10 +491,12 @@ func TestKafkaOutputConfig_UnmarshalYAML(t *testing.T) { { name: "valid minimal kafka", yaml: ` +name: primary brokers: [a:9092, b:9092] topic: amgr-events `, check: func(t *testing.T, c KafkaOutputConfig) { + require.Equal(t, "primary", c.Name) require.Equal(t, "amgr-events", c.Topic) // Format defaults to "json" when omitted. require.Equal(t, kafka.FormatJSON, c.Format) @@ -463,6 +505,7 @@ topic: amgr-events { name: "valid full kafka", yaml: ` +name: secondary brokers: [a:9092] topic: t client_id: amgr @@ -481,32 +524,37 @@ buffer_size: 4096 }, { name: "missing brokers", - yaml: "topic: t\n", + yaml: "name: test\ntopic: t\n", wantErr: true, }, { name: "missing topic", - yaml: "brokers: [a:9092]\n", + yaml: "name: test\nbrokers: [a:9092]\n", + wantErr: true, + }, + { + name: "missing name", + yaml: "brokers: [a:9092]\ntopic: t\n", wantErr: true, }, { name: "empty broker entry", - yaml: "brokers: ['']\ntopic: t\n", + yaml: "name: test\nbrokers: ['']\ntopic: t\n", wantErr: true, }, { name: "bad format", - yaml: "brokers: [a:9092]\ntopic: t\nformat: yaml\n", + yaml: "name: test\nbrokers: [a:9092]\ntopic: t\nformat: yaml\n", wantErr: true, }, { name: "bad acks", - yaml: "brokers: [a:9092]\ntopic: t\nacks: majority\n", + yaml: "name: test\nbrokers: [a:9092]\ntopic: t\nacks: majority\n", wantErr: true, }, { name: "bad compression", - yaml: "brokers: [a:9092]\ntopic: t\ncompression: deflate\n", + yaml: "name: test\nbrokers: [a:9092]\ntopic: t\ncompression: deflate\n", wantErr: true, }, } diff --git a/eventrecorder/recorder.go b/eventrecorder/recorder.go index 7ce27ed218..8e1a6fddbe 100644 --- a/eventrecorder/recorder.go +++ b/eventrecorder/recorder.go @@ -110,9 +110,9 @@ type cfgUpdateMsg struct { // destination a pre-encoded JSON blob — avoids the footgun of, say, a // protobuf-configured Kafka output silently shipping a JSON payload. type Destination interface { - // Name returns a stable identifier for this destination, suitable - // for use as a Prometheus label value (e.g. "file:/var/log/events.jsonl" - // or "webhook:https://example.com/hook"). + // Name returns the type and configured name for this destination, + // suitable for use as a Prometheus label value (e.g. "file:archive" + // or "webhook:primary"). Name() string // SendEvent encodes and delivers the event. It returns the number // of payload bytes written (for the bytes-written metric) and any @@ -171,11 +171,15 @@ func NewRecorderFromConfig(cfg Config, instance string, logger *slog.Logger, r p // buildOutputs creates Destination implementations from the given config. func buildOutputs(cfg Config, instance string, m *metrics, logger *slog.Logger) []Destination { + if err := cfg.validate(); err != nil { + logger.Error("Invalid event recorder output configuration") + return nil + } var outputs []Destination for _, fc := range cfg.FileOutputs { - fo, err := NewFileOutput(fc.Path, logger) + fo, err := NewFileOutput(fc, logger) if err != nil { - logger.Error("Failed to create file event recorder output", "path", fc.Path, "err", err) + logger.Error("Failed to create file event recorder output", "output", safeOutputIdentifier("file", fc.Name), "path", fc.Path, "err", err) continue } outputs = append(outputs, fo) @@ -183,7 +187,7 @@ func buildOutputs(cfg Config, instance string, m *metrics, logger *slog.Logger) for _, wc := range cfg.WebhookOutputs { wo, err := NewWebhookOutput(wc, m.outputDrops, logger) if err != nil { - logger.Error("Failed to create webhook event recorder output", "url", sanitizeSecretURL(wc.URL), "err", err) + logger.Error("Failed to create webhook event recorder output", "output", safeOutputIdentifier("webhook", wc.Name)) continue } outputs = append(outputs, wo) @@ -191,13 +195,18 @@ func buildOutputs(cfg Config, instance string, m *metrics, logger *slog.Logger) for _, kc := range cfg.KafkaOutputs { ko, err := NewKafkaOutput(kc, instance, m.outputDrops, m.kafkaProduceErrors, logger) if err != nil { - logger.Error("Failed to create kafka event recorder output", "brokers", kc.Brokers, "topic", kc.Topic, "err", err) + logger.Error("Failed to create kafka event recorder output", "output", safeOutputIdentifier("kafka", kc.Name)) continue } outputs = append(outputs, ko) } - for range cfg.StdoutOutputs { - outputs = append(outputs, &StdoutOutput{}) + for _, sc := range cfg.StdoutOutputs { + so, err := NewStdoutOutput(sc) + if err != nil { + logger.Error("Failed to create stdout event recorder output", "output", safeOutputIdentifier("stdout", sc.Name)) + continue + } + outputs = append(outputs, so) } return outputs } @@ -216,7 +225,7 @@ func (c *sharedRecorder) writeLoop(outputs []Destination, currentCfg Config) { defer func() { for _, out := range outputs { if err := out.Close(); err != nil && c.logger != nil { - c.logger.Error("Failed to close event recorder output", "err", err) + c.logger.Error("Failed to close event recorder output", "output", out.Name()) } } }() @@ -234,7 +243,7 @@ func (c *sharedRecorder) writeLoop(outputs []Destination, currentCfg Config) { c.logger.Error("Failed to reload event recorder outputs; keeping existing outputs") for _, out := range newOutputs { if err := out.Close(); err != nil { - c.logger.Error("Failed to close partially-built event recorder output", "err", err) + c.logger.Error("Failed to close partially-built event recorder output", "output", out.Name()) } } close(update.done) @@ -245,7 +254,7 @@ func (c *sharedRecorder) writeLoop(outputs []Destination, currentCfg Config) { currentCfg = update.cfg for _, out := range oldOutputs { if err := out.Close(); err != nil { - c.logger.Error("Failed to close old event recorder output", "err", err) + c.logger.Error("Failed to close old event recorder output", "output", out.Name()) } } c.logger.Info("Event recorder configuration reloaded", "outputs", len(outputs)) @@ -281,7 +290,7 @@ func (c *sharedRecorder) marshalAndSend(req writeRequest, outputs []Destination) c.metrics.eventSerializeErrors.WithLabelValues(req.eventType).Inc() } c.metrics.eventsRecorded.WithLabelValues(req.eventType, name, "error").Inc() - c.logger.Error("Failed to write event", "event_type", req.eventType, "output", name, "err", err) + c.logger.Error("Failed to write event", "event_type", req.eventType, "output", name) continue } c.metrics.eventsRecorded.WithLabelValues(req.eventType, name, "success").Inc() diff --git a/eventrecorder/recorder_test.go b/eventrecorder/recorder_test.go index 9dfa797a15..a54bc62ee0 100644 --- a/eventrecorder/recorder_test.go +++ b/eventrecorder/recorder_test.go @@ -14,6 +14,7 @@ package eventrecorder import ( + "bytes" "context" "log/slog" "sync" @@ -21,6 +22,7 @@ import ( "time" "github.com/stretchr/testify/require" + "gopkg.in/yaml.v2" "github.com/prometheus/alertmanager/eventrecorder/eventrecorderpb" ) @@ -134,6 +136,58 @@ func TestNewRecorderFromConfig_NilLogger(t *testing.T) { }) } +func TestBuildOutputs_LogsFilePathButNotKafkaConfig(t *testing.T) { + var logs bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logs, nil)) + cfg := Config{ + FileOutputs: []FileOutputConfig{{ + Name: "archive", + Path: "/missing/private/file/events.jsonl", + }}, + KafkaOutputs: []KafkaOutputConfig{{ + Name: "events", + Brokers: []string{"private-broker.example:9092"}, + Topic: "private-topic", + Format: "invalid", + }}, + } + + outputs := buildOutputs(cfg, "test", newMetrics(nil), logger) + require.Empty(t, outputs) + require.Contains(t, logs.String(), "file:archive") + require.Contains(t, logs.String(), "kafka:events") + require.Contains(t, logs.String(), "/missing/private/file/events.jsonl") + require.NotContains(t, logs.String(), "private-broker.example:9092") + require.NotContains(t, logs.String(), "private-topic") +} + +func TestEventRecorderConfigRejectsDuplicateOutputNames(t *testing.T) { + var cfg Config + err := yaml.Unmarshal([]byte(` +file_outputs: +- name: archive + path: /tmp/one +- name: archive + path: /tmp/two +`), &cfg) + require.ErrorContains(t, err, "duplicated") +} + +func TestOutputIdentifier(t *testing.T) { + id, err := outputIdentifier("webhook", "primary.eu-1") + require.NoError(t, err) + require.Equal(t, "webhook:primary.eu-1", id) + + for _, name := range []string{"", "private/path", "https://secret.example", "line\nbreak"} { + _, err := outputIdentifier("webhook", name) + require.Error(t, err) + if name != "" { + require.NotContains(t, err.Error(), name) + } + require.Equal(t, "webhook:", safeOutputIdentifier("webhook", name)) + } +} + func TestRecordingNotEnabledByDefault(t *testing.T) { out := newMockDestination("test:mock") rec := newTestRecorder(out) @@ -171,7 +225,7 @@ func TestApplyConfig(t *testing.T) { } func TestEventRecorderConfigEqual_OutputCount(t *testing.T) { - a := Config{FileOutputs: []FileOutputConfig{{Path: "/tmp/a"}}} + a := Config{FileOutputs: []FileOutputConfig{{Name: "a", Path: "/tmp/a"}}} b := Config{} require.False(t, configEqual(a, b), "configs with different output counts must compare unequal") @@ -180,8 +234,8 @@ func TestEventRecorderConfigEqual_OutputCount(t *testing.T) { func TestEventRecorderConfigEqual_TypeMismatch(t *testing.T) { // Same total output count but in different per-type lists must // compare unequal. - a := Config{FileOutputs: []FileOutputConfig{{Path: "/tmp/a"}}} - b := Config{WebhookOutputs: []WebhookOutputConfig{{URL: mustParseURL(t, "https://example.com/h")}}} + a := Config{FileOutputs: []FileOutputConfig{{Name: "a", Path: "/tmp/a"}}} + b := Config{WebhookOutputs: []WebhookOutputConfig{{Name: "b", URL: mustParseURL(t, "https://example.com/h")}}} require.False(t, configEqual(a, b), "outputs of different types must compare unequal") } diff --git a/eventrecorder/stdout.go b/eventrecorder/stdout.go index 1b2652c468..e97949bdce 100644 --- a/eventrecorder/stdout.go +++ b/eventrecorder/stdout.go @@ -14,6 +14,7 @@ package eventrecorder import ( + "errors" "os" "google.golang.org/protobuf/encoding/protojson" @@ -22,13 +23,24 @@ import ( ) // StdoutOutputConfig configures a stdout event recorder output. -// There are no required fields; the presence of an entry in -// stdout_outputs is sufficient to enable the output. -type StdoutOutputConfig struct{} +type StdoutOutputConfig struct { + // Name identifies this output in metrics and logs. + Name string `yaml:"name" json:"name"` +} + +// UnmarshalYAML implements the yaml.Unmarshaler interface, validating +// the stdout output configuration. +func (c *StdoutOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { + type plain StdoutOutputConfig + if err := unmarshal((*plain)(c)); err != nil { + return errors.New("invalid event_recorder stdout output configuration") + } + _, err := outputIdentifier("stdout", c.Name) + return err +} // equal reports whether two stdout output configs are semantically equal. -// All StdoutOutputConfig values are identical since the type carries no fields. -func (c StdoutOutputConfig) equal(_ StdoutOutputConfig) bool { return true } +func (c StdoutOutputConfig) equal(o StdoutOutputConfig) bool { return c.Name == o.Name } // StdoutOutput writes events as newline-delimited JSON to os.Stdout. // This is the recommended output for container deployments where stdout @@ -36,10 +48,21 @@ func (c StdoutOutputConfig) equal(_ StdoutOutputConfig) bool { return true } // // Each event is serialized with protojson and followed by a newline so // log collectors receive one self-contained JSON object per line. -type StdoutOutput struct{} +type StdoutOutput struct { + name string +} + +// NewStdoutOutput creates a stdout event recorder output. +func NewStdoutOutput(cfg StdoutOutputConfig) (*StdoutOutput, error) { + name, err := outputIdentifier("stdout", cfg.Name) + if err != nil { + return nil, err + } + return &StdoutOutput{name: name}, nil +} // Name returns the stable identifier used in Prometheus metric labels. -func (s *StdoutOutput) Name() string { return "stdout" } +func (s *StdoutOutput) Name() string { return s.name } // SendEvent serializes the event as a JSON line and writes it to stdout. // It returns the byte count written (including the trailing newline) and diff --git a/eventrecorder/stdout_test.go b/eventrecorder/stdout_test.go index cdf2286337..affb9df802 100644 --- a/eventrecorder/stdout_test.go +++ b/eventrecorder/stdout_test.go @@ -47,12 +47,14 @@ func captureStdout(t *testing.T, fn func()) string { } func TestStdoutOutput_Name(t *testing.T) { - out := &StdoutOutput{} - require.Equal(t, "stdout", out.Name()) + out, err := NewStdoutOutput(StdoutOutputConfig{Name: "primary"}) + require.NoError(t, err) + require.Equal(t, "stdout:primary", out.Name()) } func TestStdoutOutput_SendEvent(t *testing.T) { - out := &StdoutOutput{} + out, err := NewStdoutOutput(StdoutOutputConfig{Name: "send"}) + require.NoError(t, err) got := captureStdout(t, func() { n, err := out.SendEvent(sampleEvent()) @@ -67,7 +69,8 @@ func TestStdoutOutput_SendEvent(t *testing.T) { } func TestStdoutOutput_SendEventTwice(t *testing.T) { - out := &StdoutOutput{} + out, err := NewStdoutOutput(StdoutOutputConfig{Name: "twice"}) + require.NoError(t, err) got := captureStdout(t, func() { _, err := out.SendEvent(sampleEvent()) @@ -85,7 +88,8 @@ func TestStdoutOutput_SendEventTwice(t *testing.T) { } func TestStdoutOutput_Close(t *testing.T) { - out := &StdoutOutput{} + out, err := NewStdoutOutput(StdoutOutputConfig{Name: "close"}) + require.NoError(t, err) require.NoError(t, out.Close(), "Close must be a no-op and return nil") } @@ -109,7 +113,9 @@ func TestStdoutOutput_IntegrationWithRecorder(t *testing.T) { os.Stdout = w t.Cleanup(func() { os.Stdout = old }) - rec := newTestRecorder(&StdoutOutput{}, mirror) + out, err := NewStdoutOutput(StdoutOutputConfig{Name: "integration"}) + require.NoError(t, err) + rec := newTestRecorder(out, mirror) defer rec.Close() rec.RecordEvent(recordCtx(), startupEvent) @@ -132,22 +138,23 @@ func TestStdoutOutput_IntegrationWithRecorder(t *testing.T) { // --- config tests. func TestStdoutOutputConfig_Equal(t *testing.T) { - // All StdoutOutputConfig values compare equal since the type has no fields. - a := StdoutOutputConfig{} - b := StdoutOutputConfig{} + a := StdoutOutputConfig{Name: "primary"} + b := StdoutOutputConfig{Name: "primary"} require.True(t, a.equal(b)) + b.Name = "secondary" + require.False(t, a.equal(b)) } func TestEventRecorderConfig_StdoutInTotalOutputs(t *testing.T) { cfg := Config{ - StdoutOutputs: []StdoutOutputConfig{{}}, + StdoutOutputs: []StdoutOutputConfig{{Name: "primary"}}, } require.Equal(t, 1, cfg.totalOutputs()) } func TestEventRecorderConfigEqual_Stdout(t *testing.T) { - a := Config{StdoutOutputs: []StdoutOutputConfig{{}}} - b := Config{StdoutOutputs: []StdoutOutputConfig{{}}} + a := Config{StdoutOutputs: []StdoutOutputConfig{{Name: "primary"}}} + b := Config{StdoutOutputs: []StdoutOutputConfig{{Name: "primary"}}} require.True(t, configEqual(a, b)) // Removing the stdout output makes them unequal. @@ -157,16 +164,17 @@ func TestEventRecorderConfigEqual_Stdout(t *testing.T) { func TestEventRecorderConfigEqual_StdoutVsFile(t *testing.T) { // Same total count but in different per-type lists must compare unequal. - a := Config{StdoutOutputs: []StdoutOutputConfig{{}}} - b := Config{FileOutputs: []FileOutputConfig{{Path: "/tmp/events.jsonl"}}} + a := Config{StdoutOutputs: []StdoutOutputConfig{{Name: "primary"}}} + b := Config{FileOutputs: []FileOutputConfig{{Name: "primary", Path: "/tmp/events.jsonl"}}} require.False(t, configEqual(a, b)) } func TestStdoutOutputConfig_UnmarshalYAML(t *testing.T) { - // An empty map is the natural YAML representation of a - // StdoutOutputConfig since it carries no fields. - raw := "stdout_outputs:\n - {}\n" + raw := "stdout_outputs:\n - name: primary\n" var cfg Config require.NoError(t, yaml.Unmarshal([]byte(raw), &cfg)) require.Len(t, cfg.StdoutOutputs, 1) + require.Equal(t, "primary", cfg.StdoutOutputs[0].Name) + + require.Error(t, yaml.Unmarshal([]byte("stdout_outputs:\n - {}\n"), &cfg)) } diff --git a/eventrecorder/webhook.go b/eventrecorder/webhook.go index 1a4972ae1b..08e786b951 100644 --- a/eventrecorder/webhook.go +++ b/eventrecorder/webhook.go @@ -20,7 +20,6 @@ import ( "io" "log/slog" "net/http" - "net/url" "reflect" "sync" "time" @@ -36,6 +35,8 @@ import ( // WebhookOutputConfig configures an HTTP webhook event recorder output. type WebhookOutputConfig struct { + // Name identifies this output in metrics and logs. + Name string `yaml:"name" json:"name"` // URL is the endpoint to POST each event to. URL *amcommoncfg.SecretURL `yaml:"url" json:"url"` // HTTPConfig configures the HTTP client used for webhook delivery. @@ -72,6 +73,13 @@ type WebhookOutputConfig struct { func (c *WebhookOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { type plain WebhookOutputConfig if err := unmarshal((*plain)(c)); err != nil { + return errors.New("invalid event_recorder webhook output configuration") + } + return c.validate() +} + +func (c WebhookOutputConfig) validate() error { + if _, err := outputIdentifier("webhook", c.Name); err != nil { return err } if c.URL == nil || c.URL.URL == nil { @@ -92,6 +100,9 @@ func (c *WebhookOutputConfig) UnmarshalYAML(unmarshal func(any) error) error { // equal reports whether two webhook output configs are semantically // equal. func (c WebhookOutputConfig) equal(o WebhookOutputConfig) bool { + if c.Name != o.Name { + return false + } aURL, bURL := "", "" if c.URL != nil { aURL = c.URL.String() @@ -169,6 +180,9 @@ type httpBatchConfig struct { // NewWebhookOutput creates a new webhook-based event recorder output. func NewWebhookOutput(cfg WebhookOutputConfig, dropsCounter *prometheus.CounterVec, logger *slog.Logger) (*WebhookOutput, error) { + if err := cfg.validate(); err != nil { + return nil, err + } var batch *httpBatchConfig if cfg.Batch { batch = newHTTPBatchConfig(cfg.BatchMaxEvents, cfg.BatchMaxBytes, cfg.BatchFlushInterval) @@ -231,7 +245,10 @@ func newWebhookOutput(cfg WebhookOutputConfig, kind string, batch *httpBatchConf } urlStr := cfg.URL.String() - name := fmt.Sprintf("%s:%s", kind, sanitizeURL(urlStr)) + name, err := outputIdentifier(kind, cfg.Name) + if err != nil { + return nil, err + } wo := &WebhookOutput{ client: client, url: urlStr, @@ -266,29 +283,7 @@ func newWebhookOutput(cfg WebhookOutputConfig, kind string, batch *httpBatchConf return wo, nil } -// sanitizeURL strips userinfo and query parameters from a URL string, -// returning only scheme://host/path. This prevents credentials from -// leaking into metrics labels and log messages. -func sanitizeURL(raw string) string { - u, err := url.Parse(raw) - if err != nil { - return "" - } - u.User = nil - u.RawQuery = "" - u.Fragment = "" - return u.String() -} - -func sanitizeSecretURL(u *amcommoncfg.SecretURL) string { - if u == nil || u.URL == nil { - return "" - } - return sanitizeURL(u.String()) -} - -// Name returns a stable identifier for this output. The URL is -// sanitized to avoid leaking credentials. +// Name returns a stable identifier for this output. func (wo *WebhookOutput) Name() string { return wo.name } @@ -421,7 +416,7 @@ func (wo *WebhookOutput) postWithRetry(data []byte) { if err == nil { return } - wo.logger.Warn("Event recorder HTTP output POST failed", "output", wo.name, "attempt", attempt+1, "err", err) + wo.logger.Warn("Event recorder HTTP output POST failed", "output", wo.name, "attempt", attempt+1) if attempt < wo.maxRetries-1 { backoff := min(wo.retryBackoff<\n", + yaml: "name: placeholder\nurl: \n", wantErr: true, }, { @@ -433,7 +471,7 @@ func TestWebhookOutputConfig_UnmarshalYAML(t *testing.T) { // itself (ParseURL only accepts http/https), so the error // surfaces before our validator runs. name: "non-http scheme", - yaml: "url: ftp://example.com/\n", + yaml: "name: ftp\nurl: ftp://example.com/\n", wantErr: true, }, } @@ -453,14 +491,35 @@ func TestWebhookOutputConfig_UnmarshalYAML(t *testing.T) { } } +func TestWebhookOutputConfig_MalformedURLDoesNotLeak(t *testing.T) { + const secret = "user:password@%zz/private?token=secret" + var cfg WebhookOutputConfig + err := yaml.Unmarshal([]byte("name: primary\nurl: https://"+secret+"\n"), &cfg) + require.Error(t, err) + require.NotContains(t, err.Error(), secret) + require.NotContains(t, err.Error(), "user") + require.NotContains(t, err.Error(), "password") + require.NotContains(t, err.Error(), "token=secret") +} + +func TestNewWebhookOutput_ValidatesProgrammaticConfig(t *testing.T) { + _, err := NewWebhookOutput(WebhookOutputConfig{Name: "primary"}, testWebhookDrops(), slog.Default()) + require.Error(t, err) + + _, err = NewWebhookOutput(WebhookOutputConfig{URL: mustParseURL(t, "https://example.com/hook")}, testWebhookDrops(), slog.Default()) + require.Error(t, err) +} + func TestEventRecorderConfigEqual_Webhook(t *testing.T) { a := Config{WebhookOutputs: []WebhookOutputConfig{{ + Name: "primary", URL: mustParseURL(t, "https://example.com/hook"), Timeout: model.Duration(10 * time.Second), Workers: 4, MaxRetries: 3, }}} b := Config{WebhookOutputs: []WebhookOutputConfig{{ + Name: "primary", URL: mustParseURL(t, "https://example.com/hook"), Timeout: model.Duration(10 * time.Second), Workers: 4, @@ -477,6 +536,9 @@ func TestEventRecorderConfigEqual_Webhook(t *testing.T) { b.WebhookOutputs[0].Workers = 8 require.False(t, configEqual(a, b)) b.WebhookOutputs[0].Workers = a.WebhookOutputs[0].Workers + b.WebhookOutputs[0].Name = "secondary" + require.False(t, configEqual(a, b)) + b.WebhookOutputs[0].Name = a.WebhookOutputs[0].Name b.WebhookOutputs[0].Batch = true require.False(t, configEqual(a, b))