New label processor

processor/labelsprocessor/README.md

@@ -0,0 +1,23 @@
+# Labels Processor
+
+Supported pipeline types: metrics
+
+The labels processor can be used to add data point labels to all metrics that pass through it.
+If any specified labels already exist in the metric, the value will be updated.
+
+Please refer to [config.go](./config.go) for the config spec.
+
+Example:
+
+```yaml
+processors:
+ labels_processor:
+ labels:
+ - key: label1
+ value: value1
+ - key: label2
+ value: value2
+```
+
+Refer to [config.yaml](./testdata/config.yaml) for detailed
+examples on using the processor.

processor/labelsprocessor/config.go

@@ -0,0 +1,15 @@
+package labelsprocessor
+
+import "go.opentelemetry.io/collector/config/configmodels"
+
+// Config defines configuration for Labels processor.
+type Config struct {
+ configmodels.ProcessorSettings `mapstructure:",squash"`
+ Labels []LabelConfig `mapstructure:"labels"`
+}
+
+// LabelConfig defines configuration for provided labels
+type LabelConfig struct {
+ Key string `mapstructure:"key"`
+ Value string `mapstructure:"value"`
+}

processor/labelsprocessor/config_test.go

@@ -0,0 +1,50 @@
+package labelsprocessor
+
+import (
+ "os"
+ "path"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "go.opentelemetry.io/collector/component/componenttest"
+ "go.opentelemetry.io/collector/config/configmodels"
+ "go.opentelemetry.io/collector/config/configtest"
+)
+
+func TestLoadConfig(t *testing.T) {
+
+ factories, err := componenttest.ExampleComponents()
+ assert.NoError(t, err)
+
+ factories.Processors[typeStr] = NewFactory()
+
+ os.Setenv("VALUE_1", "first_val")
+ os.Setenv("VALUE_2", "second_val")
+
+ cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
+ assert.NoError(t, err)
+ assert.NotNil(t, cfg)
+
+ assert.Equal(t, cfg.Processors["labels_processor"], &Config{
+ ProcessorSettings: configmodels.ProcessorSettings{
+ TypeVal: "labels_processor",
+ NameVal: "labels_processor",
+ },
+ Labels: []LabelConfig{
+ {Key: "label1", Value: "value1"},
+ {Key: "label2", Value: "value2"},
+ },
+ })
+
+ assert.Equal(t, cfg.Processors["labels_processor/env_vars"], &Config{
+ ProcessorSettings: configmodels.ProcessorSettings{
+ TypeVal: "labels_processor",
+ NameVal: "labels_processor/env_vars",
+ },
+ Labels: []LabelConfig{
+ {Key: "label1", Value: "first_val"},
+ {Key: "label2", Value: "second_val"},
+ },
+ })
+
+}

processor/labelsprocessor/factory.go

@@ -0,0 +1,49 @@
+package labelsprocessor
+
+import (
+ "context"
+
+ "go.opentelemetry.io/collector/component"
+ "go.opentelemetry.io/collector/config/configmodels"
+ "go.opentelemetry.io/collector/consumer"
+ "go.opentelemetry.io/collector/processor/processorhelper"
+)
+
+const (
+ typeStr = "labels_processor"
+)
+
+var processorCapabilities = component.ProcessorCapabilities{MutatesConsumedData: true}
+
+// NewFactory returns a new factory for the Labels processor.
+func NewFactory() component.ProcessorFactory {
+ return processorhelper.NewFactory(
+ typeStr,
+ createDefaultConfig,
+ processorhelper.WithMetrics(createMetricsProcessor))
+}
+
+func createDefaultConfig() configmodels.Processor {
+ return &Config{
+ ProcessorSettings: configmodels.ProcessorSettings{
+ TypeVal: typeStr,
+ NameVal: typeStr,
+ },
+ }
+}
+
+func createMetricsProcessor(
+ _ context.Context,
+ _ component.ProcessorCreateParams,
+ cfg configmodels.Processor,
+ nextConsumer consumer.MetricsConsumer) (component.MetricsProcessor, error) {
+ lp, err := newLabelMetricProcessor(cfg.(*Config))
+ if err != nil {
+ return nil, err
+ }
+ return processorhelper.NewMetricsProcessor(
+ cfg,
+ nextConsumer,
+ lp,
+ processorhelper.WithCapabilities(processorCapabilities))
+}

processor/labelsprocessor/factory_test.go

@@ -0,0 +1,51 @@
+package labelsprocessor
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "go.uber.org/zap"
+
+ "go.opentelemetry.io/collector/component"
+ "go.opentelemetry.io/collector/config/configcheck"
+ "go.opentelemetry.io/collector/config/configmodels"
+ "go.opentelemetry.io/collector/exporter/exportertest"
+)
+
+func TestType(t *testing.T) {
+ factory := NewFactory()
+ pType := factory.Type()
+ assert.Equal(t, pType, configmodels.Type("labels_processor"))
+}
+
+func TestCreateDefaultConfig(t *testing.T) {
+ factory := NewFactory()
+ cfg := factory.CreateDefaultConfig()
+ assert.Equal(t, cfg, &Config{
+ ProcessorSettings: configmodels.ProcessorSettings{
+ NameVal: typeStr,
+ TypeVal: typeStr,
+ },
+ })
+ assert.NoError(t, configcheck.ValidateConfig(cfg))
+}
+
+func TestCreateProcessor(t *testing.T) {
+
+ factory := NewFactory()
+ cfg := &Config{
+ ProcessorSettings: configmodels.ProcessorSettings{
+ TypeVal: "labels_processor",
+ NameVal: "labels_processor",
+ },
+ Labels: []LabelConfig{
+ {Key: "label1", Value: "value1"},
+ },
+ }
+
+ mp, mErr := factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, cfg, exportertest.NewNopMetricsExporter())
+ assert.NoError(t, mErr)
+ assert.NotNil(t, mp)
+
+}

