Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2025-05-25 12:27
📁 discord	-	2025-05-25 12:27
📁 email	-	2025-05-25 12:27
📁 jira	-	2025-05-25 12:27
📁 msteams	-	2025-05-25 12:27
📁 msteamsv2	-	2025-05-25 12:27
📄 notify.go // Copyright 2015 Prometheus Team // 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" "errors" "fmt" "log/slog" "sort" "sync" "time" "github.com/cenkalti/backoff/v4" "github.com/cespare/xxhash/v2" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/model" "github.com/prometheus/alertmanager/featurecontrol" "github.com/prometheus/alertmanager/inhibit" "github.com/prometheus/alertmanager/nflog" "github.com/prometheus/alertmanager/nflog/nflogpb" "github.com/prometheus/alertmanager/silence" "github.com/prometheus/alertmanager/timeinterval" "github.com/prometheus/alertmanager/types" ) // ResolvedSender returns true if resolved notifications should be sent. type ResolvedSender interface { SendResolved() bool } // Peer represents the cluster node from where we are the sending the notification. type Peer interface { // WaitReady waits until the node silences and notifications have settled before attempting to send a notification. WaitReady(context.Context) error } // MinTimeout is the minimum timeout that is set for the context of a call // to a notification pipeline. const MinTimeout = 10 * time.Second // Notifier notifies about alerts under constraints of the given context. It // returns an error if unsuccessful and a flag whether the error is // recoverable. This information is useful for a retry logic. type Notifier interface { Notify(context.Context, ...types.Alert) (bool, 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 { notifier Notifier rs ResolvedSender name string idx int receiverName string } // NewIntegration returns a new integration. func NewIntegration(notifier Notifier, rs ResolvedSender, name string, idx int, receiverName string) Integration { return Integration{ notifier: notifier, rs: rs, name: name, idx: idx, receiverName: receiverName, } } // Notify implements the Notifier interface. func (i Integration) Notify(ctx context.Context, alerts ...types.Alert) (bool, error) { return i.notifier.Notify(ctx, alerts...) } // SendResolved implements the ResolvedSender interface. func (i Integration) SendResolved() bool { return i.rs.SendResolved() } // Name returns the name of the integration. func (i Integration) Name() string { return i.name } // Index returns the index of the integration. func (i Integration) Index() int { return i.idx } // String implements the Stringer interface. func (i Integration) String() string { return fmt.Sprintf("%s[%d]", i.name, i.idx) } // notifyKey defines a custom type with which a context is populated to // avoid accidental collisions. type notifyKey int const ( keyReceiverName notifyKey = iota keyRepeatInterval keyGroupLabels keyGroupKey keyFiringAlerts keyResolvedAlerts keyNow keyMuteTimeIntervals keyActiveTimeIntervals keyRouteID ) // WithReceiverName populates a context with a receiver name. func WithReceiverName(ctx context.Context, rcv string) context.Context { return context.WithValue(ctx, keyReceiverName, rcv) } // WithGroupKey populates a context with a group key. func WithGroupKey(ctx context.Context, s string) context.Context { return context.WithValue(ctx, keyGroupKey, s) } // WithFiringAlerts populates a context with a slice of firing alerts. func WithFiringAlerts(ctx context.Context, alerts []uint64) context.Context { return context.WithValue(ctx, keyFiringAlerts, alerts) } // WithResolvedAlerts populates a context with a slice of resolved alerts. func WithResolvedAlerts(ctx context.Context, alerts []uint64) context.Context { return context.WithValue(ctx, keyResolvedAlerts, alerts) } // WithGroupLabels populates a context with grouping labels. func WithGroupLabels(ctx context.Context, lset model.LabelSet) context.Context { return context.WithValue(ctx, keyGroupLabels, lset) } // WithNow populates a context with a now timestamp. func WithNow(ctx context.Context, t time.Time) context.Context { return context.WithValue(ctx, keyNow, t) } // WithRepeatInterval populates a context with a repeat interval. func WithRepeatInterval(ctx context.Context, t time.Duration) context.Context { return context.WithValue(ctx, keyRepeatInterval, t) } // WithMuteTimeIntervals populates a context with a slice of mute time names. func WithMuteTimeIntervals(ctx context.Context, mt []string) context.Context { return context.WithValue(ctx, keyMuteTimeIntervals, mt) } func WithActiveTimeIntervals(ctx context.Context, at []string) context.Context { return context.WithValue(ctx, keyActiveTimeIntervals, at) } func WithRouteID(ctx context.Context, routeID string) context.Context { return context.WithValue(ctx, keyRouteID, routeID) } // RepeatInterval extracts a repeat interval from the context. Iff none exists, the // second argument is false. func RepeatInterval(ctx context.Context) (time.Duration, bool) { v, ok := ctx.Value(keyRepeatInterval).(time.Duration) return v, ok } // ReceiverName extracts a receiver name from the context. Iff none exists, the // second argument is false. func ReceiverName(ctx context.Context) (string, bool) { v, ok := ctx.Value(keyReceiverName).(string) return v, ok } // GroupKey extracts a group key from the context. Iff none exists, the // second argument is false. func GroupKey(ctx context.Context) (string, bool) { v, ok := ctx.Value(keyGroupKey).(string) return v, ok } // GroupLabels extracts grouping label set from the context. Iff none exists, the // second argument is false. func GroupLabels(ctx context.Context) (model.LabelSet, bool) { v, ok := ctx.Value(keyGroupLabels).(model.LabelSet) return v, ok } // Now extracts a now timestamp from the context. Iff none exists, the // second argument is false. func Now(ctx context.Context) (time.Time, bool) { v, ok := ctx.Value(keyNow).(time.Time) return v, ok } // FiringAlerts extracts a slice of firing alerts from the context. // Iff none exists, the second argument is false. func FiringAlerts(ctx context.Context) ([]uint64, bool) { v, ok := ctx.Value(keyFiringAlerts).([]uint64) return v, ok } // ResolvedAlerts extracts a slice of firing alerts from the context. // Iff none exists, the second argument is false. func ResolvedAlerts(ctx context.Context) ([]uint64, bool) { v, ok := ctx.Value(keyResolvedAlerts).([]uint64) return v, ok } // MuteTimeIntervalNames extracts a slice of mute time names from the context. If and only if none exists, the // second argument is false. func MuteTimeIntervalNames(ctx context.Context) ([]string, bool) { v, ok := ctx.Value(keyMuteTimeIntervals).([]string) return v, ok } // ActiveTimeIntervalNames extracts a slice of active time names from the context. If none exists, the // second argument is false. func ActiveTimeIntervalNames(ctx context.Context) ([]string, bool) { v, ok := ctx.Value(keyActiveTimeIntervals).([]string) return v, ok } // RouteID extracts a RouteID from the context. Iff none exists, the // // second argument is false. func RouteID(ctx context.Context) (string, bool) { v, ok := ctx.Value(keyRouteID).(string) return v, ok } // A Stage processes alerts under the constraints of the given context. type Stage interface { Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) } // StageFunc wraps a function to represent a Stage. type StageFunc func(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) // Exec implements Stage interface. func (f StageFunc) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { return f(ctx, l, alerts...) } type NotificationLog interface { Log(r nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error Query(params ...nflog.QueryParam) ([]nflogpb.Entry, error) } type Metrics struct { numNotifications prometheus.CounterVec numTotalFailedNotifications prometheus.CounterVec numNotificationRequestsTotal prometheus.CounterVec numNotificationRequestsFailedTotal prometheus.CounterVec numNotificationSuppressedTotal prometheus.CounterVec notificationLatencySeconds prometheus.HistogramVec ff featurecontrol.Flagger } func NewMetrics(r prometheus.Registerer, ff featurecontrol.Flagger) Metrics { labels := []string{"integration"} if ff.EnableReceiverNamesInMetrics() { labels = append(labels, "receiver_name") } m := &Metrics{ numNotifications: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "alertmanager", Name: "notifications_total", Help: "The total number of attempted notifications.", }, labels), numTotalFailedNotifications: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "alertmanager", Name: "notifications_failed_total", Help: "The total number of failed notifications.", }, append(labels, "reason")), numNotificationRequestsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "alertmanager", Name: "notification_requests_total", Help: "The total number of attempted notification requests.", }, labels), numNotificationRequestsFailedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "alertmanager", Name: "notification_requests_failed_total", Help: "The total number of failed notification requests.", }, labels), numNotificationSuppressedTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "alertmanager", Name: "notifications_suppressed_total", Help: "The total number of notifications suppressed for being silenced, inhibited, outside of active time intervals or within muted time intervals.", }, []string{"reason"}), notificationLatencySeconds: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "alertmanager", Name: "notification_latency_seconds", Help: "The latency of notifications in seconds.", Buckets: []float64{1, 5, 10, 15, 20}, NativeHistogramBucketFactor: 1.1, NativeHistogramMaxBucketNumber: 100, NativeHistogramMinResetDuration: 1 time.Hour, }, labels), ff: ff, } r.MustRegister( m.numNotifications, m.numTotalFailedNotifications, m.numNotificationRequestsTotal, m.numNotificationRequestsFailedTotal, m.numNotificationSuppressedTotal, m.notificationLatencySeconds, ) return m } func (m Metrics) InitializeFor(receiver map[string][]Integration) { if m.ff.EnableReceiverNamesInMetrics() { // Reset the vectors to take into account receiver names changing after hot reloads. m.numNotifications.Reset() m.numNotificationRequestsTotal.Reset() m.numNotificationRequestsFailedTotal.Reset() m.notificationLatencySeconds.Reset() m.numTotalFailedNotifications.Reset() for name, integrations := range receiver { for _, integration := range integrations { m.numNotifications.WithLabelValues(integration.Name(), name) m.numNotificationRequestsTotal.WithLabelValues(integration.Name(), name) m.numNotificationRequestsFailedTotal.WithLabelValues(integration.Name(), name) m.notificationLatencySeconds.WithLabelValues(integration.Name(), name) for _, reason := range possibleFailureReasonCategory { m.numTotalFailedNotifications.WithLabelValues(integration.Name(), name, reason) } } } return } // When the feature flag is not enabled, we just carry on registering _all_ the integrations. for _, integration := range []string{ "email", "pagerduty", "wechat", "pushover", "slack", "opsgenie", "webhook", "victorops", "sns", "telegram", "discord", "webex", "msteams", "msteamsv2", "jira", "rocketchat", } { m.numNotifications.WithLabelValues(integration) m.numNotificationRequestsTotal.WithLabelValues(integration) m.numNotificationRequestsFailedTotal.WithLabelValues(integration) m.notificationLatencySeconds.WithLabelValues(integration) for _, reason := range possibleFailureReasonCategory { m.numTotalFailedNotifications.WithLabelValues(integration, reason) } } } type PipelineBuilder struct { metrics Metrics ff featurecontrol.Flagger } func NewPipelineBuilder(r prometheus.Registerer, ff featurecontrol.Flagger) PipelineBuilder { return &PipelineBuilder{ metrics: NewMetrics(r, ff), ff: ff, } } // New returns a map of receivers to Stages. func (pb PipelineBuilder) New( receivers map[string][]Integration, wait func() time.Duration, inhibitor inhibit.Inhibitor, silencer silence.Silencer, intervener timeinterval.Intervener, marker types.GroupMarker, notificationLog NotificationLog, peer Peer, ) RoutingStage { rs := make(RoutingStage, len(receivers)) ms := NewGossipSettleStage(peer) is := NewMuteStage(inhibitor, pb.metrics) tas := NewTimeActiveStage(intervener, marker, pb.metrics) tms := NewTimeMuteStage(intervener, marker, pb.metrics) ss := NewMuteStage(silencer, pb.metrics) for name := range receivers { st := createReceiverStage(name, receivers[name], wait, notificationLog, pb.metrics) rs[name] = MultiStage{ms, is, tas, tms, ss, st} } pb.metrics.InitializeFor(receivers) return rs } // createReceiverStage creates a pipeline of stages for a receiver. func createReceiverStage( name string, integrations []Integration, wait func() time.Duration, notificationLog NotificationLog, metrics Metrics, ) Stage { var fs FanoutStage for i := range integrations { recv := &nflogpb.Receiver{ GroupName: name, Integration: integrations[i].Name(), Idx: uint32(integrations[i].Index()), } var s MultiStage s = append(s, NewWaitStage(wait)) s = append(s, NewDedupStage(&integrations[i], notificationLog, recv)) s = append(s, NewRetryStage(integrations[i], name, metrics)) s = append(s, NewSetNotifiesStage(notificationLog, recv)) fs = append(fs, s) } return fs } // RoutingStage executes the inner stages based on the receiver specified in // the context. type RoutingStage map[string]Stage // Exec implements the Stage interface. func (rs RoutingStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { receiver, ok := ReceiverName(ctx) if !ok { return ctx, nil, errors.New("receiver missing") } s, ok := rs[receiver] if !ok { return ctx, nil, errors.New("stage for receiver missing") } return s.Exec(ctx, l, alerts...) } // A MultiStage executes a series of stages sequentially. type MultiStage []Stage // Exec implements the Stage interface. func (ms MultiStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { var err error for _, s := range ms { if len(alerts) == 0 { return ctx, nil, nil } ctx, alerts, err = s.Exec(ctx, l, alerts...) if err != nil { return ctx, nil, err } } return ctx, alerts, nil } // FanoutStage executes its stages concurrently. type FanoutStage []Stage // Exec attempts to execute all stages concurrently and discards the results. // It returns its input alerts and a types.MultiError if one or more stages fail. func (fs FanoutStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { var ( wg sync.WaitGroup me types.MultiError ) wg.Add(len(fs)) for _, s := range fs { go func(s Stage) { if _, _, err := s.Exec(ctx, l, alerts...); err != nil { me.Add(err) } wg.Done() }(s) } wg.Wait() if me.Len() > 0 { return ctx, alerts, &me } return ctx, alerts, nil } // GossipSettleStage waits until the Gossip has settled to forward alerts. type GossipSettleStage struct { peer Peer } // NewGossipSettleStage returns a new GossipSettleStage. func NewGossipSettleStage(p Peer) GossipSettleStage { return &GossipSettleStage{peer: p} } func (n GossipSettleStage) Exec(ctx context.Context, _ slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { if n.peer != nil { if err := n.peer.WaitReady(ctx); err != nil { return ctx, nil, err } } return ctx, alerts, nil } const ( SuppressedReasonSilence = "silence" SuppressedReasonInhibition = "inhibition" SuppressedReasonMuteTimeInterval = "mute_time_interval" SuppressedReasonActiveTimeInterval = "active_time_interval" ) // MuteStage filters alerts through a Muter. type MuteStage struct { muter types.Muter metrics Metrics } // NewMuteStage return a new MuteStage. func NewMuteStage(m types.Muter, metrics Metrics) MuteStage { return &MuteStage{muter: m, metrics: metrics} } // Exec implements the Stage interface. func (n MuteStage) Exec(ctx context.Context, logger slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { var ( filtered []types.Alert muted []types.Alert ) for _, a := range alerts { // TODO(fabxc): increment total alerts counter. // Do not send the alert if muted. if n.muter.Mutes(a.Labels) { muted = append(muted, a) } else { filtered = append(filtered, a) } // TODO(fabxc): increment muted alerts counter if muted. } if len(muted) > 0 { var reason string switch n.muter.(type) { case silence.Silencer: reason = SuppressedReasonSilence case inhibit.Inhibitor: reason = SuppressedReasonInhibition default: } n.metrics.numNotificationSuppressedTotal.WithLabelValues(reason).Add(float64(len(muted))) logger.Debug("Notifications will not be sent for muted alerts", "alerts", fmt.Sprintf("%v", muted), "reason", reason) } return ctx, filtered, nil } // WaitStage waits for a certain amount of time before continuing or until the // context is done. type WaitStage struct { wait func() time.Duration } // NewWaitStage returns a new WaitStage. func NewWaitStage(wait func() time.Duration) WaitStage { return &WaitStage{ wait: wait, } } // Exec implements the Stage interface. func (ws WaitStage) Exec(ctx context.Context, _ slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { select { case <-time.After(ws.wait()): case <-ctx.Done(): return ctx, nil, ctx.Err() } return ctx, alerts, nil } // DedupStage filters alerts. // Filtering happens based on a notification log. type DedupStage struct { rs ResolvedSender nflog NotificationLog recv nflogpb.Receiver now func() time.Time hash func(types.Alert) uint64 } // NewDedupStage wraps a DedupStage that runs against the given notification log. func NewDedupStage(rs ResolvedSender, l NotificationLog, recv nflogpb.Receiver) DedupStage { return &DedupStage{ rs: rs, nflog: l, recv: recv, now: utcNow, hash: hashAlert, } } func utcNow() time.Time { return time.Now().UTC() } // Wrap a slice in a struct so we can store a pointer in sync.Pool. type hashBuffer struct { buf []byte } var hashBuffers = sync.Pool{ New: func() interface{} { return &hashBuffer{buf: make([]byte, 0, 1024)} }, } func hashAlert(a types.Alert) uint64 { const sep = '\xff' hb := hashBuffers.Get().(hashBuffer) defer hashBuffers.Put(hb) b := hb.buf[:0] names := make(model.LabelNames, 0, len(a.Labels)) for ln := range a.Labels { names = append(names, ln) } sort.Sort(names) for _, ln := range names { b = append(b, string(ln)...) b = append(b, sep) b = append(b, string(a.Labels[ln])...) b = append(b, sep) } hash := xxhash.Sum64(b) return hash } func (n DedupStage) needsUpdate(entry nflogpb.Entry, firing, resolved map[uint64]struct{}, repeat time.Duration) bool { // If we haven't notified about the alert group before, notify right away // unless we only have resolved alerts. if entry == nil { return len(firing) > 0 } if !entry.IsFiringSubset(firing) { return true } // Notify about all alerts being resolved. // This is done irrespective of the send_resolved flag to make sure that // the firing alerts are cleared from the notification log. if len(firing) == 0 { // If the current alert group and last notification contain no firing // alert, it means that some alerts have been fired and resolved during the // last interval. In this case, there is no need to notify the receiver // since it doesn't know about them. return len(entry.FiringAlerts) > 0 } if n.rs.SendResolved() && !entry.IsResolvedSubset(resolved) { return true } // Nothing changed, only notify if the repeat interval has passed. return entry.Timestamp.Before(n.now().Add(-repeat)) } // Exec implements the Stage interface. func (n DedupStage) Exec(ctx context.Context, _ slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { gkey, ok := GroupKey(ctx) if !ok { return ctx, nil, errors.New("group key missing") } repeatInterval, ok := RepeatInterval(ctx) if !ok { return ctx, nil, errors.New("repeat interval missing") } firingSet := map[uint64]struct{}{} resolvedSet := map[uint64]struct{}{} firing := []uint64{} resolved := []uint64{} var hash uint64 for _, a := range alerts { hash = n.hash(a) if a.Resolved() { resolved = append(resolved, hash) resolvedSet[hash] = struct{}{} } else { firing = append(firing, hash) firingSet[hash] = struct{}{} } } ctx = WithFiringAlerts(ctx, firing) ctx = WithResolvedAlerts(ctx, resolved) entries, err := n.nflog.Query(nflog.QGroupKey(gkey), nflog.QReceiver(n.recv)) if err != nil && !errors.Is(err, nflog.ErrNotFound) { return ctx, nil, err } var entry nflogpb.Entry switch len(entries) { case 0: case 1: entry = entries[0] default: return ctx, nil, fmt.Errorf("unexpected entry result size %d", len(entries)) } if n.needsUpdate(entry, firingSet, resolvedSet, repeatInterval) { return ctx, alerts, nil } return ctx, nil, nil } // RetryStage notifies via passed integration with exponential backoff until it // succeeds. It aborts if the context is canceled or timed out. type RetryStage struct { integration Integration groupName string metrics Metrics labelValues []string } // NewRetryStage returns a new instance of a RetryStage. func NewRetryStage(i Integration, groupName string, metrics Metrics) RetryStage { labelValues := []string{i.Name()} if metrics.ff.EnableReceiverNamesInMetrics() { labelValues = append(labelValues, i.receiverName) } return &RetryStage{ integration: i, groupName: groupName, metrics: metrics, labelValues: labelValues, } } func (r RetryStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { r.metrics.numNotifications.WithLabelValues(r.labelValues...).Inc() ctx, alerts, err := r.exec(ctx, l, alerts...) failureReason := DefaultReason.String() if err != nil { var e ErrorWithReason if errors.As(err, &e) { failureReason = e.Reason.String() } r.metrics.numTotalFailedNotifications.WithLabelValues(append(r.labelValues, failureReason)...).Inc() } return ctx, alerts, err } func (r RetryStage) exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { var sent []types.Alert // If we shouldn't send notifications for resolved alerts, but there are only // resolved alerts, report them all as successfully notified (we still want the // notification log to log them for the next run of DedupStage). if !r.integration.SendResolved() { firing, ok := FiringAlerts(ctx) if !ok { return ctx, nil, errors.New("firing alerts missing") } if len(firing) == 0 { return ctx, alerts, nil } for _, a := range alerts { if a.Status() != model.AlertResolved { sent = append(sent, a) } } } else { sent = alerts } b := backoff.NewExponentialBackOff() b.MaxElapsedTime = 0 // Always retry. tick := backoff.NewTicker(b) defer tick.Stop() var ( i = 0 iErr error ) l = l.With("receiver", r.groupName, "integration", r.integration.String()) if groupKey, ok := GroupKey(ctx); ok { l = l.With("aggrGroup", groupKey) } for { i++ // Always check the context first to not notify again. select { case <-ctx.Done(): if iErr == nil { iErr = ctx.Err() if errors.Is(iErr, context.Canceled) { iErr = NewErrorWithReason(ContextCanceledReason, iErr) } else if errors.Is(iErr, context.DeadlineExceeded) { iErr = NewErrorWithReason(ContextDeadlineExceededReason, iErr) } } if iErr != nil { return ctx, nil, fmt.Errorf("%s/%s: notify retry canceled after %d attempts: %w", r.groupName, r.integration.String(), i, iErr) } return ctx, nil, nil default: } select { case <-tick.C: now := time.Now() retry, err := r.integration.Notify(ctx, sent...) dur := time.Since(now) r.metrics.notificationLatencySeconds.WithLabelValues(r.labelValues...).Observe(dur.Seconds()) r.metrics.numNotificationRequestsTotal.WithLabelValues(r.labelValues...).Inc() if err != nil { r.metrics.numNotificationRequestsFailedTotal.WithLabelValues(r.labelValues...).Inc() if !retry { return ctx, alerts, fmt.Errorf("%s/%s: notify retry canceled due to unrecoverable error after %d attempts: %w", r.groupName, r.integration.String(), i, err) } if ctx.Err() == nil { if iErr == nil \|\| err.Error() != iErr.Error() { // Log the error if the context isn't done and the error isn't the same as before. l.Warn("Notify attempt failed, will retry later", "attempts", i, "err", err) } // Save this error to be able to return the last seen error by an // integration upon context timeout. iErr = err } } else { l := l.With("attempts", i, "duration", dur) if i <= 1 { l = l.With("alerts", fmt.Sprintf("%v", alerts)) l.Debug("Notify success") } else { l.Info("Notify success") } return ctx, alerts, nil } case <-ctx.Done(): if iErr == nil { iErr = ctx.Err() if errors.Is(iErr, context.Canceled) { iErr = NewErrorWithReason(ContextCanceledReason, iErr) } else if errors.Is(iErr, context.DeadlineExceeded) { iErr = NewErrorWithReason(ContextDeadlineExceededReason, iErr) } } if iErr != nil { return ctx, nil, fmt.Errorf("%s/%s: notify retry canceled after %d attempts: %w", r.groupName, r.integration.String(), i, iErr) } return ctx, nil, nil } } } // SetNotifiesStage sets the notification information about passed alerts. The // passed alerts should have already been sent to the receivers. type SetNotifiesStage struct { nflog NotificationLog recv nflogpb.Receiver } // NewSetNotifiesStage returns a new instance of a SetNotifiesStage. func NewSetNotifiesStage(l NotificationLog, recv nflogpb.Receiver) SetNotifiesStage { return &SetNotifiesStage{ nflog: l, recv: recv, } } // Exec implements the Stage interface. func (n SetNotifiesStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { gkey, ok := GroupKey(ctx) if !ok { return ctx, nil, errors.New("group key missing") } firing, ok := FiringAlerts(ctx) if !ok { return ctx, nil, errors.New("firing alerts missing") } resolved, ok := ResolvedAlerts(ctx) if !ok { return ctx, nil, errors.New("resolved alerts missing") } repeat, ok := RepeatInterval(ctx) if !ok { return ctx, nil, errors.New("repeat interval missing") } expiry := 2 * repeat return ctx, alerts, n.nflog.Log(n.recv, gkey, firing, resolved, expiry) } type timeStage struct { muter types.TimeMuter marker types.GroupMarker metrics Metrics } type TimeMuteStage timeStage func NewTimeMuteStage(muter types.TimeMuter, marker types.GroupMarker, metrics Metrics) TimeMuteStage { return &TimeMuteStage{muter, marker, metrics} } // Exec implements the stage interface for TimeMuteStage. // TimeMuteStage is responsible for muting alerts whose route is not in an active time. func (tms TimeMuteStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { routeID, ok := RouteID(ctx) if !ok { return ctx, nil, errors.New("route ID missing") } gkey, ok := GroupKey(ctx) if !ok { return ctx, nil, errors.New("group key missing") } muteTimeIntervalNames, ok := MuteTimeIntervalNames(ctx) if !ok { return ctx, alerts, nil } now, ok := Now(ctx) if !ok { return ctx, alerts, errors.New("missing now timestamp") } // Skip this stage if there are no mute timings. if len(muteTimeIntervalNames) == 0 { return ctx, alerts, nil } muted, mutedBy, err := tms.muter.Mutes(muteTimeIntervalNames, now) if err != nil { return ctx, alerts, err } // If muted is false then mutedBy is nil and the muted marker is removed. tms.marker.SetMuted(routeID, gkey, mutedBy) // If the current time is inside a mute time, all alerts are removed from the pipeline. if muted { tms.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonMuteTimeInterval).Add(float64(len(alerts))) l.Debug("Notifications not sent, route is within mute time", "alerts", len(alerts)) return ctx, nil, nil } return ctx, alerts, nil } type TimeActiveStage timeStage func NewTimeActiveStage(muter types.TimeMuter, marker types.GroupMarker, metrics Metrics) TimeActiveStage { return &TimeActiveStage{muter, marker, metrics} } // Exec implements the stage interface for TimeActiveStage. // TimeActiveStage is responsible for muting alerts whose route is not in an active time. func (tas TimeActiveStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []*types.Alert, error) { routeID, ok := RouteID(ctx) if !ok { return ctx, nil, errors.New("route ID missing") } gkey, ok := GroupKey(ctx) if !ok { return ctx, nil, errors.New("group key missing") } activeTimeIntervalNames, ok := ActiveTimeIntervalNames(ctx) if !ok { return ctx, alerts, nil } // if we don't have active time intervals at all it is always active. if len(activeTimeIntervalNames) == 0 { return ctx, alerts, nil } now, ok := Now(ctx) if !ok { return ctx, alerts, errors.New("missing now timestamp") } active, _, err := tas.muter.Mutes(activeTimeIntervalNames, now) if err != nil { return ctx, alerts, err } var mutedBy []string if !active { // If the group is muted, then it must be muted by all active time intervals. // Otherwise, the group must be in at least one active time interval for it // to be active. mutedBy = activeTimeIntervalNames } tas.marker.SetMuted(routeID, gkey, mutedBy) // If the current time is not inside an active time, all alerts are removed from the pipeline if !active { tas.metrics.numNotificationSuppressedTotal.WithLabelValues(SuppressedReasonActiveTimeInterval).Add(float64(len(alerts))) l.Debug("Notifications not sent, route is not within active time", "alerts", len(alerts)) return ctx, nil, nil } return ctx, alerts, nil }	30.89 KB	2025-05-25 12:27
📄 notify_test.go // Copyright 2015 Prometheus Team // 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" "errors" "fmt" "io" "log/slog" "reflect" "sort" "strings" "testing" "time" "github.com/prometheus/client_golang/prometheus" prom_testutil "github.com/prometheus/client_golang/prometheus/testutil" "github.com/prometheus/common/model" "github.com/prometheus/common/promslog" "github.com/stretchr/testify/require" "github.com/prometheus/alertmanager/featurecontrol" "github.com/prometheus/alertmanager/nflog" "github.com/prometheus/alertmanager/nflog/nflogpb" "github.com/prometheus/alertmanager/silence" "github.com/prometheus/alertmanager/silence/silencepb" "github.com/prometheus/alertmanager/timeinterval" "github.com/prometheus/alertmanager/types" ) type sendResolved bool func (s sendResolved) SendResolved() bool { return bool(s) } type notifierFunc func(ctx context.Context, alerts ...types.Alert) (bool, error) func (f notifierFunc) Notify(ctx context.Context, alerts ...types.Alert) (bool, error) { return f(ctx, alerts...) } type failStage struct{} func (s failStage) Exec(ctx context.Context, l slog.Logger, as ...types.Alert) (context.Context, []types.Alert, error) { return ctx, nil, fmt.Errorf("some error") } type testNflog struct { qres []nflogpb.Entry qerr error logFunc func(r nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error } func (l testNflog) Query(p ...nflog.QueryParam) ([]nflogpb.Entry, error) { return l.qres, l.qerr } func (l testNflog) Log(r nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error { return l.logFunc(r, gkey, firingAlerts, resolvedAlerts, expiry) } func (l testNflog) GC() (int, error) { return 0, nil } func (l testNflog) Snapshot(w io.Writer) (int, error) { return 0, nil } func alertHashSet(hashes ...uint64) map[uint64]struct{} { res := map[uint64]struct{}{} for _, h := range hashes { res[h] = struct{}{} } return res } func TestDedupStageNeedsUpdate(t testing.T) { now := utcNow() cases := []struct { entry nflogpb.Entry firingAlerts map[uint64]struct{} resolvedAlerts map[uint64]struct{} repeat time.Duration resolve bool res bool }{ { // No matching nflog entry should update. entry: nil, firingAlerts: alertHashSet(2, 3, 4), res: true, }, { // No matching nflog entry shouldn't update if no alert fires. entry: nil, resolvedAlerts: alertHashSet(2, 3, 4), res: false, }, { // Different sets of firing alerts should update. entry: &nflogpb.Entry{FiringAlerts: []uint64{1, 2, 3}}, firingAlerts: alertHashSet(2, 3, 4), res: true, }, { // Zero timestamp in the nflog entry should always update. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2, 3}, Timestamp: time.Time{}, }, firingAlerts: alertHashSet(1, 2, 3), res: true, }, { // Identical sets of alerts shouldn't update before repeat_interval. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2, 3}, Timestamp: now.Add(-9 time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(1, 2, 3), res: false, }, { // Identical sets of alerts should update after repeat_interval. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2, 3}, Timestamp: now.Add(-11 * time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(1, 2, 3), res: true, }, { // Different sets of resolved alerts without firing alerts shouldn't update after repeat_interval. entry: &nflogpb.Entry{ ResolvedAlerts: []uint64{1, 2, 3}, Timestamp: now.Add(-11 * time.Minute), }, repeat: 10 * time.Minute, resolvedAlerts: alertHashSet(3, 4, 5), resolve: true, res: false, }, { // Different sets of resolved alerts shouldn't update when resolve is false. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2}, ResolvedAlerts: []uint64{3}, Timestamp: now.Add(-9 * time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(1), resolvedAlerts: alertHashSet(2, 3), resolve: false, res: false, }, { // Different sets of resolved alerts should update when resolve is true. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2}, ResolvedAlerts: []uint64{3}, Timestamp: now.Add(-9 * time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(1), resolvedAlerts: alertHashSet(2, 3), resolve: true, res: true, }, { // Empty set of firing alerts should update when resolve is false. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2}, ResolvedAlerts: []uint64{3}, Timestamp: now.Add(-9 * time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(), resolvedAlerts: alertHashSet(1, 2, 3), resolve: false, res: true, }, { // Empty set of firing alerts should update when resolve is true. entry: &nflogpb.Entry{ FiringAlerts: []uint64{1, 2}, ResolvedAlerts: []uint64{3}, Timestamp: now.Add(-9 * time.Minute), }, repeat: 10 * time.Minute, firingAlerts: alertHashSet(), resolvedAlerts: alertHashSet(1, 2, 3), resolve: true, res: true, }, } for i, c := range cases { t.Log("case", i) s := &DedupStage{ now: func() time.Time { return now }, rs: sendResolved(c.resolve), } res := s.needsUpdate(c.entry, c.firingAlerts, c.resolvedAlerts, c.repeat) require.Equal(t, c.res, res) } } func TestDedupStage(t testing.T) { i := 0 now := utcNow() s := &DedupStage{ hash: func(a types.Alert) uint64 { res := uint64(i) i++ return res }, now: func() time.Time { return now }, rs: sendResolved(false), } ctx := context.Background() _, _, err := s.Exec(ctx, promslog.NewNopLogger()) require.EqualError(t, err, "group key missing") ctx = WithGroupKey(ctx, "1") _, _, err = s.Exec(ctx, promslog.NewNopLogger()) require.EqualError(t, err, "repeat interval missing") ctx = WithRepeatInterval(ctx, time.Hour) alerts := []types.Alert{{}, {}, {}} // Must catch notification log query errors. s.nflog = &testNflog{ qerr: errors.New("bad things"), } ctx, _, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.EqualError(t, err, "bad things") // ... but skip ErrNotFound. s.nflog = &testNflog{ qerr: nflog.ErrNotFound, } ctx, res, err := s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err, "unexpected error on not found log entry") require.Equal(t, alerts, res, "input alerts differ from result alerts") s.nflog = &testNflog{ qerr: nil, qres: []nflogpb.Entry{ {FiringAlerts: []uint64{0, 1, 2}}, {FiringAlerts: []uint64{1, 2, 3}}, }, } ctx, _, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.Contains(t, err.Error(), "result size") // Must return no error and no alerts no need to update. i = 0 s.nflog = &testNflog{ qerr: nflog.ErrNotFound, qres: []nflogpb.Entry{ { FiringAlerts: []uint64{0, 1, 2}, Timestamp: now, }, }, } ctx, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Nil(t, res, "unexpected alerts returned") // Must return no error and all input alerts on changes. i = 0 s.nflog = &testNflog{ qerr: nil, qres: []nflogpb.Entry{ { FiringAlerts: []uint64{1, 2, 3, 4}, Timestamp: now, }, }, } _, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res, "unexpected alerts returned") } func TestMultiStage(t testing.T) { var ( alerts1 = []types.Alert{{}} alerts2 = []types.Alert{{}, {}} alerts3 = []types.Alert{{}, {}, {}} ) stage := MultiStage{ StageFunc(func(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { if !reflect.DeepEqual(alerts, alerts1) { t.Fatal("Input not equal to input of MultiStage") } //nolint:staticcheck // Ignore SA1029 ctx = context.WithValue(ctx, "key", "value") return ctx, alerts2, nil }), StageFunc(func(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { if !reflect.DeepEqual(alerts, alerts2) { t.Fatal("Input not equal to output of previous stage") } v, ok := ctx.Value("key").(string) if !ok \|\| v != "value" { t.Fatalf("Expected value %q for key %q but got %q", "value", "key", v) } return ctx, alerts3, nil }), } _, alerts, err := stage.Exec(context.Background(), promslog.NewNopLogger(), alerts1...) if err != nil { t.Fatalf("Exec failed: %s", err) } if !reflect.DeepEqual(alerts, alerts3) { t.Fatal("Output of MultiStage is not equal to the output of the last stage") } } func TestMultiStageFailure(t testing.T) { var ( ctx = context.Background() s1 = failStage{} stage = MultiStage{s1} ) _, _, err := stage.Exec(ctx, promslog.NewNopLogger(), nil) if err.Error() != "some error" { t.Fatal("Errors were not propagated correctly by MultiStage") } } func TestRoutingStage(t testing.T) { var ( alerts1 = []types.Alert{{}} alerts2 = []types.Alert{{}, {}} ) stage := RoutingStage{ "name": StageFunc(func(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { if !reflect.DeepEqual(alerts, alerts1) { t.Fatal("Input not equal to input of RoutingStage") } return ctx, alerts2, nil }), "not": failStage{}, } ctx := WithReceiverName(context.Background(), "name") _, alerts, err := stage.Exec(ctx, promslog.NewNopLogger(), alerts1...) if err != nil { t.Fatalf("Exec failed: %s", err) } if !reflect.DeepEqual(alerts, alerts2) { t.Fatal("Output of RoutingStage is not equal to the output of the inner stage") } } func TestRetryStageWithError(t testing.T) { fail, retry := true, true sent := []types.Alert{} i := Integration{ notifier: notifierFunc(func(ctx context.Context, alerts ...types.Alert) (bool, error) { if fail { fail = false return retry, errors.New("fail to deliver notification") } sent = append(sent, alerts...) return false, nil }), rs: sendResolved(false), } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{})) alerts := []types.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), }, }, } ctx := context.Background() ctx = WithFiringAlerts(ctx, []uint64{0}) // Notify with a recoverable error should retry and succeed. resctx, res, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.Equal(t, alerts, sent) require.NotNil(t, resctx) // Notify with an unrecoverable error should fail. sent = sent[:0] fail = true retry = false resctx, _, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.Error(t, err) require.NotNil(t, resctx) } func TestRetryStageWithErrorCode(t testing.T) { testcases := map[string]struct { isNewErrorWithReason bool reason Reason reasonlabel string expectedCount int }{ "for clientError": {isNewErrorWithReason: true, reason: ClientErrorReason, reasonlabel: ClientErrorReason.String(), expectedCount: 1}, "for serverError": {isNewErrorWithReason: true, reason: ServerErrorReason, reasonlabel: ServerErrorReason.String(), expectedCount: 1}, "for unexpected code": {isNewErrorWithReason: false, reason: DefaultReason, reasonlabel: DefaultReason.String(), expectedCount: 1}, } for _, testData := range testcases { retry := false testData := testData i := Integration{ name: "test", notifier: notifierFunc(func(ctx context.Context, alerts ...types.Alert) (bool, error) { if !testData.isNewErrorWithReason { return retry, errors.New("fail to deliver notification") } return retry, NewErrorWithReason(testData.reason, errors.New("fail to deliver notification")) }), rs: sendResolved(false), } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{})) alerts := []types.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), }, }, } ctx := context.Background() ctx = WithFiringAlerts(ctx, []uint64{0}) // Notify with a non-recoverable error. resctx, _, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...) counter := r.metrics.numTotalFailedNotifications require.Equal(t, testData.expectedCount, int(prom_testutil.ToFloat64(counter.WithLabelValues(r.integration.Name(), testData.reasonlabel)))) require.Error(t, err) require.NotNil(t, resctx) } } func TestRetryStageWithContextCanceled(t testing.T) { ctx, cancel := context.WithCancel(context.Background()) i := Integration{ name: "test", notifier: notifierFunc(func(ctx context.Context, alerts ...types.Alert) (bool, error) { cancel() return true, errors.New("request failed: context canceled") }), rs: sendResolved(false), } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{})) alerts := []types.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), }, }, } ctx = WithFiringAlerts(ctx, []uint64{0}) // Notify with a non-recoverable error. resctx, _, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...) counter := r.metrics.numTotalFailedNotifications require.Equal(t, 1, int(prom_testutil.ToFloat64(counter.WithLabelValues(r.integration.Name(), ContextCanceledReason.String())))) require.Error(t, err) require.NotNil(t, resctx) } func TestRetryStageNoResolved(t testing.T) { sent := []types.Alert{} i := Integration{ notifier: notifierFunc(func(ctx context.Context, alerts ...types.Alert) (bool, error) { sent = append(sent, alerts...) return false, nil }), rs: sendResolved(false), } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{})) alerts := []types.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(-time.Hour), }, }, { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), }, }, } ctx := context.Background() resctx, res, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.EqualError(t, err, "firing alerts missing") require.Nil(t, res) require.NotNil(t, resctx) ctx = WithFiringAlerts(ctx, []uint64{0}) resctx, res, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.Equal(t, []types.Alert{alerts[1]}, sent) require.NotNil(t, resctx) // All alerts are resolved. sent = sent[:0] ctx = WithFiringAlerts(ctx, []uint64{}) alerts[1].Alert.EndsAt = time.Now().Add(-time.Hour) resctx, res, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.Equal(t, []types.Alert{}, sent) require.NotNil(t, resctx) } func TestRetryStageSendResolved(t testing.T) { sent := []types.Alert{} i := Integration{ notifier: notifierFunc(func(ctx context.Context, alerts ...types.Alert) (bool, error) { sent = append(sent, alerts...) return false, nil }), rs: sendResolved(true), } r := NewRetryStage(i, "", NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{})) alerts := []types.Alert{ { Alert: model.Alert{ EndsAt: time.Now().Add(-time.Hour), }, }, { Alert: model.Alert{ EndsAt: time.Now().Add(time.Hour), }, }, } ctx := context.Background() ctx = WithFiringAlerts(ctx, []uint64{0}) resctx, res, err := r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.Equal(t, alerts, sent) require.NotNil(t, resctx) // All alerts are resolved. sent = sent[:0] ctx = WithFiringAlerts(ctx, []uint64{}) alerts[1].Alert.EndsAt = time.Now().Add(-time.Hour) resctx, res, err = r.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.Equal(t, alerts, sent) require.NotNil(t, resctx) } func TestSetNotifiesStage(t testing.T) { tnflog := &testNflog{} s := &SetNotifiesStage{ recv: &nflogpb.Receiver{GroupName: "test"}, nflog: tnflog, } alerts := []types.Alert{{}, {}, {}} ctx := context.Background() resctx, res, err := s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.EqualError(t, err, "group key missing") require.Nil(t, res) require.NotNil(t, resctx) ctx = WithGroupKey(ctx, "1") resctx, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.EqualError(t, err, "firing alerts missing") require.Nil(t, res) require.NotNil(t, resctx) ctx = WithFiringAlerts(ctx, []uint64{0, 1, 2}) resctx, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.EqualError(t, err, "resolved alerts missing") require.Nil(t, res) require.NotNil(t, resctx) ctx = WithResolvedAlerts(ctx, []uint64{}) ctx = WithRepeatInterval(ctx, time.Hour) tnflog.logFunc = func(r nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error { require.Equal(t, s.recv, r) require.Equal(t, "1", gkey) require.Equal(t, []uint64{0, 1, 2}, firingAlerts) require.Equal(t, []uint64{}, resolvedAlerts) require.Equal(t, 2time.Hour, expiry) return nil } resctx, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.NotNil(t, resctx) ctx = WithFiringAlerts(ctx, []uint64{}) ctx = WithResolvedAlerts(ctx, []uint64{0, 1, 2}) tnflog.logFunc = func(r nflogpb.Receiver, gkey string, firingAlerts, resolvedAlerts []uint64, expiry time.Duration) error { require.Equal(t, s.recv, r) require.Equal(t, "1", gkey) require.Equal(t, []uint64{}, firingAlerts) require.Equal(t, []uint64{0, 1, 2}, resolvedAlerts) require.Equal(t, 2time.Hour, expiry) return nil } resctx, res, err = s.Exec(ctx, promslog.NewNopLogger(), alerts...) require.NoError(t, err) require.Equal(t, alerts, res) require.NotNil(t, resctx) } func TestMuteStage(t testing.T) { // Mute all label sets that have a "mute" key. muter := types.MuteFunc(func(lset model.LabelSet) bool { _, ok := lset["mute"] return ok }) metrics := NewMetrics(prometheus.NewRegistry(), featurecontrol.NoopFlags{}) stage := NewMuteStage(muter, metrics) in := []model.LabelSet{ {}, {"test": "set"}, {"mute": "me"}, {"foo": "bar", "test": "set"}, {"foo": "bar", "mute": "me"}, {}, {"not": "muted"}, } out := []model.LabelSet{ {}, {"test": "set"}, {"foo": "bar", "test": "set"}, {}, {"not": "muted"}, } var inAlerts []types.Alert for _, lset := range in { inAlerts = append(inAlerts, &types.Alert{ Alert: model.Alert{Labels: lset}, }) } _, alerts, err := stage.Exec(context.Background(), promslog.NewNopLogger(), inAlerts...) if err != nil { t.Fatalf("Exec failed: %s", err) } var got []model.LabelSet for _, a := range alerts { got = append(got, a.Labels) } if !reflect.DeepEqual(got, out) { t.Fatalf("Muting failed, expected: %v\ngot %v", out, got) } suppressed := int(prom_testutil.ToFloat64(metrics.numNotificationSuppressedTotal)) if (len(in) - len(got)) != suppressed { t.Fatalf("Expected %d alerts counted in suppressed metric but got %d", (len(in) - len(got)), suppressed) } } func TestMuteStageWithSilences(t testing.T) { silences, err := silence.New(silence.Options{Retention: time.Hour}) if err != nil { t.Fatal(err) } sil := &silencepb.Silence{ EndsAt: utcNow().Add(time.Hour), Matchers: []silencepb.Matcher{{Name: "mute", Pattern: "me"}}, } if err = silences.Set(sil); err != nil { t.Fatal(err) } reg := prometheus.NewRegistry() marker := types.NewMarker(reg) silencer := silence.NewSilencer(silences, marker, promslog.NewNopLogger()) metrics := NewMetrics(reg, featurecontrol.NoopFlags{}) stage := NewMuteStage(silencer, metrics) in := []model.LabelSet{ {}, {"test": "set"}, {"mute": "me"}, {"foo": "bar", "test": "set"}, {"foo": "bar", "mute": "me"}, {}, {"not": "muted"}, } out := []model.LabelSet{ {}, {"test": "set"}, {"foo": "bar", "test": "set"}, {}, {"not": "muted"}, } var inAlerts []types.Alert for _, lset := range in { inAlerts = append(inAlerts, &types.Alert{ Alert: model.Alert{Labels: lset}, }) } // Set the second alert as previously silenced with an old version // number. This is expected to get unsilenced by the stage. marker.SetActiveOrSilenced(inAlerts[1].Fingerprint(), 0, []string{"123"}, nil) _, alerts, err := stage.Exec(context.Background(), promslog.NewNopLogger(), inAlerts...) if err != nil { t.Fatalf("Exec failed: %s", err) } var got []model.LabelSet for _, a := range alerts { got = append(got, a.Labels) } if !reflect.DeepEqual(got, out) { t.Fatalf("Muting failed, expected: %v\ngot %v", out, got) } suppressedRoundOne := int(prom_testutil.ToFloat64(metrics.numNotificationSuppressedTotal)) if (len(in) - len(got)) != suppressedRoundOne { t.Fatalf("Expected %d alerts counted in suppressed metric but got %d", (len(in) - len(got)), suppressedRoundOne) } // Do it again to exercise the version tracking of silences. _, alerts, err = stage.Exec(context.Background(), promslog.NewNopLogger(), inAlerts...) if err != nil { t.Fatalf("Exec failed: %s", err) } got = got[:0] for _, a := range alerts { got = append(got, a.Labels) } if !reflect.DeepEqual(got, out) { t.Fatalf("Muting failed, expected: %v\ngot %v", out, got) } suppressedRoundTwo := int(prom_testutil.ToFloat64(metrics.numNotificationSuppressedTotal)) if (len(in) - len(got) + suppressedRoundOne) != suppressedRoundTwo { t.Fatalf("Expected %d alerts counted in suppressed metric but got %d", (len(in) - len(got)), suppressedRoundTwo) } // Expire the silence and verify that no alerts are silenced now. if err := silences.Expire(sil.Id); err != nil { t.Fatal(err) } _, alerts, err = stage.Exec(context.Background(), promslog.NewNopLogger(), inAlerts...) if err != nil { t.Fatalf("Exec failed: %s", err) } got = got[:0] for _, a := range alerts { got = append(got, a.Labels) } if !reflect.DeepEqual(got, in) { t.Fatalf("Unmuting failed, expected: %v\ngot %v", in, got) } suppressedRoundThree := int(prom_testutil.ToFloat64(metrics.numNotificationSuppressedTotal)) if (len(in) - len(got) + suppressedRoundTwo) != suppressedRoundThree { t.Fatalf("Expected %d alerts counted in suppressed metric but got %d", (len(in) - len(got)), suppressedRoundThree) } } func TestTimeMuteStage(t testing.T) { sydney, err := time.LoadLocation("Australia/Sydney") if err != nil { t.Fatalf("Failed to load location Australia/Sydney: %s", err) } eveningsAndWeekends := map[string][]timeinterval.TimeInterval{ "evenings": {{ Times: []timeinterval.TimeRange{{ StartMinute: 0, // 00:00 EndMinute: 540, // 09:00 }, { StartMinute: 1020, // 17:00 EndMinute: 1440, // 24:00 }}, Location: &timeinterval.Location{Location: sydney}, }}, "weekends": {{ Weekdays: []timeinterval.WeekdayRange{{ InclusiveRange: timeinterval.InclusiveRange{Begin: 6, End: 6}, // Saturday }, { InclusiveRange: timeinterval.InclusiveRange{Begin: 0, End: 0}, // Sunday }}, Location: &timeinterval.Location{Location: sydney}, }}, } tests := []struct { name string intervals map[string][]timeinterval.TimeInterval now time.Time alerts []types.Alert mutedBy []string }{{ name: "Should be muted outside working hours", intervals: eveningsAndWeekends, now: time.Date(2024, 1, 1, 0, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"evenings"}, }, { name: "Should not be muted during workings hours", intervals: eveningsAndWeekends, now: time.Date(2024, 1, 1, 9, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: nil, }, { name: "Should be muted during weekends", intervals: eveningsAndWeekends, now: time.Date(2024, 1, 6, 10, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"weekends"}, }, { name: "Should be muted at 12pm UTC on a weekday", intervals: eveningsAndWeekends, now: time.Date(2024, 1, 1, 10, 0, 0, 0, time.UTC), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"evenings"}, }, { name: "Should be muted at 12pm UTC on a weekend", intervals: eveningsAndWeekends, now: time.Date(2024, 1, 6, 10, 0, 0, 0, time.UTC), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"evenings", "weekends"}, }} for _, test := range tests { t.Run(test.name, func(t testing.T) { r := prometheus.NewRegistry() marker := types.NewMarker(r) metrics := NewMetrics(r, featurecontrol.NoopFlags{}) intervener := timeinterval.NewIntervener(test.intervals) st := NewTimeMuteStage(intervener, marker, metrics) // Get the names of all time intervals for the context. muteTimeIntervalNames := make([]string, 0, len(test.intervals)) for name := range test.intervals { muteTimeIntervalNames = append(muteTimeIntervalNames, name) } // Sort the names so we can compare mutedBy with test.mutedBy. sort.Strings(muteTimeIntervalNames) ctx := context.Background() ctx = WithNow(ctx, test.now) ctx = WithGroupKey(ctx, "group1") ctx = WithActiveTimeIntervals(ctx, nil) ctx = WithMuteTimeIntervals(ctx, muteTimeIntervalNames) ctx = WithRouteID(ctx, "route1") _, active, err := st.Exec(ctx, promslog.NewNopLogger(), test.alerts...) require.NoError(t, err) if len(test.mutedBy) == 0 { // All alerts should be active. require.Equal(t, len(test.alerts), len(active)) // The group should not be marked. mutedBy, isMuted := marker.Muted("route1", "group1") require.False(t, isMuted) require.Empty(t, mutedBy) // The metric for total suppressed notifications should not // have been incremented, which means it will not be collected. require.NoError(t, prom_testutil.GatherAndCompare(r, strings.NewReader(` # HELP alertmanager_marked_alerts How many alerts by state are currently marked in the Alertmanager regardless of their expiry. # TYPE alertmanager_marked_alerts gauge alertmanager_marked_alerts{state="active"} 0 alertmanager_marked_alerts{state="suppressed"} 0 alertmanager_marked_alerts{state="unprocessed"} 0 `))) } else { // All alerts should be muted. require.Empty(t, active) // The group should be marked as muted. mutedBy, isMuted := marker.Muted("route1", "group1") require.True(t, isMuted) require.Equal(t, test.mutedBy, mutedBy) // Gets the metric for total suppressed notifications. require.NoError(t, prom_testutil.GatherAndCompare(r, strings.NewReader(fmt.Sprintf(` # HELP alertmanager_marked_alerts How many alerts by state are currently marked in the Alertmanager regardless of their expiry. # TYPE alertmanager_marked_alerts gauge alertmanager_marked_alerts{state="active"} 0 alertmanager_marked_alerts{state="suppressed"} 0 alertmanager_marked_alerts{state="unprocessed"} 0 # HELP alertmanager_notifications_suppressed_total The total number of notifications suppressed for being silenced, inhibited, outside of active time intervals or within muted time intervals. # TYPE alertmanager_notifications_suppressed_total counter alertmanager_notifications_suppressed_total{reason="mute_time_interval"} %d `, len(test.alerts))))) } }) } } func TestTimeActiveStage(t testing.T) { sydney, err := time.LoadLocation("Australia/Sydney") if err != nil { t.Fatalf("Failed to load location Australia/Sydney: %s", err) } weekdays := map[string][]timeinterval.TimeInterval{ "weekdays": {{ Weekdays: []timeinterval.WeekdayRange{{ InclusiveRange: timeinterval.InclusiveRange{ Begin: 1, // Monday End: 5, // Friday }, }}, Times: []timeinterval.TimeRange{{ StartMinute: 540, // 09:00 EndMinute: 1020, // 17:00 }}, Location: &timeinterval.Location{Location: sydney}, }}, } tests := []struct { name string intervals map[string][]timeinterval.TimeInterval now time.Time alerts []types.Alert mutedBy []string }{{ name: "Should be muted outside working hours", intervals: weekdays, now: time.Date(2024, 1, 1, 0, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"weekdays"}, }, { name: "Should not be muted during workings hours", intervals: weekdays, now: time.Date(2024, 1, 1, 9, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: nil, }, { name: "Should be muted during weekends", intervals: weekdays, now: time.Date(2024, 1, 6, 10, 0, 0, 0, sydney), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"weekdays"}, }, { name: "Should be muted at 12pm UTC", intervals: weekdays, now: time.Date(2024, 1, 6, 10, 0, 0, 0, time.UTC), alerts: []types.Alert{{Alert: model.Alert{Labels: model.LabelSet{"foo": "bar"}}}}, mutedBy: []string{"weekdays"}, }} for _, test := range tests { t.Run(test.name, func(t testing.T) { r := prometheus.NewRegistry() marker := types.NewMarker(r) metrics := NewMetrics(r, featurecontrol.NoopFlags{}) intervener := timeinterval.NewIntervener(test.intervals) st := NewTimeActiveStage(intervener, marker, metrics) // Get the names of all time intervals for the context. activeTimeIntervalNames := make([]string, 0, len(test.intervals)) for name := range test.intervals { activeTimeIntervalNames = append(activeTimeIntervalNames, name) } // Sort the names so we can compare mutedBy with test.mutedBy. sort.Strings(activeTimeIntervalNames) ctx := context.Background() ctx = WithNow(ctx, test.now) ctx = WithGroupKey(ctx, "group1") ctx = WithActiveTimeIntervals(ctx, activeTimeIntervalNames) ctx = WithMuteTimeIntervals(ctx, nil) ctx = WithRouteID(ctx, "route1") _, active, err := st.Exec(ctx, promslog.NewNopLogger(), test.alerts...) require.NoError(t, err) if len(test.mutedBy) == 0 { // All alerts should be active. require.Equal(t, len(test.alerts), len(active)) // The group should not be marked. mutedBy, isMuted := marker.Muted("route1", "group1") require.False(t, isMuted) require.Empty(t, mutedBy) // The metric for total suppressed notifications should not // have been incremented, which means it will not be collected. require.NoError(t, prom_testutil.GatherAndCompare(r, strings.NewReader(` # HELP alertmanager_marked_alerts How many alerts by state are currently marked in the Alertmanager regardless of their expiry. # TYPE alertmanager_marked_alerts gauge alertmanager_marked_alerts{state="active"} 0 alertmanager_marked_alerts{state="suppressed"} 0 alertmanager_marked_alerts{state="unprocessed"} 0 `))) } else { // All alerts should be muted. require.Empty(t, active) // The group should be marked as muted. mutedBy, isMuted := marker.Muted("route1", "group1") require.True(t, isMuted) require.Equal(t, test.mutedBy, mutedBy) // Gets the metric for total suppressed notifications. require.NoError(t, prom_testutil.GatherAndCompare(r, strings.NewReader(fmt.Sprintf(` # HELP alertmanager_marked_alerts How many alerts by state are currently marked in the Alertmanager regardless of their expiry. # TYPE alertmanager_marked_alerts gauge alertmanager_marked_alerts{state="active"} 0 alertmanager_marked_alerts{state="suppressed"} 0 alertmanager_marked_alerts{state="unprocessed"} 0 # HELP alertmanager_notifications_suppressed_total The total number of notifications suppressed for being silenced, inhibited, outside of active time intervals or within muted time intervals. # TYPE alertmanager_notifications_suppressed_total counter alertmanager_notifications_suppressed_total{reason="active_time_interval"} %d `, len(test.alerts))))) } }) } } func BenchmarkHashAlert(b testing.B) { alert := &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{"foo": "the_first_value", "bar": "the_second_value", "another": "value"}, }, } for i := 0; i < b.N; i++ { hashAlert(alert) } }	32.69 KB	2025-05-25 12:27
📁 opsgenie	-	2025-05-25 12:27
📁 pagerduty	-	2025-05-25 12:27
📁 pushover	-	2025-05-25 12:27
📁 rocketchat	-	2025-05-25 12:27
📁 slack	-	2025-05-25 12:27
📁 sns	-	2025-05-25 12:27
📁 telegram	-	2025-05-25 12:27
📁 test	-	2025-05-25 12:27
📄 util.go // Copyright 2019 Prometheus Team // 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" "crypto/sha256" "errors" "fmt" "io" "log/slog" "net/http" "net/url" "strings" "github.com/prometheus/common/version" "github.com/prometheus/alertmanager/template" "github.com/prometheus/alertmanager/types" ) // truncationMarker is the character used to represent a truncation. const truncationMarker = "…" // UserAgentHeader is the default User-Agent for notification requests. var UserAgentHeader = version.ComponentUserAgent("Alertmanager") // RedactURL removes the URL part from an error of url.Error type. func RedactURL(err error) error { var e url.Error if !errors.As(err, &e) { return err } e.URL = "<redacted>" return e } // Get sends a GET request to the given URL. func Get(ctx context.Context, client http.Client, url string) (http.Response, error) { return request(ctx, client, http.MethodGet, url, "", nil) } // PostJSON sends a POST request with JSON payload to the given URL. func PostJSON(ctx context.Context, client http.Client, url string, body io.Reader) (http.Response, error) { return post(ctx, client, url, "application/json", body) } // PostText sends a POST request with text payload to the given URL. func PostText(ctx context.Context, client http.Client, url string, body io.Reader) (http.Response, error) { return post(ctx, client, url, "text/plain", body) } func post(ctx context.Context, client http.Client, url, bodyType string, body io.Reader) (http.Response, error) { return request(ctx, client, http.MethodPost, url, bodyType, body) } func request(ctx context.Context, client http.Client, method, url, bodyType string, body io.Reader) (http.Response, error) { req, err := http.NewRequest(method, url, body) if err != nil { return nil, err } req.Header.Set("User-Agent", UserAgentHeader) if bodyType != "" { req.Header.Set("Content-Type", bodyType) } return client.Do(req.WithContext(ctx)) } // Drain consumes and closes the response's body to make sure that the // HTTP client can reuse existing connections. func Drain(r http.Response) { io.Copy(io.Discard, r.Body) r.Body.Close() } // TruncateInRunes truncates a string to fit the given size in Runes. func TruncateInRunes(s string, n int) (string, bool) { r := []rune(s) if len(r) <= n { return s, false } if n <= 3 { return string(r[:n]), true } return string(r[:n-1]) + truncationMarker, true } // TruncateInBytes truncates a string to fit the given size in Bytes. func TruncateInBytes(s string, n int) (string, bool) { // First, measure the string the w/o a to-rune conversion. if len(s) <= n { return s, false } // The truncationMarker itself is 3 bytes, we can't return any part of the string when it's less than 3. if n <= 3 { switch n { case 3: return truncationMarker, true default: return strings.Repeat(".", n), true } } // Now, to ensure we don't butcher the string we need to remove using runes. r := []rune(s) truncationTarget := n - 3 // Next, let's truncate the runes to the lower possible number. truncatedRunes := r[:truncationTarget] for len(string(truncatedRunes)) > truncationTarget { truncatedRunes = r[:len(truncatedRunes)-1] } return string(truncatedRunes) + truncationMarker, true } // TmplText is using monadic error handling in order to make string templating // less verbose. Use with care as the final error checking is easily missed. func TmplText(tmpl template.Template, data template.Data, err error) func(string) string { return func(name string) (s string) { if err != nil { return } s, err = tmpl.ExecuteTextString(name, data) return s } } // TmplHTML is using monadic error handling in order to make string templating // less verbose. Use with care as the final error checking is easily missed. func TmplHTML(tmpl template.Template, data template.Data, err error) func(string) string { return func(name string) (s string) { if err != nil { return } s, err = tmpl.ExecuteHTMLString(name, data) return s } } // Key is a string that can be hashed. type Key string // ExtractGroupKey gets the group key from the context. func ExtractGroupKey(ctx context.Context) (Key, error) { key, ok := GroupKey(ctx) if !ok { return "", fmt.Errorf("group key missing") } return Key(key), nil } // Hash returns the sha256 for a group key as integrations may have // maximum length requirements on deduplication keys. func (k Key) Hash() string { h := sha256.New() // hash.Hash.Write never returns an error. //nolint: errcheck h.Write([]byte(string(k))) return fmt.Sprintf("%x", h.Sum(nil)) } func (k Key) String() string { return string(k) } // GetTemplateData creates the template data from the context and the alerts. func GetTemplateData(ctx context.Context, tmpl template.Template, alerts []types.Alert, l slog.Logger) template.Data { recv, ok := ReceiverName(ctx) if !ok { l.Error("Missing receiver") } groupLabels, ok := GroupLabels(ctx) if !ok { l.Error("Missing group labels") } return tmpl.Data(recv, groupLabels, alerts...) } func readAll(r io.Reader) string { if r == nil { return "" } bs, err := io.ReadAll(r) if err != nil { return "" } return string(bs) } // Retrier knows when to retry an HTTP request to a receiver. 2xx status codes // are successful, anything else is a failure and only 5xx status codes should // be retried. type Retrier struct { // Function to return additional information in the error message. CustomDetailsFunc func(code int, body io.Reader) string // Additional HTTP status codes that should be retried. RetryCodes []int } // Check returns a boolean indicating whether the request should be retried // and an optional error if the request has failed. If body is not nil, it will // be included in the error message. func (r Retrier) Check(statusCode int, body io.Reader) (bool, error) { // 2xx responses are considered to be always successful. if statusCode/100 == 2 { return false, nil } // 5xx responses are considered to be always retried. retry := statusCode/100 == 5 if !retry { for _, code := range r.RetryCodes { if code == statusCode { retry = true break } } } s := fmt.Sprintf("unexpected status code %v", statusCode) var details string if r.CustomDetailsFunc != nil { details = r.CustomDetailsFunc(statusCode, body) } else { details = readAll(body) } if details != "" { s = fmt.Sprintf("%s: %s", s, details) } return retry, errors.New(s) } type ErrorWithReason struct { Err error Reason Reason } func NewErrorWithReason(reason Reason, err error) ErrorWithReason { return &ErrorWithReason{ Err: err, Reason: reason, } } func (e ErrorWithReason) Error() string { return e.Err.Error() } // Reason is the failure reason. type Reason int const ( DefaultReason Reason = iota ClientErrorReason ServerErrorReason ContextCanceledReason ContextDeadlineExceededReason ) func (s Reason) String() string { switch s { case DefaultReason: return "other" case ClientErrorReason: return "clientError" case ServerErrorReason: return "serverError" case ContextCanceledReason: return "contextCanceled" case ContextDeadlineExceededReason: return "contextDeadlineExceeded" default: panic(fmt.Sprintf("unknown Reason: %d", s)) } } // possibleFailureReasonCategory is a list of possible failure reason. var possibleFailureReasonCategory = []string{DefaultReason.String(), ClientErrorReason.String(), ServerErrorReason.String(), ContextCanceledReason.String(), ContextDeadlineExceededReason.String()} // GetFailureReasonFromStatusCode returns the reason for the failure based on the status code provided. func GetFailureReasonFromStatusCode(statusCode int) Reason { if statusCode/100 == 4 { return ClientErrorReason } if statusCode/100 == 5 { return ServerErrorReason } return DefaultReason }	8.24 KB	2025-05-25 12:27
📄 util_test.go // Copyright 2019 Prometheus Team // 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 ( "bytes" "fmt" "io" "net/http" "path" "reflect" "runtime" "testing" "github.com/stretchr/testify/require" ) func TestTruncate(t testing.T) { type expect struct { out string trunc bool } testCases := []struct { in string n int runes expect bytes expect }{ { in: "", n: 5, runes: expect{out: "", trunc: false}, bytes: expect{out: "", trunc: false}, }, { in: "abcde", n: 2, runes: expect{out: "ab", trunc: true}, bytes: expect{out: "..", trunc: true}, }, { in: "abcde", n: 4, runes: expect{out: "abc…", trunc: true}, bytes: expect{out: "a…", trunc: true}, }, { in: "abcde", n: 5, runes: expect{out: "abcde", trunc: false}, bytes: expect{out: "abcde", trunc: false}, }, { in: "abcdefgh", n: 5, runes: expect{out: "abcd…", trunc: true}, bytes: expect{out: "ab…", trunc: true}, }, { in: "a⌘cde", n: 5, runes: expect{out: "a⌘cde", trunc: false}, bytes: expect{out: "a…", trunc: true}, }, { in: "a⌘cdef", n: 5, runes: expect{out: "a⌘cd…", trunc: true}, bytes: expect{out: "a…", trunc: true}, }, { in: "世界cdef", n: 3, runes: expect{out: "世界c", trunc: true}, bytes: expect{out: "…", trunc: true}, }, { in: "❤️✅🚀🔥❌❤️✅🚀🔥❌❤️✅🚀🔥❌❤️✅🚀🔥❌", n: 19, runes: expect{out: "❤️✅🚀🔥❌❤️✅🚀🔥❌❤️✅🚀🔥❌…", trunc: true}, bytes: expect{out: "❤️✅🚀…", trunc: true}, }, } type truncateFunc func(string, int) (string, bool) for _, tc := range testCases { for _, fn := range []truncateFunc{TruncateInBytes, TruncateInRunes} { var truncated bool var out string fnPath := runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name() fnName := path.Base(fnPath) switch fnName { case "notify.TruncateInRunes": truncated = tc.runes.trunc out = tc.runes.out case "notify.TruncateInBytes": truncated = tc.bytes.trunc out = tc.bytes.out default: t.Fatalf("unknown function") } t.Run(fmt.Sprintf("%s(%s,%d)", fnName, tc.in, tc.n), func(t testing.T) { s, trunc := fn(tc.in, tc.n) require.Equal(t, out, s) require.Equal(t, truncated, trunc) }) } } } type brokenReader struct{} func (b brokenReader) Read([]byte) (int, error) { return 0, fmt.Errorf("some error") } func TestRetrierCheck(t testing.T) { for _, tc := range []struct { retrier Retrier status int body io.Reader retry bool expectedErr string }{ { retrier: Retrier{}, status: http.StatusOK, body: bytes.NewBuffer([]byte("ok")), retry: false, }, { retrier: Retrier{}, status: http.StatusNoContent, retry: false, }, { retrier: Retrier{}, status: http.StatusBadRequest, retry: false, expectedErr: "unexpected status code 400", }, { retrier: Retrier{RetryCodes: []int{http.StatusTooManyRequests}}, status: http.StatusBadRequest, body: bytes.NewBuffer([]byte("invalid request")), retry: false, expectedErr: "unexpected status code 400: invalid request", }, { retrier: Retrier{RetryCodes: []int{http.StatusTooManyRequests}}, status: http.StatusTooManyRequests, retry: true, expectedErr: "unexpected status code 429", }, { retrier: Retrier{}, status: http.StatusServiceUnavailable, body: bytes.NewBuffer([]byte("retry later")), retry: true, expectedErr: "unexpected status code 503: retry later", }, { retrier: Retrier{}, status: http.StatusBadGateway, body: &brokenReader{}, retry: true, expectedErr: "unexpected status code 502", }, { retrier: Retrier{CustomDetailsFunc: func(status int, b io.Reader) string { if status != http.StatusServiceUnavailable { return "invalid" } bs, _ := io.ReadAll(b) return fmt.Sprintf("server response is %q", string(bs)) }}, status: http.StatusServiceUnavailable, body: bytes.NewBuffer([]byte("retry later")), retry: true, expectedErr: "unexpected status code 503: server response is \"retry later\"", }, } { t.Run("", func(t testing.T) { retry, err := tc.retrier.Check(tc.status, tc.body) require.Equal(t, tc.retry, retry) if tc.expectedErr == "" { require.NoError(t, err) return } require.EqualError(t, err, tc.expectedErr) }) } }	5 KB	2025-05-25 12:27
📁 victorops	-	2025-05-25 12:27
📁 webex	-	2025-05-25 12:27
📁 webhook	-	2025-05-25 12:27
📁 wechat	-	2025-05-25 12:27

Edit File:

Terminal