Add implementation of Attributes processor config

- Added factory and config loading for Attributes processor.
- Added ability to create processors in ProcessorFactory interface.

Testing done: automated tests

Author: tigrannajaryan

internal/configv2/example_factories.go

@@ -20,6 +20,7 @@ import (
 	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/configmodels"
 	"github.com/census-instrumentation/opencensus-service/internal/factories"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 )
 
@@ -192,6 +193,22 @@ func (f *ExampleProcessorFactory) CreateDefaultConfig() configmodels.Processor {
 	}
 }
 
+// CreateTraceProcessor creates a trace processor based on this config.
+func (f *ExampleProcessorFactory) CreateTraceProcessor(
+	nextConsumer consumer.TraceConsumer,
+	cfg configmodels.Processor,
+) (processor.TraceProcessor, error) {
+	return nil, factories.ErrDataTypeIsNotSupported
+}
+
+// CreateMetricsProcessor creates a metrics processor based on this config.
+func (f *ExampleProcessorFactory) CreateMetricsProcessor(
+	nextConsumer consumer.MetricsConsumer,
+	cfg configmodels.Processor,
+) (processor.MetricsProcessor, error) {
+	return nil, factories.ErrDataTypeIsNotSupported
+}
+
 // RegisterTestFactories registers example factories. This is only used by tests.
 func RegisterTestFactories() error {
 	_ = factories.RegisterReceiverFactory(&ExampleReceiverFactory{})

internal/factories/factories.go

@@ -23,14 +23,14 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/configmodels"
+	"github.com/census-instrumentation/opencensus-service/processor"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 )
 
 ///////////////////////////////////////////////////////////////////////////////
 // Receiver factory and its registry.
 