processor/labelsprocessor/labels_processor.go

@@ -0,0 +1,118 @@
+package labelsprocessor
+
+import (
+ "context"
+ "fmt"
+
+ "go.opentelemetry.io/collector/consumer/pdata"
+ v11 "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/common/v1"
+ v1 "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/metrics/v1"
+)
+
+type labelMetricProcessor struct {
+ cfg *Config
+}
+
+func newLabelMetricProcessor(cfg *Config) (*labelMetricProcessor, error) {
+ err := validateConfig(cfg)
+ if err != nil {
+ return nil, err
+ }
+ return &labelMetricProcessor{cfg: cfg}, nil
+}
+
+func validateConfig(cfg *Config) error {
+ for _, elem := range cfg.Labels {
+ if elem.Key == "" || elem.Value == "" {
+ return fmt.Errorf("Labels Processor configuration contains an empty key or value")
+ }
+ }
+
+ keys := make(map[string]bool)
+ for _, elem := range cfg.Labels {
+ _, value := keys[elem.Key]
+ if value {
+ return fmt.Errorf("Labels Processor configuration contains duplicate keys")
+ }
+ keys[elem.Key] = true
+ }
+
+ return nil
+}
+
+func (lp *labelMetricProcessor) ProcessMetrics(_ context.Context, md pdata.Metrics) (pdata.Metrics, error) {
+
+ otlpMetrics := pdata.MetricsToOtlp(md)
+
+ for _, otlpMetric := range otlpMetrics {
+ for _, instrMetric := range otlpMetric.GetInstrumentationLibraryMetrics() {
+ for _, metric := range instrMetric.GetMetrics() {
+
+ // Multiple types of Data Points exists, and each of them must be handled differently
+ if metric.GetIntSum() != nil {
+ intDataPoints := metric.GetIntSum().GetDataPoints()
+ handleIntDataPoints(intDataPoints, lp)
+ } else if metric.GetIntGauge() != nil {
+ intDataPoints := metric.GetIntGauge().GetDataPoints()
+ handleIntDataPoints(intDataPoints, lp)
+ } else if metric.GetDoubleGauge() != nil {
+ doubleDataPoints := metric.GetDoubleGauge().GetDataPoints()
+ handleDoubleDataPoints(doubleDataPoints, lp)
+ } else if metric.GetDoubleSum() != nil {
+ doubleDataPoints := metric.GetDoubleSum().GetDataPoints()
+ handleDoubleDataPoints(doubleDataPoints, lp)
+ } else if metric.GetIntHistogram() != nil {
+ intHistogramDataPoints := metric.GetIntHistogram().GetDataPoints()
+ handleIntHistogramDataPoints(intHistogramDataPoints, lp)
+ } else if metric.GetDoubleHistogram() != nil {
+ doubleHistogramDataPoints := metric.GetDoubleHistogram().GetDataPoints()
+ handleDoubleHistogramDataPoints(doubleHistogramDataPoints, lp)
+ }
+
+ }
+ }
+ }
+
+ return md, nil
+}
+
+func handleIntDataPoints(intDataPoints []*v1.IntDataPoint, lp *labelMetricProcessor) {
+ for _, dataPoint := range intDataPoints {
+ upsertLabels(&dataPoint.Labels, lp)
+ }
+}
+
+func handleDoubleDataPoints(doubleDataPoints []*v1.DoubleDataPoint, lp *labelMetricProcessor) {
+ for _, dataPoint := range doubleDataPoints {
+ upsertLabels(&dataPoint.Labels, lp)
+ }
+}
+
+func handleIntHistogramDataPoints(intHistogramDataPoints []*v1.IntHistogramDataPoint, lp *labelMetricProcessor) {
+ for _, dataPoint := range intHistogramDataPoints {
+ upsertLabels(&dataPoint.Labels, lp)
+ }
+}
+
+func handleDoubleHistogramDataPoints(doubleHistogramDataPoints []*v1.DoubleHistogramDataPoint, lp *labelMetricProcessor) {
+ for _, dataPoint := range doubleHistogramDataPoints {
+ upsertLabels(&dataPoint.Labels, lp)
+ }
+
+}
+
+func upsertLabels(labels *[]*v11.StringKeyValue, lp *labelMetricProcessor) {
+ for _, label := range lp.cfg.Labels {
+ var updated bool = false
+ for _, elem := range *labels {
+ if elem.Key == label.Key {
+ elem.Value = label.Value
+ updated = true
+ break
+ }
+ }
+ if !updated {
+ *labels = append(*labels, &v11.StringKeyValue{Key: label.Key, Value: label.Value})
+ }
+ }
+}