Ruler: retry notifications the Alertmanager failed to accept

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873
 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889
+* [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909
 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869
 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880

docs/configuration/config-file-reference.md

@@ -6043,6 +6043,26 @@ ruler_client:
 # CLI flag: -ruler.notification-timeout
 [notification_timeout: <duration> | default = 10s]
 
+notification_backoff:
+  # Minimum delay between notification retries. Only applies when
+  # -ruler.notification.backoff-retries is greater than 0.
+  # CLI flag: -ruler.notification.backoff-min-period
+  [min_period: <duration> | default = 100ms]
+
+  # Maximum delay between notification retries. Only applies when
+  # -ruler.notification.backoff-retries is greater than 0.
+  # CLI flag: -ruler.notification.backoff-max-period
+  [max_period: <duration> | default = 1s]
+
+  # Maximum number of times to retry a notification that the Alertmanager failed
+  # to accept with a retriable error, which is a transport error or a 5xx
+  # response. A notification that is not retried is dropped and the alert is not
+  # delivered until the next evaluation of its rule group. Retries happen within
+  # -ruler.notification-timeout and delay the other notifications of the same
+  # tenant, so keep this number small. Set to 0 to disable the retries.
+  # CLI flag: -ruler.notification.backoff-retries
+  [max_retries: <int> | default = 0]
+
 alertmanager_client:
   # Path to the client certificate file, which will be used for authenticating
   # with the server. Also requires the key path to be configured.

pkg/ruler/manager.go

@@ -28,9 +28,14 @@ import (
 	"github.com/cortexproject/cortex/pkg/parser"
 	"github.com/cortexproject/cortex/pkg/ring/client"
 	"github.com/cortexproject/cortex/pkg/ruler/rulespb"
+	"github.com/cortexproject/cortex/pkg/util/backoff"
 	"github.com/cortexproject/cortex/pkg/util/users"
 )
 
+// maxNotificationErrorMessageLength is the number of characters of the Alertmanager
+// response body kept in the error logged for a notification it did not accept.
+const maxNotificationErrorMessageLength = 150
+
 type DefaultMultiTenantManager struct {
 	cfg             Config
 	notifierCfg     *config.Config
@@ -64,6 +69,7 @@ type DefaultMultiTenantManager struct {
 	lastReloadSuccessful          *prometheus.GaugeVec
 	lastReloadSuccessfulTimestamp *prometheus.GaugeVec
 	configUpdatesTotal            *prometheus.CounterVec
+	notificationRetriesTotal      *prometheus.CounterVec
 	registry                      prometheus.Registerer
 	logger                        log.Logger
 
@@ -137,6 +143,11 @@ func NewDefaultMultiTenantManager(cfg Config, limits RulesLimits, managerFactory
 			Name:      "ruler_config_updates_total",
 			Help:      "Total number of config updates triggered by a user",
 		}, []string{"user"}),
+		notificationRetriesTotal: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+			Namespace: "cortex",
+			Name:      "ruler_notification_retries_total",
+			Help:      "Total number of notifications to the Alertmanager retried after a retriable failure.",
+		}, []string{"user"}),
 		registry:               reg,
 		logger:                 logger,
 		ruleGroupIterationFunc: defaultRuleGroupIterationFunc,
