Support for building Prometheus Alertmanager integrations (#339)

* extract receiver builders to functions to reuse in Mimir
* add conversion logic for ClientOptions to upstream
* add support for prometheus integrations

Author: yuri-tceretian

go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -57,7 +58,6 @@ require (
 	github.com/hashicorp/go-msgpack v0.5.5 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-sockaddr v1.0.6 // indirect
-	github.com/hashicorp/go-uuid v1.0.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/hashicorp/memberlist v0.5.0 // indirect
@@ -87,9 +87,11 @@ require (
 	github.com/shopspring/decimal v1.2.0 // indirect
 	github.com/shurcooL/httpfs v0.0.0-20190707220628-8d4bc4ba7749 // indirect
 	github.com/shurcooL/vfsgen v0.0.0-20200824052919-0d455de96546 // indirect
-	github.com/spf13/cast v1.3.1 // indirect
+	github.com/spf13/cast v1.5.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	go.mongodb.org/mongo-driver v1.13.1 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/httptrace/otelhttptrace v0.55.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.55.0 // indirect
 	go.opentelemetry.io/otel v1.30.0 // indirect
 	go.opentelemetry.io/otel/metric v1.30.0 // indirect
 	go.opentelemetry.io/otel/trace v1.30.0 // indirect
@@ -102,6 +104,7 @@ require (
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
+	gopkg.in/telebot.v3 v3.2.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
 

go.sum

@@ -15,15 +15,17 @@ import (
 	"github.com/benbjohnson/clock"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	commoncfg "github.com/prometheus/common/config"
 
 	"github.com/grafana/alerting/receivers"
 )
 
 var ErrInvalidMethod = errors.New("webhook only supports HTTP methods PUT or POST")
 
 type clientConfiguration struct {
-	userAgent string
-	dialer    net.Dialer // We use Dialer here instead of DialContext as our mqtt client doesn't support DialContext.
+	userAgent    string
+	dialer       net.Dialer // We use Dialer here instead of DialContext as our mqtt client doesn't support DialContext.
+	customDialer bool
 }
 
 // defaultDialTimeout is the default timeout for the dialer, 30 seconds to match http.DefaultTransport.
@@ -63,9 +65,28 @@ func WithUserAgent(userAgent string) ClientOption {
 func WithDialer(dialer net.Dialer) ClientOption {
 	return func(c *clientConfiguration) {
 		c.dialer = dialer
+		c.customDialer = true
 	}
 }
 
+func ToHTTPClientOption(option ...ClientOption) []commoncfg.HTTPClientOption {
+	cfg := clientConfiguration{}
+	for _, opt := range option {
+		if opt == nil {
+			continue
+		}
+		opt(&cfg)
+	}
+	result := make([]commoncfg.HTTPClientOption, 0, len(option))
+	if cfg.userAgent != "" {
+		result = append(result, commoncfg.WithUserAgent(cfg.userAgent))
+	}
+	if cfg.customDialer {
+		result = append(result, commoncfg.WithDialContextFunc(cfg.dialer.DialContext))
+	}
+	return result
+}
+
 func (ns *Client) SendWebhook(ctx context.Context, l log.Logger, webhook *receivers.SendWebhookSettings) error {
 	// This method was moved from https://github.com/grafana/grafana/blob/71d04a326be9578e2d678f23c1efa61768e0541f/pkg/services/notifications/webhook.go#L38
 	if webhook.HTTPMethod == "" {

http/client.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"reflect"
 	"testing"
 	"time"
 
@@ -190,3 +191,24 @@ func TestSendWebhookHMAC(t *testing.T) {
 		require.NotEmpty(t, timestamp)
 	})
 }
+
+func TestToHTTPClientOption(t *testing.T) {
+	// this test guards against adding new fields to the configuration structure without updating the conversion function
+	t.Run("empty converts to empty", func(t *testing.T) {
+		require.Empty(t, ToHTTPClientOption())
+		require.Empty(t, ToHTTPClientOption(nil))
+	})
+
+	var f ClientOption = func(configuration *clientConfiguration) {
+		configuration.userAgent = "test"
+		configuration.dialer = net.Dialer{Timeout: 5 * time.Second}
+		configuration.customDialer = true
+	}
+	actual := ToHTTPClientOption(f)
+	require.Len(t, actual, 2)
+
+	// Verify number of fields using reflection
+	tp := reflect.TypeOf(clientConfiguration{})
+	// You need to increase the number of fields covered in this test, if you add a new field to the configuration struct.
+	require.Equalf(t, 3, tp.NumField(), "Not all fields are converted to HTTPClientOption, which means that the configuration will not be supported in upstream integrations")
+}

http/client_test.go

@@ -1,11 +1,35 @@
 package notify
 
 import (
+	"context"
+	"fmt"
+	"sync"
+
 	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/notify"
+	"github.com/prometheus/alertmanager/types"
+	commoncfg "github.com/prometheus/common/config"
+
+	promDiscord "github.com/prometheus/alertmanager/notify/discord"
+	promEmail "github.com/prometheus/alertmanager/notify/email"
+	promMsteams "github.com/prometheus/alertmanager/notify/msteams"
+	promOpsgenie "github.com/prometheus/alertmanager/notify/opsgenie"
+	promPagerduty "github.com/prometheus/alertmanager/notify/pagerduty"
+	promPushover "github.com/prometheus/alertmanager/notify/pushover"
+	promSlack "github.com/prometheus/alertmanager/notify/slack"
+	promSns "github.com/prometheus/alertmanager/notify/sns"
+	promTelegram "github.com/prometheus/alertmanager/notify/telegram"
+	promVictorops "github.com/prometheus/alertmanager/notify/victorops"
+	promWebex "github.com/prometheus/alertmanager/notify/webex"
+	promWebhook "github.com/prometheus/alertmanager/notify/webhook"
+	promWechat "github.com/prometheus/alertmanager/notify/wechat"
+	"github.com/prometheus/alertmanager/template"
 
 	"github.com/grafana/alerting/http"
 	"github.com/grafana/alerting/images"
+	"github.com/grafana/alerting/notify/nfstatus"
 	"github.com/grafana/alerting/receivers"
 	"github.com/grafana/alerting/receivers/alertmanager"
 	"github.com/grafana/alerting/receivers/dinding"
@@ -35,9 +59,11 @@ import (
 
 type WrapNotifierFunc func(integrationName string, notifier notify.Notifier) notify.Notifier
 
-// BuildReceiverIntegrations creates integrations for each configured notification channel in GrafanaReceiverConfig.
+var NoWrap WrapNotifierFunc = func(_ string, notifier notify.Notifier) notify.Notifier { return notifier }
+
+// BuildGrafanaReceiverIntegrations creates integrations for each configured notification channel in GrafanaReceiverConfig.
 // It returns a slice of Integration objects, one for each notification channel, along with any errors that occurred.
-func BuildReceiverIntegrations(
+func BuildGrafanaReceiverIntegrations(
 	receiver GrafanaReceiverConfig,
 	tmpl *templates.Template,
 	img images.Provider,
@@ -133,3 +159,170 @@ func BuildReceiverIntegrations(
 	}
 	return integrations
 }
+
+// BuildPrometheusReceiverIntegrations builds a list of integration notifiers off of a receiver config.
+// Taken from https://github.com/grafana/mimir/blob/fa489e696481fe0b7b97598077565dc5027afa84/pkg/alertmanager/alertmanager.go#L754
+// which is taken from https://github.com/prometheus/alertmanager/blob/94d875f1227b29abece661db1a68c001122d1da5/cmd/alertmanager/main.go#L112-L159.
+func BuildPrometheusReceiverIntegrations(
+	nc config.Receiver,
+	tmplProvider TemplatesProvider,
+	httpClientOptions []http.ClientOption,
+	logger log.Logger,
+	wrapper WrapNotifierFunc,
+) ([]*nfstatus.Integration, error) {
+	var (
+		errs         types.MultiError
+		integrations []*nfstatus.Integration
+		tmpl         *template.Template
+		httpOps      []commoncfg.HTTPClientOption
+		initOnce     = sync.OnceFunc(func() { // lazy evaluate template so we do not create one if we don't need it
+			httpOps = http.ToHTTPClientOption(httpClientOptions...)
+			t, err := tmplProvider.GetTemplate(templates.MimirKind)
+			if err != nil {
+				errs.Add(err)
+				return
+			}
+			tmpl = t
+		})
+		add = func(name string, i int, rs notify.ResolvedSender, f func(l log.Logger) (notify.Notifier, error)) {
+			initOnce()
+			n, err := f(log.With(logger, "integration", name))
+			if err != nil {
+				errs.Add(err)
+				return
+			}
+			if wrapper != nil {
+				n = wrapper(name, n)
+			}
+			integrations = append(integrations, nfstatus.NewIntegration(n, rs, name, i, nc.Name))
+		}
+	)
+
+	for i, c := range nc.WebhookConfigs {
+		add("webhook", i, c, func(l log.Logger) (notify.Notifier, error) { return promWebhook.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.EmailConfigs {
+		add("email", i, c, func(l log.Logger) (notify.Notifier, error) { return promEmail.New(c, tmpl, l), nil })
+	}
+	for i, c := range nc.PagerdutyConfigs {
+		add("pagerduty", i, c, func(l log.Logger) (notify.Notifier, error) { return promPagerduty.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.OpsGenieConfigs {
+		add("opsgenie", i, c, func(l log.Logger) (notify.Notifier, error) { return promOpsgenie.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.WechatConfigs {
+		add("wechat", i, c, func(l log.Logger) (notify.Notifier, error) { return promWechat.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.SlackConfigs {
+		add("slack", i, c, func(l log.Logger) (notify.Notifier, error) { return promSlack.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.VictorOpsConfigs {
+		add("victorops", i, c, func(l log.Logger) (notify.Notifier, error) { return promVictorops.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.PushoverConfigs {
+		add("pushover", i, c, func(l log.Logger) (notify.Notifier, error) { return promPushover.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.SNSConfigs {
+		add("sns", i, c, func(l log.Logger) (notify.Notifier, error) { return promSns.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.TelegramConfigs {
+		add("telegram", i, c, func(l log.Logger) (notify.Notifier, error) { return promTelegram.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.DiscordConfigs {
+		add("discord", i, c, func(l log.Logger) (notify.Notifier, error) { return promDiscord.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.WebexConfigs {
+		add("webex", i, c, func(l log.Logger) (notify.Notifier, error) { return promWebex.New(c, tmpl, l, httpOps...) })
+	}
+	for i, c := range nc.MSTeamsConfigs {
+		add("msteams", i, c, func(l log.Logger) (notify.Notifier, error) { return promMsteams.New(c, tmpl, l, httpOps...) })
+	}
+	// If we add support for more integrations, we need to add them to validation as well. See validation.allowedIntegrationNames field.
+	if errs.Len() > 0 {
+		return nil, &errs
+	}
+	return integrations, nil
+}
+
+// BuildReceiversIntegrations builds integrations for the provided API receivers and returns them mapped by receiver name.
+// It ensures uniqueness of receivers by the name, overwriting duplicates and logs warnings.
+// Returns an error if any integration fails during its construction.
+func BuildReceiversIntegrations(
+	tenantID int64,
+	apiReceivers []*APIReceiver,
+	templ TemplatesProvider,
+	images images.Provider,
+	decryptFn GetDecryptedValueFn,
+	emailSender receivers.EmailSender,
+	httpClientOptions []http.ClientOption,
+	notifierFunc WrapNotifierFunc,
+	version string,
+	logger log.Logger,
+) (map[string][]*Integration, error) {
+	nameToReceiver := make(map[string]*APIReceiver, len(apiReceivers))
+	for _, receiver := range apiReceivers {
+		if existing, ok := nameToReceiver[receiver.Name]; ok {
+			itypes := make([]string, 0, len(existing.GrafanaIntegrations.Integrations))
+			for _, i := range existing.GrafanaIntegrations.Integrations {
+				itypes = append(itypes, i.Type)
+			}
+			level.Warn(logger).Log("msg", "receiver with same name is defined multiple times. Only the last one will be used", "receiver_name", receiver.Name, "overwritten_integrations", itypes)
+		}
+		nameToReceiver[receiver.Name] = receiver
+	}
+
+	integrationsMap := make(map[string][]*Integration, len(apiReceivers))
+	for name, apiReceiver := range nameToReceiver {
+		integrations, err := BuildReceiverIntegrations(tenantID, apiReceiver, templ, images, decryptFn, emailSender, httpClientOptions, notifierFunc, version, logger)
+		if err != nil {
+			return nil, fmt.Errorf("failed to build receiver %s: %w", name, err)
+		}
+		integrationsMap[name] = integrations
+	}
+	return integrationsMap, nil
+}
+
+// BuildReceiverIntegrations builds integrations for the provided API receiver and returns them.
+// It supports both Prometheus and Grafana integrations and ensures that both of them use only templates dedicated for the kind.
+func BuildReceiverIntegrations(
+	tenantID int64,
+	receiver *APIReceiver,
+	tmpls TemplatesProvider,
+	images images.Provider,
+	decryptFn GetDecryptedValueFn,
+	emailSender receivers.EmailSender,
+	httpClientOptions []http.ClientOption,
+	wrapNotifierFunc WrapNotifierFunc,
+	version string,
+	logger log.Logger,
+) ([]*Integration, error) {
+	var integrations []*Integration
+	if len(receiver.Integrations) > 0 {
+		receiverCfg, err := BuildReceiverConfiguration(context.Background(), receiver, DecodeSecretsFromBase64, decryptFn)
+		if err != nil {
+			return nil, err
+		}
+		tmpl, err := tmpls.GetTemplate(templates.GrafanaKind)
+		if err != nil {
+			return nil, err
+		}
+		integrations = BuildGrafanaReceiverIntegrations(
+			receiverCfg,
+			tmpl,
+			images,
+			logger,
+			emailSender,
+			wrapNotifierFunc,
+			tenantID,
+			version,
+			httpClientOptions...,
+		)
+	}
+	mimir, err := BuildPrometheusReceiverIntegrations(receiver.ConfigReceiver, tmpls, httpClientOptions, logger, wrapNotifierFunc)
+	if err != nil {
+		return nil, err
+	}
+	integrations = append(integrations, mimir...)
+
+	return integrations, nil
+}