Add factories and component types to support logs data type

We are planning on experimenting with logs data type in the Collector.
This commit introduces factories and components types for logs and
implements pipeline building for logs.

The data.Log struct is for now empty. We need to wait for OTLP protocol
to define the log data format and we will use it in data.Log.

data.Log is in internal package since logs support is experimental
and is not intended for public usage.

Resolves #957

Author: Tigran Najaryan

component/exporter.go

@@ -62,6 +62,12 @@ type MetricsExporter interface {
 	MetricsExporterBase
 }
 
+// LogExporter is an LogConsumer that is also an Exporter.
+type LogExporter interface {
+	Exporter
+	consumer.LogConsumer
+}
+
 // ExporterFactoryBase defines the common functions for all exporter factories.
 type ExporterFactoryBase interface {
 	Factory
@@ -111,3 +117,17 @@ type ExporterFactory interface {
 	CreateMetricsExporter(ctx context.Context, params ExporterCreateParams,
 		cfg configmodels.Exporter) (MetricsExporter, error)
 }
+
+// LogExporterFactory can create Exporter.
+type LogExporterFactory interface {
+	ExporterFactoryBase
+
+	// CreateLogExporter creates an exporter based on this config.
+	// If the exporter type does not support the data type or if the config is not valid
+	// error will be returned instead.
+	CreateLogExporter(
+		ctx context.Context,
+		params ExporterCreateParams,
+		cfg configmodels.Exporter,
+	) (LogExporter, error)
+}

component/processor.go

@@ -66,6 +66,12 @@ type MetricsProcessor interface {
 	MetricsProcessorBase
 }
 
+// LogProcessor composes LogConsumer with some additional processor-specific functions.
+type LogProcessor interface {
+	Processor
+	consumer.LogConsumer
+}
+
 // ProcessorCapabilities describes the capabilities of a Processor.
 type ProcessorCapabilities struct {
 	// MutatesConsumedData is set to true if Consume* function of the
@@ -131,3 +137,18 @@ type ProcessorFactory interface {
 	CreateMetricsProcessor(ctx context.Context, params ProcessorCreateParams,
 		nextConsumer consumer.MetricsConsumer, cfg configmodels.Processor) (MetricsProcessor, error)
 }
+
+// LogProcessorFactory is factory interface for processors.
+type LogProcessorFactory interface {
+	ProcessorFactoryBase
+
+	// CreateLogProcessor creates a processor based on this config.
+	// If the processor type does not support the data type or if the config is not valid
+	// error will be returned instead.
+	CreateLogProcessor(
+		ctx context.Context,
+		params ProcessorCreateParams,
+		cfg configmodels.Processor,
+		nextConsumer consumer.LogConsumer,
+	) (LogProcessor, error)
+}

component/receiver.go

@@ -49,6 +49,13 @@ type MetricsReceiver interface {
 	Receiver
 }
 
+// A LogReceiver is a "log data"-to-"internal format" converter.
+// Its purpose is to translate data from the wild into internal data format.
+// LogReceiver feeds a consumer.LogConsumer with data.
+type LogReceiver interface {
+	Receiver
+}
+
 // ReceiverFactoryBase defines the common functions for all receiver factories.
 type ReceiverFactoryBase interface {
 	Factory
@@ -117,3 +124,18 @@ type ReceiverFactory interface {
 	CreateMetricsReceiver(ctx context.Context, params ReceiverCreateParams,
 		cfg configmodels.Receiver, nextConsumer consumer.MetricsConsumer) (MetricsReceiver, error)
 }
+
+// LogReceiverFactory can create a LogReceiver.
+type LogReceiverFactory interface {
+	ReceiverFactoryBase
+
+	// CreateLogReceiver creates a log receiver based on this config.
+	// If the receiver type does not support the data type or if the config is not valid
+	// error will be returned instead.
+	CreateLogReceiver(
+		ctx context.Context,
+		params ReceiverCreateParams,
+		cfg configmodels.Receiver,
+		nextConsumer consumer.LogConsumer,
+	) (LogReceiver, error)
+}

config/config.go

@@ -539,17 +539,7 @@ func loadPipelines(v *viper.Viper) (configmodels.Pipelines, error) {
 		var pipelineCfg configmodels.Pipeline
 
 		// Set the type.
-		switch typeStr {
-		case configmodels.TracesDataTypeStr:
-			pipelineCfg.InputType = configmodels.TracesDataType
-		case configmodels.MetricsDataTypeStr:
-			pipelineCfg.InputType = configmodels.MetricsDataType
-		default:
-			return nil, &configError{
-				code: errInvalidPipelineType,
-				msg:  fmt.Sprintf("invalid pipeline type %q (must be metrics or traces)", typeStr),
-			}
-		}
+		pipelineCfg.InputType = configmodels.DataType(typeStr)
 
 		pipelineConfig := ViperSub(pipelinesConfig, key)
 

config/config_test.go

@@ -100,7 +100,7 @@ func TestDecodeConfig(t *testing.T) {
 	assert.Equal(t, 1, len(config.Processors), "Incorrect processors count")
 
 	assert.Equal(t,
-		&ExampleProcessor{
+		&ExampleProcessorCfg{
 			ProcessorSettings: configmodels.ProcessorSettings{
 				TypeVal: "exampleprocessor",
 				NameVal: "exampleprocessor",
@@ -258,7 +258,7 @@ func TestSimpleConfig(t *testing.T) {
 		assert.Equalf(t, 1, len(config.Processors), "TEST[%s]", test.name)
 
 		assert.Equalf(t,
-			&ExampleProcessor{
+			&ExampleProcessorCfg{
 				ProcessorSettings: configmodels.ProcessorSettings{
 					TypeVal: "exampleprocessor",
 					NameVal: "exampleprocessor",
@@ -367,7 +367,6 @@ func TestDecodeConfig_Invalid(t *testing.T) {
 		{name: "unknown-processor-type", expected: errUnknownProcessorType},
 		{name: "invalid-service-extensions-value", expected: errUnmarshalErrorOnService},
 		{name: "invalid-sequence-value", expected: errUnmarshalErrorOnPipeline},
-		{name: "invalid-pipeline-type", expected: errInvalidPipelineType},
 		{name: "invalid-pipeline-type-and-name", expected: errInvalidTypeAndNameKey},
 		{name: "duplicate-extension", expected: errDuplicateExtensionName},
 		{name: "duplicate-receiver", expected: errDuplicateReceiverName},

config/configmodels/configmodels.go

@@ -79,37 +79,21 @@ type Processors map[string]Processor
 
 // DataType is the data type that is supported for collection. We currently support
 // collecting metrics and traces, this can expand in the future (e.g. logs, events, etc).
-type DataType int
+
+type DataType string
 
 // Currently supported data types. Add new data types here when new types are supported in the future.
 const (
-	_ DataType = iota // skip 0, start types from 1.
-
 	// TracesDataType is the data type tag for traces.
-	TracesDataType
+	TracesDataType DataType = "traces"
 
 	// MetricsDataType is the data type tag for metrics.
-	MetricsDataType
-)
+	MetricsDataType DataType = "metrics"
 
-// Data type strings.
-const (
-	TracesDataTypeStr  = "traces"
-	MetricsDataTypeStr = "metrics"
+	// LogsDataType is the data type tag for logs.
+	LogsDataType DataType = "logs"
 )
 
-// GetString converts data type to string.
-func (dataType DataType) GetString() string {
-	switch dataType {
-	case TracesDataType:
-		return TracesDataTypeStr
-	case MetricsDataType:
-		return MetricsDataTypeStr
-	default:
-		panic("unknown data type")
-	}
-}
-
 // Pipeline defines a single pipeline.
 type Pipeline struct {
 	Name       string   `mapstructure:"-"`

config/example_factories.go

@@ -25,6 +25,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
 	"github.com/open-telemetry/opentelemetry-collector/consumer"
 	"github.com/open-telemetry/opentelemetry-collector/consumer/consumerdata"
+	"github.com/open-telemetry/opentelemetry-collector/internal/data"
 )
 
 // ExampleReceiver is for testing purposes. We are defining an example config and factory
@@ -60,7 +61,8 @@ func (f *ExampleReceiverFactory) CustomUnmarshaler() component.CustomUnmarshaler
 func (f *ExampleReceiverFactory) CreateDefaultConfig() configmodels.Receiver {
 	return &ExampleReceiver{
 		ReceiverSettings: configmodels.ReceiverSettings{
-			TypeVal:  "examplereceiver",
+			TypeVal:  f.Type(),
+			NameVal:  string(f.Type()),
 			Endpoint: "localhost:1000",
 		},
 		ExtraSetting:     "some string",
@@ -80,7 +82,7 @@ func (f *ExampleReceiverFactory) CreateTraceReceiver(
 		return nil, configerror.ErrDataTypeIsNotSupported
 	}
 
-	// There must be one receiver for both metrics and traces. We maintain a map of
+	// There must be one receiver for all data types. We maintain a map of
 	// receivers per config.
 
 	// Check to see if there is already a receiver for this config.
@@ -105,7 +107,7 @@ func (f *ExampleReceiverFactory) CreateMetricsReceiver(
 		return nil, configerror.ErrDataTypeIsNotSupported
 	}
 
-	// There must be one receiver for both metrics and traces. We maintain a map of
+	// There must be one receiver for all data types. We maintain a map of
 	// receivers per config.
 
 	// Check to see if there is already a receiver for this config.
@@ -120,12 +122,35 @@ func (f *ExampleReceiverFactory) CreateMetricsReceiver(
 	return receiver, nil
 }
 
+func (f *ExampleReceiverFactory) CreateLogReceiver(
+	ctx context.Context,
+	params component.ReceiverCreateParams,
+	cfg configmodels.Receiver,
+	nextConsumer consumer.LogConsumer,
+) (component.LogReceiver, error) {
+	// There must be one receiver for all data types. We maintain a map of
+	// receivers per config.
+
+	// Check to see if there is already a receiver for this config.
+	receiver, ok := exampleReceivers[cfg]
+	if !ok {
+		receiver = &ExampleReceiverProducer{}
+		// Remember the receiver in the map
+		exampleReceivers[cfg] = receiver
+	}
+	receiver.LogConsumer = nextConsumer
+
+	return receiver, nil
+
+}
+
 // ExampleReceiverProducer allows producing traces and metrics for testing purposes.
 type ExampleReceiverProducer struct {
-	TraceConsumer   consumer.TraceConsumerOld
 	Started         bool
 	Stopped         bool
+	TraceConsumer   consumer.TraceConsumerOld
 	MetricsConsumer consumer.MetricsConsumerOld
+	LogConsumer     consumer.LogConsumer
 }
 
 // Start tells the receiver to start its processing.
@@ -199,7 +224,8 @@ func (f *MultiProtoReceiverFactory) CustomUnmarshaler() component.CustomUnmarsha
 // CreateDefaultConfig creates the default configuration for the Receiver.
 func (f *MultiProtoReceiverFactory) CreateDefaultConfig() configmodels.Receiver {
 	return &MultiProtoReceiver{
-		TypeVal: "multireceiver",
+		TypeVal: f.Type(),
+		NameVal: string(f.Type()),
 		Protocols: map[string]MultiProtoReceiverOneCfg{
 			"http": {
 				Endpoint:     "example.com:8888",
@@ -256,7 +282,10 @@ func (f *ExampleExporterFactory) Type() configmodels.Type {
 // CreateDefaultConfig creates the default configuration for the Exporter.
 func (f *ExampleExporterFactory) CreateDefaultConfig() configmodels.Exporter {
 	return &ExampleExporter{
-		ExporterSettings: configmodels.ExporterSettings{TypeVal: f.Type()},
+		ExporterSettings: configmodels.ExporterSettings{
+			TypeVal: f.Type(),
+			NameVal: string(f.Type()),
+		},
 		ExtraSetting:     "some export string",
 		ExtraMapSetting:  nil,
 		ExtraListSetting: nil,
@@ -273,10 +302,19 @@ func (f *ExampleExporterFactory) CreateMetricsExporter(logger *zap.Logger, cfg c
 	return &ExampleExporterConsumer{}, nil
 }
 
+func (f *ExampleExporterFactory) CreateLogExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogExporter, error) {
+	return &ExampleExporterConsumer{}, nil
+}
+
 // ExampleExporterConsumer stores consumed traces and metrics for testing purposes.
 type ExampleExporterConsumer struct {
 	Traces           []consumerdata.TraceData
 	Metrics          []consumerdata.MetricsData
+	Logs             []data.Logs
 	ExporterStarted  bool
 	ExporterShutdown bool
 }
@@ -301,6 +339,11 @@ func (exp *ExampleExporterConsumer) ConsumeMetricsData(ctx context.Context, md c
 	return nil
 }
 
+func (exp *ExampleExporterConsumer) ConsumeLogs(ctx context.Context, ld data.Logs) error {
+	exp.Logs = append(exp.Logs, ld)
+	return nil
+}
+
 // Name returns the name of the exporter.
 func (exp *ExampleExporterConsumer) Name() string {
 	return "exampleexporter"
@@ -312,9 +355,9 @@ func (exp *ExampleExporterConsumer) Shutdown(context.Context) error {
 	return nil
 }
 
-// ExampleProcessor is for testing purposes. We are defining an example config and factory
+// ExampleProcessorCfg is for testing purposes. We are defining an example config and factory
 // for "exampleprocessor" processor type.
-type ExampleProcessor struct {
+type ExampleProcessorCfg struct {
 	configmodels.ProcessorSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
 	ExtraSetting                   string                   `mapstructure:"extra"`
 	ExtraMapSetting                map[string]string        `mapstructure:"extra_map"`
@@ -332,11 +375,14 @@ func (f *ExampleProcessorFactory) Type() configmodels.Type {
 
 // CreateDefaultConfig creates the default configuration for the Processor.
 func (f *ExampleProcessorFactory) CreateDefaultConfig() configmodels.Processor {
-	return &ExampleProcessor{
-		ProcessorSettings: configmodels.ProcessorSettings{},
-		ExtraSetting:      "some export string",
-		ExtraMapSetting:   nil,
-		ExtraListSetting:  nil,
+	return &ExampleProcessorCfg{
+		ProcessorSettings: configmodels.ProcessorSettings{
+			TypeVal: f.Type(),
+			NameVal: string(f.Type()),
+		},
+		ExtraSetting:     "some export string",
+		ExtraMapSetting:  nil,
+		ExtraListSetting: nil,
 	}
 }
 
@@ -358,6 +404,35 @@ func (f *ExampleProcessorFactory) CreateMetricsProcessor(
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
 
+func (f *ExampleProcessorFactory) CreateLogProcessor(
+	ctx context.Context,
+	params component.ProcessorCreateParams,
+	cfg configmodels.Processor,
+	nextConsumer consumer.LogConsumer,
+) (component.LogProcessor, error) {
+	return &ExampleProcessor{nextConsumer}, nil
+}
+
+type ExampleProcessor struct {
+	nextConsumer consumer.LogConsumer
+}
+
+func (ep *ExampleProcessor) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func (ep *ExampleProcessor) Shutdown(ctx context.Context) error {
+	return nil
+}
+
+func (ep *ExampleProcessor) GetCapabilities() component.ProcessorCapabilities {
+	return component.ProcessorCapabilities{MutatesConsumedData: false}
+}
+
+func (ep *ExampleProcessor) ConsumeLogs(ctx context.Context, ld data.Logs) error {
+	return ep.nextConsumer.ConsumeLogs(ctx, ld)
+}
+
 // ExampleExtensionCfg is for testing purposes. We are defining an example config and factory
 // for "exampleextension" extension type.
 type ExampleExtensionCfg struct {
@@ -387,10 +462,13 @@ func (f *ExampleExtensionFactory) Type() configmodels.Type {
 // CreateDefaultConfig creates the default configuration for the Extension.
 func (f *ExampleExtensionFactory) CreateDefaultConfig() configmodels.Extension {
 	return &ExampleExtensionCfg{
-		ExtensionSettings: configmodels.ExtensionSettings{TypeVal: f.Type()},
-		ExtraSetting:      "extra string setting",
-		ExtraMapSetting:   nil,
-		ExtraListSetting:  nil,
+		ExtensionSettings: configmodels.ExtensionSettings{
+			TypeVal: f.Type(),
+			NameVal: string(f.Type()),
+		},
+		ExtraSetting:     "extra string setting",
+		ExtraMapSetting:  nil,
+		ExtraListSetting: nil,
 	}
 }