Add tests for AlertLifeCycleObserver

Signed-off-by: Emmanuel Lodovice <[email protected]>

alertobserver/testing.go

@@ -0,0 +1,69 @@
+// Copyright 2023 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 alertobserver
+
+import (
+	"context"
+
+	"github.com/prometheus/alertmanager/types"
+)
+
+type FakeAlertLifeCycleObserver struct {
+	ReceivedAlerts      []*types.Alert
+	RejectedAlerts      []*types.Alert
+	SentAlerts          []*types.Alert
+	FailedAlerts        []*types.Alert
+	AggrGroupAlerts     []*types.Alert
+	PipelineAlerts      []*types.Alert
+	PipelineStageAlerts map[string][]*types.Alert
+}
+
+func (o *FakeAlertLifeCycleObserver) Received(alerts ...*types.Alert) {
+	o.ReceivedAlerts = append(o.ReceivedAlerts, alerts...)
+}
+
+func (o *FakeAlertLifeCycleObserver) Rejected(reason string, alerts ...*types.Alert) {
+	o.RejectedAlerts = append(o.RejectedAlerts, alerts...)
+}
+
+func (o *FakeAlertLifeCycleObserver) AddedAggrGroup(groupKey string, alert *types.Alert) {
+	o.AggrGroupAlerts = append(o.AggrGroupAlerts, alert)
+}
+
+func (o *FakeAlertLifeCycleObserver) PipelineStart(ctx context.Context, alerts ...*types.Alert) {
+	o.PipelineAlerts = append(o.PipelineAlerts, alerts...)
+}
+
+func (o *FakeAlertLifeCycleObserver) Sent(ctx context.Context, integration string, alerts ...*types.Alert) {
+	o.SentAlerts = append(o.SentAlerts, alerts...)
+}
+
+func (o *FakeAlertLifeCycleObserver) SendFailed(
+	ctx context.Context,
+	integration string,
+	reason string,
+	alerts ...*types.Alert,
+) {
+	o.FailedAlerts = append(o.FailedAlerts, alerts...)
+}
+
+func (o *FakeAlertLifeCycleObserver) PipelinePassStage(ctx context.Context, stageName string, alerts ...*types.Alert) {
+	o.PipelineStageAlerts[stageName] = append(o.PipelineStageAlerts[stageName], alerts...)
+}
+
+func NewFakeAlertLifeCycleObserver() *FakeAlertLifeCycleObserver {
+	return &FakeAlertLifeCycleObserver{
+		PipelineStageAlerts: map[string][]*types.Alert{},
+	}
+}

api/v1/api.go

@@ -452,7 +452,7 @@ func (api *API) insertAlerts(w http.ResponseWriter, r *http.Request, alerts ...*
 			validationErrs.Add(err)
 			api.m.Invalid().Inc()
 			if api.alertLCObserver != nil {
-				api.alertLCObserver.Rejected("Invalid", a)
+				api.alertLCObserver.Rejected(err.Error(), a)
 			}
 			continue
 		}
@@ -464,7 +464,7 @@ func (api *API) insertAlerts(w http.ResponseWriter, r *http.Request, alerts ...*
 			err: err,
 		}, nil)
 		if api.alertLCObserver != nil {
-			api.alertLCObserver.Rejected("Failed to create", validAlerts...)
+			api.alertLCObserver.Rejected(err.Error(), validAlerts...)
 		}
 		return
 	}

api/v1/api_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 
+	"github.com/prometheus/alertmanager/alertobserver"
 	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/dispatch"
 	"github.com/prometheus/alertmanager/pkg/labels"
