[metricstransformprocessor] add new operation- convert_resource_attributes_to_metric_labels

processor/metricstransformprocessor/README.md

@@ -5,7 +5,7 @@ Supported pipeline types: metrics
  - e.g. If want to rename a metric or label to `new_name` while there is already a metric or label called `new_name`, this operation will not take any effect. There will also be an error logged
 
 ## Description
-The metrics transform processor can be used to rename metrics, labels, or label values. It can also be used to perform aggregations on metrics across labels or label values.
+The metrics transform processor can be used to rename metrics, labels, or label values. It can also be used to perform aggregations on metrics across labels or label values. Additionally, it supports converting resource attributes to metric labels. Customers can select which resource attributes they want to set as metric labels.
 
 ## Capabilities
 - Rename metrics (e.g. rename `cpu/usage` to `cpu/usage_time`)
@@ -17,12 +17,15 @@ The metrics transform processor can be used to rename metrics, labels, or label
  - Aggregation_type: sum, mean, max
 - Add label to an existing metric
 - When adding or updating a label value, specify `{{version}}` to include the application version number
+- Convert resource attributes to metric labels
 
 ## Configuration
 ```yaml
 # transforms is a list of transformations with each element transforming a metric selected by metric name
 transforms:
  # name is used to match with the metric to operate on. This implementation doesn’t utilize the filtermetric’s MatchProperties struct because it doesn’t match well with what I need at this phase. All is needed for this processor at this stage is a single name string that can be used to match with selected metrics. The list of metric names and the match type in the filtermetric’s MatchProperties struct are unnecessary. Also, based on the issue about improving filtering configuration, it seems like this struct is subject to be slightly modified.
+
+ # it also accepts `all_metrics` as the value of 'metric_name' field. This batch operation applies changes to all the metrics. However, currently, it only works and is tested for the 'convert_resource_attributes_to_labels' operation.
  - metric_name: <current_metric_name>
 
  # action specifies if the operations are performed on the current copy of the metric or on a newly created metric that will be inserted
@@ -55,6 +58,14 @@ transforms:
  aggregated_values: [values...]
  new_value: <new_value> 
  aggregation_type: {sum, mean, max}
+
+ # convert_resource_attributes_to_labels action converts the resourde attributes to metric labels.
+ - action: convert_resource_attributes_to_labels
+ # customers can select which resource attributes they want to set as metric labels. If `resource_attributes` are not set or the list is empty,
+ # it convert all the resource attributes to metric labels by default.
+ resource_attributes:
+ - resource_attribute_1
+ - resource_attribute_2
 ```
 
 ## Examples
