Small cleanups in service/builder (#3693)

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

Author: bogdandrutu

service/internal/builder/exporters_builder.go

@@ -139,31 +139,38 @@ type dataTypeRequirement struct {
 type dataTypeRequirements map[config.DataType]dataTypeRequirement
 
 // Data type requirements for all exporters.
-type exportersRequiredDataTypes map[config.Exporter]dataTypeRequirements
+type exportersRequiredDataTypes map[config.ComponentID]dataTypeRequirements
 
 // BuildExporters builds Exporters from config.
 func BuildExporters(
 	logger *zap.Logger,
 	tracerProvider trace.TracerProvider,
 	buildInfo component.BuildInfo,
-	config *config.Config,
+	cfg *config.Config,
 	factories map[config.Type]component.ExporterFactory,
 ) (Exporters, error) {
 	logger = logger.With(zap.String(zapKindKey, zapKindLogExporter))
 
 	// We need to calculate required input data types for each exporter so that we know
 	// which data type must be started for each exporter.
-	exporterInputDataTypes := calcExportersRequiredDataTypes(config)
+	exporterInputDataTypes := calcExportersRequiredDataTypes(cfg)
 
 	exporters := make(Exporters)
-	// BuildExporters exporters based on configuration and required input data types.
-	for _, expCfg := range config.Exporters {
+
+	// Build exporters exporters based on configuration and required input data types.
+	for expID, expCfg := range cfg.Exporters {
 		set := component.ExporterCreateSettings{
-			Logger:         logger.With(zap.Stringer(zapNameKey, expCfg.ID())),
+			Logger:         logger.With(zap.String(zapNameKey, expID.String())),
 			TracerProvider: tracerProvider,
 			BuildInfo:      buildInfo,
 		}
-		exp, err := buildExporter(context.Background(), factories, set, expCfg, exporterInputDataTypes)
+
+		factory, exists := factories[expID.Type()]
+		if !exists || factory == nil {
+			return nil, fmt.Errorf("exporter factory not found for type: %s", expID.Type())
+		}
+
+		exp, err := buildExporter(context.Background(), factory, set, expCfg, exporterInputDataTypes[expID])
 		if err != nil {
 			return nil, err
 		}
@@ -174,7 +181,7 @@ func BuildExporters(
 	return exporters, nil
 }
 
-func calcExportersRequiredDataTypes(config *config.Config) exportersRequiredDataTypes {
+func calcExportersRequiredDataTypes(cfg *config.Config) exportersRequiredDataTypes {
 	// Go over all pipelines. The data type of the pipeline defines what data type
 	// each exporter is expected to receive. Collect all required types for each
 	// exporter.
@@ -187,43 +194,34 @@ func calcExportersRequiredDataTypes(config *config.Config) exportersRequiredData
 	result := make(exportersRequiredDataTypes)
 
 	// Iterate over pipelines.
-	for _, pipeline := range config.Service.Pipelines {
+	for _, pipeline := range cfg.Service.Pipelines {
 		// Iterate over all exporters for this pipeline.
-		for _, expName := range pipeline.Exporters {
-			// Find the exporter config by name.
-			exporter := config.Exporters[expName]
-
-			// Create the data type requirement for the exporter if it does not exist.
-			if result[exporter] == nil {
-				result[exporter] = make(dataTypeRequirements)
+		for _, expID := range pipeline.Exporters {
+			// Create the data type requirement for the expCfg if it does not exist.
+			if _, ok := result[expID]; !ok {
+				result[expID] = make(dataTypeRequirements)
 			}
 
-			// Remember that this data type is required for the exporter and also which
+			// Remember that this data type is required for the expCfg and also which
 			// pipeline the requirement is coming from.
-			result[exporter][pipeline.InputType] = dataTypeRequirement{pipeline}
+			result[expID][pipeline.InputType] = dataTypeRequirement{pipeline}
 		}
 	}
 	return result
 }
 
 func buildExporter(
 	ctx context.Context,
-	factories map[config.Type]component.ExporterFactory,
+	factory component.ExporterFactory,
 	set component.ExporterCreateSettings,
 	cfg config.Exporter,
-	exportersInputDataTypes exportersRequiredDataTypes,
+	inputDataTypes dataTypeRequirements,
 ) (*builtExporter, error) {
-	factory := factories[cfg.ID().Type()]
-	if factory == nil {
-		return nil, fmt.Errorf("exporter factory not found for type: %s", cfg.ID().Type())
-	}
-
 	exporter := &builtExporter{
 		logger:        set.Logger,
 		expByDataType: make(map[config.DataType]component.Exporter, 3),
 	}
 
-	inputDataTypes := exportersInputDataTypes[cfg]
 	if inputDataTypes == nil {
 		set.Logger.Info("Ignoring exporter as it is not used by any pipeline")
 		return exporter, nil

service/internal/builder/exporters_builder_test.go

@@ -83,7 +83,7 @@ func TestBuildExporters(t *testing.T) {
 		// Since the endpoint of opencensus exporter doesn't actually exist, e1 may
 		// already stop because it cannot connect.
 		// The test should stop running if this isn't the error cause.
-		require.Equal(t, err.Error(), "rpc error: code = Canceled desc = grpc: the client connection is closing")
+		require.EqualError(t, err, "rpc error: code = Canceled desc = grpc: the client connection is closing")
 	}
 
 	// Remove the pipeline so that the exporter is not attached to any pipeline.

service/internal/builder/extensions_builder.go

@@ -119,20 +119,26 @@ func BuildExtensions(
 	config *config.Config,
 	factories map[config.Type]component.ExtensionFactory,
 ) (Extensions, error) {
-	logger = logger.With(zap.String(zapKindKey, zapKindExtension))
 	extensions := make(Extensions)
-	for _, extName := range config.Service.Extensions {
-		extCfg, exists := config.Extensions[extName]
-		if !exists {
-			return nil, fmt.Errorf("extension %q is not configured", extName)
+	for _, extID := range config.Service.Extensions {
+		extCfg, existsCfg := config.Extensions[extID]
+		if !existsCfg {
+			return nil, fmt.Errorf("extension %q is not configured", extID)
+		}
+
+		factory, existsFactory := factories[extID.Type()]
+		if !existsFactory {
+			return nil, fmt.Errorf("extension factory for type %q is not configured", extID.Type())
 		}
 
 		set := component.ExtensionCreateSettings{
-			Logger:         logger.With(zap.Stringer(zapNameKey, extCfg.ID())),
+			Logger: logger.With(
+				zap.String(zapKindKey, zapKindExtension),
+				zap.String(zapNameKey, extID.String())),
 			TracerProvider: tracerProvider,
 			BuildInfo:      buildInfo,
 		}
-		ext, err := buildExtension(context.Background(), factories, set, extCfg)
+		ext, err := buildExtension(context.Background(), factory, set, extCfg)
 		if err != nil {
 			return nil, err
 		}
@@ -143,12 +149,7 @@ func BuildExtensions(
 	return extensions, nil
 }
 
-func buildExtension(ctx context.Context, factories map[config.Type]component.ExtensionFactory, creationSet component.ExtensionCreateSettings, cfg config.Extension) (*builtExtension, error) {
-	factory := factories[cfg.ID().Type()]
-	if factory == nil {
-		return nil, fmt.Errorf("extension factory for type %q is not configured", cfg.ID().Type())
-	}
-
+func buildExtension(ctx context.Context, factory component.ExtensionFactory, creationSet component.ExtensionCreateSettings, cfg config.Extension) (*builtExtension, error) {
 	ext := &builtExtension{
 		logger: creationSet.Logger,
 	}

service/internal/builder/extensions_builder_test.go

@@ -119,7 +119,7 @@ func TestService_setupExtensions(t *testing.T) {
 			ext, err := BuildExtensions(zap.NewNop(), trace.NewNoopTracerProvider(), component.DefaultBuildInfo(), tt.config, tt.factories.Extensions)
 
 			assert.Error(t, err)
-			assert.Equal(t, tt.wantErrMsg, err.Error())
+			assert.EqualError(t, err, tt.wantErrMsg)
 			assert.Equal(t, 0, len(ext))
 		})
 	}

service/internal/builder/pipelines_builder.go

@@ -144,17 +144,24 @@ func (pb *pipelinesBuilder) buildPipeline(ctx context.Context, pipelineCfg *conf
 	// the processor itself becomes a consumer for the one that precedes it in
 	// in the pipeline and so on.
 	for i := len(pipelineCfg.Processors) - 1; i >= 0; i-- {
-		procName := pipelineCfg.Processors[i]
-		procCfg := pb.config.Processors[procName]
+		procID := pipelineCfg.Processors[i]
 
-		factory := pb.factories[procCfg.ID().Type()]
+		procCfg, existsCfg := pb.config.Processors[procID]
+		if !existsCfg {
+			return nil, fmt.Errorf("processor %q is not configured", procID)
+		}
+
+		factory, existsFactory := pb.factories[procID.Type()]
+		if !existsFactory {
+			return nil, fmt.Errorf("processor factory for type %q is not configured", procID.Type())
+		}
 
 		// This processor must point to the next consumer and then
 		// it becomes the next for the previous one (previous in the pipeline,
 		// which we will build in the next loop iteration).
 		var err error
 		set := component.ProcessorCreateSettings{
-			Logger:         pb.logger.With(zap.String(zapKindKey, zapKindProcessor), zap.Stringer(zapNameKey, procCfg.ID())),
+			Logger:         pb.logger.With(zap.String(zapKindKey, zapKindProcessor), zap.String(zapNameKey, procID.String())),
 			TracerProvider: pb.tracerProvider,
 			BuildInfo:      pb.buildInfo,
 		}
@@ -188,17 +195,17 @@ func (pb *pipelinesBuilder) buildPipeline(ctx context.Context, pipelineCfg *conf
 
 		default:
 			return nil, fmt.Errorf("error creating processor %q in pipeline %q, data type %s is not supported",
-				procName, pipelineCfg.Name, pipelineCfg.InputType)
+				procID, pipelineCfg.Name, pipelineCfg.InputType)
 		}
 
 		if err != nil {
 			return nil, fmt.Errorf("error creating processor %q in pipeline %q: %v",
-				procName, pipelineCfg.Name, err)
+				procID, pipelineCfg.Name, err)
 		}
 
 		// Check if the factory really created the processor.
 		if tc == nil && mc == nil && lc == nil {
-			return nil, fmt.Errorf("factory for %v produced a nil processor", procCfg.ID())
+			return nil, fmt.Errorf("factory for %v produced a nil processor", procID)
 		}
 	}
 

service/internal/builder/receivers_builder.go

@@ -80,8 +80,6 @@ func (rcvs Receivers) StartAll(ctx context.Context, host component.Host) error {
 
 // receiversBuilder builds receivers from config.
 type receiversBuilder struct {
-	logger         *zap.Logger
-	buildInfo      component.BuildInfo
 	config         *config.Config
 	builtPipelines BuiltPipelines
 	factories      map[config.Type]component.ReceiverFactory
@@ -92,19 +90,20 @@ func BuildReceivers(
 	logger *zap.Logger,
 	tracerProvider trace.TracerProvider,
 	buildInfo component.BuildInfo,
-	config *config.Config,
+	cfg *config.Config,
 	builtPipelines BuiltPipelines,
 	factories map[config.Type]component.ReceiverFactory,
 ) (Receivers, error) {
-	rb := &receiversBuilder{logger.With(zap.String(zapKindKey, zapKindReceiver)), buildInfo, config, builtPipelines, factories}
+	rb := &receiversBuilder{cfg, builtPipelines, factories}
 
 	receivers := make(Receivers)
-	for _, recvCfg := range rb.config.Receivers {
+	for recvID, recvCfg := range cfg.Receivers {
 		set := component.ReceiverCreateSettings{
-			Logger:         rb.logger.With(zap.Stringer(zapNameKey, recvCfg.ID())),
+			Logger:         logger.With(zap.String(zapKindKey, zapKindReceiver), zap.String(zapNameKey, recvID.String())),
 			TracerProvider: tracerProvider,
 			BuildInfo:      buildInfo,
 		}
+
 		rcv, err := rb.buildReceiver(context.Background(), set, recvCfg)
 		if err != nil {
 			if err == errUnusedReceiver {
@@ -199,13 +198,10 @@ func attachReceiverToPipelines(
 	if err != nil {
 		if err == componenterror.ErrDataTypeIsNotSupported {
 			return fmt.Errorf(
-				"receiver %v does not support %s but it was used in a "+
-					"%s pipeline",
-				cfg.ID(),
-				dataType,
-				dataType)
+				"receiver %v does not support %s but it was used in a %s pipeline",
+				cfg.ID(), dataType, dataType)
 		}
-		return fmt.Errorf("cannot create receiver %v: %s", cfg.ID(), err.Error())
+		return fmt.Errorf("cannot create receiver %v: %w", cfg.ID(), err)
 	}
 
 	// Check if the factory really created the receiver.