Migrate Resource Processor to internal data model (open-telemetry#1315)

This commit migrates resource processor to internal data model. Existing processor configuration is relevant only to OpenCensus format, so the configuration schema has to be changed. Taking this opportunity, this commit adds existing logic for attributes manipulation from attributes processor to resource processor. 
New config uses "attributes" field which represents actions that can be made on resource attributes. Supported actions: INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT.
In order to migrate existing resource processor config:
1. Move key/values from "labels" field to "attributes" with action="upsert".
2. Add value from "type" field to "attributes" with key="opencensus.resourcetype"  and action="upsert".

internal/processor/attraction/attraction.go

@@ -26,7 +26,7 @@ import (
 // Settings
 type Settings struct {
 	// Actions specifies the list of attributes to act on.
-	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE}.
+	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT}.
 	// This is a required field.
 	Actions []ActionKeyValue `mapstructure:"actions"`
 }

processor/attributesprocessor/config.go

@@ -33,7 +33,7 @@ type Config struct {
 	filterspan.MatchConfig `mapstructure:",squash"`
 
 	// Specifies the list of attributes to act on.
-	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE}.
+	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT}.
 	// This is a required field.
 	attraction.Settings `mapstructure:",squash"`
 }

processor/resourceprocessor/README.md

@@ -2,25 +2,26 @@
 
 Supported pipeline types: metrics, traces
 
-The resource processor can be used to override a resource.
+The resource processor can be used to apply changes on resource attributes.
 Please refer to [config.go](./config.go) for the config spec.
 
-The following configuration options are required:
-- `type`: Resource type to be applied. If specified, this value overrides the
-original resource type. Otherwise, the original resource type is kept.
-- `labels`: Map of key/value pairs that should be added to the resource.
+`attributes` represents actions that can be applied on resource attributes.
+See processor/attributesprocessor/README.md for more details on supported attributes actions.
 
 Examples:
 
 ```yaml
 processors:
   resource:
-    type: "host"
-    labels: {
-      "cloud.zone": "zone-1",
-      "k8s.cluster.name": "k8s-cluster",
-      "host.name": "k8s-node",
-    }
+    attributes:
+    - key: cloud.zone
+      value: "zone-1"
+      action: upsert
+    - key: k8s.cluster.name
+      from_attribute: k8s-cluster
+      action: insert
+    - key: redundant-attribute
+      action: delete
 ```
 
 Refer to [config.yaml](./testdata/config.yaml) for detailed

processor/resourceprocessor/config.go

@@ -16,14 +16,20 @@ package resourceprocessor
 
 import (
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/internal/processor/attraction"
 )
 
 // Config defines configuration for Resource processor.
 type Config struct {
 	configmodels.ProcessorSettings `mapstructure:",squash"`
-	// ResourceType overrides the original resource type.
+
+	// AttributesActions specifies the list of actions to be applied on resource attributes.
+	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT}.
+	AttributesActions []attraction.ActionKeyValue `mapstructure:"attributes"`
+
+	// ResourceType field is deprecated. Set "opencensus.type" key in "attributes.upsert" map instead.
 	ResourceType string `mapstructure:"type"`
-	// Labels specify static labels to be added to resource.
-	// In case of a conflict the label will be overridden.
+
+	// Labels field is deprecated. Use "attributes.upsert" instead.
 	Labels map[string]string `mapstructure:"labels"`
 }

processor/resourceprocessor/config_test.go