@@ -153,6 +154,74 @@ func TestAddAlerts(t *testing.T) {
 		body, _ := io.ReadAll(res.Body)
 
 		require.Equal(t, tc.code, w.Code, fmt.Sprintf("test case: %d, StartsAt %v, EndsAt %v, Response: %s", i, tc.start, tc.end, string(body)))
+
+		observer := alertobserver.NewFakeAlertLifeCycleObserver()
+		api.alertLCObserver = observer
+		r, err = http.NewRequest("POST", "/api/v1/alerts", bytes.NewReader(b))
+		w = httptest.NewRecorder()
+		if err != nil {
+			t.Errorf("Unexpected error %v", err)
+		}
+		api.addAlerts(w, r)
+		if tc.code == 200 {
+			require.Equal(t, observer.ReceivedAlerts[0].Fingerprint(), alerts[0].Fingerprint())
+		} else {
+			require.Equal(t, observer.RejectedAlerts[0].Fingerprint(), alerts[0].Fingerprint())
+		}
+	}
+}
+
+func TestAddAlertsWithAlertLCObserver(t *testing.T) {
+	now := func(offset int) time.Time {
+		return time.Now().Add(time.Duration(offset) * time.Second)
+	}
+
+	for i, tc := range []struct {
+		start, end time.Time
+		err        bool
+		code       int
+	}{
+		{time.Time{}, time.Time{}, false, 200},
+		{now(1), now(0), false, 400},
+		{now(0), time.Time{}, true, 500},
+	} {
+		alerts := []model.Alert{{
+			StartsAt:    tc.start,
+			EndsAt:      tc.end,
+			Labels:      model.LabelSet{"label1": "test1"},
+			Annotations: model.LabelSet{"annotation1": "some text"},
+		}}
+		b, err := json.Marshal(&alerts)
+		if err != nil {
+			t.Errorf("Unexpected error %v", err)
+		}
+
+		alertsProvider := newFakeAlerts([]*types.Alert{}, tc.err)
+		observer := alertobserver.NewFakeAlertLifeCycleObserver()
+		api := New(alertsProvider, nil, newGetAlertStatus(alertsProvider), nil, nil, nil, observer)
+		defaultGlobalConfig := config.DefaultGlobalConfig()
+		route := config.Route{}
+		api.Update(&config.Config{
+			Global: &defaultGlobalConfig,
+			Route:  &route,
+		})
+
+		r, err := http.NewRequest("POST", "/api/v1/alerts", bytes.NewReader(b))
+		w := httptest.NewRecorder()
+		if err != nil {
+			t.Errorf("Unexpected error %v", err)
+		}
+
+		api.addAlerts(w, r)
+		res := w.Result()
+		body, _ := io.ReadAll(res.Body)
+
+		require.Equal(t, tc.code, w.Code, fmt.Sprintf("test case: %d, StartsAt %v, EndsAt %v, Response: %s", i, tc.start, tc.end, string(body)))
+		if tc.code == 200 {
+			require.Equal(t, observer.ReceivedAlerts[0].Fingerprint(), alerts[0].Fingerprint())
+		} else {
+			require.Equal(t, observer.RejectedAlerts[0].Fingerprint(), alerts[0].Fingerprint())
+		}
 	}
 }
 

api/v2/api.go

@@ -355,7 +355,7 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 			validationErrs.Add(err)
 			api.m.Invalid().Inc()
 			if api.alertLCObserver != nil {
-				api.alertLCObserver.Rejected("Invalid", a)
+				api.alertLCObserver.Rejected(err.Error(), a)
 			}
 			continue
 		}
@@ -364,7 +364,7 @@ func (api *API) postAlertsHandler(params alert_ops.PostAlertsParams) middleware.
 	if err := api.alerts.Put(validAlerts...); err != nil {
 		level.Error(logger).Log("msg", "Failed to create alerts", "err", err)
 		if api.alertLCObserver != nil {
-			api.alertLCObserver.Rejected("Failed to create", validAlerts...)
+			api.alertLCObserver.Rejected(err.Error(), validAlerts...)
 		}
 		return alert_ops.NewPostAlertsInternalServerError().WithPayload(err.Error())
 	}

