Implement metadata api query params

CHANGELOG.md

@@ -6,6 +6,7 @@
 * [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526
 * [FEATURE] Update prometheus alertmanager version to v0.28.0 and add new integration msteamsv2, jira, and rocketchat. #6590
 * [FEATURE] Ingester: Support out-of-order native histogram ingestion. It automatically enabled when `-ingester.out-of-order-time-window > 0` and `-blocks-storage.tsdb.enable-native-histograms=true`. #6626 #6663
+* [ENHANCEMENT] Querier: Support query parameters to metadata api (/api/v1/metadata) to allow user to limit metadata to return. #6681
 * [ENHANCEMENT] Query Frontend: Add new limit `-frontend.max-query-response-size` for total query response size after decompression in query frontend. #6607
 * [ENHANCEMENT] Alertmanager: Add nflog and silences maintenance metrics. #6659
 * [ENHANCEMENT] Querier: limit label APIs to query only ingesters if `start` param is not been specified. #6618

integration/e2ecortex/client.go

@@ -115,9 +115,9 @@ func NewPromQueryClient(address string) (*Client, error) {
 }
 
 // Push the input timeseries to the remote endpoint
-func (c *Client) Push(timeseries []prompb.TimeSeries) (*http.Response, error) {
+func (c *Client) Push(timeseries []prompb.TimeSeries, metadata ...prompb.MetricMetadata) (*http.Response, error) {
 	// Create write request
-	data, err := proto.Marshal(&prompb.WriteRequest{Timeseries: timeseries})
+	data, err := proto.Marshal(&prompb.WriteRequest{Timeseries: timeseries, Metadata: metadata})
 	if err != nil {
 		return nil, err
 	}

integration/ingester_metadata_test.go

@@ -0,0 +1,200 @@
+//go:build requires_docker
+// +build requires_docker
+
+package integration
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/prompb"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/e2e"
+	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+)
+
+func TestIngesterMetadata(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	require.NoError(t, s.StartAndWaitReady(consul))
+
+	baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags())
+
+	minio := e2edb.NewMinio(9000, baseFlags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	flags := mergeFlags(baseFlags, map[string]string{
+		// alert manager
+		"-alertmanager.web.external-url": "http://localhost/alertmanager",
+		// consul
+		"-ring.store":      "consul",
+		"-consul.hostname": consul.NetworkHTTPEndpoint(),
+	})
+
+	// Start Cortex components
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, querier))
+
+	// Wait until distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	// Wait until querier has updated the ring.
+	require.NoError(t, querier.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	client, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", userID)
+	require.NoError(t, err)
+
+	metadataMetricNum := 5
+	metadataPerMetrics := 2
+	metadata := make([]prompb.MetricMetadata, 0, metadataMetricNum)
+	for i := 0; i < metadataMetricNum; i++ {
+		for j := 0; j < metadataPerMetrics; j++ {
+			metadata = append(metadata, prompb.MetricMetadata{
+				MetricFamilyName: fmt.Sprintf("metadata_name_%d", i),
+				Help:             fmt.Sprintf("metadata_help_%d_%d", i, j),
+				Unit:             fmt.Sprintf("metadata_unit_%d_%d", i, j),
+			})
+		}
+	}
+	res, err := client.Push(nil, metadata...)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	testMetadataQueryParams(t, client, metadataMetricNum, metadataPerMetrics)
+}
+
+func TestIngesterMetadataWithTenantFederation(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	require.NoError(t, s.StartAndWaitReady(consul))
+
+	baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags())
+
+	minio := e2edb.NewMinio(9000, baseFlags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	flags := mergeFlags(baseFlags, map[string]string{
+		// tenant federation
+		"-tenant-federation.enabled": "true",
+		// alert manager
+		"-alertmanager.web.external-url": "http://localhost/alertmanager",
+		// consul
+		"-ring.store":      "consul",
+		"-consul.hostname": consul.NetworkHTTPEndpoint(),
+	})
+
+	// Start Cortex components
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, querier))
+
+	// Wait until distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	// Wait until querier has updated the ring.
+	require.NoError(t, querier.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	metadataMetricNum := 5
+	metadataPerMetrics := 2
+	metadata := make([]prompb.MetricMetadata, 0, metadataMetricNum)
+	for i := 0; i < metadataMetricNum; i++ {
+		for j := 0; j < metadataPerMetrics; j++ {
+			metadata = append(metadata, prompb.MetricMetadata{
+				MetricFamilyName: fmt.Sprintf("metadata_name_%d", i),
+				Help:             fmt.Sprintf("metadata_help_%d_%d", i, j),
+				Unit:             fmt.Sprintf("metadata_unit_%d_%d", i, j),
+			})
+		}
+	}
+
+	numUsers := 2
+	tenantIDs := make([]string, numUsers)
+	for u := 0; u < numUsers; u++ {
+		tenantIDs[u] = fmt.Sprintf("user-%d", u)
+		c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", tenantIDs[u])
+		require.NoError(t, err)
+
+		res, err := c.Push(nil, metadata...)
+		require.NoError(t, err)
+		require.Equal(t, 200, res.StatusCode)
+	}
+
+	client, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), querier.HTTPEndpoint(), "", "", strings.Join(tenantIDs, "|"))
+	require.NoError(t, err)
+
+	testMetadataQueryParams(t, client, metadataMetricNum, metadataPerMetrics)
+}
+
+func testMetadataQueryParams(t *testing.T, client *e2ecortex.Client, metadataMetricNum, metadataPerMetrics int) {
+	t.Run("test no parameter", func(t *testing.T) {
+		result, err := client.Metadata("", "")
+		require.NoError(t, err)
+		require.Equal(t, metadataMetricNum, len(result))
+
+		for _, v := range result {
+			require.Equal(t, metadataPerMetrics, len(v))
+		}
+	})
+
+	t.Run("test name parameter", func(t *testing.T) {
+		t.Run("existing name", func(t *testing.T) {
+			name := "metadata_name_0"
+			result, err := client.Metadata(name, "")
+			require.NoError(t, err)
+			m, ok := result[name]
+			require.True(t, ok)
+			require.Equal(t, metadataPerMetrics, len(m))
+		})
+		t.Run("existing name with limit 0", func(t *testing.T) {
+			name := "metadata_name_0"
+			result, err := client.Metadata(name, "0")
+			require.NoError(t, err)
+			require.Equal(t, 0, len(result))
+		})
+		t.Run("non-existing name", func(t *testing.T) {
+			result, err := client.Metadata("dummy", "")
+			require.NoError(t, err)
+			require.Equal(t, 0, len(result))
+		})
+	})
+
+	t.Run("test limit parameter", func(t *testing.T) {
+		t.Run("less than length of metadata", func(t *testing.T) {
+			result, err := client.Metadata("", "3")
+			require.NoError(t, err)
+			require.Equal(t, 3, len(result))
+		})
+		t.Run("limit: 0", func(t *testing.T) {
+			result, err := client.Metadata("", "0")
+			require.NoError(t, err)
+			require.Equal(t, 0, len(result))
+		})
+		t.Run("invalid limit", func(t *testing.T) {
+			_, err := client.Metadata("", "dummy")
+			require.Error(t, err)
+		})
+	})
+}

