Release 1.15.1 (#5291)

* Handle grpc code resource exhausted for store gateway (#5286)

* handle grpc code resource exhausted for store gateway

Signed-off-by: Ben Ye <[email protected]>

* fix lint

Signed-off-by: Ben Ye <[email protected]>

* update changelog

Signed-off-by: Ben Ye <[email protected]>

* try fixing test

Signed-off-by: Ben Ye <[email protected]>

* try to fix E2E test

Signed-off-by: Ben Ye <[email protected]>

* lint

Signed-off-by: Ben Ye <[email protected]>

* try again

Signed-off-by: Ben Ye <[email protected]>

* fix message

Signed-off-by: Ben Ye <[email protected]>

* remove labels API

Signed-off-by: Ben Ye <[email protected]>

* remove logic to check string contains

Signed-off-by: Ben Ye <[email protected]>

* make limiter vars private

Signed-off-by: Ben Ye <[email protected]>

---------

Signed-off-by: Ben Ye <[email protected]>

* Validating new fields on the PagerDuty AM config (#5290)

* Not Allowing to set the new  service_key_file and routing_key_file on AM config

Signed-off-by: Alan Protasio <[email protected]>

* changelog

Signed-off-by: Alan Protasio <[email protected]>

* fix multiples types

Signed-off-by: Alan Protasio <[email protected]>

---------

Signed-off-by: Alan Protasio <[email protected]>

---------

Signed-off-by: Ben Ye <[email protected]>
Signed-off-by: Alan Protasio <[email protected]>
Co-authored-by: Ben Ye <[email protected]>

Author: alanprot

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## master / unreleased
 
+## 1.15.1 2023-04-26
+
+* [CHANGE] Alertmanager: Validating new fields on the PagerDuty AM config. #5290
+* [BUGFIX] Querier: Convert gRPC `ResourceExhausted` status code from store gateway to 422 limit error. #5286
+
 ## 1.15.0 2023-04-19
 
 * [CHANGE] Storage: Make Max exemplars config per tenant instead of global configuration. #5080 #5122

integration/e2ecortex/client.go

@@ -150,10 +150,88 @@ func (c *Client) QueryRangeRaw(query string, start, end time.Time, step time.Dur
 }
 
 // QueryRaw runs a query directly against the querier API.
-func (c *Client) QueryRaw(query string) (*http.Response, []byte, error) {
-	addr := fmt.Sprintf("http://%s/api/prom/api/v1/query?query=%s", c.querierAddress, url.QueryEscape(query))
+func (c *Client) QueryRaw(query string, ts time.Time) (*http.Response, []byte, error) {
+	u := &url.URL{
+		Scheme: "http",
+		Path:   fmt.Sprintf("%s/api/prom/api/v1/query", c.querierAddress),
+	}
+	q := u.Query()
+	q.Set("query", query)
 
-	return c.query(addr)
+	if !ts.IsZero() {
+		q.Set("time", FormatTime(ts))
+	}
+	u.RawQuery = q.Encode()
+	return c.query(u.String())
+}
+
+// SeriesRaw runs a series request directly against the querier API.
+func (c *Client) SeriesRaw(matches []string, startTime, endTime time.Time) (*http.Response, []byte, error) {
+	u := &url.URL{
+		Scheme: "http",
+		Path:   fmt.Sprintf("%s/api/prom/api/v1/series", c.querierAddress),
+	}
+	q := u.Query()
+
+	for _, m := range matches {
+		q.Add("match[]", m)
+	}
+
+	if !startTime.IsZero() {
+		q.Set("start", FormatTime(startTime))
+	}
+	if !endTime.IsZero() {
+		q.Set("end", FormatTime(endTime))
+	}
+
+	u.RawQuery = q.Encode()
+	return c.query(u.String())
+}
+
+// LabelNamesRaw runs a label names request directly against the querier API.
+func (c *Client) LabelNamesRaw(matches []string, startTime, endTime time.Time) (*http.Response, []byte, error) {
+	u := &url.URL{
+		Scheme: "http",
+		Path:   fmt.Sprintf("%s/api/prom/api/v1/labels", c.querierAddress),
+	}
+	q := u.Query()
+
+	for _, m := range matches {
+		q.Add("match[]", m)
+	}
+
+	if !startTime.IsZero() {
+		q.Set("start", FormatTime(startTime))
+	}
+	if !endTime.IsZero() {
+		q.Set("end", FormatTime(endTime))
+	}
+
+	u.RawQuery = q.Encode()
+	return c.query(u.String())
+}
+
+// LabelValuesRaw runs a label values request directly against the querier API.
+func (c *Client) LabelValuesRaw(label string, matches []string, startTime, endTime time.Time) (*http.Response, []byte, error) {
+	u := &url.URL{
+		Scheme: "http",
+		Path:   fmt.Sprintf("%s/api/prom/api/v1/label/%s/values", c.querierAddress, label),
+	}
+	q := u.Query()
+
+	for _, m := range matches {
+		q.Add("match[]", m)
+	}
+
+	if !startTime.IsZero() {
+		q.Set("start", FormatTime(startTime))
+	}
+	if !endTime.IsZero() {
+		q.Set("end", FormatTime(endTime))
+	}
+
+	u.RawQuery = q.Encode()
+	return c.query(u.String())
 }
 
 // RemoteRead runs a remote read query.
@@ -259,8 +337,8 @@ func (c *Client) LabelValues(label string, start, end time.Time, matches []strin
 }
 
 // LabelNames gets label names
-func (c *Client) LabelNames(start, end time.Time) ([]string, error) {
-	result, _, err := c.querierClient.LabelNames(context.Background(), nil, start, end)
+func (c *Client) LabelNames(start, end time.Time, matchers ...string) ([]string, error) {
+	result, _, err := c.querierClient.LabelNames(context.Background(), matchers, start, end)
 	return result, err
 }
 

integration/querier_test.go

@@ -5,6 +5,7 @@ package integration
 
 import (
 	"fmt"
+	"net/http"
 	"strconv"
 	"strings"
 	"testing"
@@ -818,6 +819,90 @@ func TestQuerierWithBlocksStorageOnMissingBlocksFromStorage(t *testing.T) {
 	assert.Contains(t, err.Error(), "500")
 }
 
+func TestQuerierWithBlocksStorageLimits(t *testing.T) {
+	const blockRangePeriod = 5 * time.Second
+
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Configure the blocks storage to frequently compact TSDB head
+	// and ship blocks to the storage.
+	flags := mergeFlags(BlocksStorageFlags(), map[string]string{
+		"-blocks-storage.tsdb.block-ranges-period": blockRangePeriod.String(),
+		"-blocks-storage.tsdb.ship-interval":       "1s",
+		"-blocks-storage.tsdb.retention-period":    ((blockRangePeriod * 2) - 1).String(),
+	})
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	// Start Cortex components for the write path.
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester))
+
+	// Wait until the distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	// Push some series to Cortex.
+	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", "user-1")
+	require.NoError(t, err)
+
+	seriesTimestamp := time.Now()
+	series2Timestamp := seriesTimestamp.Add(blockRangePeriod * 2)
+	series1, _ := generateSeries("series_1", seriesTimestamp, prompb.Label{Name: "job", Value: "test"})
+	series2, _ := generateSeries("series_2", series2Timestamp, prompb.Label{Name: "job", Value: "test"})
+
+	res, err := c.Push(series1)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	res, err = c.Push(series2)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	// Wait until the TSDB head is compacted and shipped to the storage.
+	// The shipped block contains the 1st series, while the 2ns series in in the head.
+	require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(1), "cortex_ingester_shipper_uploads_total"))
+	require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(2), "cortex_ingester_memory_series_created_total"))
+	require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(1), "cortex_ingester_memory_series_removed_total"))
+	require.NoError(t, ingester.WaitSumMetrics(e2e.Equals(1), "cortex_ingester_memory_series"))
+
+	// Start the querier and store-gateway, and configure them to frequently sync blocks fast enough to trigger consistency check.
+	storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-blocks-storage.bucket-store.sync-interval": "5s",
+		"-querier.max-fetched-series-per-query":      "1",
+	}), "")
+	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-blocks-storage.bucket-store.sync-interval": "5s",
+		"-querier.max-fetched-series-per-query":      "1",
+	}), "")
+	require.NoError(t, s.StartAndWaitReady(querier, storeGateway))
+
+	// Wait until the querier and store-gateway have updated the ring, and wait until the blocks are old enough for consistency check
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512*2), "cortex_ring_tokens_total"))
+	require.NoError(t, storeGateway.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, querier.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(4), []string{"cortex_querier_blocks_scan_duration_seconds"}, e2e.WithMetricCount))
+
+	// Query back the series.
+	c, err = e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	// We expect all queries hitting 422 exceeded series limit
+	resp, body, err := c.QueryRaw(`{job="test"}`, series2Timestamp)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode)
+	require.Contains(t, string(body), "max number of series limit")
+
+	resp, body, err = c.SeriesRaw([]string{`{job="test"}`}, series2Timestamp.Add(-time.Hour), series2Timestamp)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusUnprocessableEntity, resp.StatusCode)
+	require.Contains(t, string(body), "max number of series limit")
+}
+
 func TestQueryLimitsWithBlocksStorageRunningInMicroServices(t *testing.T) {
 	const blockRangePeriod = 5 * time.Second
 

integration/query_frontend_test.go

@@ -293,7 +293,7 @@ func runQueryFrontendTest(t *testing.T, cfg queryFrontendTestConfig) {
 
 		// No need to repeat the test on missing metric name for each user.
 		if userID == 0 && cfg.testMissingMetricName {
-			res, body, err := c.QueryRaw("{instance=~\"hello.*\"}")
+			res, body, err := c.QueryRaw("{instance=~\"hello.*\"}", time.Now())
 			require.NoError(t, err)
 			require.Equal(t, 422, res.StatusCode)
 			require.Contains(t, string(body), "query must contain metric name")
@@ -317,7 +317,7 @@ func runQueryFrontendTest(t *testing.T, cfg queryFrontendTestConfig) {
 
 		// No need to repeat the test on Server-Timing header for each user.
 		if userID == 0 && cfg.queryStatsEnabled {
-			res, _, err := c.QueryRaw("{instance=~\"hello.*\"}")
+			res, _, err := c.QueryRaw("{instance=~\"hello.*\"}", time.Now())
 			require.NoError(t, err)
 			require.Regexp(t, "querier_wall_time;dur=[0-9.]*, response_time;dur=[0-9.]*$", res.Header.Values("Server-Timing")[0])
 		}

integration/zone_aware_test.go

@@ -135,7 +135,7 @@ func TestZoneAwareReplication(t *testing.T) {
 	require.NoError(t, ingester3.Kill())
 
 	// Query back any series => fail (either because of a timeout or 500)
-	result, _, err := client.QueryRaw("series_1")
+	result, _, err := client.QueryRaw("series_1", time.Now())
 	if !errors.Is(err, context.DeadlineExceeded) {
 		require.NoError(t, err)
 		require.Equal(t, 500, result.StatusCode)

pkg/alertmanager/api.go

@@ -40,12 +40,14 @@ const (
 )
 
 var (
-	errPasswordFileNotAllowed        = errors.New("setting password_file, bearer_token_file and credentials_file is not allowed")
-	errOAuth2SecretFileNotAllowed    = errors.New("setting OAuth2 client_secret_file is not allowed")
-	errTLSFileNotAllowed             = errors.New("setting TLS ca_file, cert_file and key_file is not allowed")
-	errSlackAPIURLFileNotAllowed     = errors.New("setting Slack api_url_file and global slack_api_url_file is not allowed")
-	errVictorOpsAPIKeyFileNotAllowed = errors.New("setting VictorOps api_key_file is not allowed")
-	errOpsGenieAPIKeyFileNotAllowed  = errors.New("setting OpsGenie api_key_file is not allowed")
+	errPasswordFileNotAllowed            = errors.New("setting password_file, bearer_token_file and credentials_file is not allowed")
+	errOAuth2SecretFileNotAllowed        = errors.New("setting OAuth2 client_secret_file is not allowed")
+	errTLSFileNotAllowed                 = errors.New("setting TLS ca_file, cert_file and key_file is not allowed")
+	errSlackAPIURLFileNotAllowed         = errors.New("setting Slack api_url_file and global slack_api_url_file is not allowed")
+	errVictorOpsAPIKeyFileNotAllowed     = errors.New("setting VictorOps api_key_file is not allowed")
+	errOpsGenieAPIKeyFileNotAllowed      = errors.New("setting OpsGenie api_key_file is not allowed")
+	errPagerDutyRoutingKeyFileNotAllowed = errors.New("setting PagerDuty routing_key_file is not allowed")
+	errPagerDutyServiceKeyFileNotAllowed = errors.New("setting PagerDuty service_key_file is not allowed")
 )
 
 // UserConfig is used to communicate a users alertmanager configs
@@ -356,6 +358,11 @@ func validateAlertmanagerConfig(cfg interface{}) error {
 		if err := validateVictorOpsConfig(v.Interface().(config.VictorOpsConfig)); err != nil {
 			return err
 		}
+
+	case reflect.TypeOf(config.PagerdutyConfig{}):
+		if err := validatePagerdutyConfig(v.Interface().(config.PagerdutyConfig)); err != nil {
+			return err
+		}
 	}
 
 	// If the input config is a struct, recursively iterate on all fields.
@@ -430,7 +437,7 @@ func validateReceiverTLSConfig(cfg commoncfg.TLSConfig) error {
 }
 
 // validateGlobalConfig validates the Global config and returns an error if it contains
-// settings now allowed by Cortex.
+// settings not allowed by Cortex.
 func validateGlobalConfig(cfg config.GlobalConfig) error {
 	if cfg.OpsGenieAPIKeyFile != "" {
 		return errOpsGenieAPIKeyFileNotAllowed
@@ -442,7 +449,7 @@ func validateGlobalConfig(cfg config.GlobalConfig) error {
 }
 
 // validateOpsGenieConfig validates the OpsGenie config and returns an error if it contains
-// settings now allowed by Cortex.
+// settings not allowed by Cortex.
 func validateOpsGenieConfig(cfg config.OpsGenieConfig) error {
 	if cfg.APIKeyFile != "" {
 		return errOpsGenieAPIKeyFileNotAllowed
@@ -451,7 +458,7 @@ func validateOpsGenieConfig(cfg config.OpsGenieConfig) error {
 }
 
 // validateSlackConfig validates the Slack config and returns an error if it contains
-// settings now allowed by Cortex.
+// settings not allowed by Cortex.
 func validateSlackConfig(cfg config.SlackConfig) error {
 	if cfg.APIURLFile != "" {
 		return errSlackAPIURLFileNotAllowed
@@ -460,10 +467,24 @@ func validateSlackConfig(cfg config.SlackConfig) error {
 }
 
 // validateVictorOpsConfig validates the VictorOps config and returns an error if it contains
-// settings now allowed by Cortex.
+// settings not allowed by Cortex.
 func validateVictorOpsConfig(cfg config.VictorOpsConfig) error {
 	if cfg.APIKeyFile != "" {
 		return errVictorOpsAPIKeyFileNotAllowed
 	}
 	return nil
 }
+
+// validatePagerdutyConfig validates the pager duty config and returns an error if it contains
+// settings not allowed by Cortex.
+func validatePagerdutyConfig(cfg config.PagerdutyConfig) error {
+	if cfg.RoutingKeyFile != "" {
+		return errPagerDutyRoutingKeyFileNotAllowed
+	}
+
+	if cfg.ServiceKeyFile != "" {
+		return errPagerDutyServiceKeyFileNotAllowed
+	}
+
+	return nil
+}

pkg/alertmanager/api_test.go

@@ -563,6 +563,34 @@ template_files:
 			maxTemplateSize: 20,
 			err:             nil,
 		},
+		{
+			name: "Should return error if PagerDuty routing_key_file is set",
+			cfg: `
+alertmanager_config: |
+  receivers:
+    - name: default-receiver
+      pagerduty_configs:
+        - routing_key_file: /secrets
+
+  route:
+    receiver: 'default-receiver'
+`,
+			err: errors.Wrap(errPagerDutyRoutingKeyFileNotAllowed, "error validating Alertmanager config"),
+		},
+		{
+			name: "Should return error if PagerDuty service_key_file is set",
+			cfg: `
+alertmanager_config: |
+  receivers:
+    - name: default-receiver
+      pagerduty_configs:
+        - service_key_file: /secrets
+
+  route:
+    receiver: 'default-receiver'
+`,
+			err: errors.Wrap(errPagerDutyServiceKeyFileNotAllowed, "error validating Alertmanager config"),
+		},
 	}
 
 	limits := &mockAlertManagerLimits{}