Adjust alert subscription behavior.

Rename IterActive to Subscribe. The returned iterator now
returns all alerts with pending notifications which may also include
resolved alerts.

dispatch.go

@@ -60,19 +60,23 @@ func (d *Dispatcher) Run() {
 
 	d.ctx, d.cancel = context.WithCancel(context.Background())
 
-	updates := d.alerts.IterActive()
-	defer updates.Close()
-
-	d.run(updates.Next())
+	d.run(d.alerts.Subscribe())
 }
 
-func (d *Dispatcher) run(updates <-chan *types.Alert) {
+func (d *Dispatcher) run(it provider.AlertIterator) {
 	cleanup := time.NewTicker(15 * time.Second)
 	defer cleanup.Stop()
 
+	defer it.Close()
+
 	for {
 		select {
-		case alert := <-updates:
+		case alert := <-it.Next():
+			// Log errors but keep trying
+			if err := it.Err(); err != nil {
+				log.Errorf("Error on alert update: %s", err)
+				continue
+			}
 			d.mtx.RLock()
 			routes := d.routes.Match(alert.Labels)
 			d.mtx.RUnlock()

provider/mem.go

@@ -65,15 +65,27 @@ func NewMemAlerts(data *MemData) *MemAlerts {
 	}
 }
 
-func (a *MemAlerts) IterActive() AlertIterator {
+func (a *MemAlerts) Subscribe() AlertIterator {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 	a.data.mtx.Lock()
 	defer a.data.mtx.Unlock()
 
+	// Get fingerprints for all alerts that have pending notifications.
+	fps := map[model.Fingerprint]struct{}{}
+	for _, ns := range a.data.notifies {
+		for fp, notify := range ns {
+			if !notify.Delivered {
+				fps[fp] = struct{}{}
+			}
+		}
+	}
+
+	// All alerts that have pending notifications are part of the
+	// new scubscription.
 	var alerts []*types.Alert
 	for _, a := range a.data.alerts {
-		if !a.Resolved() {
+		if _, ok := fps[a.Fingerprint()]; ok {
 			alerts = append(alerts, a)
 		}
 	}

provider/provider.go

@@ -32,9 +32,10 @@ type AlertIterator interface {
 
 // Alerts gives access to a set of alerts.
 type Alerts interface {
-	// IterActive returns an iterator over active alerts from the
-	// beginning of time. They are not guaranteed to be in chronological order.
-	IterActive() AlertIterator
+	// 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.
+	Subscribe() AlertIterator
 	// All returns a list of all existing alerts.
 	// TODO(fabxc): this is not a scalable solution
 	All() ([]*types.Alert, error)