api/v2/api_test.go

@@ -28,7 +28,10 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 
+	"github.com/prometheus/alertmanager/alertobserver"
+	"github.com/prometheus/alertmanager/api/metrics"
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
+	alert_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/alert"
 	general_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/general"
 	receiver_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/receiver"
 	silence_ops "github.com/prometheus/alertmanager/api/v2/restapi/operations/silence"
@@ -510,3 +513,64 @@ receivers:
 		require.Equal(t, tc.body, string(body))
 	}
 }
+
+func TestPostAlertHandler(t *testing.T) {
+	now := time.Now()
+	for i, tc := range []struct {
+		start, end time.Time
+		err        bool
+		code       int
+	}{
+		{time.Time{}, time.Time{}, false, 200},
+		{now, time.Time{}, false, 200},
+		{time.Time{}, now.Add(time.Duration(-1) * time.Second), false, 200},
+		{time.Time{}, now, false, 200},
+		{time.Time{}, now.Add(time.Duration(1) * time.Second), false, 200},
+		{now.Add(time.Duration(-2) * time.Second), now.Add(time.Duration(-1) * time.Second), false, 200},
+		{now.Add(time.Duration(1) * time.Second), now.Add(time.Duration(2) * time.Second), false, 200},
+		{now.Add(time.Duration(1) * time.Second), now, false, 400},
+	} {
+		alerts, alertsBytes := createAlert(t, tc.start, tc.end)
+		api := API{
+			uptime: time.Now(),
+			alerts: newFakeAlerts([]*types.Alert{}),
+			logger: log.NewNopLogger(),
+			m:      metrics.NewAlerts("v2", nil),
+		}
+		api.Update(&config.Config{
+			Global: &config.GlobalConfig{
+				ResolveTimeout: model.Duration(5),
+			},
+			Route: &config.Route{},
+		}, nil)
+
+		r, err := http.NewRequest("POST", "/api/v2/alerts", bytes.NewReader(alertsBytes))
+		require.NoError(t, err)
+
+		w := httptest.NewRecorder()
+		p := runtime.TextProducer()
+		responder := api.postAlertsHandler(alert_ops.PostAlertsParams{
+			HTTPRequest: r,
+			Alerts:      alerts,
+		})
+		responder.WriteResponse(w, p)
+		body, _ := io.ReadAll(w.Result().Body)
+
+		require.Equal(t, tc.code, w.Code, fmt.Sprintf("test case: %d, response: %s", i, string(body)))
+
+		observer := alertobserver.NewFakeAlertLifeCycleObserver()
+		api.alertLCObserver = observer
+		r, err = http.NewRequest("POST", "/api/v2/alerts", bytes.NewReader(alertsBytes))
+		require.NoError(t, err)
+		api.postAlertsHandler(alert_ops.PostAlertsParams{
+			HTTPRequest: r,
+			Alerts:      alerts,
+		})
+		amAlert := OpenAPIAlertsToAlerts(alerts)
+		if tc.code == 200 {
+			require.Equal(t, observer.ReceivedAlerts[0].Fingerprint(), amAlert[0].Fingerprint())
+		} else {
+			require.Equal(t, observer.RejectedAlerts[0].Fingerprint(), amAlert[0].Fingerprint())
+		}
+	}
+}

api/v2/testing.go

