// Copyright 2016 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 mem import ( "context" "sync" "time" "github.com/go-kit/kit/log" "github.com/go-kit/kit/log/level" "github.com/prometheus/common/model" "github.com/prometheus/alertmanager/provider" "github.com/prometheus/alertmanager/store" "github.com/prometheus/alertmanager/types" ) const alertChannelLength = 200 // Alerts gives access to a set of alerts. All methods are goroutine-safe. type Alerts struct { alerts *store.Alerts cancel context.CancelFunc mtx sync.Mutex listeners map[int]listeningAlerts next int logger log.Logger } type listeningAlerts struct { alerts chan *types.Alert done chan struct{} } // NewAlerts returns a new alert provider. func NewAlerts(ctx context.Context, m types.Marker, intervalGC time.Duration, l log.Logger) (*Alerts, error) { ctx, cancel := context.WithCancel(ctx) a := &Alerts{ alerts: store.NewAlerts(intervalGC), cancel: cancel, listeners: map[int]listeningAlerts{}, next: 0, logger: log.With(l, "component", "provider"), } a.alerts.SetGCCallback(func(alerts []*types.Alert) { for _, alert := range alerts { // As we don't persist alerts, we no longer consider them after // they are resolved. Alerts waiting for resolved notifications are // held in memory in aggregation groups redundantly. m.Delete(alert.Fingerprint()) } a.mtx.Lock() for i, l := range a.listeners { select { case <-l.done: delete(a.listeners, i) close(l.alerts) default: // listener is not closed yet, hence proceed. } } a.mtx.Unlock() }) a.alerts.Run(ctx) return a, nil } // Close the alert provider. func (a *Alerts) Close() { if a.cancel != nil { a.cancel() } } func max(a, b int) int { if a > b { return a } return b } // Subscribe returns an iterator over active alerts that have not been // resolved and successfully notified about. // They are not guaranteed to be in chronological order. func (a *Alerts) Subscribe() provider.AlertIterator { var ( ch = make(chan *types.Alert, max(a.alerts.Count(), alertChannelLength)) done = make(chan struct{}) ) for a := range a.alerts.List() { ch <- a } a.mtx.Lock() i := a.next a.next++ a.listeners[i] = listeningAlerts{alerts: ch, done: done} a.mtx.Unlock() return provider.NewAlertIterator(ch, done, nil) } // GetPending returns an iterator over all alerts that have // pending notifications. func (a *Alerts) GetPending() provider.AlertIterator { var ( ch = make(chan *types.Alert, alertChannelLength) done = make(chan struct{}) ) go func() { defer close(ch) for a := range a.alerts.List() { select { case ch <- a: case <-done: return } } }() return provider.NewAlertIterator(ch, done, nil) } // Get returns the alert for a given fingerprint. func (a *Alerts) Get(fp model.Fingerprint) (*types.Alert, error) { return a.alerts.Get(fp) } // Put adds the given alert to the set. func (a *Alerts) Put(alerts ...*types.Alert) error { for _, alert := range alerts { fp := alert.Fingerprint() // Check that there's an alert existing within the store before // trying to merge. if old, err := a.alerts.Get(fp); err == nil { // Merge alerts if there is an overlap in activity range. if (alert.EndsAt.After(old.StartsAt) && alert.EndsAt.Before(old.EndsAt)) || (alert.StartsAt.After(old.StartsAt) && alert.StartsAt.Before(old.EndsAt)) { alert = old.Merge(alert) } } if err := a.alerts.Set(alert); err != nil { level.Error(a.logger).Log("msg", "error on set alert", "err", err) continue } a.mtx.Lock() for _, l := range a.listeners { select { case l.alerts <- alert: case <-l.done: } } a.mtx.Unlock() } return nil }