Implement quorum reads and merging for /v2/silences and /v2/silence/id

integration/alertmanager_test.go

@@ -386,29 +386,58 @@ func TestAlertmanagerSharding(t *testing.T) {
 				assert.Equal(t, s3, ids[id3].Status.State)
 			}
 
-			// Endpoint: GET /silences
+			// Endpoint: GET /v1/silences
 			{
 				for _, c := range clients {
-					list, err := c.GetSilences(context.Background())
+					list, err := c.GetSilencesV1(context.Background())
 					require.NoError(t, err)
 					assertSilences(list, types.SilenceStateActive, types.SilenceStateActive, types.SilenceStateActive)
 				}
 			}
 
-			// Endpoint: GET /silence/{id}
+			// Endpoint: GET /v2/silences
 			{
 				for _, c := range clients {
-					sil1, err := c.GetSilence(context.Background(), id1)
+					list, err := c.GetSilencesV2(context.Background())
+					require.NoError(t, err)
+					assertSilences(list, types.SilenceStateActive, types.SilenceStateActive, types.SilenceStateActive)
+				}
+			}
+
+			// Endpoint: GET /v1/silence/{id}
+			{
+				for _, c := range clients {
+					sil1, err := c.GetSilenceV1(context.Background(), id1)
+					require.NoError(t, err)
+					assert.Equal(t, comment(1), sil1.Comment)
+					assert.Equal(t, types.SilenceStateActive, sil1.Status.State)
+
+					sil2, err := c.GetSilenceV1(context.Background(), id2)
+					require.NoError(t, err)
+					assert.Equal(t, comment(2), sil2.Comment)
+					assert.Equal(t, types.SilenceStateActive, sil2.Status.State)
+
+					sil3, err := c.GetSilenceV1(context.Background(), id3)
+					require.NoError(t, err)
+					assert.Equal(t, comment(3), sil3.Comment)
+					assert.Equal(t, types.SilenceStateActive, sil3.Status.State)
+				}
+			}
+
+			// Endpoint: GET /v2/silence/{id}
+			{
+				for _, c := range clients {
+					sil1, err := c.GetSilenceV2(context.Background(), id1)
 					require.NoError(t, err)
 					assert.Equal(t, comment(1), sil1.Comment)
 					assert.Equal(t, types.SilenceStateActive, sil1.Status.State)
 
-					sil2, err := c.GetSilence(context.Background(), id2)
+					sil2, err := c.GetSilenceV2(context.Background(), id2)
 					require.NoError(t, err)
 					assert.Equal(t, comment(2), sil2.Comment)
 					assert.Equal(t, types.SilenceStateActive, sil2.Status.State)
 
-					sil3, err := c.GetSilence(context.Background(), id3)
+					sil3, err := c.GetSilenceV2(context.Background(), id3)
 					require.NoError(t, err)
 					assert.Equal(t, comment(3), sil3.Comment)
 					assert.Equal(t, types.SilenceStateActive, sil3.Status.State)
@@ -445,7 +474,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 				require.NoError(t, waitForSilences("expired", 1*testCfg.replicationFactor))
 
 				for _, c := range clients {
-					list, err := c.GetSilences(context.Background())
+					list, err := c.GetSilencesV2(context.Background())
 					require.NoError(t, err)
 					assertSilences(list, types.SilenceStateActive, types.SilenceStateExpired, types.SilenceStateActive)
 				}
@@ -455,7 +484,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 				require.NoError(t, waitForSilences("expired", 2*testCfg.replicationFactor))
 
 				for _, c := range clients {
-					list, err := c.GetSilences(context.Background())
+					list, err := c.GetSilencesV2(context.Background())
 					require.NoError(t, err)
 					assertSilences(list, types.SilenceStateActive, types.SilenceStateExpired, types.SilenceStateExpired)
 				}
@@ -465,7 +494,7 @@ func TestAlertmanagerSharding(t *testing.T) {
 				require.NoError(t, waitForSilences("expired", 3*testCfg.replicationFactor))
 
 				for _, c := range clients {
-					list, err := c.GetSilences(context.Background())
+					list, err := c.GetSilencesV2(context.Background())
 					require.NoError(t, err)
 					assertSilences(list, types.SilenceStateExpired, types.SilenceStateExpired, types.SilenceStateExpired)
 				}

integration/e2ecortex/client.go

@@ -651,7 +651,7 @@ func (c *Client) CreateSilence(ctx context.Context, silence types.Silence) (stri
 	return decoded.Data.SilenceID, nil
 }
 
-func (c *Client) GetSilences(ctx context.Context) ([]types.Silence, error) {
+func (c *Client) GetSilencesV1(ctx context.Context) ([]types.Silence, error) {
 	u := c.alertmanagerClient.URL("api/prom/api/v1/silences", nil)
 
 	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
@@ -689,7 +689,36 @@ func (c *Client) GetSilences(ctx context.Context) ([]types.Silence, error) {
 	return decoded.Data, nil
 }
 
-func (c *Client) GetSilence(ctx context.Context, id string) (types.Silence, error) {
+func (c *Client) GetSilencesV2(ctx context.Context) ([]types.Silence, error) {
+	u := c.alertmanagerClient.URL("api/prom/api/v2/silences", nil)
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("error creating request: %v", err)
+	}
+
+	resp, body, err := c.alertmanagerClient.Do(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, ErrNotFound
+	}
+
+	if resp.StatusCode/100 != 2 {
+		return nil, fmt.Errorf("getting silences failed with status %d and error %v", resp.StatusCode, string(body))
+	}
+
+	decoded := []types.Silence{}
+	if err := json.Unmarshal(body, &decoded); err != nil {
+		return nil, err
+	}
+
+	return decoded, nil
+}
+
+func (c *Client) GetSilenceV1(ctx context.Context, id string) (types.Silence, error) {
 	u := c.alertmanagerClient.URL(fmt.Sprintf("api/prom/api/v1/silence/%s", url.PathEscape(id)), nil)
 
 	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
@@ -727,6 +756,35 @@ func (c *Client) GetSilence(ctx context.Context, id string) (types.Silence, erro
 	return decoded.Data, nil
 }
 
+func (c *Client) GetSilenceV2(ctx context.Context, id string) (types.Silence, error) {
+	u := c.alertmanagerClient.URL(fmt.Sprintf("api/prom/api/v2/silence/%s", url.PathEscape(id)), nil)
+
+	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
+	if err != nil {
+		return types.Silence{}, fmt.Errorf("error creating request: %v", err)
+	}
+
+	resp, body, err := c.alertmanagerClient.Do(ctx, req)
+	if err != nil {
+		return types.Silence{}, err
+	}
+
+	if resp.StatusCode == http.StatusNotFound {
+		return types.Silence{}, ErrNotFound
+	}
+
+	if resp.StatusCode/100 != 2 {
+		return types.Silence{}, fmt.Errorf("getting silence failed with status %d and error %v", resp.StatusCode, string(body))
+	}
+
+	decoded := types.Silence{}
+	if err := json.Unmarshal(body, &decoded); err != nil {
+		return types.Silence{}, err
+	}
+
+	return decoded, nil
+}
+
 func (c *Client) DeleteSilence(ctx context.Context, id string) error {
 	u := c.alertmanagerClient.URL(fmt.Sprintf("api/prom/api/v1/silence/%s", url.PathEscape(id)), nil)
 

pkg/alertmanager/distributor.go

@@ -94,12 +94,18 @@ func (d *Distributor) isQuorumReadPath(p string) (bool, merger.Merger) {
 	if strings.HasSuffix(p, "/v2/alerts/groups") {
 		return true, merger.V2AlertGroups{}
 	}
+	if strings.HasSuffix(p, "/v2/silences") {
+		return true, merger.V2Silences{}
+	}
+	if strings.HasSuffix(path.Dir(p), "/v2/silence") {
+		return true, merger.V2SilenceID{}
+	}
 	return false, nil
 }
 
 func (d *Distributor) isUnaryReadPath(p string) bool {
-	return strings.HasSuffix(p, "/silences") ||
-		strings.HasSuffix(path.Dir(p), "/silence") ||
+	return strings.HasSuffix(p, "/v1/silences") ||
+		strings.HasSuffix(path.Dir(p), "/v1/silence") ||
 		strings.HasSuffix(p, "/status") ||
 		strings.HasSuffix(p, "/receivers")
 }

pkg/alertmanager/distributor_test.go

@@ -119,14 +119,24 @@ func TestDistributor_DistributeRequest(t *testing.T) {
 			headersNotPreserved: true,
 			route:               "/alerts/groups",
 		}, {
-			name:               "Read /silences is sent to only 1 AM",
+			name:               "Read /v1/silences is sent to only 1 AM",
 			numAM:              5,
 			numHappyAM:         5,
 			replicationFactor:  3,
 			isRead:             true,
 			expStatusCode:      http.StatusOK,
 			expectedTotalCalls: 1,
-			route:              "/silences",
+			route:              "/v1/silences",
+		}, {
+			name:               "Read /v2/silences is sent to 3 AMs",
+			numAM:              5,
+			numHappyAM:         5,
+			replicationFactor:  3,
+			isRead:             true,
+			expStatusCode:      http.StatusOK,
+			expectedTotalCalls: 3,
+			route:              "/v2/silences",
+			responseBody:       []byte(`[]`),
 		}, {
 			name:               "Write /silences is sent to only 1 AM",
 			numAM:              5,
@@ -136,15 +146,26 @@ func TestDistributor_DistributeRequest(t *testing.T) {
 			expectedTotalCalls: 1,
 			route:              "/silences",
 		}, {
-			name:               "Read /silence/id is sent to only 1 AM",
+			name:               "Read /v1/silence/id is sent to only 1 AM",
 			numAM:              5,
 			numHappyAM:         5,
 			replicationFactor:  3,
 			isRead:             true,
 			expStatusCode:      http.StatusOK,
 			expectedTotalCalls: 1,
-			route:              "/silence/id",
+			route:              "/v1/silence/id",
 		}, {
+			name:               "Read /v2/silence/id is sent to 3 AMs",
+			numAM:              5,
+			numHappyAM:         5,
+			replicationFactor:  3,
+			isRead:             true,
+			expStatusCode:      http.StatusOK,
+			expectedTotalCalls: 3,
+			route:              "/v2/silence/id",
+			responseBody:       []byte(`{"id":"aaa","updatedAt":"2020-01-01T00:00:00Z"}`),
+		},
+		{
 			name:                "Write /silence/id not supported",
 			numAM:               5,
 			numHappyAM:          5,

pkg/alertmanager/merger/v2_silence_id.go

@@ -0,0 +1,34 @@
+package merger
+
+import (
+	"errors"
+
+	"github.com/go-openapi/swag"
+	v2_models "github.com/prometheus/alertmanager/api/v2/models"
+)
+
+// V2SilenceID implements the Merger interface for GET /v2/silence/{id}. It returns the most
+// recently updated silence (newest UpdatedAt timestamp).
+type V2SilenceID struct{}
+
+func (V2SilenceID) MergeResponses(in [][]byte) ([]byte, error) {
+	silences := make(v2_models.GettableSilences, 0)
+	for _, body := range in {
+		parsed := &v2_models.GettableSilence{}
+		if err := swag.ReadJSON(body, parsed); err != nil {
+			return nil, err
+		}
+		silences = append(silences, parsed)
+	}
+
+	merged, err := mergeV2Silences(silences)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(merged) != 1 {
+		return nil, errors.New("unexpected mismatched silence ids")
+	}
+
+	return swag.WriteJSON(merged[0])
+}

pkg/alertmanager/merger/v2_silence_id_test.go

@@ -0,0 +1,61 @@
+package merger
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestV2SilenceId_ReturnsNewestSilence(t *testing.T) {
+
+	// We re-use MergeV2Silences so we rely on that being primarily tested elsewhere.
+
+	in := [][]byte{
+		[]byte(`{"id":"77b580dd-1d9c-4b7e-9bba-13ac173cb4e5","status":{"state":"expired"},` +
+			`"updatedAt":"2021-04-28T17:31:02.215Z","comment":"This is the newest silence",` +
+			`"createdBy":"","endsAt":"2021-04-28T17:31:02.215Z","matchers":` +
+			`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+			`"startsAt":"2021-04-28T17:31:01.725Z"}`),
+		[]byte(`{"id":"77b580dd-1d9c-4b7e-9bba-13ac173cb4e5","status":{"state":"expired"},` +
+			`"updatedAt":"2021-04-28T17:31:02.000Z","comment":"Silence Comment #1",` +
+			`"createdBy":"","endsAt":"2021-04-28T17:31:02.215Z","matchers":` +
+			`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+			`"startsAt":"2021-04-28T17:31:01.725Z"}`),
+		[]byte(`{"id":"77b580dd-1d9c-4b7e-9bba-13ac173cb4e5","status":{"state":"expired"},` +
+			`"updatedAt":"2021-04-28T17:31:02.000Z","comment":"Silence Comment #1",` +
+			`"createdBy":"","endsAt":"2021-04-28T17:31:02.215Z","matchers":` +
+			`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+			`"startsAt":"2021-04-28T17:31:01.725Z"}`),
+	}
+
+	expected := []byte(`{"id":"77b580dd-1d9c-4b7e-9bba-13ac173cb4e5","status":{"state":"expired"},` +
+		`"updatedAt":"2021-04-28T17:31:02.215Z","comment":"This is the newest silence",` +
+		`"createdBy":"","endsAt":"2021-04-28T17:31:02.215Z","matchers":` +
+		`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+		`"startsAt":"2021-04-28T17:31:01.725Z"}`)
+
+	out, err := V2SilenceID{}.MergeResponses(in)
+	require.NoError(t, err)
+	require.Equal(t, string(expected), string(out))
+}
+
+func TestV2SilenceID_InvalidDifferentIDs(t *testing.T) {
+
+	// Responses containing silences with different IDs is invalid input.
+
+	in := [][]byte{
+		[]byte(`{"id":"77b580dd-1d9c-4b7e-9bba-13ac173cb4e5","status":{"state":"expired"},` +
+			`"updatedAt":"2021-04-28T17:31:02.215Z","comment":"Silence Comment #1",` +
+			`"createdBy":"","endsAt":"2021-04-28T17:31:02.215Z","matchers":` +
+			`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+			`"startsAt":"2021-04-28T17:31:01.725Z"}`),
+		[]byte(`{"id":"261248d1-4ff7-4cf1-9957-850c65f4e48b","status":{"state":"expired"},` +
+			`"updatedAt":"2021-04-28T17:31:02.082Z","comment":"Silence Comment #3",` +
+			`"createdBy":"","endsAt":"2021-04-28T17:31:02.082Z","matchers":` +
+			`[{"isEqual":true,"isRegex":false,"name":"instance","value":"prometheus-one"}],` +
+			`"startsAt":"2021-04-28T17:31:01.735Z"}`),
+	}
+
+	_, err := V2SilenceID{}.MergeResponses(in)
+	require.Error(t, err)
+}