notify: add Kafka receiver

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## main / (unreleased)
 
+* [FEATURE] notify: Add an Apache Kafka receiver using the webhook v4 JSON message format.
 * [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.
 * [ENHANCEMENT] notify: The discord and webex integrations now report a failure `reason` on `alertmanager_notifications_failed_total`.
 * [ENHANCEMENT] eventrecorder: Add optional webhook batching.

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)
 }

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

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"
@@ -478,6 +479,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{}
@@ -979,6 +985,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"`

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",

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,12 +112,18 @@ 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...) })
 	}
 	for i, c := range nc.MattermostConfigs {
 		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
 }

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())

docs/configuration.md

@@ -845,6 +845,8 @@ msteamsv2_configs:
   [ - <msteamsv2_config>, ... ]
 jira_configs:
   [ - <jira_config>, ... ]
+kafka_configs:
+  [ - <kafka_config>, ... ]
 opsgenie_configs:
   [ - <opsgenie_config>, ... ]
 pagerduty_configs:
@@ -2008,6 +2010,45 @@ There is a list of
 [integrations](https://prometheus.io/docs/operating/integrations/#alertmanager-webhook-receiver) with
 this feature.
 
+### `<kafka_config>`
+
+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: <boolean> | default = true ]
+
+# Seed Kafka brokers in host:port form. At least one broker is required.
+brokers:
+  [ - <string> ... ]
+
+# Topic to produce notifications to.
+topic: <string>
+
+# Client identifier reported to the brokers.
+[ client_id: <string> | 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 and enables idempotent writes.
+[ 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: <tls_config> ]
+```
+
 ### `<incidentio_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).

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) |

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 (

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