[aggregator] keep metric type during the aggregation

scripts/development/m3_stack/m3coordinator-aggregator.yml

@@ -80,3 +80,5 @@ carbon:
 
 tagOptions:
   idScheme: quoted
+
+storeMetricsType: true

scripts/development/m3_stack/m3coordinator-standard.yml

@@ -36,3 +36,5 @@ carbon:
 
 tagOptions:
   idScheme: quoted
+
+storeMetricsType: true

scripts/docker-integration-tests/aggregator/m3coordinator.yml

@@ -77,3 +77,5 @@ ingest:
       retry:
         maxBackoff: 10s
         jitter: true
+
+storeMetricsType: true

scripts/docker-integration-tests/aggregator/test.sh

@@ -170,12 +170,14 @@ function prometheus_remote_write {
   local label1_value=${label1_value:-label1}
   local label2_name=${label2_name:-label2}
   local label2_value=${label2_value:-label2}
+  local metric_type=${metric_type:counter}
 
   network_name="aggregator"
   network=$(docker network ls | fgrep $network_name | tr -s ' ' | cut -f 1 -d ' ' | tail -n 1)
   out=$((docker run -it --rm --network $network           \
     $PROMREMOTECLI_IMAGE                                  \
     -u http://m3coordinator01:7202/api/v1/prom/remote/write \
+    -h M3-Prom-Type:${metric_type}                        \
     -t __name__:${metric_name}                            \
     -t ${label0_name}:${label0_value}                     \
     -t ${label1_name}:${label1_value}                     \
@@ -217,6 +219,22 @@ function prometheus_query_native {
   return $?
 }
 
+function dbnode_fetch {
+  local namespace=${namespace}
+  local id=${id}
+  local rangeStart=${rangeStart}
+  local rangeEnd=${rangeEnd}
+  local jq_path=${jq_path:-}
+  local expected_value=${expected_value:-}
+
+  result=$(curl -s                                    \
+    "0.0.0.0:9002/fetch" \
+    "-d" \
+    "{\"namespace\": \"${namespace}\", \"id\": \"${id}\", \"rangeStart\": ${rangeStart}, \"rangeEnd\": ${rangeEnd}}" | jq -r "${jq_path}")
+  test "$result" = "$expected_value"
+  return $?
+}
+
 function test_aggregated_rollup_rule {
   resolution_seconds="10"
   now=$(date +"%s")
@@ -234,6 +252,7 @@ function test_aggregated_rollup_rule {
     label0_name="app" label0_value="nginx_edge" \
       label1_name="status_code" label1_value="500" \
       label2_name="endpoint" label2_value="/foo/bar" \
+      metric_type="counter" \
       prometheus_remote_write \
       http_requests $write_at $value \
       true "Expected request to succeed" \
@@ -251,6 +270,7 @@ function test_aggregated_rollup_rule {
     label0_name="app" label0_value="nginx_edge" \
       label1_name="status_code" label1_value="500" \
       label2_name="endpoint" label2_value="/foo/baz" \
+      metric_type="gauge" \
       prometheus_remote_write \
       http_requests $write_at $value \
       true "Expected request to succeed" \
@@ -284,6 +304,38 @@ function test_aggregated_rollup_rule {
     retry_with_backoff prometheus_query_native
 }
 
+function test_metric_type_survives_aggregation {
+  now=$(date +"%s")
+
+  echo "Test metric type should be kept after aggregation"
+
+  # Emit values for endpoint /foo/bar (to ensure right values aggregated)
+  write_at="$now_truncated"
+  value="42"
+
+  metric_type="counter" \
+  prometheus_remote_write \
+  metric_type_test $now $value \
+  true "Expected request to succeed" \
+  200 "Expected request to return status code 200"
+
+  start=$(( $now - 3600 ))
+  end=$(( $now + 3600 ))
+  jq_path=".datapoints[0].annotation"
+
+  echo "Test query metric type"
+
+  # Test by metric types are stored in aggregated namespace
+  ATTEMPTS=50 TIMEOUT=2 MAX_TIMEOUT=4 \
+    namespace="agg" \
+    id='{__name__=\"metric_type_test\",label0=\"label0\",label1=\"label1\",label2=\"label2\"}' \
+    rangeStart=${start} \
+    rangeEnd=${end} \
+    jq_path="$jq_path" expected_value="CAE=" \
+    retry_with_backoff dbnode_fetch
+}
+
 echo "Run tests"
 test_aggregated_graphite_metric
 test_aggregated_rollup_rule
+test_metric_type_survives_aggregation

src/cmd/services/m3coordinator/downsample/flush_handler.go

@@ -49,6 +49,8 @@ var (
 	// GraphiteIDSchemeTagValue specifies that the graphite ID
 	// scheme should be used for a metric.
 	GraphiteIDSchemeTagValue = []byte("graphite")
+	// PromTypeTagValue specifies metric type
+	PromTypeTagValue = []byte("__promt__")
 )
 
 var (

src/cmd/services/m3coordinator/ingest/m3msg/ingest.go

@@ -157,6 +157,7 @@ type ingestOp struct {
 	sp          policy.StoragePolicy
 	callback    m3msg.Callbackable
 	tags        models.Tags
+	promType    ts.PromMetricType
 	datapoints  ts.Datapoints
 	q           storage.WriteQuery
 }
@@ -218,6 +219,7 @@ func (op *ingestOp) resetWriteQuery() error {
 		Tags:       op.tags,
 		Datapoints: op.datapoints,
 		Unit:       convert.UnitForM3DB(op.sp.Resolution().Precision),
+		Type:       op.promType,
 		Attributes: storagemetadata.Attributes{
 			MetricsType: storagemetadata.AggregatedMetricsType,
 			Resolution:  op.sp.Resolution().Window,
@@ -249,6 +251,12 @@ func (op *ingestOp) resetTags() error {
 			// or if passing for the second time
 			continue
 		}
+		if bytes.Equal(name, downsample.PromTypeTagValue) {
+			if len(value) == 1 {
+				op.promType = ts.PromMetricType(value[0])
+			}
+			continue
+		}
 
 		op.tags = op.tags.AddTagWithoutNormalizing(models.Tag{
 			Name:  name,

src/cmd/services/m3coordinator/ingest/write.go

@@ -471,6 +471,13 @@ func (d *downsamplerAndWriter) writeAggregatedBatch(
 			appender.AddTag(downsample.MetricsOptionIDSchemeTagName,
 				downsample.GraphiteIDSchemeTagValue)
 		}
+		// If prom type is set then send it to aggregator too to do not loose it.
+		// Currently it's passed as a tag and later filtered out when series will back to
+		// the coordinator.
+		// Also, see note above.
+		if value.Attributes.PromType != ts.PromMetricTypeUnknown {
+			appender.AddTag(downsample.PromTypeTagValue, []byte{byte(value.Attributes.PromType)})
+		}
 
 		opts := downsample.SampleAppenderOptions{
 			MetricType: value.Attributes.M3Type,

src/query/api/v1/handler/prometheus/remote/write.go

@@ -493,6 +493,31 @@ func (h *PromWriteHandler) parseRequest(
 		}
 	}
 
+	if promType := r.Header.Get(headers.PromTypeHeader); promType != "" {
+		var tp prompb.MetricType
+		switch strings.ToLower(promType) {
+		case "counter":
+			tp = prompb.MetricType_COUNTER
+		case "gauge":
+			tp = prompb.MetricType_GAUGE
+		case "gauge-histogram":
+			tp = prompb.MetricType_GAUGE_HISTOGRAM
+		case "histogram":
+			tp = prompb.MetricType_HISTOGRAM
+		case "info":
+			tp = prompb.MetricType_INFO
+		case "stateset":
+			tp = prompb.MetricType_STATESET
+		case "summary":
+			tp = prompb.MetricType_SUMMARY
+		default:
+			return parseRequestResult{}, fmt.Errorf("unknown prom metric type %s", promType)
+		}
+		for i := range req.Timeseries {
+			req.Timeseries[i].Type = tp
+		}
+	}
+
 	return parseRequestResult{
 		Request:        &req,
 		Options:        opts,

src/query/storage/converter.go

@@ -151,42 +151,47 @@ func PromTimeSeriesToSeriesAttributes(series prompb.TimeSeries) (ts.SeriesAttrib
 
 // SeriesAttributesToAnnotationPayload converts ts.SeriesAttributes into an annotation.Payload.
 func SeriesAttributesToAnnotationPayload(seriesAttributes ts.SeriesAttributes) (annotation.Payload, error) {
-	var metricType annotation.MetricType
+	metricType, err := PromMetricTypeToAnnotationPayloadType(seriesAttributes.PromType)
+	if err != nil {
+		return annotation.Payload{}, err
+	}
 
-	switch seriesAttributes.PromType {
+	return annotation.Payload{
+		MetricType:        metricType,
+		HandleValueResets: seriesAttributes.HandleValueResets,
+	}, nil
+}
+
+func PromMetricTypeToAnnotationPayloadType(t ts.PromMetricType) (annotation.MetricType, error) {
+	switch t {
 
 	case ts.PromMetricTypeUnknown:
-		metricType = annotation.MetricType_UNKNOWN
+		return annotation.MetricType_UNKNOWN, nil
 
 	case ts.PromMetricTypeCounter:
-		metricType = annotation.MetricType_COUNTER
+		return annotation.MetricType_COUNTER, nil
 
 	case ts.PromMetricTypeGauge:
-		metricType = annotation.MetricType_GAUGE
+		return annotation.MetricType_GAUGE, nil
 
 	case ts.PromMetricTypeHistogram:
-		metricType = annotation.MetricType_HISTOGRAM
+		return annotation.MetricType_HISTOGRAM, nil
 
 	case ts.PromMetricTypeGaugeHistogram:
-		metricType = annotation.MetricType_GAUGE_HISTOGRAM
+		return annotation.MetricType_GAUGE_HISTOGRAM, nil
 
 	case ts.PromMetricTypeSummary:
-		metricType = annotation.MetricType_SUMMARY
+		return annotation.MetricType_SUMMARY, nil
 
 	case ts.PromMetricTypeInfo:
-		metricType = annotation.MetricType_INFO
+		return annotation.MetricType_INFO, nil
 
 	case ts.PromMetricTypeStateSet:
-		metricType = annotation.MetricType_STATESET
+		return annotation.MetricType_STATESET, nil
 
 	default:
-		return annotation.Payload{}, fmt.Errorf("invalid Prometheus metric type %v", seriesAttributes.PromType)
+		return annotation.MetricType_UNKNOWN, fmt.Errorf("invalid Prometheus metric type %v", t)
 	}
-
-	return annotation.Payload{
-		MetricType:        metricType,
-		HandleValueResets: seriesAttributes.HandleValueResets,
-	}, nil
 }
 
 // PromSamplesToM3Datapoints converts Prometheus samples to M3 datapoints

src/query/storage/m3/storage.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/m3db/m3/src/dbnode/client"
+	"github.com/m3db/m3/src/dbnode/generated/proto/annotation"
 	"github.com/m3db/m3/src/dbnode/storage/index"
 	"github.com/m3db/m3/src/query/block"
 	"github.com/m3db/m3/src/query/errors"
@@ -719,6 +720,25 @@ func (s *m3storage) writeSingle(
 
 	namespaceID := namespace.NamespaceID()
 	session := namespace.Session()
+
+	annot := query.Annotation()
+	if query.Options().Type != ts.PromMetricTypeUnknown && (annot == nil || len(annot) == 0) {
+		tp, err := storage.PromMetricTypeToAnnotationPayloadType(query.Options().Type)
+		if err != nil {
+			return err
+		}
+
+		annotationPayload := annotation.Payload{MetricType: tp}
+		annot, err = annotationPayload.Marshal()
+		if err != nil {
+			return err
+		}
+
+		if len(annot) == 0 {
+			annot = nil
+		}
+	}
+
 	return session.WriteTagged(namespaceID, identID, iterator,
-		datapoint.Timestamp, datapoint.Value, query.Unit(), query.Annotation())
+		datapoint.Timestamp, datapoint.Value, query.Unit(), annot)
 }

src/query/storage/types.go

@@ -245,6 +245,7 @@ type WriteQueryOptions struct {
 	Datapoints ts.Datapoints
 	Unit       xtime.Unit
 	Annotation []byte
+	Type       ts.PromMetricType
 	Attributes storagemetadata.Attributes
 }
 

src/x/headers/headers.go

@@ -43,6 +43,10 @@ const (
 	// Valid values are "unaggregated" or "aggregated".
 	MetricsTypeHeader = M3HeaderPrefix + "Metrics-Type"
 
+	// PromTypeHeader sets the prometheus metric type. Valid values are
+	// "counter", "gauge", etc.
+	PromTypeHeader = M3HeaderPrefix + "Prom-Type"
+
 	// WriteTypeHeader is a header that controls if default
 	// writes should be written to both unaggregated and aggregated
 	// namespaces, or if unaggregated values are skipped and