Wala shell

Advanced PHP File Manager

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

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 dispatch.go // Copyright 2018 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 dispatch import ( "context" "errors" "fmt" "log/slog" "sort" "sync" "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/model" "github.com/prometheus/alertmanager/notify" "github.com/prometheus/alertmanager/provider" "github.com/prometheus/alertmanager/store" "github.com/prometheus/alertmanager/types" ) // DispatcherMetrics represents metrics associated to a dispatcher. type DispatcherMetrics struct { aggrGroups prometheus.Gauge processingDuration prometheus.Summary aggrGroupLimitReached prometheus.Counter } // NewDispatcherMetrics returns a new registered DispatchMetrics. func NewDispatcherMetrics(registerLimitMetrics bool, r prometheus.Registerer) DispatcherMetrics { m := DispatcherMetrics{ aggrGroups: prometheus.NewGauge( prometheus.GaugeOpts{ Name: "alertmanager_dispatcher_aggregation_groups", Help: "Number of active aggregation groups", }, ), processingDuration: prometheus.NewSummary( prometheus.SummaryOpts{ Name: "alertmanager_dispatcher_alert_processing_duration_seconds", Help: "Summary of latencies for the processing of alerts.", }, ), aggrGroupLimitReached: prometheus.NewCounter( prometheus.CounterOpts{ Name: "alertmanager_dispatcher_aggregation_group_limit_reached_total", Help: "Number of times when dispatcher failed to create new aggregation group due to limit.", }, ), } if r != nil { r.MustRegister(m.aggrGroups, m.processingDuration) if registerLimitMetrics { r.MustRegister(m.aggrGroupLimitReached) } } return &m } // Dispatcher sorts incoming alerts into aggregation groups and // assigns the correct notifiers to each. type Dispatcher struct { route Route alerts provider.Alerts stage notify.Stage marker types.GroupMarker metrics DispatcherMetrics limits Limits timeout func(time.Duration) time.Duration mtx sync.RWMutex aggrGroupsPerRoute map[Route]map[model.Fingerprint]aggrGroup aggrGroupsNum int done chan struct{} ctx context.Context cancel func() logger slog.Logger } // Limits describes limits used by Dispatcher. type Limits interface { // MaxNumberOfAggregationGroups returns max number of aggregation groups that dispatcher can have. // 0 or negative value = unlimited. // If dispatcher hits this limit, it will not create additional groups, but will log an error instead. MaxNumberOfAggregationGroups() int } // NewDispatcher returns a new Dispatcher. func NewDispatcher( ap provider.Alerts, r Route, s notify.Stage, mk types.GroupMarker, to func(time.Duration) time.Duration, lim Limits, l slog.Logger, m DispatcherMetrics, ) Dispatcher { if lim == nil { lim = nilLimits{} } disp := &Dispatcher{ alerts: ap, stage: s, route: r, marker: mk, timeout: to, logger: l.With("component", "dispatcher"), metrics: m, limits: lim, } return disp } // Run starts dispatching alerts incoming via the updates channel. func (d Dispatcher) Run() { d.done = make(chan struct{}) d.mtx.Lock() d.aggrGroupsPerRoute = map[Route]map[model.Fingerprint]aggrGroup{} d.aggrGroupsNum = 0 d.metrics.aggrGroups.Set(0) d.ctx, d.cancel = context.WithCancel(context.Background()) d.mtx.Unlock() d.run(d.alerts.Subscribe()) close(d.done) } func (d Dispatcher) run(it provider.AlertIterator) { maintenance := time.NewTicker(30 * time.Second) defer maintenance.Stop() defer it.Close() for { select { case alert, ok := <-it.Next(): if !ok { // Iterator exhausted for some reason. if err := it.Err(); err != nil { d.logger.Error("Error on alert update", "err", err) } return } d.logger.Debug("Received alert", "alert", alert) // Log errors but keep trying. if err := it.Err(); err != nil { d.logger.Error("Error on alert update", "err", err) continue } now := time.Now() for _, r := range d.route.Match(alert.Labels) { d.processAlert(alert, r) } d.metrics.processingDuration.Observe(time.Since(now).Seconds()) case <-maintenance.C: d.doMaintenance() case <-d.ctx.Done(): return } } } func (d Dispatcher) doMaintenance() { d.mtx.Lock() defer d.mtx.Unlock() for _, groups := range d.aggrGroupsPerRoute { for _, ag := range groups { if ag.empty() { ag.stop() d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey()) delete(groups, ag.fingerprint()) d.aggrGroupsNum-- d.metrics.aggrGroups.Dec() } } } } // AlertGroup represents how alerts exist within an aggrGroup. type AlertGroup struct { Alerts types.AlertSlice Labels model.LabelSet Receiver string GroupKey string RouteID string } type AlertGroups []AlertGroup func (ag AlertGroups) Swap(i, j int) { ag[i], ag[j] = ag[j], ag[i] } func (ag AlertGroups) Less(i, j int) bool { if ag[i].Labels.Equal(ag[j].Labels) { return ag[i].Receiver < ag[j].Receiver } return ag[i].Labels.Before(ag[j].Labels) } func (ag AlertGroups) Len() int { return len(ag) } // Groups returns a slice of AlertGroups from the dispatcher's internal state. func (d Dispatcher) Groups(routeFilter func(Route) bool, alertFilter func(types.Alert, time.Time) bool) (AlertGroups, map[model.Fingerprint][]string) { groups := AlertGroups{} d.mtx.RLock() defer d.mtx.RUnlock() // Keep a list of receivers for an alert to prevent checking each alert // again against all routes. The alert has already matched against this // route on ingestion. receivers := map[model.Fingerprint][]string{} now := time.Now() for route, ags := range d.aggrGroupsPerRoute { if !routeFilter(route) { continue } for _, ag := range ags { receiver := route.RouteOpts.Receiver alertGroup := &AlertGroup{ Labels: ag.labels, Receiver: receiver, GroupKey: ag.GroupKey(), RouteID: ag.routeID, } alerts := ag.alerts.List() filteredAlerts := make([]types.Alert, 0, len(alerts)) for _, a := range alerts { if !alertFilter(a, now) { continue } fp := a.Fingerprint() if r, ok := receivers[fp]; ok { // Receivers slice already exists. Add // the current receiver to the slice. receivers[fp] = append(r, receiver) } else { // First time we've seen this alert fingerprint. // Initialize a new receivers slice. receivers[fp] = []string{receiver} } filteredAlerts = append(filteredAlerts, a) } if len(filteredAlerts) == 0 { continue } alertGroup.Alerts = filteredAlerts groups = append(groups, alertGroup) } } sort.Sort(groups) for i := range groups { sort.Sort(groups[i].Alerts) } for i := range receivers { sort.Strings(receivers[i]) } return groups, receivers } // Stop the dispatcher. func (d Dispatcher) Stop() { if d == nil { return } d.mtx.Lock() if d.cancel == nil { d.mtx.Unlock() return } d.cancel() d.cancel = nil d.mtx.Unlock() <-d.done } // notifyFunc is a function that performs notification for the alert // with the given fingerprint. It aborts on context cancelation. // Returns false iff notifying failed. type notifyFunc func(context.Context, ...types.Alert) bool // processAlert determines in which aggregation group the alert falls // and inserts it. func (d Dispatcher) processAlert(alert types.Alert, route Route) { groupLabels := getGroupLabels(alert, route) fp := groupLabels.Fingerprint() d.mtx.Lock() defer d.mtx.Unlock() routeGroups, ok := d.aggrGroupsPerRoute[route] if !ok { routeGroups = map[model.Fingerprint]aggrGroup{} d.aggrGroupsPerRoute[route] = routeGroups } ag, ok := routeGroups[fp] if ok { ag.insert(alert) return } // If the group does not exist, create it. But check the limit first. if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && d.aggrGroupsNum >= limit { d.metrics.aggrGroupLimitReached.Inc() d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", d.aggrGroupsNum, "limit", limit, "alert", alert.Name()) return } ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger) routeGroups[fp] = ag d.aggrGroupsNum++ d.metrics.aggrGroups.Inc() // Insert the 1st alert in the group before starting the group's run() // function, to make sure that when the run() will be executed the 1st // alert is already there. ag.insert(alert) go ag.run(func(ctx context.Context, alerts ...types.Alert) bool { _, _, err := d.stage.Exec(ctx, d.logger, alerts...) if err != nil { logger := d.logger.With("num_alerts", len(alerts), "err", err) if errors.Is(ctx.Err(), context.Canceled) { // It is expected for the context to be canceled on // configuration reload or shutdown. In this case, the // message should only be logged at the debug level. logger.Debug("Notify for alerts failed") } else { logger.Error("Notify for alerts failed") } } return err == nil }) } func getGroupLabels(alert types.Alert, route Route) model.LabelSet { groupLabels := model.LabelSet{} for ln, lv := range alert.Labels { if _, ok := route.RouteOpts.GroupBy[ln]; ok \|\| route.RouteOpts.GroupByAll { groupLabels[ln] = lv } } return groupLabels } // aggrGroup aggregates alert fingerprints into groups to which a // common set of routing options applies. // It emits notifications in the specified intervals. type aggrGroup struct { labels model.LabelSet opts RouteOpts logger slog.Logger routeID string routeKey string alerts store.Alerts ctx context.Context cancel func() done chan struct{} next time.Timer timeout func(time.Duration) time.Duration mtx sync.RWMutex hasFlushed bool } // newAggrGroup returns a new aggregation group. func newAggrGroup(ctx context.Context, labels model.LabelSet, r Route, to func(time.Duration) time.Duration, logger slog.Logger) aggrGroup { if to == nil { to = func(d time.Duration) time.Duration { return d } } ag := &aggrGroup{ labels: labels, routeID: r.ID(), routeKey: r.Key(), opts: &r.RouteOpts, timeout: to, alerts: store.NewAlerts(), done: make(chan struct{}), } ag.ctx, ag.cancel = context.WithCancel(ctx) ag.logger = logger.With("aggrGroup", ag) // Set an initial one-time wait before flushing // the first batch of notifications. ag.next = time.NewTimer(ag.opts.GroupWait) return ag } func (ag aggrGroup) fingerprint() model.Fingerprint { return ag.labels.Fingerprint() } func (ag aggrGroup) GroupKey() string { return fmt.Sprintf("%s:%s", ag.routeKey, ag.labels) } func (ag aggrGroup) String() string { return ag.GroupKey() } func (ag aggrGroup) run(nf notifyFunc) { defer close(ag.done) defer ag.next.Stop() for { select { case now := <-ag.next.C: // Give the notifications time until the next flush to // finish before terminating them. ctx, cancel := context.WithTimeout(ag.ctx, ag.timeout(ag.opts.GroupInterval)) // The now time we retrieve from the ticker is the only reliable // point of time reference for the subsequent notification pipeline. // Calculating the current time directly is prone to flaky behavior, // which usually only becomes apparent in tests. ctx = notify.WithNow(ctx, now) // Populate context with information needed along the pipeline. ctx = notify.WithGroupKey(ctx, ag.GroupKey()) ctx = notify.WithGroupLabels(ctx, ag.labels) ctx = notify.WithReceiverName(ctx, ag.opts.Receiver) ctx = notify.WithRepeatInterval(ctx, ag.opts.RepeatInterval) ctx = notify.WithMuteTimeIntervals(ctx, ag.opts.MuteTimeIntervals) ctx = notify.WithActiveTimeIntervals(ctx, ag.opts.ActiveTimeIntervals) ctx = notify.WithRouteID(ctx, ag.routeID) // Wait the configured interval before calling flush again. ag.mtx.Lock() ag.next.Reset(ag.opts.GroupInterval) ag.hasFlushed = true ag.mtx.Unlock() ag.flush(func(alerts ...types.Alert) bool { return nf(ctx, alerts...) }) cancel() case <-ag.ctx.Done(): return } } } func (ag aggrGroup) stop() { // Calling cancel will terminate all in-process notifications // and the run() loop. ag.cancel() <-ag.done } // insert inserts the alert into the aggregation group. func (ag aggrGroup) insert(alert types.Alert) { if err := ag.alerts.Set(alert); err != nil { ag.logger.Error("error on set alert", "err", err) } // Immediately trigger a flush if the wait duration for this // alert is already over. ag.mtx.Lock() defer ag.mtx.Unlock() if !ag.hasFlushed && alert.StartsAt.Add(ag.opts.GroupWait).Before(time.Now()) { ag.next.Reset(0) } } func (ag aggrGroup) empty() bool { return ag.alerts.Empty() } // flush sends notifications for all new alerts. func (ag aggrGroup) flush(notify func(...types.Alert) bool) { if ag.empty() { return } var ( alerts = ag.alerts.List() alertsSlice = make(types.AlertSlice, 0, len(alerts)) resolvedSlice = make(types.AlertSlice, 0, len(alerts)) now = time.Now() ) for _, alert := range alerts { a := alert // Ensure that alerts don't resolve as time move forwards. if a.ResolvedAt(now) { resolvedSlice = append(resolvedSlice, &a) } else { a.EndsAt = time.Time{} } alertsSlice = append(alertsSlice, &a) } sort.Stable(alertsSlice) ag.logger.Debug("flushing", "alerts", fmt.Sprintf("%v", alertsSlice)) if notify(alertsSlice...) { // Delete all resolved alerts as we just sent a notification for them, // and we don't want to send another one. However, we need to make sure // that each resolved alert has not fired again during the flush as then // we would delete an active alert thinking it was resolved. if err := ag.alerts.DeleteIfNotModified(resolvedSlice); err != nil { ag.logger.Error("error on delete alerts", "err", err) } } } type nilLimits struct{} func (n nilLimits) MaxNumberOfAggregationGroups() int { return 0 }	14.15 KB	2025-05-25 12:27
📄 dispatch_test.go // Copyright 2018 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 dispatch import ( "context" "fmt" "log/slog" "reflect" "sort" "sync" "testing" "time" "github.com/prometheus/client_golang/prometheus" "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/config" "github.com/prometheus/alertmanager/notify" "github.com/prometheus/alertmanager/provider/mem" "github.com/prometheus/alertmanager/types" ) func TestAggrGroup(t testing.T) { lset := model.LabelSet{ "a": "v1", "b": "v2", } opts := &RouteOpts{ Receiver: "n1", GroupBy: map[model.LabelName]struct{}{ "a": {}, "b": {}, }, GroupWait: 1 time.Second, GroupInterval: 300 * time.Millisecond, RepeatInterval: 1 * time.Hour, } route := &Route{ RouteOpts: opts, } var ( a1 = &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{ "a": "v1", "b": "v2", "c": "v3", }, StartsAt: time.Now().Add(time.Minute), EndsAt: time.Now().Add(time.Hour), }, UpdatedAt: time.Now(), } a2 = &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{ "a": "v1", "b": "v2", "c": "v4", }, StartsAt: time.Now().Add(-time.Hour), EndsAt: time.Now().Add(2 time.Hour), }, UpdatedAt: time.Now(), } a3 = &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{ "a": "v1", "b": "v2", "c": "v5", }, StartsAt: time.Now().Add(time.Minute), EndsAt: time.Now().Add(5 * time.Minute), }, UpdatedAt: time.Now(), } ) var ( last = time.Now() current = time.Now() lastCurMtx = &sync.Mutex{} alertsCh = make(chan types.AlertSlice) ) ntfy := func(ctx context.Context, alerts ...types.Alert) bool { // Validate that the context is properly populated. if _, ok := notify.Now(ctx); !ok { t.Errorf("now missing") } if _, ok := notify.GroupKey(ctx); !ok { t.Errorf("group key missing") } if lbls, ok := notify.GroupLabels(ctx); !ok \|\| !reflect.DeepEqual(lbls, lset) { t.Errorf("wrong group labels: %q", lbls) } if rcv, ok := notify.ReceiverName(ctx); !ok \|\| rcv != opts.Receiver { t.Errorf("wrong receiver: %q", rcv) } if ri, ok := notify.RepeatInterval(ctx); !ok \|\| ri != opts.RepeatInterval { t.Errorf("wrong repeat interval: %q", ri) } lastCurMtx.Lock() last = current // Subtract a millisecond to allow for races. current = time.Now().Add(-time.Millisecond) lastCurMtx.Unlock() alertsCh <- types.AlertSlice(alerts) return true } removeEndsAt := func(as types.AlertSlice) types.AlertSlice { for i, a := range as { ac := a ac.EndsAt = time.Time{} as[i] = &ac } return as } // Test regular situation where we wait for group_wait to send out alerts. ag := newAggrGroup(context.Background(), lset, route, nil, promslog.NewNopLogger()) go ag.run(ntfy) ag.insert(a1) select { case <-time.After(2 * opts.GroupWait): t.Fatalf("expected initial batch after group_wait") case batch := <-alertsCh: lastCurMtx.Lock() s := time.Since(last) lastCurMtx.Unlock() if s < opts.GroupWait { t.Fatalf("received batch too early after %v", s) } exp := removeEndsAt(types.AlertSlice{a1}) sort.Sort(batch) if !reflect.DeepEqual(batch, exp) { t.Fatalf("expected alerts %v but got %v", exp, batch) } } for i := 0; i < 3; i++ { // New alert should come in after group interval. ag.insert(a3) select { case <-time.After(2 * opts.GroupInterval): t.Fatalf("expected new batch after group interval but received none") case batch := <-alertsCh: lastCurMtx.Lock() s := time.Since(last) lastCurMtx.Unlock() if s < opts.GroupInterval { t.Fatalf("received batch too early after %v", s) } exp := removeEndsAt(types.AlertSlice{a1, a3}) sort.Sort(batch) if !reflect.DeepEqual(batch, exp) { t.Fatalf("expected alerts %v but got %v", exp, batch) } } } ag.stop() // Add an alert that started more than group_interval in the past. We expect // immediate flushing. // Finally, set all alerts to be resolved. After successful notify the aggregation group // should empty itself. ag = newAggrGroup(context.Background(), lset, route, nil, promslog.NewNopLogger()) go ag.run(ntfy) ag.insert(a1) ag.insert(a2) // a2 lies way in the past so the initial group_wait should be skipped. select { case <-time.After(opts.GroupWait / 2): t.Fatalf("expected immediate alert but received none") case batch := <-alertsCh: exp := removeEndsAt(types.AlertSlice{a1, a2}) sort.Sort(batch) if !reflect.DeepEqual(batch, exp) { t.Fatalf("expected alerts %v but got %v", exp, batch) } } for i := 0; i < 3; i++ { // New alert should come in after group interval. ag.insert(a3) select { case <-time.After(2 * opts.GroupInterval): t.Fatalf("expected new batch after group interval but received none") case batch := <-alertsCh: lastCurMtx.Lock() s := time.Since(last) lastCurMtx.Unlock() if s < opts.GroupInterval { t.Fatalf("received batch too early after %v", s) } exp := removeEndsAt(types.AlertSlice{a1, a2, a3}) sort.Sort(batch) if !reflect.DeepEqual(batch, exp) { t.Fatalf("expected alerts %v but got %v", exp, batch) } } } // Resolve an alert, and it should be removed after the next batch was sent. a1r := a1 a1r.EndsAt = time.Now() ag.insert(&a1r) exp := append(types.AlertSlice{&a1r}, removeEndsAt(types.AlertSlice{a2, a3})...) select { case <-time.After(2 opts.GroupInterval): t.Fatalf("expected new batch after group interval but received none") case batch := <-alertsCh: lastCurMtx.Lock() s := time.Since(last) lastCurMtx.Unlock() if s < opts.GroupInterval { t.Fatalf("received batch too early after %v", s) } sort.Sort(batch) if !reflect.DeepEqual(batch, exp) { t.Fatalf("expected alerts %v but got %v", exp, batch) } } // Resolve all remaining alerts, they should be removed after the next batch was sent. // Do not add a1r as it should have been deleted following the previous batch. a2r, a3r := a2, a3 resolved := types.AlertSlice{&a2r, &a3r} for _, a := range resolved { a.EndsAt = time.Now() ag.insert(a) } select { case <-time.After(2 * opts.GroupInterval): t.Fatalf("expected new batch after group interval but received none") case batch := <-alertsCh: lastCurMtx.Lock() s := time.Since(last) lastCurMtx.Unlock() if s < opts.GroupInterval { t.Fatalf("received batch too early after %v", s) } sort.Sort(batch) if !reflect.DeepEqual(batch, resolved) { t.Fatalf("expected alerts %v but got %v", resolved, batch) } if !ag.empty() { t.Fatalf("Expected aggregation group to be empty after resolving alerts: %v", ag) } } ag.stop() } func TestGroupLabels(t testing.T) { a := &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{ "a": "v1", "b": "v2", "c": "v3", }, }, } route := &Route{ RouteOpts: RouteOpts{ GroupBy: map[model.LabelName]struct{}{ "a": {}, "b": {}, }, GroupByAll: false, }, } expLs := model.LabelSet{ "a": "v1", "b": "v2", } ls := getGroupLabels(a, route) if !reflect.DeepEqual(ls, expLs) { t.Fatalf("expected labels are %v, but got %v", expLs, ls) } } func TestGroupByAllLabels(t testing.T) { a := &types.Alert{ Alert: model.Alert{ Labels: model.LabelSet{ "a": "v1", "b": "v2", "c": "v3", }, }, } route := &Route{ RouteOpts: RouteOpts{ GroupBy: map[model.LabelName]struct{}{}, GroupByAll: true, }, } expLs := model.LabelSet{ "a": "v1", "b": "v2", "c": "v3", } ls := getGroupLabels(a, route) if !reflect.DeepEqual(ls, expLs) { t.Fatalf("expected labels are %v, but got %v", expLs, ls) } } func TestGroups(t testing.T) { confData := `receivers: - name: 'kafka' - name: 'prod' - name: 'testing' route: group_by: ['alertname'] group_wait: 10ms group_interval: 10ms receiver: 'prod' routes: - match: env: 'testing' receiver: 'testing' group_by: ['alertname', 'service'] - match: env: 'prod' receiver: 'prod' group_by: ['alertname', 'service', 'cluster'] continue: true - match: kafka: 'yes' receiver: 'kafka' group_by: ['alertname', 'service', 'cluster']` conf, err := config.Load(confData) if err != nil { t.Fatal(err) } logger := promslog.NewNopLogger() route := NewRoute(conf.Route, nil) marker := types.NewMarker(prometheus.NewRegistry()) alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil) if err != nil { t.Fatal(err) } defer alerts.Close() timeout := func(d time.Duration) time.Duration { return time.Duration(0) } recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]types.Alert)} dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry())) go dispatcher.Run() defer dispatcher.Stop() // Create alerts. the dispatcher will automatically create the groups. inputAlerts := []types.Alert{ // Matches the parent route. newAlert(model.LabelSet{"alertname": "OtherAlert", "cluster": "cc", "service": "dd"}), // Matches the first sub-route. newAlert(model.LabelSet{"env": "testing", "alertname": "TestingAlert", "service": "api", "instance": "inst1"}), // Matches the second sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "aa", "service": "api", "instance": "inst1"}), newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "aa", "service": "api", "instance": "inst2"}), // Matches the second sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "bb", "service": "api", "instance": "inst1"}), // Matches the second and third sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighLatency", "cluster": "bb", "service": "db", "kafka": "yes", "instance": "inst3"}), } alerts.Put(inputAlerts...) // Let alerts get processed. for i := 0; len(recorder.Alerts()) != 7 && i < 10; i++ { time.Sleep(200 time.Millisecond) } require.Len(t, recorder.Alerts(), 7) alertGroups, receivers := dispatcher.Groups( func(Route) bool { return true }, func(types.Alert, time.Time) bool { return true }, ) require.Equal(t, AlertGroups{ &AlertGroup{ Alerts: []types.Alert{inputAlerts[0]}, Labels: model.LabelSet{ "alertname": "OtherAlert", }, Receiver: "prod", GroupKey: "{}:{alertname=\"OtherAlert\"}", RouteID: "{}", }, &AlertGroup{ Alerts: []types.Alert{inputAlerts[1]}, Labels: model.LabelSet{ "alertname": "TestingAlert", "service": "api", }, Receiver: "testing", GroupKey: "{}/{env=\"testing\"}:{alertname=\"TestingAlert\", service=\"api\"}", RouteID: "{}/{env=\"testing\"}/0", }, &AlertGroup{ Alerts: []types.Alert{inputAlerts[2], inputAlerts[3]}, Labels: model.LabelSet{ "alertname": "HighErrorRate", "service": "api", "cluster": "aa", }, Receiver: "prod", GroupKey: "{}/{env=\"prod\"}:{alertname=\"HighErrorRate\", cluster=\"aa\", service=\"api\"}", RouteID: "{}/{env=\"prod\"}/1", }, &AlertGroup{ Alerts: []types.Alert{inputAlerts[4]}, Labels: model.LabelSet{ "alertname": "HighErrorRate", "service": "api", "cluster": "bb", }, Receiver: "prod", GroupKey: "{}/{env=\"prod\"}:{alertname=\"HighErrorRate\", cluster=\"bb\", service=\"api\"}", RouteID: "{}/{env=\"prod\"}/1", }, &AlertGroup{ Alerts: []types.Alert{inputAlerts[5]}, Labels: model.LabelSet{ "alertname": "HighLatency", "service": "db", "cluster": "bb", }, Receiver: "kafka", GroupKey: "{}/{kafka=\"yes\"}:{alertname=\"HighLatency\", cluster=\"bb\", service=\"db\"}", RouteID: "{}/{kafka=\"yes\"}/2", }, &AlertGroup{ Alerts: []types.Alert{inputAlerts[5]}, Labels: model.LabelSet{ "alertname": "HighLatency", "service": "db", "cluster": "bb", }, Receiver: "prod", GroupKey: "{}/{env=\"prod\"}:{alertname=\"HighLatency\", cluster=\"bb\", service=\"db\"}", RouteID: "{}/{env=\"prod\"}/1", }, }, alertGroups) require.Equal(t, map[model.Fingerprint][]string{ inputAlerts[0].Fingerprint(): {"prod"}, inputAlerts[1].Fingerprint(): {"testing"}, inputAlerts[2].Fingerprint(): {"prod"}, inputAlerts[3].Fingerprint(): {"prod"}, inputAlerts[4].Fingerprint(): {"prod"}, inputAlerts[5].Fingerprint(): {"kafka", "prod"}, }, receivers) } func TestGroupsWithLimits(t testing.T) { confData := `receivers: - name: 'kafka' - name: 'prod' - name: 'testing' route: group_by: ['alertname'] group_wait: 10ms group_interval: 10ms receiver: 'prod' routes: - match: env: 'testing' receiver: 'testing' group_by: ['alertname', 'service'] - match: env: 'prod' receiver: 'prod' group_by: ['alertname', 'service', 'cluster'] continue: true - match: kafka: 'yes' receiver: 'kafka' group_by: ['alertname', 'service', 'cluster']` conf, err := config.Load(confData) if err != nil { t.Fatal(err) } logger := promslog.NewNopLogger() route := NewRoute(conf.Route, nil) marker := types.NewMarker(prometheus.NewRegistry()) alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil) if err != nil { t.Fatal(err) } defer alerts.Close() timeout := func(d time.Duration) time.Duration { return time.Duration(0) } recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]types.Alert)} lim := limits{groups: 6} m := NewDispatcherMetrics(true, prometheus.NewRegistry()) dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, lim, logger, m) go dispatcher.Run() defer dispatcher.Stop() // Create alerts. the dispatcher will automatically create the groups. inputAlerts := []types.Alert{ // Matches the parent route. newAlert(model.LabelSet{"alertname": "OtherAlert", "cluster": "cc", "service": "dd"}), // Matches the first sub-route. newAlert(model.LabelSet{"env": "testing", "alertname": "TestingAlert", "service": "api", "instance": "inst1"}), // Matches the second sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "aa", "service": "api", "instance": "inst1"}), newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "aa", "service": "api", "instance": "inst2"}), // Matches the second sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighErrorRate", "cluster": "bb", "service": "api", "instance": "inst1"}), // Matches the second and third sub-route. newAlert(model.LabelSet{"env": "prod", "alertname": "HighLatency", "cluster": "bb", "service": "db", "kafka": "yes", "instance": "inst3"}), } err = alerts.Put(inputAlerts...) if err != nil { t.Fatal(err) } // Let alerts get processed. for i := 0; len(recorder.Alerts()) != 7 && i < 10; i++ { time.Sleep(200 time.Millisecond) } require.Len(t, recorder.Alerts(), 7) routeFilter := func(Route) bool { return true } alertFilter := func(types.Alert, time.Time) bool { return true } alertGroups, _ := dispatcher.Groups(routeFilter, alertFilter) require.Len(t, alertGroups, 6) require.Equal(t, 0.0, testutil.ToFloat64(m.aggrGroupLimitReached)) // Try to store new alert. This time, we will hit limit for number of groups. err = alerts.Put(newAlert(model.LabelSet{"env": "prod", "alertname": "NewAlert", "cluster": "new-cluster", "service": "db"})) if err != nil { t.Fatal(err) } // Let alert get processed. for i := 0; testutil.ToFloat64(m.aggrGroupLimitReached) == 0 && i < 10; i++ { time.Sleep(200 * time.Millisecond) } require.Equal(t, 1.0, testutil.ToFloat64(m.aggrGroupLimitReached)) // Verify there are still only 6 groups. alertGroups, _ = dispatcher.Groups(routeFilter, alertFilter) require.Len(t, alertGroups, 6) } type recordStage struct { mtx sync.RWMutex alerts map[string]map[model.Fingerprint]types.Alert } func (r recordStage) Alerts() []types.Alert { r.mtx.RLock() defer r.mtx.RUnlock() alerts := make([]types.Alert, 0) for k := range r.alerts { for _, a := range r.alerts[k] { alerts = append(alerts, a) } } return alerts } func (r recordStage) Exec(ctx context.Context, l slog.Logger, alerts ...types.Alert) (context.Context, []types.Alert, error) { r.mtx.Lock() defer r.mtx.Unlock() gk, ok := notify.GroupKey(ctx) if !ok { panic("GroupKey not present!") } if _, ok := r.alerts[gk]; !ok { r.alerts[gk] = make(map[model.Fingerprint]types.Alert) } for _, a := range alerts { r.alerts[gk][a.Fingerprint()] = a } return ctx, nil, nil } var ( // Set the start time in the past to trigger a flush immediately. t0 = time.Now().Add(-time.Minute) // Set the end time in the future to avoid deleting the alert. t1 = t0.Add(2 time.Minute) ) func newAlert(labels model.LabelSet) types.Alert { return &types.Alert{ Alert: model.Alert{ Labels: labels, Annotations: model.LabelSet{"foo": "bar"}, StartsAt: t0, EndsAt: t1, GeneratorURL: "http://example.com/prometheus", }, UpdatedAt: t0, Timeout: false, } } func TestDispatcherRace(t testing.T) { logger := promslog.NewNopLogger() marker := types.NewMarker(prometheus.NewRegistry()) alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil) if err != nil { t.Fatal(err) } defer alerts.Close() timeout := func(d time.Duration) time.Duration { return time.Duration(0) } dispatcher := NewDispatcher(alerts, nil, nil, marker, timeout, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry())) go dispatcher.Run() dispatcher.Stop() } func TestDispatcherRaceOnFirstAlertNotDeliveredWhenGroupWaitIsZero(t testing.T) { const numAlerts = 5000 logger := promslog.NewNopLogger() marker := types.NewMarker(prometheus.NewRegistry()) alerts, err := mem.NewAlerts(context.Background(), marker, time.Hour, nil, logger, nil) if err != nil { t.Fatal(err) } defer alerts.Close() route := &Route{ RouteOpts: RouteOpts{ Receiver: "default", GroupBy: map[model.LabelName]struct{}{"alertname": {}}, GroupWait: 0, GroupInterval: 1 time.Hour, // Should never hit in this test. RepeatInterval: 1 * time.Hour, // Should never hit in this test. }, } timeout := func(d time.Duration) time.Duration { return d } recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]types.Alert)} dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, nil, logger, NewDispatcherMetrics(false, prometheus.NewRegistry())) go dispatcher.Run() defer dispatcher.Stop() // Push all alerts. for i := 0; i < numAlerts; i++ { alert := newAlert(model.LabelSet{"alertname": model.LabelValue(fmt.Sprintf("Alert_%d", i))}) require.NoError(t, alerts.Put(alert)) } // Wait until the alerts have been notified or the waiting timeout expires. for deadline := time.Now().Add(5 time.Second); time.Now().Before(deadline); { if len(recorder.Alerts()) >= numAlerts { break } // Throttle. time.Sleep(10 * time.Millisecond) } // We expect all alerts to be notified immediately, since they all belong to different groups. require.Len(t, recorder.Alerts(), numAlerts) } type limits struct { groups int } func (l limits) MaxNumberOfAggregationGroups() int { return l.groups } func TestDispatcher_DoMaintenance(t testing.T) { r := prometheus.NewRegistry() marker := types.NewMarker(r) alerts, err := mem.NewAlerts(context.Background(), marker, time.Minute, nil, promslog.NewNopLogger(), nil) if err != nil { t.Fatal(err) } route := &Route{ RouteOpts: RouteOpts{ GroupBy: map[model.LabelName]struct{}{"alertname": {}}, GroupWait: 0, GroupInterval: 5 time.Minute, // Should never hit in this test. }, } timeout := func(d time.Duration) time.Duration { return d } recorder := &recordStage{alerts: make(map[string]map[model.Fingerprint]types.Alert)} ctx := context.Background() dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, nil, promslog.NewNopLogger(), NewDispatcherMetrics(false, r)) aggrGroups := make(map[Route]map[model.Fingerprint]aggrGroup) aggrGroups[route] = make(map[model.Fingerprint]aggrGroup) // Insert an aggregation group with no alerts. labels := model.LabelSet{"alertname": "1"} aggrGroup1 := newAggrGroup(ctx, labels, route, timeout, promslog.NewNopLogger()) aggrGroups[route][aggrGroup1.fingerprint()] = aggrGroup1 dispatcher.aggrGroupsPerRoute = aggrGroups // Must run otherwise doMaintenance blocks on aggrGroup1.stop(). go aggrGroup1.run(func(context.Context, ...*types.Alert) bool { return true }) // Insert a marker for the aggregation group's group key. marker.SetMuted(route.ID(), aggrGroup1.GroupKey(), []string{"weekends"}) mutedBy, isMuted := marker.Muted(route.ID(), aggrGroup1.GroupKey()) require.True(t, isMuted) require.Equal(t, []string{"weekends"}, mutedBy) // Run the maintenance and the marker should be removed. dispatcher.doMaintenance() mutedBy, isMuted = marker.Muted(route.ID(), aggrGroup1.GroupKey()) require.False(t, isMuted) require.Empty(t, mutedBy) }	21.46 KB	2025-05-25 12:27
📄 route.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 dispatch import ( "encoding/json" "fmt" "sort" "strconv" "strings" "time" "github.com/prometheus/common/model" "github.com/prometheus/alertmanager/config" "github.com/prometheus/alertmanager/pkg/labels" ) // DefaultRouteOpts are the defaulting routing options which apply // to the root route of a routing tree. var DefaultRouteOpts = RouteOpts{ GroupWait: 30 * time.Second, GroupInterval: 5 * time.Minute, RepeatInterval: 4 * time.Hour, GroupBy: map[model.LabelName]struct{}{}, GroupByAll: false, MuteTimeIntervals: []string{}, } // A Route is a node that contains definitions of how to handle alerts. type Route struct { parent Route // The configuration parameters for matches of this route. RouteOpts RouteOpts // Matchers an alert has to fulfill to match // this route. Matchers labels.Matchers // If true, an alert matches further routes on the same level. Continue bool // Children routes of this route. Routes []Route } // NewRoute returns a new route. func NewRoute(cr config.Route, parent Route) Route { // Create default and overwrite with configured settings. opts := DefaultRouteOpts if parent != nil { opts = parent.RouteOpts } if cr.Receiver != "" { opts.Receiver = cr.Receiver } if cr.GroupBy != nil { opts.GroupBy = map[model.LabelName]struct{}{} for _, ln := range cr.GroupBy { opts.GroupBy[ln] = struct{}{} } opts.GroupByAll = false } else { if cr.GroupByAll { opts.GroupByAll = cr.GroupByAll } } if cr.GroupWait != nil { opts.GroupWait = time.Duration(cr.GroupWait) } if cr.GroupInterval != nil { opts.GroupInterval = time.Duration(cr.GroupInterval) } if cr.RepeatInterval != nil { opts.RepeatInterval = time.Duration(cr.RepeatInterval) } // Build matchers. var matchers labels.Matchers // cr.Match will be deprecated. This for loop appends matchers. for ln, lv := range cr.Match { matcher, err := labels.NewMatcher(labels.MatchEqual, ln, lv) if err != nil { // This error must not happen because the config already validates the yaml. panic(err) } matchers = append(matchers, matcher) } // cr.MatchRE will be deprecated. This for loop appends regex matchers. for ln, lv := range cr.MatchRE { matcher, err := labels.NewMatcher(labels.MatchRegexp, ln, lv.String()) if err != nil { // This error must not happen because the config already validates the yaml. panic(err) } matchers = append(matchers, matcher) } // We append the new-style matchers. This can be simplified once the deprecated matcher syntax is removed. matchers = append(matchers, cr.Matchers...) sort.Sort(matchers) opts.MuteTimeIntervals = cr.MuteTimeIntervals opts.ActiveTimeIntervals = cr.ActiveTimeIntervals route := &Route{ parent: parent, RouteOpts: opts, Matchers: matchers, Continue: cr.Continue, } route.Routes = NewRoutes(cr.Routes, route) return route } // NewRoutes returns a slice of routes. func NewRoutes(croutes []config.Route, parent Route) []Route { res := []Route{} for _, cr := range croutes { res = append(res, NewRoute(cr, parent)) } return res } // Match does a depth-first left-to-right search through the route tree // and returns the matching routing nodes. func (r Route) Match(lset model.LabelSet) []Route { if !r.Matchers.Matches(lset) { return nil } var all []Route for _, cr := range r.Routes { matches := cr.Match(lset) all = append(all, matches...) if matches != nil && !cr.Continue { break } } // If no child nodes were matches, the current node itself is a match. if len(all) == 0 { all = append(all, r) } return all } // Key returns a key for the route. It does not uniquely identify the route in general. func (r Route) Key() string { b := strings.Builder{} if r.parent != nil { b.WriteString(r.parent.Key()) b.WriteRune('/') } b.WriteString(r.Matchers.String()) return b.String() } // ID returns a unique identifier for the route. func (r Route) ID() string { b := strings.Builder{} if r.parent != nil { b.WriteString(r.parent.ID()) b.WriteRune('/') } b.WriteString(r.Matchers.String()) if r.parent != nil { for i := range r.parent.Routes { if r == r.parent.Routes[i] { b.WriteRune('/') b.WriteString(strconv.Itoa(i)) break } } } return b.String() } // Walk traverses the route tree in depth-first order. func (r Route) Walk(visit func(Route)) { visit(r) for i := range r.Routes { r.Routes[i].Walk(visit) } } // RouteOpts holds various routing options necessary for processing alerts // that match a given route. type RouteOpts struct { // The identifier of the associated notification configuration. Receiver string // What labels to group alerts by for notifications. GroupBy map[model.LabelName]struct{} // Use all alert labels to group. GroupByAll bool // How long to wait to group matching alerts before sending // a notification. GroupWait time.Duration GroupInterval time.Duration RepeatInterval time.Duration // A list of time intervals for which the route is muted. MuteTimeIntervals []string // A list of time intervals for which the route is active. ActiveTimeIntervals []string } func (ro RouteOpts) String() string { var labels []model.LabelName for ln := range ro.GroupBy { labels = append(labels, ln) } return fmt.Sprintf("<RouteOpts send_to:%q group_by:%q group_by_all:%t timers:%q\|%q>", ro.Receiver, labels, ro.GroupByAll, ro.GroupWait, ro.GroupInterval) } // MarshalJSON returns a JSON representation of the routing options. func (ro *RouteOpts) MarshalJSON() ([]byte, error) { v := struct { Receiver string `json:"receiver"` GroupBy model.LabelNames `json:"groupBy"` GroupByAll bool `json:"groupByAll"` GroupWait time.Duration `json:"groupWait"` GroupInterval time.Duration `json:"groupInterval"` RepeatInterval time.Duration `json:"repeatInterval"` }{ Receiver: ro.Receiver, GroupByAll: ro.GroupByAll, GroupWait: ro.GroupWait, GroupInterval: ro.GroupInterval, RepeatInterval: ro.RepeatInterval, } for ln := range ro.GroupBy { v.GroupBy = append(v.GroupBy, ln) } return json.Marshal(&v) }	6.69 KB	2025-05-25 12:27
📄 route_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 dispatch import ( "reflect" "testing" "time" "github.com/prometheus/common/model" "github.com/stretchr/testify/require" "gopkg.in/yaml.v2" "github.com/prometheus/alertmanager/config" ) func TestRouteMatch(t testing.T) { in := ` receiver: 'notify-def' routes: - match: owner: 'team-A' receiver: 'notify-A' routes: - match: env: 'testing' receiver: 'notify-testing' group_by: [...] - match: env: "production" receiver: 'notify-productionA' group_wait: 1m continue: true - match_re: env: "produ." job: "." receiver: 'notify-productionB' group_wait: 30s group_interval: 5m repeat_interval: 1h group_by: ['job'] - match_re: owner: 'team-(B\|C)' group_by: ['foo', 'bar'] group_wait: 2m receiver: 'notify-BC' - match: group_by: 'role' group_by: ['role'] routes: - match: env: 'testing' receiver: 'notify-testing' routes: - match: wait: 'long' group_wait: 2m ` var ctree config.Route if err := yaml.UnmarshalStrict([]byte(in), &ctree); err != nil { t.Fatal(err) } var ( def = DefaultRouteOpts tree = NewRoute(&ctree, nil) ) lset := func(labels ...string) map[model.LabelName]struct{} { s := map[model.LabelName]struct{}{} for _, ls := range labels { s[model.LabelName(ls)] = struct{}{} } return s } tests := []struct { input model.LabelSet result []RouteOpts keys []string }{ { input: model.LabelSet{ "owner": "team-A", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "unset", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-C", }, result: []RouteOpts{ { Receiver: "notify-BC", GroupBy: lset("foo", "bar"), GroupByAll: false, GroupWait: 2 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=~\"^(?:team-(B\|C))$\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "testing", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset(), GroupByAll: true, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=\"team-A\"}/{env=\"testing\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "production", }, result: []RouteOpts{ { Receiver: "notify-productionA", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: 1 * time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, { Receiver: "notify-productionB", GroupBy: lset("job"), GroupByAll: false, GroupWait: 30 * time.Second, GroupInterval: 5 * time.Minute, RepeatInterval: 1 * time.Hour, }, }, keys: []string{ "{}/{owner=\"team-A\"}/{env=\"production\"}", "{}/{owner=\"team-A\"}/{env=~\"^(?:produ.)$\",job=~\"^(?:.)$\"}", }, }, { input: model.LabelSet{ "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-def", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", "wait": "long", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: 2 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}/{wait=\"long\"}"}, }, } for _, test := range tests { var matches []RouteOpts var keys []string for _, r := range tree.Match(test.input) { matches = append(matches, &r.RouteOpts) keys = append(keys, r.Key()) } if !reflect.DeepEqual(matches, test.result) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.result, matches) } if !reflect.DeepEqual(keys, test.keys) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.keys, keys) } } } func TestRouteWalk(t testing.T) { in := ` receiver: 'notify-def' routes: - match: owner: 'team-A' receiver: 'notify-A' routes: - match: env: 'testing' receiver: 'notify-testing' group_by: [...] - match: env: "production" receiver: 'notify-productionA' group_wait: 1m continue: true - match_re: env: "produ." job: "." receiver: 'notify-productionB' group_wait: 30s group_interval: 5m repeat_interval: 1h group_by: ['job'] - match_re: owner: 'team-(B\|C)' group_by: ['foo', 'bar'] group_wait: 2m receiver: 'notify-BC' - match: group_by: 'role' group_by: ['role'] routes: - match: env: 'testing' receiver: 'notify-testing' routes: - match: wait: 'long' group_wait: 2m ` var ctree config.Route if err := yaml.UnmarshalStrict([]byte(in), &ctree); err != nil { t.Fatal(err) } tree := NewRoute(&ctree, nil) expected := []string{ "notify-def", "notify-A", "notify-testing", "notify-productionA", "notify-productionB", "notify-BC", "notify-def", "notify-testing", "notify-testing", } var got []string tree.Walk(func(r Route) { got = append(got, r.RouteOpts.Receiver) }) if !reflect.DeepEqual(got, expected) { t.Errorf("\nexpected:\n%v\ngot:\n%v", expected, got) } } func TestInheritParentGroupByAll(t testing.T) { in := ` routes: - match: env: 'parent' group_by: ['...'] routes: - match: env: 'child1' - match: env: 'child2' group_by: ['foo'] ` var ctree config.Route if err := yaml.UnmarshalStrict([]byte(in), &ctree); err != nil { t.Fatal(err) } tree := NewRoute(&ctree, nil) parent := tree.Routes[0] child1 := parent.Routes[0] child2 := parent.Routes[1] require.True(t, parent.RouteOpts.GroupByAll) require.True(t, child1.RouteOpts.GroupByAll) require.False(t, child2.RouteOpts.GroupByAll) } func TestRouteMatchers(t testing.T) { in := ` receiver: 'notify-def' routes: - matchers: ['{owner="team-A"}', '{level!="critical"}'] receiver: 'notify-A' routes: - matchers: ['{env="testing"}', '{baz!~".quux"}'] receiver: 'notify-testing' group_by: [...] - matchers: ['{env="production"}'] receiver: 'notify-productionA' group_wait: 1m continue: true - matchers: [ env=~"produ.", job=~"."] receiver: 'notify-productionB' group_wait: 30s group_interval: 5m repeat_interval: 1h group_by: ['job'] - matchers: [owner=~"team-(B\|C)"] group_by: ['foo', 'bar'] group_wait: 2m receiver: 'notify-BC' - matchers: [group_by="role"] group_by: ['role'] routes: - matchers: ['{env="testing"}'] receiver: 'notify-testing' routes: - matchers: [wait="long"] group_wait: 2m ` var ctree config.Route if err := yaml.UnmarshalStrict([]byte(in), &ctree); err != nil { t.Fatal(err) } var ( def = DefaultRouteOpts tree = NewRoute(&ctree, nil) ) lset := func(labels ...string) map[model.LabelName]struct{} { s := map[model.LabelName]struct{}{} for _, ls := range labels { s[model.LabelName(ls)] = struct{}{} } return s } tests := []struct { input model.LabelSet result []RouteOpts keys []string }{ { input: model.LabelSet{ "owner": "team-A", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "unset", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-C", }, result: []RouteOpts{ { Receiver: "notify-BC", GroupBy: lset("foo", "bar"), GroupByAll: false, GroupWait: 2 * time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=~\"team-(B\|C)\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "testing", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset(), GroupByAll: true, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}/{baz!~\".quux\",env=\"testing\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "production", }, result: []RouteOpts{ { Receiver: "notify-productionA", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: 1 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, { Receiver: "notify-productionB", GroupBy: lset("job"), GroupByAll: false, GroupWait: 30 * time.Second, GroupInterval: 5 * time.Minute, RepeatInterval: 1 * time.Hour, }, }, keys: []string{ "{}/{level!=\"critical\",owner=\"team-A\"}/{env=\"production\"}", "{}/{level!=\"critical\",owner=\"team-A\"}/{env=~\"produ.\",job=~\".\"}", }, }, { input: model.LabelSet{ "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-def", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", "wait": "long", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: 2 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}/{wait=\"long\"}"}, }, } for _, test := range tests { var matches []RouteOpts var keys []string for _, r := range tree.Match(test.input) { matches = append(matches, &r.RouteOpts) keys = append(keys, r.Key()) } if !reflect.DeepEqual(matches, test.result) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.result, matches) } if !reflect.DeepEqual(keys, test.keys) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.keys, keys) } } } func TestRouteMatchersAndMatch(t testing.T) { in := ` receiver: 'notify-def' routes: - matchers: ['{owner="team-A"}', '{level!="critical"}'] receiver: 'notify-A' routes: - matchers: ['{env="testing"}', '{baz!~".quux"}'] receiver: 'notify-testing' group_by: [...] - match: env: "production" receiver: 'notify-productionA' group_wait: 1m continue: true - matchers: [ env=~"produ.", job=~"."] receiver: 'notify-productionB' group_wait: 30s group_interval: 5m repeat_interval: 1h group_by: ['job'] - match_re: owner: 'team-(B\|C)' group_by: ['foo', 'bar'] group_wait: 2m receiver: 'notify-BC' - matchers: [group_by="role"] group_by: ['role'] routes: - matchers: ['{env="testing"}'] receiver: 'notify-testing' routes: - matchers: [wait="long"] group_wait: 2m ` var ctree config.Route if err := yaml.UnmarshalStrict([]byte(in), &ctree); err != nil { t.Fatal(err) } var ( def = DefaultRouteOpts tree = NewRoute(&ctree, nil) ) lset := func(labels ...string) map[model.LabelName]struct{} { s := map[model.LabelName]struct{}{} for _, ls := range labels { s[model.LabelName(ls)] = struct{}{} } return s } tests := []struct { input model.LabelSet result []RouteOpts keys []string }{ { input: model.LabelSet{ "owner": "team-A", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "unset", }, result: []RouteOpts{ { Receiver: "notify-A", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}"}, }, { input: model.LabelSet{ "owner": "team-C", }, result: []RouteOpts{ { Receiver: "notify-BC", GroupBy: lset("foo", "bar"), GroupByAll: false, GroupWait: 2 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{owner=~\"^(?:team-(B\|C))$\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "testing", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset(), GroupByAll: true, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{level!=\"critical\",owner=\"team-A\"}/{baz!~\".quux\",env=\"testing\"}"}, }, { input: model.LabelSet{ "owner": "team-A", "env": "production", }, result: []RouteOpts{ { Receiver: "notify-productionA", GroupBy: def.GroupBy, GroupByAll: false, GroupWait: 1 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, { Receiver: "notify-productionB", GroupBy: lset("job"), GroupByAll: false, GroupWait: 30 * time.Second, GroupInterval: 5 * time.Minute, RepeatInterval: 1 * time.Hour, }, }, keys: []string{ "{}/{level!=\"critical\",owner=\"team-A\"}/{env=\"production\"}", "{}/{level!=\"critical\",owner=\"team-A\"}/{env=~\"produ.\",job=~\".\"}", }, }, { input: model.LabelSet{ "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-def", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: def.GroupWait, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}"}, }, { input: model.LabelSet{ "env": "testing", "group_by": "role", "wait": "long", }, result: []RouteOpts{ { Receiver: "notify-testing", GroupBy: lset("role"), GroupByAll: false, GroupWait: 2 time.Minute, GroupInterval: def.GroupInterval, RepeatInterval: def.RepeatInterval, }, }, keys: []string{"{}/{group_by=\"role\"}/{env=\"testing\"}/{wait=\"long\"}"}, }, } for _, test := range tests { var matches []RouteOpts var keys []string for _, r := range tree.Match(test.input) { matches = append(matches, &r.RouteOpts) keys = append(keys, r.Key()) } if !reflect.DeepEqual(matches, test.result) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.result, matches) } if !reflect.DeepEqual(keys, test.keys) { t.Errorf("\nexpected:\n%v\ngot:\n%v", test.keys, keys) } } } func TestRouteID(t testing.T) { in := ` receiver: default routes: - continue: true matchers: - foo=bar receiver: test1 routes: - matchers: - bar=baz - continue: true matchers: - foo=bar receiver: test1 routes: - matchers: - bar=baz - continue: true matchers: - foo=bar receiver: test2 routes: - matchers: - bar=baz - continue: true matchers: - bar=baz receiver: test3 routes: - matchers: - baz=qux - matchers: - qux=corge - continue: true matchers: - qux=~"[a-zA-Z0-9]+" - continue: true matchers: - corge!~"[0-9]+" ` cr := config.Route{} require.NoError(t, yaml.Unmarshal([]byte(in), &cr)) r := NewRoute(&cr, nil) expected := []string{ "{}", "{}/{foo=\"bar\"}/0", "{}/{foo=\"bar\"}/0/{bar=\"baz\"}/0", "{}/{foo=\"bar\"}/1", "{}/{foo=\"bar\"}/1/{bar=\"baz\"}/0", "{}/{foo=\"bar\"}/2", "{}/{foo=\"bar\"}/2/{bar=\"baz\"}/0", "{}/{bar=\"baz\"}/3", "{}/{bar=\"baz\"}/3/{baz=\"qux\"}/0", "{}/{bar=\"baz\"}/3/{qux=\"corge\"}/1", "{}/{qux=~\"[a-zA-Z0-9]+\"}/4", "{}/{corge!~\"[0-9]+\"}/5", } var actual []string r.Walk(func(r *Route) { actual = append(actual, r.ID()) }) require.ElementsMatch(t, actual, expected) }	19.44 KB	2025-05-25 12:27

Terminal