integration/otlp_test.go

@@ -88,7 +88,7 @@ func TestOTLP(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []string{"__name__", "foo"}, labelNames)
 
-	metadataResult, err := c.Metadata("series_1", "")
+	metadataResult, err := c.Metadata("series_1_total", "")
 	require.NoError(t, err)
 	require.Equal(t, 1, len(metadataResult))
 

pkg/distributor/distributor.go

@@ -1460,13 +1460,12 @@ func (d *Distributor) metricsForLabelMatchersCommon(ctx context.Context, from, t
 }
 
 // MetricsMetadata returns all metric metadata of a user.
-func (d *Distributor) MetricsMetadata(ctx context.Context) ([]scrape.MetricMetadata, error) {
+func (d *Distributor) MetricsMetadata(ctx context.Context, req *ingester_client.MetricsMetadataRequest) ([]scrape.MetricMetadata, error) {
 	replicationSet, err := d.GetIngestersForMetadata(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	req := &ingester_client.MetricsMetadataRequest{}
 	// TODO(gotjosh): We only need to look in all the ingesters if shardByAllLabels is enabled.
 	resps, err := d.ForReplicationSet(ctx, replicationSet, d.cfg.ZoneResultsQuorumMetadata, false, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
 		return client.MetricsMetadata(ctx, req)

pkg/distributor/distributor_test.go

@@ -2774,7 +2774,7 @@ func TestDistributor_MetricsMetadata(t *testing.T) {
 			require.NoError(t, err)
 
 			// Assert on metric metadata
-			metadata, err := ds[0].MetricsMetadata(ctx)
+			metadata, err := ds[0].MetricsMetadata(ctx, &client.MetricsMetadataRequest{Limit: -1, LimitPerMetric: -1, Metric: ""})
 			require.NoError(t, err)
 			assert.Equal(t, 10, len(metadata))