@@ -22,33 +22,35 @@ import (
 
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/internal/processor/attraction"
 )
 
 func TestLoadConfig(t *testing.T) {
 	factories, err := config.ExampleComponents()
 	assert.NoError(t, err)
 
-	factory := &Factory{}
 	factories.Processors[typeStr] = &Factory{}
 
 	cfg, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
 	assert.NoError(t, err)
 	assert.NotNil(t, cfg)
 
-	p1 := cfg.Processors["resource"]
-	assert.Equal(t, p1, factory.CreateDefaultConfig())
-
-	p2 := cfg.Processors["resource/2"]
-	assert.Equal(t, p2, &Config{
+	assert.Equal(t, cfg.Processors["resource"], &Config{
 		ProcessorSettings: configmodels.ProcessorSettings{
 			TypeVal: "resource",
-			NameVal: "resource/2",
+			NameVal: "resource",
+		},
+		AttributesActions: []attraction.ActionKeyValue{
+			{Key: "cloud.zone", Value: "zone-1", Action: attraction.UPSERT},
+			{Key: "k8s.cluster.name", FromAttribute: "k8s-cluster", Action: attraction.INSERT},
+			{Key: "redundant-attribute", Action: attraction.DELETE},
 		},
-		ResourceType: "host",
-		Labels: map[string]string{
-			"cloud.zone":       "zone-1",
-			"k8s.cluster.name": "k8s-cluster",
-			"host.name":        "k8s-node",
+	})
+
+	assert.Equal(t, cfg.Processors["resource/invalid"], &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: "resource",
+			NameVal: "resource/invalid",
 		},
 	})
 }

processor/resourceprocessor/factory.go

@@ -15,11 +15,16 @@
 package resourceprocessor
 
 import (
+	"context"
+	"fmt"
+
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/internal/processor/attraction"
+	"go.opentelemetry.io/collector/translator/conventions"
 )
 
 const (
@@ -32,30 +37,83 @@ type Factory struct {
 }
 
 // Type gets the type of the Option config created by this factory.
-func (Factory) Type() configmodels.Type {
+func (*Factory) Type() configmodels.Type {
 	return typeStr
 }
 
 // CreateDefaultConfig creates the default configuration for processor.
-func (Factory) CreateDefaultConfig() configmodels.Processor {
+// Note: This isn't a valid configuration because the processor would do no work.
+func (*Factory) CreateDefaultConfig() configmodels.Processor {
 	return &Config{
 		ProcessorSettings: configmodels.ProcessorSettings{
 			TypeVal: typeStr,
 			NameVal: typeStr,
 		},
-		ResourceType: "",
-		Labels:       map[string]string{},
 	}
 }
 
 // CreateTraceProcessor creates a trace processor based on this config.
-func (Factory) CreateTraceProcessor(logger *zap.Logger, nextConsumer consumer.TraceConsumerOld, cfg configmodels.Processor) (component.TraceProcessorOld, error) {
-	oCfg := cfg.(*Config)
-	return newResourceTraceProcessor(nextConsumer, oCfg), nil
+func (*Factory) CreateTraceProcessor(
+	ctx context.Context,
+	params component.ProcessorCreateParams,
+	nextConsumer consumer.TraceConsumer,
+	cfg configmodels.Processor,
+) (component.TraceProcessor, error) {
+	attrProc, err := createAttrProcessor(cfg.(*Config), params.Logger)
+	if err != nil {
+		return nil, err
+	}
+	return newResourceTraceProcessor(nextConsumer, attrProc), nil
 }
 
 // CreateMetricsProcessor creates a metrics processor based on this config.
-func (Factory) CreateMetricsProcessor(logger *zap.Logger, nextConsumer consumer.MetricsConsumerOld, cfg configmodels.Processor) (component.MetricsProcessorOld, error) {
-	oCfg := cfg.(*Config)
-	return newResourceMetricProcessor(nextConsumer, oCfg), nil
+func (*Factory) CreateMetricsProcessor(
+	ctx context.Context,
+	params component.ProcessorCreateParams,
+	nextConsumer consumer.MetricsConsumer,
+	cfg configmodels.Processor,
+) (component.MetricsProcessor, error) {
+	attrProc, err := createAttrProcessor(cfg.(*Config), params.Logger)
+	if err != nil {
+		return nil, err
+	}
+	return newResourceMetricProcessor(nextConsumer, attrProc), nil
+}
+
+func createAttrProcessor(cfg *Config, logger *zap.Logger) (*attraction.AttrProc, error) {
+	handleDeprecatedFields(cfg, logger)
+	if len(cfg.AttributesActions) == 0 {
+		return nil, fmt.Errorf("error creating \"%q\" processor due to missing required field \"attributes\"", cfg.Name())
+	}
+	attrProc, err := attraction.NewAttrProc(&attraction.Settings{Actions: cfg.AttributesActions})
+	if err != nil {
+		return nil, fmt.Errorf("error creating \"%q\" processor: %w", cfg.Name(), err)
+	}
+	return attrProc, nil
+}
+
+// handleDeprecatedFields converts deprecated ResourceType and Labels fields into Attributes.Upsert
+func handleDeprecatedFields(cfg *Config, logger *zap.Logger) {
+
+	// Upsert value from deprecated ResourceType config to resource attributes with "opencensus.type" key
+	if cfg.ResourceType != "" {
+		logger.Warn("[DEPRECATED] \"type\" field is deprecated and will be removed in future release. " +
+			"Please set the value to \"attributes\" with key=opencensus.type and action=upsert.")
+		upsertResourceType := attraction.ActionKeyValue{
+			Action: attraction.UPSERT,
+			Key:    conventions.OCAttributeResourceType,
+			Value:  cfg.ResourceType,
+		}
+		cfg.AttributesActions = append(cfg.AttributesActions, upsertResourceType)
+	}
+
+	// Upsert values from deprecated Labels config to resource attributes
+	if len(cfg.Labels) > 0 {
+		logger.Warn("[DEPRECATED] \"labels\" field is deprecated and will be removed in future release. " +
+			"Please use \"attributes\" field instead.")
+		for k, v := range cfg.Labels {
+			action := attraction.ActionKeyValue{Action: attraction.UPSERT, Key: k, Value: v}
+			cfg.AttributesActions = append(cfg.AttributesActions, action)
+		}
+	}
 }

processor/resourceprocessor/factory_test.go

@@ -15,12 +15,16 @@
 package resourceprocessor
 
 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/internal/processor/attraction"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
@@ -32,13 +36,81 @@ func TestCreateDefaultConfig(t *testing.T) {
 
 func TestCreateProcessor(t *testing.T) {
 	var factory Factory
-	cfg := factory.CreateDefaultConfig()
+	cfg := &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: "resource",
+			NameVal: "resource",
+		},
+		AttributesActions: []attraction.ActionKeyValue{
+			{Key: "cloud.zone", Value: "zone-1", Action: attraction.UPSERT},
+		},
+	}
 
-	tp, err := factory.CreateTraceProcessor(zap.NewNop(), nil, cfg)
+	tp, err := factory.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
 	assert.NoError(t, err)
 	assert.NotNil(t, tp)
 
-	mp, err := factory.CreateMetricsProcessor(zap.NewNop(), nil, cfg)
+	mp, err := factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
 	assert.NoError(t, err)
 	assert.NotNil(t, mp)
 }
+
+func TestInvalidEmptyActions(t *testing.T) {
+	var factory Factory
+	cfg := factory.CreateDefaultConfig()
+
+	_, err := factory.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
+	assert.Error(t, err)
+
+	_, err = factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
+	assert.Error(t, err)
+}
+
+func TestInvalidAttributeActions(t *testing.T) {
+	var factory Factory
+	cfg := &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: "resource",
+			NameVal: "resource",
+		},
+		AttributesActions: []attraction.ActionKeyValue{
+			{Key: "k", Value: "v", Action: "invalid-action"},
+		},
+	}
+
+	_, err := factory.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
+	assert.Error(t, err)
+
+	_, err = factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, nil, cfg)
+	assert.Error(t, err)
+}
+
+func TestDeprecatedConfig(t *testing.T) {
+	cfg := &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: "resource",
+			NameVal: "resource",
+		},
+		ResourceType: "host",
+		Labels: map[string]string{
+			"cloud.zone": "zone-1",
+		},
+	}
+
+	handleDeprecatedFields(cfg, zap.NewNop())
+
+	assert.EqualValues(t, &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: "resource",
+			NameVal: "resource",
+		},
+		ResourceType: "host",
+		Labels: map[string]string{
+			"cloud.zone": "zone-1",
+		},
+		AttributesActions: []attraction.ActionKeyValue{
+			{Key: "opencensus.resourcetype", Value: "host", Action: attraction.UPSERT},
+			{Key: "cloud.zone", Value: "zone-1", Action: attraction.UPSERT},
+		},
+	}, cfg)
+}