@@ -184,6 +195,7 @@ func (r *DefaultMultiTenantManager) SyncRuleGroups(ctx context.Context, ruleGrou
 			r.lastReloadSuccessful.DeleteLabelValues(userID)
 			r.lastReloadSuccessfulTimestamp.DeleteLabelValues(userID)
 			r.configUpdatesTotal.DeleteLabelValues(userID)
+			r.notificationRetriesTotal.DeleteLabelValues(userID)
 			r.userManagerMetrics.RemoveUserRegistry(userID)
 			if r.ruleEvalMetrics != nil {
 				r.ruleEvalMetrics.deletePerUserMetrics(userID)
@@ -420,6 +432,7 @@ func (r *DefaultMultiTenantManager) getOrCreateNotifier(userID string, userManag
 	}
 
 	logger := log.With(r.logger, "user", userID)
+	notificationRetries := r.notificationRetriesTotal.WithLabelValues(userID)
 
 	n = newRulerNotifier(&notifier.Options{
 		QueueCapacity: r.cfg.NotificationQueueCapacity,
@@ -437,25 +450,7 @@ func (r *DefaultMultiTenantManager) getOrCreateNotifier(userID string, userManag
 			defer sp.Finish()
 			ctx = ot.ContextWithSpan(ctx, sp)
 			_ = ot.GlobalTracer().Inject(sp.Context(), ot.HTTPHeaders, ot.HTTPHeadersCarrier(req.Header))
-			resp, err := ctxhttp.Do(ctx, client, req)
-			if err != nil {
-				level.Error(logger).Log("msg", "error occurred while sending alerts", "error", err)
-				return resp, err
-			}
-			defer resp.Body.Close()
-			if resp.StatusCode/100 != 2 {
-				bodyBytes, err := io.ReadAll(resp.Body)
-				if err != nil {
-					level.Error(logger).Log("msg", "error reading response body", "error", err, "response code", resp.StatusCode)
-					return resp, err
-				}
-				customErrorMessage := string(bodyBytes)
-				if len(customErrorMessage) >= 150 {
-					customErrorMessage = customErrorMessage[:150]
-				}
-				level.Error(logger).Log("msg", "error occurred sending notification", "error", customErrorMessage, "response code", resp.StatusCode)
-			}
-			return resp, err
+			return sendNotification(ctx, client, req, r.cfg.NotificationBackoff, notificationRetries, logger)
 		},
 	}, r.cfg.NameValidationScheme, logger, userManagerRegistry, r.notifiersDiscoveryMetrics)
 
@@ -470,6 +465,108 @@ func (r *DefaultMultiTenantManager) getOrCreateNotifier(userID string, userManag
 	return n.notifier, nil
 }
 
+// sendNotification sends the alerts of a tenant to one Alertmanager, retrying
+// the failures that another attempt may recover from.
+//
+// The Prometheus notifier does not retry: it logs the error returned here and
+// moves on to the next batch, so a notification failing once is never delivered
+// again until the rule group is evaluated again. Two failures recover on a
+// second attempt, and both are routine when the ruler keeps connections alive:
+// the Alertmanager closing an idle connection just as it is reused, and an
+// Alertmanager replying 503 because it is shutting down while the connection is
+// still pinned to it.
+//
+// Retries are disabled unless backoffCfg.MaxRetries is greater than 0, and are
+// always bounded by the deadline of ctx, which carries the notification timeout.
+func sendNotification(ctx context.Context, client *http.Client, req *http.Request, backoffCfg backoff.Config, retries prometheus.Counter, logger log.Logger) (*http.Response, error) {
+	// A MaxRetries of 0 means infinite retries to backoff.Backoff, so the retries
+	// have to be skipped here instead of relying on Ongoing() to stop them.
+	var retry *backoff.Backoff
+	if backoffCfg.MaxRetries > 0 {
+		retry = backoff.New(ctx, backoffCfg)
+	}
+
+	for {
+		resp, retriable, err := doNotification(ctx, client, req, logger)
+		if !retriable || retry == nil || !retry.Ongoing() {
+			return resp, err
+		}
+
+		retryReq, rewindErr := rewindNotificationRequest(req)
+		if rewindErr != nil {
+			level.Error(logger).Log("msg", "notification cannot be retried", "error", rewindErr)
+			return resp, err
+		}
+		req = retryReq
+
+		// The connection the failed attempt used may be half-closed by the
+		// Alertmanager, or pinned to an instance that is refusing the alerts while it
+		// shuts down. Dropping the idle connections of this tenant's client makes the
+		// next attempt open a new one, which is also how it reaches another instance
+		// when the Alertmanagers are behind a load balancer.
+		client.CloseIdleConnections()
+
+		retries.Inc()
+		retry.Wait()
+	}
+}
+
+// doNotification sends the alerts once and reports whether the failure, if any,
+// is worth another attempt. The response body is always closed, so the returned
+// response carries the status and nothing else.
+func doNotification(ctx context.Context, client *http.Client, req *http.Request, logger log.Logger) (*http.Response, bool, error) {
+	resp, err := ctxhttp.Do(ctx, client, req)
+	if err != nil {
+		level.Error(logger).Log("msg", "error occurred while sending alerts", "error", err)
+		return resp, isRetriableNotificationError(err), err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode/100 != 2 {
+		bodyBytes, err := io.ReadAll(resp.Body)
+		if err != nil {
+			level.Error(logger).Log("msg", "error reading response body", "error", err, "response code", resp.StatusCode)
+			return resp, false, err
+		}
+		customErrorMessage := string(bodyBytes)
+		if len(customErrorMessage) >= maxNotificationErrorMessageLength {
+			customErrorMessage = customErrorMessage[:maxNotificationErrorMessageLength]
+		}
+		level.Error(logger).Log("msg", "error occurred sending notification", "error", customErrorMessage, "response code", resp.StatusCode)
+	}
+
+	// Only a server error can succeed on another attempt. A request the
+	// Alertmanager rejected is rejected again, however many times it is sent.
+	return resp, resp.StatusCode/100 == 5, nil
+}
+
+// isRetriableNotificationError returns whether the transport error may be
+// recovered from by sending the alerts again. An error caused by the context
+// being cancelled or its deadline being exceeded is not retriable, there is no
+// time left to retry.
+func isRetriableNotificationError(err error) bool {
+	return !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)
+}
+
+// rewindNotificationRequest returns a copy of req that can be sent again. The
+// body of a request that has already been sent is consumed and closed.
+func rewindNotificationRequest(req *http.Request) (*http.Request, error) {
+	// GetBody is always set by the Prometheus notifier, which builds the request
+	// over a bytes.Reader, but Do is a public option of the notifier.
+	if req.GetBody == nil {
+		return nil, errors.New("the request body cannot be replayed")
+	}
+
+	body, err := req.GetBody()
+	if err != nil {
+		return nil, err
+	}
+
+	rewound := req.Clone(req.Context())
+	rewound.Body = body
+	return rewound, nil
+}
+
 func (r *DefaultMultiTenantManager) notifierApplyExternalLabels(userID string, externalLabels labels.Labels) error {
 	r.notifiersMtx.Lock()
 	defer r.notifiersMtx.Unlock()