@@ -148,3 +159,14 @@ operation:
  label: label
  label_value: value
 ```
+
+### Convert Resource Attributes to Metric Labels
+```yaml
+# converts resource attributes to metric labels. 
+...
+operation:
+ - action: convert_resource_attributes_to_labels
+ resource_attributes:
+ - ecs.cluster
+ - ecs.task_id
+```

processor/metricstransformprocessor/config.go

@@ -91,6 +91,9 @@ type Operation struct {
 
  // LabelValue identifies the exact label value to operate on
  LabelValue string `mapstructure:"label_value"`
+
+ // ResourceAttributes identifies the resource attributes to convert them to metric labels
+ ResourceAttributes []string `mapstructure:"resource_attributes"`
 }
 
 // ValueAction renames label values.
@@ -138,6 +141,11 @@ const (
  // DeleteLabelValue deletes a label value by also removing all the points associated with this label value
  DeleteLabelValue OperationAction = "delete_label_value"
 
+ // ConvertResourceAttributesToLabels converts resource attributes to metric labels.
+ // TODO: right now it converts all the attributes to metric labels by default. Make it
+ // configurable to select specific resource attributes as metric labels.
+ ConvertResourceAttributesToLabels OperationAction = "convert_resource_attributes_to_labels"
+
  // Mean indicates taking the mean of the aggregated data.
  Mean AggregationType = "mean"
 

processor/metricstransformprocessor/config_test.go

@@ -76,6 +76,16 @@ var (
  NewName: "new_name",
  Operations: testDataOperations,
  },
+ {
+ MetricName: "all_metrics",
+ Action: Update,
+ Operations: []Operation{
+ {
+ Action: ConvertResourceAttributesToLabels,
+ ResourceAttributes: []string{"ecs.cluster"},
+ },
+ },
+ },
  },
  },
  },

processor/metricstransformprocessor/metrics_testcase_builder_test.go

@@ -16,6 +16,7 @@ package metricstransformprocessor
 
 import (
  metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
+ resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
  "google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -33,6 +34,17 @@ func metricBuilder() builder {
  }
 }
 
+// metricBuilderWithResource is used to build metrics with Resource for testing
+func metricBuilderWithResource() builder {
+ return builder{
+ metric: &metricspb.Metric{
+ MetricDescriptor: &metricspb.MetricDescriptor{},
+ Timeseries: make([]*metricspb.TimeSeries, 0),
+ Resource: &resourcepb.Resource{},
+ },
+ }
+}
+
 // setName sets the name of the metric
 func (b builder) setName(name string) builder {
  b.metric.MetricDescriptor.Name = name
@@ -51,6 +63,12 @@ func (b builder) setLabels(labels []string) builder {
  return b
 }
 
+// setResourceAttributes sets resource attributes
+func (b builder) setResourceAttributes(attributes map[string]string) builder {
+ b.metric.Resource.Labels = attributes
+ return b
+}
+
 // addTimeseries adds new timeseries with the labelValuesVal and startTimestamp
 func (b builder) addTimeseries(startTimestampSeconds int64, labelValuesVal []string) builder {
  labelValues := make([]*metricspb.LabelValue, len(labelValuesVal))

processor/metricstransformprocessor/metrics_transform_processor.go

@@ -27,10 +27,11 @@ import (
 )
 
 type internalTransform struct {
- MetricName string
- Action ConfigAction
- NewName string
- Operations []internalOperation
+ MetricName string
+ Action ConfigAction
+ NewName string
+ Operations []internalOperation
+ resourceAttributeMap map[string]string
 }
 
 type internalOperation struct {
@@ -65,7 +66,24 @@ func (mtp *metricsTransformProcessor) ProcessMetrics(_ context.Context, md pdata
  nameToMetricMapping[metric.MetricDescriptor.Name] = metric
  }
 
+ resourceAttributes := make(map[string]string)
+ if data.Resource != nil {
+ resourceAttributes = data.Resource.Labels
+ }
  for _, transform := range mtp.transforms {
+ transform.resourceAttributeMap = resourceAttributes
+
+ // When user sets "metric_name=all_metrics", these operations are going to be applied
+ // to all metric data points. Right now, this batch operation only works and is tested for
+ // `convert_resource_attributes_to_labels` operation, and does not break existing experience
+ // for other operations. After updating the metrics, it will continue for the next iteration.
+ if transform.MetricName == "all_metrics" {
+ for _, metric := range data.Metrics {
+ mtp.update(metric, transform)
+ }
+ continue
+ }
+
  metric, ok := nameToMetricMapping[transform.MetricName]
  if !ok {
  continue
@@ -110,6 +128,8 @@ func (mtp *metricsTransformProcessor) update(metric *metricspb.Metric, transform
  mtp.addLabelOp(metric, op)
  case DeleteLabelValue:
  mtp.deleteLabelValueOp(metric, op)
+ case ConvertResourceAttributesToLabels:
+ mtp.convertResourceAttributesToLabels(metric, transform.resourceAttributeMap, op)
  }
  }
 }

processor/metricstransformprocessor/metrics_transform_processor_testcases_test.go

@@ -570,6 +570,129 @@ var (
  build(),
  },
  },
+ {
+ name: "all_metrics_batch_operation_test",
+ transforms: []internalTransform{
+ {
+ MetricName: "all_metrics",
+ Action: Update,
+ Operations: []internalOperation{
+ {
+ configOperation: Operation{
+ Action: UpdateLabel,
+ Label: "label1",
+ NewLabel: "new/label1",
+ },
+ },
+ },
+ },
+ },
+ in: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ out: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"new/label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ },
+ {
+ name: "convert_resource_attributes_to_labels_with_nil_resource",
+ transforms: []internalTransform{
+ {
+ MetricName: "all_metrics",
+ Action: Update,
+ Operations: []internalOperation{
+ {
+ configOperation: Operation{
+ Action: ConvertResourceAttributesToLabels,
+ },
+ },
+ },
+ },
+ },
+ in: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ out: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ },
+ {
+ name: "convert_resource_attributes_to_labels_empty_list_config",
+ transforms: []internalTransform{
+ {
+ MetricName: "all_metrics",
+ Action: Update,
+ Operations: []internalOperation{
+ {
+ configOperation: Operation{
+ Action: ConvertResourceAttributesToLabels,
+ },
+ },
+ },
+ },
+ },
+ in: []*metricspb.Metric{
+ metricBuilderWithResource().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setResourceAttributes(map[string]string{"label1": "value1"}).
+ addTimeseries(1, []string{}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ out: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ },
+ {
+ name: "convert_resource_attributes_to_labels_with_config_value",
+ transforms: []internalTransform{
+ {
+ MetricName: "all_metrics",
+ Action: Update,
+ Operations: []internalOperation{
+ {
+ configOperation: Operation{
+ Action: ConvertResourceAttributesToLabels,
+ ResourceAttributes: []string{"label1"},
+ },
+ },
+ },
+ },
+ },
+ in: []*metricspb.Metric{
+ metricBuilderWithResource().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setResourceAttributes(map[string]string{"label1": "value1"}).
+ addTimeseries(1, []string{}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ out: []*metricspb.Metric{
+ metricBuilder().setName("metric1").
+ setDataType(metricspb.MetricDescriptor_CUMULATIVE_INT64).
+ setLabels([]string{"label1"}).
+ addTimeseries(1, []string{"value1"}).
+ addInt64Point(0, 3, 2).build(),
+ },
+ },
  // INSERT
  {
  name: "metric_name_insert",

processor/metricstransformprocessor/operation_convert_resource_attributes_to_labels.go

@@ -0,0 +1,49 @@
+// Copyright 2020 OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metricstransformprocessor
+
+import (
+ metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
+)
+
+func (mtp *metricsTransformProcessor) convertResourceAttributesToLabels(metric *metricspb.Metric, resourceAttributes map[string]string, op internalOperation) {
+ expectedLabelMap := make(map[string]string)
+
+ if len(op.configOperation.ResourceAttributes) > 0 {
+ for _, attribute := range op.configOperation.ResourceAttributes {
+ attributeValue, ok := resourceAttributes[attribute]
+ if ok {
+ expectedLabelMap[attribute] = attributeValue
+ }
+ }
+ } else {
+ expectedLabelMap = resourceAttributes
+ }
+
+ for key, value := range expectedLabelMap {
+ lablelKey := &metricspb.LabelKey{
+ Key: key,
+ }
+ labelValue := &metricspb.LabelValue{
+ Value: value,
+ HasValue: true,
+ }
+
+ metric.MetricDescriptor.LabelKeys = append(metric.MetricDescriptor.LabelKeys, lablelKey)
+ for _, ts := range metric.Timeseries {
+ ts.LabelValues = append(ts.LabelValues, labelValue)
+ }
+ }
+}

processor/metricstransformprocessor/testdata/config_full.yaml

@@ -22,6 +22,13 @@ processors:
  aggregated_values: [value1, value2]
  new_value: new_value
  aggregation_type: sum
+ - metric_name: all_metrics
+ action: update
+ operations:
+ - action: convert_resource_attributes_to_labels
+ resource_attributes:
+ - ecs.cluster
+
  metricstransform/addlabel:
  transforms:
  - metric_name: some_name