Add e2e test

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

integration/e2ecortex/client.go

@@ -270,7 +270,7 @@ func convertTimeseriesToMetrics(timeseries []prompb.TimeSeries, metadata []promp
 	return metrics
 }
 
-func otlpWriteRequest(name string, temporality pmetric.AggregationTemporality, labels ...prompb.Label) pmetricotlp.ExportRequest {
+func otlpWriteRequest(name, unit string, temporality pmetric.AggregationTemporality, labels ...prompb.Label) pmetricotlp.ExportRequest {
 	d := pmetric.NewMetrics()
 
 	// Generate One Counter, One Gauge, One Histogram, One Exponential-Histogram
@@ -292,6 +292,7 @@ func otlpWriteRequest(name string, temporality pmetric.AggregationTemporality, l
 	// Generate One Counter
 	counterMetric := scopeMetric.Metrics().AppendEmpty()
 	counterMetric.SetName(name)
+	counterMetric.SetUnit(unit)
 	counterMetric.SetDescription("test-counter-description")
 
 	counterMetric.SetEmptySum()
@@ -310,8 +311,8 @@ func otlpWriteRequest(name string, temporality pmetric.AggregationTemporality, l
 	return pmetricotlp.NewExportRequestFromMetrics(d)
 }
 
-func (c *Client) OTLPPushExemplar(name string, temporality pmetric.AggregationTemporality, labels ...prompb.Label) (*http.Response, error) {
-	data, err := otlpWriteRequest(name, temporality, labels...).MarshalProto()
+func (c *Client) OTLPPushExemplar(name, unit string, temporality pmetric.AggregationTemporality, labels ...prompb.Label) (*http.Response, error) {
+	data, err := otlpWriteRequest(name, unit, temporality, labels...).MarshalProto()
 	if err != nil {
 		return nil, err
 	}

integration/otlp_test.go

@@ -150,7 +150,7 @@ func TestOTLPIngestExemplar(t *testing.T) {
 	c, err := e2ecortex.NewClient(cortex.HTTPEndpoint(), cortex.HTTPEndpoint(), "", "", "user-1")
 	require.NoError(t, err)
 
-	res, err := c.OTLPPushExemplar("exemplar_1", pmetric.AggregationTemporalityCumulative)
+	res, err := c.OTLPPushExemplar("exemplar_1", "", pmetric.AggregationTemporalityCumulative)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 
@@ -242,15 +242,15 @@ func TestOTLPPromoteResourceAttributesPerTenant(t *testing.T) {
 		{Name: "attr3", Value: "value"},
 	}
 
-	res, err := c1.OTLPPushExemplar("series_1", pmetric.AggregationTemporalityCumulative, labels...)
+	res, err := c1.OTLPPushExemplar("series_1", "", pmetric.AggregationTemporalityCumulative, labels...)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 
-	res, err = c2.OTLPPushExemplar("series_1", pmetric.AggregationTemporalityCumulative, labels...)
+	res, err = c2.OTLPPushExemplar("series_1", "", pmetric.AggregationTemporalityCumulative, labels...)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 
-	res, err = c3.OTLPPushExemplar("series_1", pmetric.AggregationTemporalityCumulative, labels...)
+	res, err = c3.OTLPPushExemplar("series_1", "", pmetric.AggregationTemporalityCumulative, labels...)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 
@@ -267,6 +267,65 @@ func TestOTLPPromoteResourceAttributesPerTenant(t *testing.T) {
 	require.Equal(t, labelSet3, []string{"__name__", "attr1", "attr2", "attr3", "instance", "job"})
 }
 
+func TestOTLPEnableTypeAndUnitLabels(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	minio := e2edb.NewMinio(9000, bucketName)
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	// Configure the blocks storage to frequently compact TSDB head
+	// and ship blocks to the storage.
+	flags := mergeFlags(BlocksStorageFlags(), map[string]string{
+		"-auth.enabled": "true",
+
+		// OTLP
+		"-distributor.otlp.enable-type-and-unit-labels": "true",
+
+		// alert manager
+		"-alertmanager.web.external-url":   "http://localhost/alertmanager",
+		"-alertmanager-storage.backend":    "local",
+		"-alertmanager-storage.local.path": filepath.Join(e2e.ContainerSharedDir, "alertmanager_configs"),
+	})
+
+	// make alert manager config dir
+	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{}))
+
+	require.NoError(t, copyFileToSharedDir(s, "docs/configuration/single-process-config-blocks-local.yaml", cortexConfigFile))
+
+	// start cortex and assert runtime-config is loaded correctly
+	cortex := e2ecortex.NewSingleBinaryWithConfigFile("cortex", cortexConfigFile, flags, "", 9009, 9095)
+	require.NoError(t, s.StartAndWaitReady(cortex))
+
+	c, err := e2ecortex.NewClient(cortex.HTTPEndpoint(), cortex.HTTPEndpoint(), "", "", "user-1")
+	require.NoError(t, err)
+
+	// Push some series to Cortex.
+	now := time.Now()
+
+	labels := []prompb.Label{
+		{Name: "service.name", Value: "test-service"},
+		{Name: "attr1", Value: "value"},
+	}
+
+	res, err := c.OTLPPushExemplar("series_1", "seconds", pmetric.AggregationTemporalityCumulative, labels...)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	value, err := c.Query("series_1_seconds", now)
+	require.NoError(t, err)
+	vector, ok := value.(model.Vector)
+	fmt.Println("vector", vector)
+	require.True(t, ok)
+	require.Equal(t, 1, len(vector))
+
+	metric := vector[0].Metric
+	require.Equal(t, model.LabelValue("seconds"), metric["__unit__"])
+	require.Equal(t, model.LabelValue("gauge"), metric["__type__"])
+}
+
 func TestOTLPPushDeltaTemporality(t *testing.T) {
 	s, err := e2e.NewScenario(networkName)
 	require.NoError(t, err)
@@ -310,7 +369,7 @@ func TestOTLPPushDeltaTemporality(t *testing.T) {
 		{Name: "attr1", Value: "value"},
 	}
 
-	res, err := c.OTLPPushExemplar("series_1", pmetric.AggregationTemporalityDelta, labels...)
+	res, err := c.OTLPPushExemplar("series_1", "", pmetric.AggregationTemporalityDelta, labels...)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)