Add helper processor factory to reduce boilerplate (#1339)

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

processor/attributesprocessor/attributes_test.go

@@ -135,7 +135,7 @@ func TestSpanProcessor_NilEmptyData(t *testing.T) {
 			output: testdata.GenerateTraceDataOneEmptyOneNilInstrumentationLibrary(),
 		},
 	}
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Settings.Actions = []attraction.ActionKeyValue{
@@ -195,7 +195,7 @@ func TestAttributes_FilterSpans(t *testing.T) {
 		},
 	}
 
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{
@@ -265,7 +265,7 @@ func TestAttributes_FilterSpansByNameStrict(t *testing.T) {
 		},
 	}
 
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{
@@ -333,7 +333,7 @@ func TestAttributes_FilterSpansByNameRegexp(t *testing.T) {
 		},
 	}
 
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{
@@ -396,7 +396,7 @@ func TestAttributes_Hash(t *testing.T) {
 		},
 	}
 
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{
@@ -441,7 +441,7 @@ func BenchmarkAttributes_FilterSpansByName(b *testing.B) {
 		},
 	}
 
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{

processor/attributesprocessor/config_test.go

@@ -32,7 +32,7 @@ func TestLoadingConifg(t *testing.T) {
 	factories, err := config.ExampleComponents()
 	assert.NoError(t, err)
 
-	factory := &Factory{}
+	factory := NewFactory()
 	factories.Processors[typeStr] = factory
 	cfg, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
 	assert.NoError(t, err)

processor/attributesprocessor/factory.go

@@ -19,30 +19,28 @@ import (
 	"fmt"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/config/configerror"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/internal/processor/attraction"
 	"go.opentelemetry.io/collector/internal/processor/filterspan"
+	"go.opentelemetry.io/collector/processor/processorhelper"
 )
 
 const (
 	// typeStr is the value of "type" key in configuration.
 	typeStr = "attributes"
 )
 
-// Factory is the factory for Attributes processor.
-type Factory struct {
+// NewFactory returns a new factory for the Attributes processor.
+func NewFactory() component.ProcessorFactory {
+	return processorhelper.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		processorhelper.WithTraceProcessor(createTraceProcessor))
 }
 
-// Type gets the type of the config created by this factory.
-func (f *Factory) Type() configmodels.Type {
-	return typeStr
-}
-
-// CreateDefaultConfig creates the default configuration for the processor.
 // Note: This isn't a valid configuration because the processor would do no work.
-func (f *Factory) CreateDefaultConfig() configmodels.Processor {
+func createDefaultConfig() configmodels.Processor {
 	return &Config{
 		ProcessorSettings: configmodels.ProcessorSettings{
 			TypeVal: typeStr,
@@ -51,14 +49,12 @@ func (f *Factory) CreateDefaultConfig() configmodels.Processor {
 	}
 }
 
-// CreateTraceProcessor creates a trace processor based on this config.
-func (f *Factory) CreateTraceProcessor(
+func createTraceProcessor(
 	_ context.Context,
 	_ component.ProcessorCreateParams,
-	nextConsumer consumer.TraceConsumer,
 	cfg configmodels.Processor,
+	nextConsumer consumer.TraceConsumer,
 ) (component.TraceProcessor, error) {
-
 	oCfg := cfg.(*Config)
 	if len(oCfg.Actions) == 0 {
 		return nil, fmt.Errorf("error creating \"attributes\" processor due to missing required field \"actions\" of processor %q", cfg.Name())
@@ -77,13 +73,3 @@ func (f *Factory) CreateTraceProcessor(
 	}
 	return newTraceProcessor(nextConsumer, attrProc, include, exclude)
 }
-
-// CreateMetricsProcessor creates a metrics processor based on this config.
-func (f *Factory) CreateMetricsProcessor(
-	_ context.Context,
-	_ component.ProcessorCreateParams,
-	_ consumer.MetricsConsumer,
-	_ configmodels.Processor,
-) (component.MetricsProcessor, error) {
-	return nil, configerror.ErrDataTypeIsNotSupported
-}

processor/attributesprocessor/factory_test.go

@@ -30,12 +30,12 @@ import (
 )
 
 func TestFactory_Type(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	assert.Equal(t, factory.Type(), configmodels.Type(typeStr))
 }
 
 func TestFactory_CreateDefaultConfig(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	assert.Equal(t, cfg, &Config{
 		ProcessorSettings: configmodels.ProcessorSettings{
@@ -47,15 +47,15 @@ func TestFactory_CreateDefaultConfig(t *testing.T) {
 }
 
 func TestFactoryCreateTraceProcessor_EmptyActions(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	ap, err := factory.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{}, exportertest.NewNopTraceExporter(), cfg)
 	assert.Error(t, err)
 	assert.Nil(t, ap)
 }
 
 func TestFactoryCreateTraceProcessor_InvalidActions(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	// Missing key
@@ -68,7 +68,7 @@ func TestFactoryCreateTraceProcessor_InvalidActions(t *testing.T) {
 }
 
 func TestFactoryCreateTraceProcessor(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 	oCfg := cfg.(*Config)
 	oCfg.Actions = []attraction.ActionKeyValue{
@@ -95,7 +95,7 @@ func TestFactoryCreateTraceProcessor(t *testing.T) {
 }
 
 func TestFactory_CreateMetricsProcessor(t *testing.T) {
-	factory := Factory{}
+	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 
 	mp, err := factory.CreateMetricsProcessor(

processor/batchprocessor/batch_processor_test.go

@@ -36,7 +36,7 @@ import (
 
 func TestBatchProcessorSpansDelivered(t *testing.T) {
 	sink := &exportertest.SinkTraceExporter{}
-	cfg := generateDefaultConfig()
+	cfg := createDefaultConfig().(*Config)
 	cfg.SendBatchSize = 128
 	creationParams := component.ProcessorCreateParams{Logger: zap.NewNop()}
 	batcher := newBatchTracesProcessor(creationParams, sink, cfg)
@@ -80,7 +80,7 @@ func TestBatchProcessorSentBySize(t *testing.T) {
 	defer view.Unregister(views...)
 
 	sink := &exportertest.SinkTraceExporter{}
-	cfg := generateDefaultConfig()
+	cfg := createDefaultConfig().(*Config)
 	sendBatchSize := 20
 	cfg.SendBatchSize = uint32(sendBatchSize)
 	cfg.Timeout = 500 * time.Millisecond
@@ -128,7 +128,7 @@ func TestBatchProcessorSentBySize(t *testing.T) {
 
 func TestBatchProcessorSentByTimeout(t *testing.T) {
 	sink := &exportertest.SinkTraceExporter{}
-	cfg := generateDefaultConfig()
+	cfg := createDefaultConfig().(*Config)
 	sendBatchSize := 100
 	cfg.SendBatchSize = uint32(sendBatchSize)
 	cfg.Timeout = 100 * time.Millisecond

processor/batchprocessor/config_test.go

@@ -30,8 +30,8 @@ func TestLoadConfig(t *testing.T) {
 	factories, err := config.ExampleComponents()
 	assert.NoError(t, err)
 
-	factory := &Factory{}
-	factories.Processors[typeStr] = &Factory{}
+	factory := NewFactory()
+	factories.Processors[typeStr] = factory
 	cfg, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
 
 	require.Nil(t, err)

processor/batchprocessor/factory.go

@@ -21,6 +21,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor/processorhelper"
 )
 
 const (
@@ -31,49 +32,42 @@ const (
 	defaultTimeout       = 200 * time.Millisecond
 )
 
-// Factory is the factory for batch processor.
-type Factory struct {
+// NewFactory returns a new factory for the Batch processor.
+func NewFactory() component.ProcessorFactory {
+	return processorhelper.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		processorhelper.WithTraceProcessor(createTraceProcessor),
+		processorhelper.WithMetricsProcessor(createMetricsProcessor))
 }
 
-// Type gets the type of the config created by this factory.
-func (f *Factory) Type() configmodels.Type {
-	return typeStr
-}
-
-// CreateDefaultConfig creates the default configuration for processor.
-func (f *Factory) CreateDefaultConfig() configmodels.Processor {
-	return generateDefaultConfig()
+func createDefaultConfig() configmodels.Processor {
+	return &Config{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: typeStr,
+			NameVal: typeStr,
+		},
+		SendBatchSize: defaultSendBatchSize,
+		Timeout:       defaultTimeout,
+	}
 }
 
-// CreateTraceProcessor creates a trace processor based on this config.
-func (f *Factory) CreateTraceProcessor(
-	ctx context.Context,
+func createTraceProcessor(
+	_ context.Context,
 	params component.ProcessorCreateParams,
+	cfg configmodels.Processor,
 	nextConsumer consumer.TraceConsumer,
-	c configmodels.Processor,
 ) (component.TraceProcessor, error) {
-	cfg := c.(*Config)
-	return newBatchTracesProcessor(params, nextConsumer, cfg), nil
+	oCfg := cfg.(*Config)
+	return newBatchTracesProcessor(params, nextConsumer, oCfg), nil
 }
 
-// CreateMetricsProcessor creates a metrics processor based on this config.
-func (f *Factory) CreateMetricsProcessor(
-	ctx context.Context,
+func createMetricsProcessor(
+	_ context.Context,
 	params component.ProcessorCreateParams,
+	cfg configmodels.Processor,
 	nextConsumer consumer.MetricsConsumer,
-	c configmodels.Processor,
 ) (component.MetricsProcessor, error) {
-	cfg := c.(*Config)
-	return newBatchMetricsProcessor(params, nextConsumer, cfg), nil
-}
-
-func generateDefaultConfig() *Config {
-	return &Config{
-		ProcessorSettings: configmodels.ProcessorSettings{
-			TypeVal: typeStr,
-			NameVal: typeStr,
-		},
-		SendBatchSize: defaultSendBatchSize,
-		Timeout:       defaultTimeout,
-	}
+	oCfg := cfg.(*Config)
+	return newBatchMetricsProcessor(params, nextConsumer, oCfg), nil
 }

processor/batchprocessor/factory_test.go

@@ -26,15 +26,15 @@ import (
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
-	factory := &Factory{}
+	factory := NewFactory()
 
 	cfg := factory.CreateDefaultConfig()
 	assert.NotNil(t, cfg, "failed to create default config")
 	assert.NoError(t, configcheck.ValidateConfig(cfg))
 }
 
 func TestCreateProcessor(t *testing.T) {
-	factory := &Factory{}
+	factory := NewFactory()
 
 	cfg := factory.CreateDefaultConfig()
 	creationParams := component.ProcessorCreateParams{Logger: zap.NewNop()}

processor/filterprocessor/config_test.go

@@ -46,7 +46,7 @@ func TestLoadingConfigStrict(t *testing.T) {
 	factories, err := config.ExampleComponents()
 	assert.Nil(t, err)
 
-	factory := &Factory{}
+	factory := NewFactory()
 	factories.Processors[configmodels.Type(typeStr)] = factory
 	config, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config_strict.yaml"), factories)
 
@@ -146,7 +146,7 @@ func TestLoadingConfigRegexp(t *testing.T) {
 	factories, err := config.ExampleComponents()
 	assert.Nil(t, err)
 
-	factory := &Factory{}
+	factory := NewFactory()
 	factories.Processors[typeStr] = factory
 	config, err := config.LoadConfigFile(t, path.Join(".", "testdata", "config_regexp.yaml"), factories)