@@ -19,11 +19,14 @@ import (
 	"time"
 
 	"github.com/go-openapi/strfmt"
+	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 
 	open_api_models "github.com/prometheus/alertmanager/api/v2/models"
 	"github.com/prometheus/alertmanager/pkg/labels"
+	"github.com/prometheus/alertmanager/provider"
 	"github.com/prometheus/alertmanager/silence/silencepb"
+	"github.com/prometheus/alertmanager/types"
 )
 
 func createSilence(t *testing.T, ID, creator string, start, ends time.Time) (open_api_models.PostableSilence, []byte) {
@@ -68,3 +71,48 @@ func createLabelMatcher(t *testing.T, name, value string, matchType labels.Match
 	matcher, _ := labels.NewMatcher(matchType, name, value)
 	return matcher
 }
+
+func createAlert(t *testing.T, start, ends time.Time) (open_api_models.PostableAlerts, []byte) {
+	startsAt := strfmt.DateTime(start)
+	endsAt := strfmt.DateTime(ends)
+
+	alert := open_api_models.PostableAlert{
+		StartsAt:    startsAt,
+		EndsAt:      endsAt,
+		Annotations: open_api_models.LabelSet{"annotation1": "some text"},
+		Alert: open_api_models.Alert{
+			Labels:       open_api_models.LabelSet{"label1": "test1"},
+			GeneratorURL: "http://localhost:3000",
+		},
+	}
+	alerts := open_api_models.PostableAlerts{}
+	alerts = append(alerts, &alert)
+	b, err := json.Marshal(alerts)
+	require.NoError(t, err)
+	return alerts, b
+}
+
+type fakeAlerts struct {
+	fps    map[model.Fingerprint]int
+	alerts []*types.Alert
+	err    error
+}
+
+func newFakeAlerts(alerts []*types.Alert) *fakeAlerts {
+	fps := make(map[model.Fingerprint]int)
+	for i, a := range alerts {
+		fps[a.Fingerprint()] = i
+	}
+	f := &fakeAlerts{
+		alerts: alerts,
+		fps:    fps,
+	}
+	return f
+}
+
+func (f *fakeAlerts) Subscribe() provider.AlertIterator           { return nil }
+func (f *fakeAlerts) Get(model.Fingerprint) (*types.Alert, error) { return nil, nil }
+func (f *fakeAlerts) GetPending() provider.AlertIterator          { return nil }
+func (f *fakeAlerts) Put(alerts ...*types.Alert) error {
+	return f.err
+}

dispatch/dispatch_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 
+	"github.com/prometheus/alertmanager/alertobserver"
 	"github.com/prometheus/alertmanager/config"
 	"github.com/prometheus/alertmanager/notify"
 	"github.com/prometheus/alertmanager/provider/mem"
@@ -568,6 +569,56 @@ route:
 	require.Len(t, alertGroups, 6)
 }
 
+func TestGroupsAlertLCObserver(t *testing.T) {
+	confData := `receivers:
+- name: 'testing'
+
+route:
+  group_by: ['alertname']
+  group_wait: 10ms
+  group_interval: 10ms
+  receiver: 'testing'`
+	conf, err := config.Load(confData)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	logger := log.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)}
+	m := NewDispatcherMetrics(true, prometheus.NewRegistry())
+	observer := alertobserver.NewFakeAlertLifeCycleObserver()
+	dispatcher := NewDispatcher(alerts, route, recorder, marker, timeout, nil, logger, m, observer)
+	go dispatcher.Run()
+	defer dispatcher.Stop()
+
+	// Create alerts. the dispatcher will automatically create the groups.
+	inputAlerts := []*types.Alert{
+		newAlert(model.LabelSet{"alertname": "OtherAlert", "cluster": "cc", "service": "dd"}),
+	}
+	err = alerts.Put(inputAlerts...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Let alerts get processed.
+	for i := 0; len(recorder.Alerts()) != 1 && i < 10; i++ {
+		time.Sleep(200 * time.Millisecond)
+	}
+	require.Equal(t, 1, len(recorder.Alerts()))
+	require.Equal(t, inputAlerts[0].Fingerprint(), observer.AggrGroupAlerts[0].Fingerprint())
+	o, ok := notify.AlertLCObserver(dispatcher.ctx)
+	require.True(t, ok)
+	require.Equal(t, observer, o)
+}
+
 type recordStage struct {
 	mtx    sync.RWMutex
 	alerts map[string]map[model.Fingerprint]*types.Alert