diff --git a/CHANGELOG.md b/CHANGELOG.md index a5182671a2..03e7af78ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,7 @@ ## main / (unreleased) +* [FEATURE] notify: Add an Apache Kafka receiver using the webhook v4 JSON message format. + ## 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/app/reloader.go b/app/reloader.go index d9c62ca9c8..9137606158 100644 --- a/app/reloader.go +++ b/app/reloader.go @@ -15,6 +15,7 @@ package app import ( "context" + "errors" "fmt" "log/slog" "net/url" @@ -66,8 +67,9 @@ type reloader struct { tracingMgr *tracing.Manager // Short-lived components: atomically swapped on every reload. - dispatcher atomic.Pointer[dispatch.Dispatcher] - inhibitor atomic.Pointer[inhibit.Inhibitor] + dispatcher atomic.Pointer[dispatch.Dispatcher] + inhibitor atomic.Pointer[inhibit.Inhibitor] + integrations []notify.Integration // Functions and values used during reload. waitFunc func() time.Duration @@ -115,20 +117,20 @@ func (r *reloader) reload(conf *config.Config) error { // Build the map of receiver to integrations. receivers := make(map[string][]notify.Integration, len(activeReceivers)) - var integrationsNum int + var integrations []notify.Integration for _, rcv := range conf.Receivers { if _, found := activeReceivers[rcv.Name]; !found { // No need to build a receiver if no route is using it. configLogger.Info("skipping creation of receiver not referenced by any route", "receiver", rcv.Name) continue } - integrations, err := receiver.BuildReceiverIntegrations(rcv, tmpl, r.logger) + receiverIntegrations, err := receiver.BuildReceiverIntegrations(rcv, tmpl, r.logger) if err != nil { - return err + return errors.Join(err, notify.CloseIntegrations(integrations)) } // rcv.Name is guaranteed to be unique across all receivers. - receivers[rcv.Name] = integrations - integrationsNum += len(integrations) + receivers[rcv.Name] = receiverIntegrations + integrations = append(integrations, receiverIntegrations...) } // Build the map of time interval names to time interval definitions. @@ -151,7 +153,7 @@ func (r *reloader) reload(conf *config.Config) error { // it before stopping the old components keeps them running if it // errors. if err := r.tracingMgr.ApplyConfig(conf.TracingConfig); err != nil { - return fmt.Errorf("failed to apply tracing config: %w", err) + return errors.Join(fmt.Errorf("failed to apply tracing config: %w", err), notify.CloseIntegrations(integrations)) } // Reload event recorder outputs before stopping the old dispatcher so @@ -164,6 +166,10 @@ func (r *reloader) reload(conf *config.Config) error { if old := r.dispatcher.Load(); old != nil { old.Stop() } + if err := notify.CloseIntegrations(r.integrations); err != nil { + configLogger.Warn("failed to close receiver integrations", "err", err) + } + r.integrations = nil newInhibitor := inhibit.NewInhibitor(r.alerts, conf.InhibitRules, r.logger, r.eventRecorder) @@ -187,7 +193,7 @@ func (r *reloader) reload(conf *config.Config) error { ) r.metrics.configuredReceivers.Set(float64(len(activeReceivers))) - r.metrics.configuredIntegrations.Set(float64(integrationsNum)) + r.metrics.configuredIntegrations.Set(float64(len(integrations))) r.metrics.configuredInhibitionRules.Set(float64(len(conf.InhibitRules))) r.apih.Update(conf, func(ctx context.Context, labels model.LabelSet) { @@ -246,6 +252,7 @@ func (r *reloader) reload(conf *config.Config) error { go newDispatcher.Run(r.startTime.Add(r.dispatchStartDelay)) newDispatcher.WaitForLoading() r.dispatcher.Store(newDispatcher) + r.integrations = integrations return nil } @@ -260,5 +267,7 @@ func (r *reloader) stop() error { if d := r.dispatcher.Load(); d != nil { d.Stop() } - return nil + integrations := r.integrations + r.integrations = nil + return notify.CloseIntegrations(integrations) } diff --git a/app/reloader_test.go b/app/reloader_test.go index cefa661ac2..fe471cbbef 100644 --- a/app/reloader_test.go +++ b/app/reloader_test.go @@ -171,3 +171,33 @@ func TestReloader_StopIsNilSafe(t *testing.T) { // stop before any reload (both pointers nil) must not panic. require.NoError(t, r.stop()) } + +type closeTrackingNotifier struct { + closes int +} + +func (*closeTrackingNotifier) Notify(context.Context, ...*alert.Alert) (bool, error) { + return false, nil +} + +func (n *closeTrackingNotifier) Close() error { + n.closes++ + return nil +} + +type resolvedSender bool + +func (r resolvedSender) SendResolved() bool { return bool(r) } + +func TestReloaderClosesIntegrations(t *testing.T) { + r := newTestReloader(t) + n := &closeTrackingNotifier{} + r.integrations = []notify.Integration{ + notify.NewIntegration(n, resolvedSender(true), "test", 0, "receiver"), + } + + require.NoError(t, r.reload(mustConfig(t))) + require.Equal(t, 1, n.closes) + require.NoError(t, r.stop()) + require.Equal(t, 1, n.closes, "an integration should only be closed once") +} diff --git a/config/config.go b/config/config.go index 2355a9d715..ec8921ff38 100644 --- a/config/config.go +++ b/config/config.go @@ -35,6 +35,7 @@ import ( "github.com/prometheus/alertmanager/notify/discord" "github.com/prometheus/alertmanager/notify/incidentio" "github.com/prometheus/alertmanager/notify/jira" + "github.com/prometheus/alertmanager/notify/kafka" "github.com/prometheus/alertmanager/notify/mattermost" "github.com/prometheus/alertmanager/notify/msteams" "github.com/prometheus/alertmanager/notify/msteamsv2" @@ -480,6 +481,11 @@ func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { } iio.HTTPConfig = cmp.Or(iio.HTTPConfig, c.Global.HTTPConfig) } + for _, kafka := range rcv.KafkaConfigs { + if kafka == nil { + return errors.New("missing kafka config") + } + } for _, ogc := range rcv.OpsGenieConfigs { if ogc == nil { ogc = &opsgenie.OpsGenieConfig{} @@ -981,6 +987,7 @@ type Receiver struct { DiscordConfigs []*discord.DiscordConfig `yaml:"discord_configs,omitempty" json:"discord_configs,omitempty"` EmailConfigs []*EmailConfig `yaml:"email_configs,omitempty" json:"email_configs,omitempty"` IncidentioConfigs []*incidentio.IncidentioConfig `yaml:"incidentio_configs,omitempty" json:"incidentio_configs,omitempty"` + KafkaConfigs []*kafka.Config `yaml:"kafka_configs,omitempty" json:"kafka_configs,omitempty"` PagerdutyConfigs []*pagerduty.PagerdutyConfig `yaml:"pagerduty_configs,omitempty" json:"pagerduty_configs,omitempty"` SlackConfigs []*SlackConfig `yaml:"slack_configs,omitempty" json:"slack_configs,omitempty"` WebhookConfigs []*webhook.WebhookConfig `yaml:"webhook_configs,omitempty" json:"webhook_configs,omitempty"` diff --git a/config/config_test.go b/config/config_test.go index 2a37784de3..33d57015b6 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -936,6 +936,10 @@ receivers: integration: "jira_configs", expectedErr: "missing jira config", }, + { + integration: "kafka_configs", + expectedErr: "missing kafka config", + }, { integration: "mattermost_configs", expectedErr: "missing mattermost config", diff --git a/config/receiver/receiver.go b/config/receiver/receiver.go index cdbacc7629..e8f2326356 100644 --- a/config/receiver/receiver.go +++ b/config/receiver/receiver.go @@ -26,6 +26,7 @@ import ( "github.com/prometheus/alertmanager/notify/email" "github.com/prometheus/alertmanager/notify/incidentio" "github.com/prometheus/alertmanager/notify/jira" + "github.com/prometheus/alertmanager/notify/kafka" "github.com/prometheus/alertmanager/notify/mattermost" "github.com/prometheus/alertmanager/notify/msteams" "github.com/prometheus/alertmanager/notify/msteamsv2" @@ -111,6 +112,9 @@ func BuildReceiverIntegrations(nc config.Receiver, tmpl *template.Template, logg for i, c := range nc.IncidentioConfigs { add("incidentio", i, c, func(l *slog.Logger) (notify.Notifier, error) { return incidentio.New(c, tmpl, l, httpOpts...) }) } + for i, c := range nc.KafkaConfigs { + add("kafka", i, c, func(l *slog.Logger) (notify.Notifier, error) { return kafka.New(c, tmpl, l) }) + } for i, c := range nc.RocketchatConfigs { add("rocketchat", i, c, func(l *slog.Logger) (notify.Notifier, error) { return rocketchat.New(c, tmpl, l, httpOpts...) }) } @@ -118,5 +122,8 @@ func BuildReceiverIntegrations(nc config.Receiver, tmpl *template.Template, logg add("mattermost", i, c, func(l *slog.Logger) (notify.Notifier, error) { return mattermost.New(c, tmpl, l, httpOpts...) }) } - return integrations, errs + if errs != nil { + return nil, errors.Join(errs, notify.CloseIntegrations(integrations)) + } + return integrations, nil } diff --git a/config/receiver/receiver_test.go b/config/receiver/receiver_test.go index 5e45478e32..4c99e46e91 100644 --- a/config/receiver/receiver_test.go +++ b/config/receiver/receiver_test.go @@ -24,6 +24,7 @@ import ( "github.com/prometheus/alertmanager/config" "github.com/prometheus/alertmanager/notify" + notifykafka "github.com/prometheus/alertmanager/notify/kafka" ) type sendResolved bool @@ -36,6 +37,23 @@ func TestBuildReceiverIntegrations(t *testing.T) { err bool exp []notify.Integration }{ + { + receiver: config.Receiver{ + Name: "foo", + KafkaConfigs: []*notifykafka.Config{ + { + Brokers: []string{"127.0.0.1:1"}, + Topic: "alerts", + NotifierConfig: amcommoncfg.NotifierConfig{ + VSendResolved: true, + }, + }, + }, + }, + exp: []notify.Integration{ + notify.NewIntegration(nil, sendResolved(true), "kafka", 0, "foo"), + }, + }, { receiver: config.Receiver{ Name: "foo", @@ -79,6 +97,7 @@ func TestBuildReceiverIntegrations(t *testing.T) { return } require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, notify.CloseIntegrations(integrations)) }) require.Len(t, integrations, len(tc.exp)) for i := range tc.exp { require.Equal(t, tc.exp[i].SendResolved(), integrations[i].SendResolved()) diff --git a/docs/configuration.md b/docs/configuration.md index be9a73f841..0bf4f27eca 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -845,6 +845,8 @@ msteamsv2_configs: [ - , ... ] jira_configs: [ - , ... ] +kafka_configs: + [ - , ... ] opsgenie_configs: [ - , ... ] pagerduty_configs: @@ -2008,6 +2010,45 @@ There is a list of [integrations](https://prometheus.io/docs/operating/integrations/#alertmanager-webhook-receiver) with this feature. +### `` + +The Kafka receiver produces one record for each Alertmanager notification group. +The record key is the notification's group key, keeping notifications for the +same group on the same partition. The record value is a JSON object using the +same version 4 format as the [webhook receiver](#webhook_config). The +`truncatedAlerts` field is always zero because the Kafka receiver sends every +alert in the group. + +The target topic must already exist, unless the brokers are configured to +automatically create topics. Alertmanager does not create it. + +```yaml +# Whether to notify about resolved alerts. +[ send_resolved: | default = true ] + +# Seed Kafka brokers in host:port form. At least one broker is required. +brokers: + [ - ... ] + +# Topic to produce notifications to. +topic: + +# Client identifier reported to the brokers. +[ client_id: | default = "alertmanager" ] + +# Producer acknowledgement level. With "none", broker-side delivery failures +# cannot be reported to Alertmanager. "leader" waits for the partition leader. +# "all" waits for all in-sync replicas. +[ acks: <"none" | "leader" | "all"> | default = "leader" ] + +# Compression codec for record batches. When omitted, batches are uncompressed. +[ compression: <"none" | "gzip" | "snappy" | "lz4" | "zstd"> ] + +# TLS configuration for broker connections. When unset, connections use +# PLAINTEXT. +[ tls_config: ] +``` + ### `` incident.io notifications are sent via the [incident.io Alert Sources API](https://api-docs.incident.io/tag/Alert-Sources-V2#operation/Alert%20Sources%20V2_Create). @@ -2272,9 +2313,8 @@ topic: [ format: <"json" | "protobuf"> | default = "json" ] # Producer acknowledgement level. "leader" matches the franz-go default -# and minimizes Alertmanager's exposure to Kafka latency. "all" enables -# the idempotent producer for at-least-once durability at the cost of -# higher latency. +# and minimizes Alertmanager's exposure to Kafka latency. "all" waits for +# all in-sync replicas at the cost of higher latency. [ acks: <"none" | "leader" | "all"> | default = "leader" ] # Compression codec for record batches. When omitted, batches are sent diff --git a/docs/integrations.md b/docs/integrations.md index 3bd96fcd02..997bb29705 100644 --- a/docs/integrations.md +++ b/docs/integrations.md @@ -14,6 +14,7 @@ Alertmanager supports a number of notification integrations via the [configurati | [Email](https://en.wikipedia.org/wiki/Email) | [email_config](configuration.md#email_config) | - | [SMTP](https://en.wikipedia.org/wiki/Simple_Mail_Transfer_Protocol) | | [incident.io](https://incident.io/) | [incidentio_config](configuration.md#incidentio_config) | [Alert Sources Documentation](https://api-docs.incident.io/tag/Alert-Sources-V2) | [Alert Sources V2 API](https://api-docs.incident.io/tag/Alert-Sources-V2#operation/Alert%20Sources%20V2_Create) | | [Jira](https://www.atlassian.com/software/jira) | [jira_config](configuration.md#jira_config) | [Jira Cloud Platform](https://developer.atlassian.com/cloud/jira/platform/) | [REST API v2](https://developer.atlassian.com/cloud/jira/platform/rest/v2/intro/) / [REST API v3](https://developer.atlassian.com/cloud/jira/platform/rest/v3/intro/) | +| [Apache Kafka](https://kafka.apache.org/) | [kafka_config](configuration.md#kafka_config) | [Kafka Documentation](https://kafka.apache.org/documentation/) | [Kafka Protocol](https://kafka.apache.org/protocol) | | [Mattermost](https://mattermost.com/) | [mattermost_config](configuration.md#mattermost_config) | [Incoming Webhooks](https://developers.mattermost.com/integrate/webhooks/incoming/) | [Mattermost Webhook API](https://developers.mattermost.com/integrate/webhooks/incoming/) | | [Microsoft Teams](https://www.microsoft.com/en-us/microsoft-teams/) | [msteams_config](configuration.md#msteams_config) | [Incoming Webhooks (Deprecated)](https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/what-are-webhooks-and-connectors) | [Microsoft Teams Connectors](https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/what-are-webhooks-and-connectors) | | [Microsoft Teams v2](https://www.microsoft.com/en-us/microsoft-teams/) | [msteamsv2_config](configuration.md#msteamsv2_config) | [Workflows for Teams](https://support.microsoft.com/en-gb/office/create-incoming-webhooks-with-workflows-for-microsoft-teams-8ae491c7-0394-4861-ba59-055e33f75498) | [Power Automate Flows](https://learn.microsoft.com/en-us/power-automate/teams/overview) | diff --git a/kafka/kafka.go b/kafka/kafka.go index e49d36e5ed..95b47bcd1f 100644 --- a/kafka/kafka.go +++ b/kafka/kafka.go @@ -20,10 +20,8 @@ // producer buffers, consumer group orchestration, message // serialisation) live in the calling package. // -// At the moment the only consumer of this package is the event -// recorder's Kafka output (eventrecorder/kafka.go). A future Kafka -// receiver (see github.com/prometheus/alertmanager/issues/1996) is the -// other intended user. +// The event recorder's Kafka output and the Kafka notification receiver +// both use this package. package kafka import ( diff --git a/notify/integration_close_test.go b/notify/integration_close_test.go new file mode 100644 index 0000000000..e2d8378a61 --- /dev/null +++ b/notify/integration_close_test.go @@ -0,0 +1,43 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package notify + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/prometheus/alertmanager/types" +) + +type closableNotifier struct { + closed bool +} + +func (n *closableNotifier) Notify(context.Context, ...*types.Alert) (bool, error) { + return false, nil +} + +func (n *closableNotifier) Close() error { + n.closed = true + return nil +} + +func TestIntegrationClose(t *testing.T) { + n := &closableNotifier{} + i := NewIntegration(n, sendResolved(false), "test", 0, "receiver") + require.NoError(t, i.Close()) + require.True(t, n.closed) +} diff --git a/notify/kafka/config.go b/notify/kafka/config.go new file mode 100644 index 0000000000..3282368b4d --- /dev/null +++ b/notify/kafka/config.go @@ -0,0 +1,70 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "errors" + + commoncfg "github.com/prometheus/common/config" + + amcommoncfg "github.com/prometheus/alertmanager/config/common" + sharedkafka "github.com/prometheus/alertmanager/kafka" +) + +var defaultConfig = Config{ + NotifierConfig: amcommoncfg.NotifierConfig{VSendResolved: true}, +} + +// Config configures notifications sent to a Kafka topic. +type Config struct { + amcommoncfg.NotifierConfig `yaml:",inline" json:",inline"` + + Brokers []string `yaml:"brokers" json:"brokers"` + Topic string `yaml:"topic" json:"topic"` + ClientID string `yaml:"client_id,omitempty" json:"client_id,omitempty"` + Acks sharedkafka.Acks `yaml:"acks,omitempty" json:"acks,omitempty"` + Compression sharedkafka.Compression `yaml:"compression,omitempty" json:"compression,omitempty"` + TLSConfig *commoncfg.TLSConfig `yaml:"tls_config,omitempty" json:"tls_config,omitempty"` +} + +// UnmarshalYAML implements the yaml.Unmarshaler interface. +func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { + *c = defaultConfig + type plain Config + if err := unmarshal((*plain)(c)); err != nil { + return err + } + return c.validate() +} + +func (c Config) validate() error { + if err := c.clientOptions().Validate(); err != nil { + return err + } + if c.Topic == "" { + return errors.New("kafka: topic is required") + } + return nil +} + +func (c Config) clientOptions() sharedkafka.ClientOptions { + return sharedkafka.ClientOptions{ + Brokers: c.Brokers, + Topic: c.Topic, + ClientID: c.ClientID, + Acks: c.Acks, + Compression: c.Compression, + TLSConfig: c.TLSConfig, + } +} diff --git a/notify/kafka/config_test.go b/notify/kafka/config_test.go new file mode 100644 index 0000000000..dafc092812 --- /dev/null +++ b/notify/kafka/config_test.go @@ -0,0 +1,60 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "testing" + + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v2" + + sharedkafka "github.com/prometheus/alertmanager/kafka" +) + +func TestConfigUnmarshalYAML(t *testing.T) { + var c Config + require.NoError(t, yaml.Unmarshal([]byte(` +brokers: [kafka-1:9092, kafka-2:9092] +topic: alerts +client_id: alertmanager-test +acks: all +compression: zstd +`), &c)) + require.True(t, c.SendResolved()) + require.Equal(t, []string{"kafka-1:9092", "kafka-2:9092"}, c.Brokers) + require.Equal(t, "alerts", c.Topic) + require.Equal(t, "alertmanager-test", c.ClientID) + require.Equal(t, sharedkafka.AcksAll, c.Acks) + require.Equal(t, sharedkafka.CompressionZstd, c.Compression) +} + +func TestConfigValidation(t *testing.T) { + for _, tc := range []struct { + name string + yaml string + err string + }{ + {name: "missing brokers", yaml: "topic: alerts", err: "at least one broker"}, + {name: "empty broker", yaml: "brokers: ['']\ntopic: alerts", err: "broker entries must be non-empty"}, + {name: "missing topic", yaml: "brokers: [kafka:9092]", err: "topic is required"}, + {name: "invalid acks", yaml: "brokers: [kafka:9092]\ntopic: alerts\nacks: majority", err: "unknown acks"}, + {name: "invalid compression", yaml: "brokers: [kafka:9092]\ntopic: alerts\ncompression: deflate", err: "unknown compression"}, + } { + t.Run(tc.name, func(t *testing.T) { + var c Config + err := yaml.Unmarshal([]byte(tc.yaml), &c) + require.ErrorContains(t, err, tc.err) + }) + } +} diff --git a/notify/kafka/kafka.go b/notify/kafka/kafka.go new file mode 100644 index 0000000000..86b8243175 --- /dev/null +++ b/notify/kafka/kafka.go @@ -0,0 +1,100 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package kafka provides notifications to Apache Kafka. +package kafka + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "github.com/twmb/franz-go/pkg/kgo" + + sharedkafka "github.com/prometheus/alertmanager/kafka" + "github.com/prometheus/alertmanager/notify" + "github.com/prometheus/alertmanager/notify/webhook" + "github.com/prometheus/alertmanager/template" + "github.com/prometheus/alertmanager/types" +) + +type producer interface { + ProduceSync(context.Context, ...*kgo.Record) kgo.ProduceResults + Close() +} + +// Notifier implements a notifier that produces grouped alerts to Kafka. +type Notifier struct { + conf *Config + tmpl *template.Template + logger *slog.Logger + producer producer +} + +// New returns a new Kafka notifier. +func New(conf *Config, tmpl *template.Template, logger *slog.Logger) (*Notifier, error) { + if err := conf.validate(); err != nil { + return nil, err + } + if logger == nil { + logger = slog.New(slog.DiscardHandler) + } + opts, err := sharedkafka.BuildOpts(conf.clientOptions(), logger) + if err != nil { + return nil, err + } + client, err := kgo.NewClient(opts...) + if err != nil { + return nil, fmt.Errorf("kafka: creating producer: %w", err) + } + sharedkafka.PingInBackground(client, logger) + return &Notifier{ + conf: conf, + tmpl: tmpl, + logger: logger, + producer: client, + }, nil +} + +// Notify implements the notify.Notifier interface. +func (n *Notifier) Notify(ctx context.Context, alerts ...*types.Alert) (bool, error) { + groupKey, err := notify.ExtractGroupKey(ctx) + if err != nil { + return false, err + } + msg := webhook.Message{ + Version: "4", + Data: notify.GetTemplateData(ctx, n.tmpl, alerts, n.logger), + GroupKey: groupKey.String(), + } + payload, err := json.Marshal(&msg) + if err != nil { + return false, fmt.Errorf("kafka: encoding notification: %w", err) + } + results := n.producer.ProduceSync(ctx, &kgo.Record{ + Topic: n.conf.Topic, + Key: []byte(groupKey.String()), + Value: payload, + }) + if err := results.FirstErr(); err != nil { + return true, fmt.Errorf("kafka: producing notification: %w", err) + } + return false, nil +} + +// Close releases the Kafka producer resources. +func (n *Notifier) Close() error { + n.producer.Close() + return nil +} diff --git a/notify/kafka/kafka_test.go b/notify/kafka/kafka_test.go new file mode 100644 index 0000000000..6542bf68b3 --- /dev/null +++ b/notify/kafka/kafka_test.go @@ -0,0 +1,122 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/prometheus/common/model" + "github.com/prometheus/common/promslog" + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kfake" + "github.com/twmb/franz-go/pkg/kgo" + + "github.com/prometheus/alertmanager/notify" + notifytest "github.com/prometheus/alertmanager/notify/test" + "github.com/prometheus/alertmanager/notify/webhook" + "github.com/prometheus/alertmanager/types" +) + +func TestNotifyProducesWebhookV4Message(t *testing.T) { + const topic = "alerts" + cluster, err := kfake.NewCluster(kfake.NumBrokers(1), kfake.SeedTopics(1, topic)) + require.NoError(t, err) + t.Cleanup(cluster.Close) + + n, err := New(&Config{Brokers: cluster.ListenAddrs(), Topic: topic}, notifytest.CreateTmpl(t), promslog.NewNopLogger()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, n.Close()) }) + + alert := &types.Alert{Alert: model.Alert{ + Labels: model.LabelSet{"alertname": "HighLatency", "severity": "critical"}, + Annotations: model.LabelSet{"summary": "Latency is high"}, + StartsAt: time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC), + }} + const groupKey = `{}/{alertname="HighLatency"}` + ctx := notify.WithGroupKey(context.Background(), groupKey) + ctx = notify.WithReceiverName(ctx, "kafka-alerts") + ctx = notify.WithGroupLabels(ctx, model.LabelSet{"alertname": "HighLatency"}) + + retry, err := n.Notify(ctx, alert) + require.NoError(t, err) + require.False(t, retry) + + consumer, err := kgo.NewClient( + kgo.SeedBrokers(cluster.ListenAddrs()...), + kgo.ConsumeTopics(topic), + kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), + ) + require.NoError(t, err) + defer consumer.Close() + fetchCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + fetches := consumer.PollFetches(fetchCtx) + require.NoError(t, fetches.Err()) + records := fetches.Records() + require.Len(t, records, 1) + require.Equal(t, groupKey, string(records[0].Key)) + + var msg webhook.Message + require.NoError(t, json.Unmarshal(records[0].Value, &msg)) + require.Equal(t, "4", msg.Version) + require.Equal(t, groupKey, msg.GroupKey) + require.Equal(t, "kafka-alerts", msg.Receiver) + require.Equal(t, "firing", msg.Status) + require.Len(t, msg.Alerts, 1) + require.Equal(t, "HighLatency", msg.Alerts[0].Labels["alertname"]) +} + +type fakeProducer struct { + err error + closed bool +} + +func (p *fakeProducer) ProduceSync(_ context.Context, records ...*kgo.Record) kgo.ProduceResults { + results := make(kgo.ProduceResults, len(records)) + for i, record := range records { + results[i] = kgo.ProduceResult{Record: record, Err: p.err} + } + return results +} + +func (p *fakeProducer) Close() { p.closed = true } + +func TestNotifyErrors(t *testing.T) { + n := &Notifier{ + conf: &Config{Topic: "alerts"}, + tmpl: notifytest.CreateTmpl(t), + logger: promslog.NewNopLogger(), + producer: &fakeProducer{err: errors.New("broker unavailable")}, + } + + retry, err := n.Notify(context.Background(), &types.Alert{}) + require.ErrorContains(t, err, "group key missing") + require.False(t, retry) + + ctx := notify.WithGroupKey(context.Background(), "group") + retry, err = n.Notify(ctx, &types.Alert{}) + require.ErrorContains(t, err, "broker unavailable") + require.True(t, retry) +} + +func TestClose(t *testing.T) { + p := &fakeProducer{} + n := &Notifier{producer: p} + require.NoError(t, n.Close()) + require.True(t, p.closed) +} diff --git a/notify/metrics.go b/notify/metrics.go index 33a58cf464..c073652b85 100644 --- a/notify/metrics.go +++ b/notify/metrics.go @@ -125,6 +125,7 @@ func (m *Metrics) InitializeFor(receiver map[string][]Integration) { "msteamsv2", "incidentio", "jira", + "kafka", "rocketchat", "mattermost", } { diff --git a/notify/notify.go b/notify/notify.go index 0bb2dc6e62..80b5231417 100644 --- a/notify/notify.go +++ b/notify/notify.go @@ -65,6 +65,10 @@ type Notifier interface { Notify(context.Context, ...*alert.Alert) (bool, error) } +type closer interface { + Close() error +} + // Integration wraps a notifier and its configuration to be uniquely identified // by name and index from its origin in the configuration. type Integration struct { @@ -127,6 +131,23 @@ func (i *Integration) String() string { return fmt.Sprintf("%s[%d]", i.name, i.idx) } +// Close releases resources owned by the underlying notifier, if any. +func (i *Integration) Close() error { + if c, ok := i.notifier.(closer); ok { + return c.Close() + } + return nil +} + +// CloseIntegrations releases resources owned by the supplied integrations. +func CloseIntegrations(integrations []Integration) error { + var errs error + for i := range integrations { + errs = errors.Join(errs, integrations[i].Close()) + } + return errs +} + // A Stage processes alerts under the constraints of the given context. type Stage interface { Exec(ctx context.Context, l *slog.Logger, alerts ...*alert.Alert) (context.Context, []*alert.Alert, error)