Add processor helper. In the future add metrics and tracing (#1359)

Signed-off-by: Bogdan Drutu <[email protected]>

processor/attributesprocessor/attributes.go

@@ -17,39 +17,31 @@ package attributesprocessor
 import (
 	"context"
 
-	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/component/componenterror"
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/internal/processor/attraction"
 	"go.opentelemetry.io/collector/internal/processor/filterspan"
 	"go.opentelemetry.io/collector/processor"
 )
 
 type attributesProcessor struct {
-	nextConsumer consumer.TraceConsumer
-	attrProc     *attraction.AttrProc
-	include      filterspan.Matcher
-	exclude      filterspan.Matcher
+	attrProc *attraction.AttrProc
+	include  filterspan.Matcher
+	exclude  filterspan.Matcher
 }
 
 // newTraceProcessor returns a processor that modifies attributes of a span.
 // To construct the attributes processors, the use of the factory methods are required
 // in order to validate the inputs.
-func newTraceProcessor(nextConsumer consumer.TraceConsumer, attrProc *attraction.AttrProc, include, exclude filterspan.Matcher) (component.TraceProcessor, error) {
-	if nextConsumer == nil {
-		return nil, componenterror.ErrNilNextConsumer
+func newAttributesProcessor(attrProc *attraction.AttrProc, include, exclude filterspan.Matcher) *attributesProcessor {
+	return &attributesProcessor{
+		attrProc: attrProc,
+		include:  include,
+		exclude:  exclude,
 	}
-	ap := &attributesProcessor{
-		nextConsumer: nextConsumer,
-		attrProc:     attrProc,
-		include:      include,
-		exclude:      exclude,
-	}
-	return ap, nil
 }
 
-func (a *attributesProcessor) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
+// ProcessTraces implements the TProcessor
+func (a *attributesProcessor) ProcessTraces(_ context.Context, td pdata.Traces) (pdata.Traces, error) {
 	rss := td.ResourceSpans()
 	for i := 0; i < rss.Len(); i++ {
 		rs := rss.At(i)
@@ -79,21 +71,7 @@ func (a *attributesProcessor) ConsumeTraces(ctx context.Context, td pdata.Traces
 			}
 		}
 	}
-	return a.nextConsumer.ConsumeTraces(ctx, td)
-}
-
-func (a *attributesProcessor) GetCapabilities() component.ProcessorCapabilities {
-	return component.ProcessorCapabilities{MutatesConsumedData: true}
-}
-
-// Start is invoked during service startup.
-func (a *attributesProcessor) Start(_ context.Context, _ component.Host) error {
-	return nil
-}
-
-// Shutdown is invoked during service shutdown.
-func (a *attributesProcessor) Shutdown(context.Context) error {
-	return nil
+	return td, nil
 }
 
 // skipSpan determines if a span should be processed.

processor/attributesprocessor/factory.go

@@ -31,6 +31,8 @@ const (
 	typeStr = "attributes"
 )
 
+var processorCapabilities = component.ProcessorCapabilities{MutatesConsumedData: true}
+
 // NewFactory returns a new factory for the Attributes processor.
 func NewFactory() component.ProcessorFactory {
 	return processorhelper.NewFactory(
@@ -71,5 +73,10 @@ func createTraceProcessor(
 	if err != nil {
 		return nil, err
 	}
-	return newTraceProcessor(nextConsumer, attrProc, include, exclude)
+
+	return processorhelper.NewTraceProcessor(
+		cfg,
+		nextConsumer,
+		newAttributesProcessor(attrProc, include, exclude),
+		processorhelper.WithCapabilities(processorCapabilities))
 }

processor/attributesprocessor/factory_test.go

@@ -83,15 +83,15 @@ func TestFactoryCreateTraceProcessor(t *testing.T) {
 	tp, err = factory.CreateTraceProcessor(
 		context.Background(), component.ProcessorCreateParams{}, nil, cfg)
 	assert.Nil(t, tp)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 
 	oCfg.Actions = []attraction.ActionKeyValue{
 		{Action: attraction.DELETE},
 	}
 	tp, err = factory.CreateTraceProcessor(
 		context.Background(), component.ProcessorCreateParams{}, exportertest.NewNopTraceExporter(), cfg)
 	assert.Nil(t, tp)
-	assert.NotNil(t, err)
+	assert.Error(t, err)
 }
 
 func TestFactory_CreateMetricsProcessor(t *testing.T) {

processor/filterprocessor/factory.go

@@ -28,6 +28,8 @@ const (
 	typeStr = "filter"
 )
 
+var processorCapabilities = component.ProcessorCapabilities{MutatesConsumedData: false}
+
 // NewFactory returns a new factory for the Filter processor.
 func NewFactory() component.ProcessorFactory {
 	return processorhelper.NewFactory(
@@ -51,6 +53,13 @@ func createMetricsProcessor(
 	cfg configmodels.Processor,
 	nextConsumer consumer.MetricsConsumer,
 ) (component.MetricsProcessor, error) {
-	oCfg := cfg.(*Config)
-	return newFilterMetricProcessor(nextConsumer, oCfg)
+	fp, err := newFilterMetricProcessor(cfg.(*Config))
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewMetricsProcessor(
+		cfg,
+		nextConsumer,
+		fp,
+		processorhelper.WithCapabilities(processorCapabilities))
 }

processor/filterprocessor/factory_test.go

@@ -27,6 +27,7 @@ import (
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configcheck"
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/exporter/exportertest"
 )
 
 func TestType(t *testing.T) {
@@ -81,7 +82,7 @@ func TestCreateProcessors(t *testing.T) {
 				tp, tErr := factory.CreateTraceProcessor(
 					context.Background(),
 					component.ProcessorCreateParams{Logger: zap.NewNop()},
-					nil,
+					exportertest.NewNopTraceExporter(),
 					cfg)
 				// Not implemented error
 				assert.NotNil(t, tErr)
@@ -90,9 +91,9 @@ func TestCreateProcessors(t *testing.T) {
 				mp, mErr := factory.CreateMetricsProcessor(
 					context.Background(),
 					component.ProcessorCreateParams{Logger: zap.NewNop()},
-					nil,
+					exportertest.NewNopMetricsExporter(),
 					cfg)
-				assert.Equal(t, test.succeed, mp != (*filterMetricProcessor)(nil))
+				assert.Equal(t, test.succeed, mp != nil)
 				assert.Equal(t, test.succeed, mErr == nil)
 			})
 		}

processor/filterprocessor/filter_processor.go

@@ -19,23 +19,18 @@ import (
 
 	metricspb "github.com/census-instrumentation/opencensus-proto/gen-go/metrics/v1"
 
-	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/consumer/pdatautil"
 	"go.opentelemetry.io/collector/internal/processor/filtermetric"
 )
 
 type filterMetricProcessor struct {
 	cfg     *Config
-	next    consumer.MetricsConsumer
 	include *filtermetric.Matcher
 	exclude *filtermetric.Matcher
 }
 
-var _ component.MetricsProcessor = (*filterMetricProcessor)(nil)
-
-func newFilterMetricProcessor(next consumer.MetricsConsumer, cfg *Config) (*filterMetricProcessor, error) {
+func newFilterMetricProcessor(cfg *Config) (*filterMetricProcessor, error) {
 	inc, err := createMatcher(cfg.Metrics.Include)
 	if err != nil {
 		return nil, err
@@ -48,7 +43,6 @@ func newFilterMetricProcessor(next consumer.MetricsConsumer, cfg *Config) (*filt
 
 	return &filterMetricProcessor{
 		cfg:     cfg,
-		next:    next,
 		include: inc,
 		exclude: exc,
 	}, nil
@@ -68,28 +62,8 @@ func createMatcher(mp *filtermetric.MatchProperties) (*filtermetric.Matcher, err
 	return &matcher, nil
 }
 
-// GetCapabilities returns the Capabilities assocciated with the resource processor.
-func (fmp *filterMetricProcessor) GetCapabilities() component.ProcessorCapabilities {
-	return component.ProcessorCapabilities{MutatesConsumedData: false}
-}
-
-// Start is invoked during service startup.
-func (*filterMetricProcessor) Start(_ context.Context, _ component.Host) error {
-	return nil
-}
-
-// Shutdown is invoked during service shutdown.
-func (*filterMetricProcessor) Shutdown(_ context.Context) error {
-	return nil
-}
-
-// ConsumeMetricsData implements the MetricsProcessor interface
-func (fmp *filterMetricProcessor) ConsumeMetrics(ctx context.Context, md pdata.Metrics) error {
-	return fmp.next.ConsumeMetrics(ctx, fmp.filterMetrics(md))
-}
-
-// filterMetrics filters the given spans based off the filterMetricProcessor's filters.
-func (fmp *filterMetricProcessor) filterMetrics(md pdata.Metrics) pdata.Metrics {
+// ProcessMetrics filters the given spans based off the filterMetricProcessor's filters.
+func (fmp *filterMetricProcessor) ProcessMetrics(_ context.Context, md pdata.Metrics) (pdata.Metrics, error) {
 	mds := pdatautil.MetricsToMetricsData(md)
 	for i := range mds {
 		if len(mds[i].Metrics) == 0 {
@@ -103,7 +77,7 @@ func (fmp *filterMetricProcessor) filterMetrics(md pdata.Metrics) pdata.Metrics
 		}
 		mds[i].Metrics = keep
 	}
-	return pdatautil.MetricsFromMetricsData(mds)
+	return pdatautil.MetricsFromMetricsData(mds), nil
 }
 
 // shouldKeepMetric determines whether a metric should be kept based off the filterMetricProcessor's filters.

processor/filterprocessor/filter_processor_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer/consumerdata"
 	"go.opentelemetry.io/collector/consumer/pdatautil"
@@ -199,7 +200,8 @@ func TestFilterMetricProcessor(t *testing.T) {
 					Exclude: test.exc,
 				},
 			}
-			fmp, err := newFilterMetricProcessor(next, cfg)
+			factory := NewFactory()
+			fmp, err := factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, next, cfg)
 			assert.NotNil(t, fmp)
 			assert.Nil(t, err)
 
@@ -282,7 +284,8 @@ func BenchmarkFilter_MetricNames(b *testing.B) {
 				Exclude: test.exc,
 			},
 		}
-		fmp, err := newFilterMetricProcessor(next, cfg)
+		factory := NewFactory()
+		fmp, err := factory.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, next, cfg)
 		assert.NotNil(b, fmp)
 		assert.Nil(b, err)
 

processor/memorylimiter/factory.go

@@ -17,8 +17,6 @@ package memorylimiter
 import (
 	"context"
 
-	"go.uber.org/zap"
-
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/consumer"
@@ -30,6 +28,8 @@ const (
 	typeStr = "memory_limiter"
 )
 
+var processorCapabilities = component.ProcessorCapabilities{MutatesConsumedData: false}
+
 // NewFactory returns a new factory for the Memory Limiter processor.
 func NewFactory() component.ProcessorFactory {
 	return processorhelper.NewFactory(
@@ -57,7 +57,16 @@ func createTraceProcessor(
 	cfg configmodels.Processor,
 	nextConsumer consumer.TraceConsumer,
 ) (component.TraceProcessor, error) {
-	return createProcessor(params.Logger, nextConsumer, nil, nil, cfg)
+	ml, err := newMemoryLimiter(params.Logger, cfg.(*Config))
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewTraceProcessor(
+		cfg,
+		nextConsumer,
+		ml,
+		processorhelper.WithCapabilities(processorCapabilities),
+		processorhelper.WithShutdown(ml.shutdown))
 }
 
 func createMetricsProcessor(
@@ -66,7 +75,16 @@ func createMetricsProcessor(
 	cfg configmodels.Processor,
 	nextConsumer consumer.MetricsConsumer,
 ) (component.MetricsProcessor, error) {
-	return createProcessor(params.Logger, nil, nextConsumer, nil, cfg)
+	ml, err := newMemoryLimiter(params.Logger, cfg.(*Config))
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewMetricsProcessor(
+		cfg,
+		nextConsumer,
+		ml,
+		processorhelper.WithCapabilities(processorCapabilities),
+		processorhelper.WithShutdown(ml.shutdown))
 }
 
 func createLogProcessor(
@@ -75,29 +93,14 @@ func createLogProcessor(
 	cfg configmodels.Processor,
 	nextConsumer consumer.LogConsumer,
 ) (component.LogProcessor, error) {
-	return createProcessor(params.Logger, nil, nil, nextConsumer, cfg)
-}
-
-type TripleTypeProcessor interface {
-	consumer.TraceConsumer
-	consumer.MetricsConsumer
-	consumer.LogConsumer
-	component.Processor
-}
-
-func createProcessor(
-	logger *zap.Logger,
-	traceConsumer consumer.TraceConsumer,
-	metricConsumer consumer.MetricsConsumer,
-	logConsumer consumer.LogConsumer,
-	cfg configmodels.Processor,
-) (TripleTypeProcessor, error) {
-	pCfg := cfg.(*Config)
-	return newMemoryLimiter(
-		logger,
-		traceConsumer,
-		metricConsumer,
-		logConsumer,
-		pCfg,
-	)
+	ml, err := newMemoryLimiter(params.Logger, cfg.(*Config))
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewLogProcessor(
+		cfg,
+		nextConsumer,
+		ml,
+		processorhelper.WithCapabilities(processorCapabilities),
+		processorhelper.WithShutdown(ml.shutdown))
 }