-// ReceiverFactory is factory interface for receivers. Note: only configuration-related
-// functionality exists for now. We will add more factory functionality in the future.
+// ReceiverFactory is factory interface for receivers.
 type ReceiverFactory interface {
 	// Type gets the type of the Receiver created by this factory.
 	Type() string
@@ -115,32 +115,43 @@ func GetExporterFactory(typeStr string) ExporterFactory {
 }
 
 ///////////////////////////////////////////////////////////////////////////////
-// Option factory and its registry.
+// Processor factory and its registry.
 
-// OptionFactory is factory interface for options. Note: only configuration-related
-// functionality exists for now. We will add more factory functionality in the future.
-type OptionFactory interface {
-	// Type gets the type of the Option created by this factory.
+// ProcessorFactory is factory interface for processors.
+type ProcessorFactory interface {
+	// Type gets the type of the Processor created by this factory.
 	Type() string
 
-	// CreateDefaultConfig creates the default configuration for the Option.
+	// CreateDefaultConfig creates the default configuration for the Processor.
 	CreateDefaultConfig() configmodels.Processor
+
+	// CreateTraceProcessor creates a trace processor based on this config.
+	// If the processor type does not support tracing or if the config is not valid
+	// error will be returned instead.
+	CreateTraceProcessor(nextConsumer consumer.TraceConsumer,
+		cfg configmodels.Processor) (processor.TraceProcessor, error)
+
+	// CreateMetricsProcessor creates a metrics processor based on this config.
+	// If the processor type does not support metrics or if the config is not valid
+	// error will be returned instead.
+	CreateMetricsProcessor(nextConsumer consumer.MetricsConsumer,
+		cfg configmodels.Processor) (processor.MetricsProcessor, error)
 }
 
-// List of registered option factories.
-var optionFactories = make(map[string]OptionFactory)
+// List of registered processor factories.
+var processorFactories = make(map[string]ProcessorFactory)
 
-// RegisterProcessorFactory registers a option factory.
-func RegisterProcessorFactory(factory OptionFactory) error {
-	if optionFactories[factory.Type()] != nil {
-		panic(fmt.Sprintf("duplicate option factory %q", factory.Type()))
+// RegisterProcessorFactory registers a processor factory.
+func RegisterProcessorFactory(factory ProcessorFactory) error {
+	if processorFactories[factory.Type()] != nil {
+		panic(fmt.Sprintf("duplicate processor factory %q", factory.Type()))
 	}
 
-	optionFactories[factory.Type()] = factory
+	processorFactories[factory.Type()] = factory
 	return nil
 }
 
-// GetProcessorFactory gets a option factory by type string.
-func GetProcessorFactory(typeStr string) OptionFactory {
-	return optionFactories[typeStr]
+// GetProcessorFactory gets a processor factory by type string.
+func GetProcessorFactory(typeStr string) ProcessorFactory {
+	return processorFactories[typeStr]
 }

internal/factories/factories_test.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"testing"
 
+	"github.com/census-instrumentation/opencensus-service/processor"
+
 	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 
@@ -130,21 +132,37 @@ func TestRegisterExporterFactory(t *testing.T) {
 	}
 }
 
-type ExampleOptionFactory struct {
+type ExampleProcessorFactory struct {
 }
 
-// Type gets the type of the Option config created by this factory.
-func (f *ExampleOptionFactory) Type() string {
+// Type gets the type of the Processor config created by this factory.
+func (f *ExampleProcessorFactory) Type() string {
 	return "exampleoption"
 }
 
 // CreateDefaultConfig creates the default configuration for the Processor.
-func (f *ExampleOptionFactory) CreateDefaultConfig() configmodels.Processor {
+func (f *ExampleProcessorFactory) CreateDefaultConfig() configmodels.Processor {
 	return nil
 }
 
-func TestRegisterOptionFactory(t *testing.T) {
-	f := ExampleOptionFactory{}
+// CreateTraceProcessor creates a trace processor based on this config.
+func (f *ExampleProcessorFactory) CreateTraceProcessor(
+	nextConsumer consumer.TraceConsumer,
+	cfg configmodels.Processor,
+) (processor.TraceProcessor, error) {
+	return nil, ErrDataTypeIsNotSupported
+}
+
+// CreateMetricsProcessor creates a metrics processor based on this config.
+func (f *ExampleProcessorFactory) CreateMetricsProcessor(
+	nextConsumer consumer.MetricsConsumer,
+	cfg configmodels.Processor,
+) (processor.MetricsProcessor, error) {
+	return nil, ErrDataTypeIsNotSupported
+}
+
+func TestRegisterProcessorFactory(t *testing.T) {
+	f := ExampleProcessorFactory{}
 	err := RegisterProcessorFactory(&f)
 	if err != nil {
 		t.Fatalf("cannot register factory")

processor/addattributesprocessor/config.go

@@ -0,0 +1,26 @@
+// Copyright 2019, OpenCensus 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 addattributesprocessor
+
+import (
+	"github.com/census-instrumentation/opencensus-service/internal/configmodels"
+)
+
+// ConfigV2 defines configuration for Attributes processor.
+type ConfigV2 struct {
+	configmodels.ProcessorSettings `mapstructure:",squash"`
+	Overwrite                      bool                   `mapstructure:"overwrite"`
+	Values                         map[string]interface{} `mapstructure:"values"`
+}

processor/addattributesprocessor/config_test.go

@@ -0,0 +1,51 @@
+// Copyright 2019, OpenCensus 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 addattributesprocessor
+
+import (
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/census-instrumentation/opencensus-service/internal/configv2"
+	"github.com/census-instrumentation/opencensus-service/internal/factories"
+)
+
+var _ = configv2.RegisterTestFactories()
+
+func TestLoadConfig(t *testing.T) {
+
+	factory := factories.GetProcessorFactory(typeStr)
+
+	config, err := configv2.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"))
+
+	require.Nil(t, err)
+	require.NotNil(t, config)
+
+	p0 := config.Processors["attributes"]
+	assert.Equal(t, p0, factory.CreateDefaultConfig())
+
+	p1 := config.Processors["attributes/2"]
+	assert.Equal(t, p1,
+		&ConfigV2{
+			Values: map[string]interface{}{
+				"attribute1":         123,
+				"string attribute":   "string value",
+				"attribute.with.dot": "another value",
+			},
+		})
+}

processor/addattributesprocessor/factory.go

@@ -0,0 +1,62 @@
+// Copyright 2019, OpenCensus 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 addattributesprocessor
+
+import (
+	"github.com/census-instrumentation/opencensus-service/consumer"
+	"github.com/census-instrumentation/opencensus-service/internal/configmodels"
+	"github.com/census-instrumentation/opencensus-service/internal/factories"
+	"github.com/census-instrumentation/opencensus-service/processor"
+)
+
+var _ = factories.RegisterProcessorFactory(&processorFactory{})
+
+const (
+	// The value of "type" key in configuration.
+	typeStr = "attributes"
+)
+
+// processorFactory is the factory for OpenCensus exporter.
+type processorFactory struct {
+}
+
+// Type gets the type of the Option config created by this factory.
+func (f *processorFactory) Type() string {
+	return typeStr
+}
+
+// CreateDefaultConfig creates the default configuration for exporter.
+func (f *processorFactory) CreateDefaultConfig() configmodels.Processor {
+	return &ConfigV2{
+		Values: map[string]interface{}{},
+	}
+}
+
+// CreateTraceProcessor creates a trace processor based on this config.
+func (f *processorFactory) CreateTraceProcessor(
+	nextConsumer consumer.TraceConsumer,
+	cfg configmodels.Processor,
+) (processor.TraceProcessor, error) {
+	oCfg := cfg.(*ConfigV2)
+	return NewTraceProcessor(nextConsumer, WithAttributes(oCfg.Values), WithOverwrite(oCfg.Overwrite))
+}
+
+// CreateMetricsProcessor creates a metrics processor based on this config.
+func (f *processorFactory) CreateMetricsProcessor(
+	nextConsumer consumer.MetricsConsumer,
+	cfg configmodels.Processor,
+) (processor.MetricsProcessor, error) {
+	return nil, factories.ErrDataTypeIsNotSupported
+}

processor/addattributesprocessor/factory_test.go

@@ -0,0 +1,47 @@
+// Copyright 2019, OpenCensus 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 addattributesprocessor
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/census-instrumentation/opencensus-service/internal/factories"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	factory := factories.GetProcessorFactory(typeStr)
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+	assert.NotNil(t, cfg, "failed to create default config")
+}
+
+func TestCreateProcessor(t *testing.T) {
+	factory := factories.GetProcessorFactory(typeStr)
+	require.NotNil(t, factory)
+
+	cfg := factory.CreateDefaultConfig()
+
+	tp, err := factory.CreateTraceProcessor(nil, cfg)
+	assert.NotNil(t, tp)
+	assert.NoError(t, err, "cannot create trace processor")
+
+	mp, err := factory.CreateMetricsProcessor(nil, cfg)
+	assert.Nil(t, mp)
+	assert.Error(t, err, "should not be able to create metric processor")
+}

processor/addattributesprocessor/testdata/config.yaml

@@ -0,0 +1,19 @@
+receivers:
+  examplereceiver:
+
+processors:
+  attributes:
+  attributes/2:
+    values:
+      attribute1: 123
+      "string attribute": "string value"
+      "attribute.with.dot": "another value"
+
+exporters:
+  exampleexporter:
+
+pipelines:
+  traces:
+    receivers: [examplereceiver]
+    processors: [attributes]
+    exporters: [exampleexporter]