[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/aggregator/aggregator/aggregator.go

@@ -287,6 +287,7 @@ func (agg *aggregator) AddPassthrough(
 			ChunkedID: id.ChunkedID{
 				Data: []byte(metric.ID),
 			},
+			Type:      metric.Type,
 			TimeNanos: metric.TimeNanos,
 			Value:     metric.Value,
 		},

src/aggregator/aggregator/elem_test.go

@@ -1810,6 +1810,7 @@ func testFlushLocalMetricFn() (
 	return func(
 		idPrefix []byte,
 		id id.RawID,
+		metricType metric.Type,
 		idSuffix []byte,
 		timeNanos int64,
 		value float64,

src/aggregator/aggregator/flush.go

@@ -23,6 +23,7 @@ package aggregator
 import (
 	"time"
 
+	"github.com/m3db/m3/src/metrics/metric"
 	"github.com/m3db/m3/src/metrics/metric/id"
 	"github.com/m3db/m3/src/metrics/policy"
 )
@@ -83,6 +84,7 @@ const (
 type flushLocalMetricFn func(
 	idPrefix []byte,
 	id id.RawID,
+	metricType metric.Type,
 	idSuffix []byte,
 	timeNanos int64,
 	value float64,

src/aggregator/aggregator/generic_elem.go

@@ -537,9 +537,9 @@ func (e *GenericElem) processValueWithAggregationLock(
 			for _, point := range toFlush {
 				switch e.idPrefixSuffixType {
 				case NoPrefixNoSuffix:
-					flushLocalFn(nil, e.id, nil, point.TimeNanos, point.Value, e.sp)
+					flushLocalFn(nil, e.id, metric.GaugeType, nil, point.TimeNanos, point.Value, e.sp)
 				case WithPrefixWithSuffix:
-					flushLocalFn(e.FullPrefix(e.opts), e.id, e.TypeStringFor(e.aggTypesOpts, aggType),
+					flushLocalFn(e.FullPrefix(e.opts), e.id, metric.GaugeType, e.TypeStringFor(e.aggTypesOpts, aggType),
 						point.TimeNanos, point.Value, e.sp)
 				}
 			}

src/aggregator/aggregator/handler/writer/protobuf.go

@@ -133,6 +133,7 @@ func (w *protobufWriter) prepare(mp aggregated.ChunkedMetricWithStoragePolicy) (
 	w.m.ID = append(w.m.ID, mp.Suffix...)
 	w.m.Metric.TimeNanos = mp.TimeNanos
 	w.m.Metric.Value = mp.Value
+	w.m.Metric.Type = mp.Type
 	w.m.StoragePolicy = mp.StoragePolicy
 	shard := w.shardFn(w.m.ID, w.numShards)
 	return w.m, shard

src/aggregator/aggregator/list.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/m3db/m3/src/aggregator/aggregator/handler"
 	"github.com/m3db/m3/src/aggregator/aggregator/handler/writer"
+	"github.com/m3db/m3/src/metrics/metric"
 	"github.com/m3db/m3/src/metrics/metric/aggregated"
 	metricid "github.com/m3db/m3/src/metrics/metric/id"
 	"github.com/m3db/m3/src/metrics/policy"
@@ -434,6 +435,7 @@ func (l *baseMetricList) flushBefore(beforeNanos int64, flushType flushType) {
 func (l *baseMetricList) consumeLocalMetric(
 	idPrefix []byte,
 	id metricid.RawID,
+	metricType metric.Type,
 	idSuffix []byte,
 	timeNanos int64,
 	value float64,
@@ -447,6 +449,7 @@ func (l *baseMetricList) consumeLocalMetric(
 	chunkedMetricWithPolicy := aggregated.ChunkedMetricWithStoragePolicy{
 		ChunkedMetric: aggregated.ChunkedMetric{
 			ChunkedID: chunkedID,
+			Type:      metricType,
 			TimeNanos: timeNanos,
 			Value:     value,
 		},
@@ -463,6 +466,7 @@ func (l *baseMetricList) consumeLocalMetric(
 func (l *baseMetricList) discardLocalMetric(
 	idPrefix []byte,
 	id metricid.RawID,
+	metricType metric.Type,
 	idSuffix []byte,
 	timeNanos int64,
 	value float64,

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

@@ -123,6 +123,7 @@ func NewIngester(
 func (i *Ingester) Ingest(
 	ctx context.Context,
 	id []byte,
+	metricType ts.PromMetricType,
 	metricNanos, encodeNanos int64,
 	value float64,
 	sp policy.StoragePolicy,
@@ -131,6 +132,7 @@ func (i *Ingester) Ingest(
 	op := i.p.Get().(*ingestOp)
 	op.c = ctx
 	op.id = id
+	op.metricType = metricType
 	op.metricNanos = metricNanos
 	op.value = value
 	op.sp = sp
@@ -152,6 +154,7 @@ type ingestOp struct {
 
 	c           context.Context
 	id          []byte
+	metricType  ts.PromMetricType
 	metricNanos int64
 	value       float64
 	sp          policy.StoragePolicy
@@ -216,6 +219,7 @@ func (op *ingestOp) resetWriteQuery() error {
 	op.resetDataPoints()
 	return op.q.Reset(storage.WriteQueryOptions{
 		Tags:       op.tags,
+		Type:       op.metricType,
 		Datapoints: op.datapoints,
 		Unit:       convert.UnitForM3DB(op.sp.Resolution().Precision),
 		Attributes: storagemetadata.Attributes{

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

@@ -72,7 +72,7 @@ func TestIngest(t *testing.T) {
 	callback := m3msg.NewProtobufCallback(m, protobuf.NewAggregatedDecoder(nil), &wg)
 
 	m.EXPECT().Ack()
-	ingester.Ingest(context.TODO(), id, metricNanos, 0, val, sp, callback)
+	ingester.Ingest(context.TODO(), id, ts.PromMetricTypeGauge, metricNanos, 0, val, sp, callback)
 
 	for appender.cnt() != 1 {
 		time.Sleep(100 * time.Millisecond)
@@ -93,7 +93,7 @@ func TestIngest(t *testing.T) {
 		},
 		Tags: models.NewTags(2, nil).AddTags(
 			[]models.Tag{
-				models.Tag{
+				{
 					Name:  []byte("__name__"),
 					Value: []byte("foo"),
 				},
@@ -144,7 +144,7 @@ func TestIngestNonRetryableError(t *testing.T) {
 	callback := m3msg.NewProtobufCallback(m, protobuf.NewAggregatedDecoder(nil), &wg)
 
 	m.EXPECT().Ack()
-	ingester.Ingest(context.TODO(), id, metricNanos, 0, val, sp, callback)
+	ingester.Ingest(context.TODO(), id, ts.PromMetricTypeGauge, metricNanos, 0, val, sp, callback)
 
 	for appender.cntErr() != 1 {
 		time.Sleep(100 * time.Millisecond)

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

@@ -495,14 +495,24 @@ func (d *downsamplerAndWriter) writeAggregatedBatch(
 		}
 
 		for _, dp := range value.Datapoints {
-			switch value.Attributes.M3Type {
-			case ts.M3MetricTypeGauge:
-				err = result.SamplesAppender.AppendGaugeTimedSample(dp.Timestamp, dp.Value)
-			case ts.M3MetricTypeCounter:
-				err = result.SamplesAppender.AppendCounterTimedSample(dp.Timestamp, int64(dp.Value))
-			case ts.M3MetricTypeTimer:
-				err = result.SamplesAppender.AppendTimerTimedSample(dp.Timestamp, dp.Value)
+			if value.Attributes.PromType != ts.PromMetricTypeUnknown {
+				switch value.Attributes.PromType {
+				case ts.PromMetricTypeCounter:
+					err = result.SamplesAppender.AppendCounterTimedSample(dp.Timestamp, int64(dp.Value))
+				default:
+					err = result.SamplesAppender.AppendGaugeTimedSample(dp.Timestamp, dp.Value)
+				}
+			} else {
+				switch value.Attributes.M3Type {
+				case ts.M3MetricTypeGauge:
+					err = result.SamplesAppender.AppendGaugeTimedSample(dp.Timestamp, dp.Value)
+				case ts.M3MetricTypeCounter:
+					err = result.SamplesAppender.AppendCounterTimedSample(dp.Timestamp, int64(dp.Value))
+				case ts.M3MetricTypeTimer:
+					err = result.SamplesAppender.AppendTimerTimedSample(dp.Timestamp, dp.Value)
+				}
 			}
+
 			if err != nil {
 				// If we see an error break out so we can try processing the
 				// next datapoint.

src/cmd/services/m3coordinator/server/m3msg/protobuf_handler.go

@@ -122,7 +122,7 @@ func (h *pbHandler) Process(msg consumer.Message) {
 		}
 	}
 
-	h.writeFn(h.ctx, dec.ID(), dec.TimeNanos(), dec.EncodeNanos(), dec.Value(), sp, r)
+	h.writeFn(h.ctx, dec.ID(), dec.Type(), dec.TimeNanos(), dec.EncodeNanos(), dec.Value(), sp, r)
 }
 
 func (h *pbHandler) Close() { h.wg.Wait() }

src/cmd/services/m3coordinator/server/m3msg/protobuf_handler_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/m3db/m3/src/msg/consumer"
 	"github.com/m3db/m3/src/msg/generated/proto/msgpb"
 	"github.com/m3db/m3/src/msg/protocol/proto"
+	"github.com/m3db/m3/src/query/ts"
 	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/server"
 	xtime "github.com/m3db/m3/src/x/time"
@@ -233,6 +234,7 @@ type mockWriter struct {
 func (m *mockWriter) write(
 	ctx context.Context,
 	name []byte,
+	metricType ts.PromMetricType,
 	metricNanos, encodeNanos int64,
 	value float64,
 	sp policy.StoragePolicy,

src/cmd/services/m3coordinator/server/m3msg/types.go

@@ -24,12 +24,14 @@ import (
 	"context"
 
 	"github.com/m3db/m3/src/metrics/policy"
+	"github.com/m3db/m3/src/query/ts"
 )
 
 // WriteFn is the function that writes a metric.
 type WriteFn func(
 	ctx context.Context,
 	id []byte,
+	metricType ts.PromMetricType,
 	metricNanos, encodeNanos int64,
 	value float64,
 	sp policy.StoragePolicy,

src/metrics/aggregation/type.go

@@ -139,7 +139,7 @@ func (a Type) IsValidForGauge() bool {
 // IsValidForCounter if an Type is valid for Counter.
 func (a Type) IsValidForCounter() bool {
 	switch a {
-	case Min, Max, Mean, Count, Sum, SumSq, Stdev:
+	case Min, Max, Mean, Count, Sum, SumSq, Stdev, Last:
 		return true
 	default:
 		return false