Ensure context propagated, logger enhancement (#829)

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

Author: bogdandrutu

service/builder/builder.go

@@ -23,6 +23,13 @@ const (
 	// flags
 	configCfg      = "config"
 	memBallastFlag = "mem-ballast-size-mib"
+
+	kindLogKey       = "component_kind"
+	kindLogReceiver  = "receiver"
+	kindLogProcessor = "processor"
+	kindLogExporter  = "exporter"
+	typeLogKey       = "component_type"
+	nameLogKey       = "component_name"
 )
 
 var (

service/builder/exporters_builder.go

@@ -29,22 +29,23 @@ import (
 // builtExporter is an exporter that is built based on a config. It can have
 // a trace and/or a metrics consumer and have a shutdown function.
 type builtExporter struct {
-	te component.TraceExporterBase
-	me component.MetricsExporterBase
+	logger *zap.Logger
+	te     component.TraceExporterBase
+	me     component.MetricsExporterBase
 }
 
 // Start the exporter.
 func (exp *builtExporter) Start(ctx context.Context, host component.Host) error {
 	var errors []error
 	if exp.te != nil {
-		err := exp.te.Start(context.Background(), host)
+		err := exp.te.Start(ctx, host)
 		if err != nil {
 			errors = append(errors, err)
 		}
 	}
 
 	if exp.me != nil {
-		err := exp.me.Start(context.Background(), host)
+		err := exp.me.Start(ctx, host)
 		if err != nil {
 			errors = append(errors, err)
 		}
@@ -54,16 +55,16 @@ func (exp *builtExporter) Start(ctx context.Context, host component.Host) error
 }
 
 // Shutdown the trace component and the metrics component of an exporter.
-func (exp *builtExporter) Shutdown(context.Context) error {
+func (exp *builtExporter) Shutdown(ctx context.Context) error {
 	var errors []error
 	if exp.te != nil {
-		if err := exp.te.Shutdown(context.Background()); err != nil {
+		if err := exp.te.Shutdown(ctx); err != nil {
 			errors = append(errors, err)
 		}
 	}
 
 	if exp.me != nil {
-		if err := exp.me.Shutdown(context.Background()); err != nil {
+		if err := exp.me.Shutdown(ctx); err != nil {
 			errors = append(errors, err)
 		}
 	}
@@ -75,23 +76,23 @@ func (exp *builtExporter) Shutdown(context.Context) error {
 type Exporters map[configmodels.Exporter]*builtExporter
 
 // StartAll starts all exporters.
-func (exps Exporters) StartAll(logger *zap.Logger, host component.Host) error {
-	for cfg, exp := range exps {
-		logger.Info("Exporter is starting...", zap.String("exporter", cfg.Name()))
+func (exps Exporters) StartAll(ctx context.Context, host component.Host) error {
+	for _, exp := range exps {
+		exp.logger.Info("Exporter is starting...")
 
-		if err := exp.Start(context.Background(), host); err != nil {
+		if err := exp.Start(ctx, host); err != nil {
 			return err
 		}
-		logger.Info("Exporter started.", zap.String("exporter", cfg.Name()))
+		exp.logger.Info("Exporter started.")
 	}
 	return nil
 }
 
 // ShutdownAll stops all exporters.
-func (exps Exporters) ShutdownAll() error {
+func (exps Exporters) ShutdownAll(ctx context.Context) error {
 	var errs []error
 	for _, exp := range exps {
-		err := exp.Shutdown(context.Background())
+		err := exp.Shutdown(ctx)
 		if err != nil {
 			errs = append(errs, err)
 		}
@@ -127,7 +128,7 @@ func NewExportersBuilder(
 	config *configmodels.Config,
 	factories map[string]component.ExporterFactoryBase,
 ) *ExportersBuilder {
-	return &ExportersBuilder{logger, config, factories}
+	return &ExportersBuilder{logger.With(zap.String(kindLogKey, kindLogExporter)), config, factories}
 }
 
 // BuildExporters exporters from config.
@@ -140,7 +141,8 @@ func (eb *ExportersBuilder) Build() (Exporters, error) {
 
 	// BuildExporters exporters based on configuration and required input data types.
 	for _, cfg := range eb.config.Exporters {
-		exp, err := eb.buildExporter(cfg, exporterInputDataTypes)
+		componentLogger := eb.logger.With(zap.String(typeLogKey, cfg.Type()), zap.String(nameLogKey, cfg.Name()))
+		exp, err := eb.buildExporter(componentLogger, cfg, exporterInputDataTypes)
 		if err != nil {
 			return nil, err
 		}
@@ -185,6 +187,7 @@ func (eb *ExportersBuilder) calcExportersRequiredDataTypes() exportersRequiredDa
 }
 
 func (eb *ExportersBuilder) buildExporter(
+	logger *zap.Logger,
 	config configmodels.Exporter,
 	exportersInputDataTypes exportersRequiredDataTypes,
 ) (*builtExporter, error) {
@@ -193,22 +196,23 @@ func (eb *ExportersBuilder) buildExporter(
 		return nil, fmt.Errorf("exporter factory not found for type: %s", config.Type())
 	}
 
-	exporter := &builtExporter{}
+	exporter := &builtExporter{
+		logger: logger,
+	}
 
 	inputDataTypes := exportersInputDataTypes[config]
 	if inputDataTypes == nil {
 		// TODO  https://github.com/open-telemetry/opentelemetry-collector/issues/294
 		// Move this validation to config/config.go:validateConfig
 		// No data types where requested for this exporter. This can only happen
 		// if there are no pipelines associated with the exporter.
-		eb.logger.Warn("Exporter " + config.Name() +
-			" is not associated with any pipeline and will not export data.")
+		logger.Warn("Exportee is not associated with any pipeline and will not export data.")
 		return exporter, nil
 	}
 
 	if requirement, ok := inputDataTypes[configmodels.TracesDataType]; ok {
 		// Traces data type is required. Create a trace exporter based on config.
-		te, err := createTraceExporter(factory, eb.logger, config)
+		te, err := createTraceExporter(factory, logger, config)
 		if err != nil {
 			if err == configerror.ErrDataTypeIsNotSupported {
 				// Could not create because this exporter does not support this data type.
@@ -227,7 +231,7 @@ func (eb *ExportersBuilder) buildExporter(
 
 	if requirement, ok := inputDataTypes[configmodels.MetricsDataType]; ok {
 		// Metrics data type is required. Create a trace exporter based on config.
-		me, err := createMetricsExporter(factory, eb.logger, config)
+		me, err := createMetricsExporter(factory, logger, config)
 		if err != nil {
 			if err == configerror.ErrDataTypeIsNotSupported {
 				// Could not create because this exporter does not support this data type.

service/builder/exporters_builder_test.go

@@ -73,8 +73,7 @@ func TestExportersBuilder_Build(t *testing.T) {
 	assert.Nil(t, e1.me)
 
 	// Ensure it can be started.
-	err = exporters.StartAll(zap.NewNop(), componenttest.NewNopHost())
-	assert.NoError(t, err)
+	assert.NoError(t, exporters.StartAll(context.Background(), componenttest.NewNopHost()))
 
 	// Ensure it can be stopped.
 	if err = e1.Shutdown(context.Background()); err != nil {
@@ -110,14 +109,14 @@ func TestExportersBuilder_StartAll(t *testing.T) {
 	traceExporter := &config.ExampleExporterConsumer{}
 	metricExporter := &config.ExampleExporterConsumer{}
 	exporters[expCfg] = &builtExporter{
-		te: traceExporter,
-		me: metricExporter,
+		logger: zap.NewNop(),
+		te:     traceExporter,
+		me:     metricExporter,
 	}
 	assert.False(t, traceExporter.ExporterStarted)
 	assert.False(t, metricExporter.ExporterStarted)
 
-	err := exporters.StartAll(zap.NewNop(), componenttest.NewNopHost())
-	assert.NoError(t, err)
+	assert.NoError(t, exporters.StartAll(context.Background(), componenttest.NewNopHost()))
 
 	assert.True(t, traceExporter.ExporterStarted)
 	assert.True(t, metricExporter.ExporterStarted)
@@ -129,12 +128,13 @@ func TestExportersBuilder_StopAll(t *testing.T) {
 	traceExporter := &config.ExampleExporterConsumer{}
 	metricExporter := &config.ExampleExporterConsumer{}
 	exporters[expCfg] = &builtExporter{
-		te: traceExporter,
-		me: metricExporter,
+		logger: zap.NewNop(),
+		te:     traceExporter,
+		me:     metricExporter,
 	}
 	assert.False(t, traceExporter.ExporterShutdown)
 	assert.False(t, metricExporter.ExporterShutdown)
-	exporters.ShutdownAll()
+	assert.NoError(t, exporters.ShutdownAll(context.Background()))
 
 	assert.True(t, traceExporter.ExporterShutdown)
 	assert.True(t, metricExporter.ExporterShutdown)
@@ -194,10 +194,10 @@ func (b *badExporterFactory) CreateDefaultConfig() configmodels.Exporter {
 	return &configmodels.ExporterSettings{}
 }
 
-func (b *badExporterFactory) CreateTraceExporter(logger *zap.Logger, cfg configmodels.Exporter) (component.TraceExporterOld, error) {
+func (b *badExporterFactory) CreateTraceExporter(_ *zap.Logger, _ configmodels.Exporter) (component.TraceExporterOld, error) {
 	return nil, nil
 }
 
-func (b *badExporterFactory) CreateMetricsExporter(logger *zap.Logger, cfg configmodels.Exporter) (component.MetricsExporterOld, error) {
+func (b *badExporterFactory) CreateMetricsExporter(_ *zap.Logger, _ configmodels.Exporter) (component.MetricsExporterOld, error) {
 	return nil, nil
 }

service/builder/pipelines_builder.go

@@ -32,6 +32,7 @@ import (
 // It can have a trace and/or a metrics consumer (the consumer is either the first
 // processor in the pipeline or the exporter if pipeline has no processors).
 type builtPipeline struct {
+	logger  *zap.Logger
 	firstTC consumer.TraceConsumerBase
 	firstMC consumer.MetricsConsumerBase
 
@@ -45,33 +46,33 @@ type builtPipeline struct {
 // BuiltPipelines is a map of build pipelines created from pipeline configs.
 type BuiltPipelines map[*configmodels.Pipeline]*builtPipeline
 
-func (bps BuiltPipelines) StartProcessors(logger *zap.Logger, host component.Host) error {
-	for cfg, bp := range bps {
-		logger.Info("Pipeline is starting...", zap.String("pipeline", cfg.Name))
+func (bps BuiltPipelines) StartProcessors(ctx context.Context, host component.Host) error {
+	for _, bp := range bps {
+		bp.logger.Info("Pipeline is starting...")
 		// Start in reverse order, starting from the back of processors pipeline.
 		// This is important so that processors that are earlier in the pipeline and
 		// reference processors that are later in the pipeline do not start sending
 		// data to later pipelines which are not yet started.
 		for i := len(bp.processors) - 1; i >= 0; i-- {
-			if err := bp.processors[i].Start(context.Background(), host); err != nil {
+			if err := bp.processors[i].Start(ctx, host); err != nil {
 				return err
 			}
 		}
-		logger.Info("Pipeline is started.", zap.String("pipeline", cfg.Name))
+		bp.logger.Info("Pipeline is started.")
 	}
 	return nil
 }
 
-func (bps BuiltPipelines) ShutdownProcessors(logger *zap.Logger) error {
+func (bps BuiltPipelines) ShutdownProcessors(ctx context.Context) error {
 	var errs []error
-	for cfg, bp := range bps {
-		logger.Info("Pipeline is shutting down...", zap.String("pipeline", cfg.Name))
+	for _, bp := range bps {
+		bp.logger.Info("Pipeline is shutting down...")
 		for _, p := range bp.processors {
-			if err := p.Shutdown(context.Background()); err != nil {
+			if err := p.Shutdown(ctx); err != nil {
 				errs = append(errs, err)
 			}
 		}
-		logger.Info("Pipeline is shutdown.", zap.String("pipeline", cfg.Name))
+		bp.logger.Info("Pipeline is shutdown.")
 	}
 
 	if len(errs) != 0 {
@@ -117,8 +118,7 @@ func (pb *PipelinesBuilder) Build() (BuiltPipelines, error) {
 // Builds a pipeline of processors. Returns the first processor in the pipeline.
 // The last processor in the pipeline will be plugged to fan out the data into exporters
 // that are configured for this pipeline.
-func (pb *PipelinesBuilder) buildPipeline(
-	pipelineCfg *configmodels.Pipeline,
+func (pb *PipelinesBuilder) buildPipeline(pipelineCfg *configmodels.Pipeline,
 ) (*builtPipeline, error) {
 
 	// BuildProcessors the pipeline backwards.
@@ -152,18 +152,19 @@ func (pb *PipelinesBuilder) buildPipeline(
 		// it becomes the next for the previous one (previous in the pipeline,
 		// which we will build in the next loop iteration).
 		var err error
+		componentLogger := pb.logger.With(zap.String(kindLogKey, kindLogProcessor), zap.String(typeLogKey, procCfg.Type()), zap.String(nameLogKey, procCfg.Name()))
 		switch pipelineCfg.InputType {
 		case configmodels.TracesDataType:
 			var proc component.TraceProcessorBase
-			proc, err = createTraceProcessor(factory, pb.logger, procCfg, tc)
+			proc, err = createTraceProcessor(factory, componentLogger, procCfg, tc)
 			if proc != nil {
 				mutatesConsumedData = mutatesConsumedData || proc.GetCapabilities().MutatesConsumedData
 			}
 			processors[i] = proc
 			tc = proc
 		case configmodels.MetricsDataType:
 			var proc component.MetricsProcessorBase
-			proc, err = createMetricsProcessor(factory, pb.logger, procCfg, mc)
+			proc, err = createMetricsProcessor(factory, componentLogger, procCfg, mc)
 			if proc != nil {
 				mutatesConsumedData = mutatesConsumedData || proc.GetCapabilities().MutatesConsumedData
 			}
@@ -182,9 +183,12 @@ func (pb *PipelinesBuilder) buildPipeline(
 		}
 	}
 
-	pb.logger.Info("Pipeline is enabled.", zap.String("pipelines", pipelineCfg.Name))
+	pipelineLogger := pb.logger.With(zap.String("pipeline_name", pipelineCfg.Name),
+		zap.String("pipeline_datatype", pipelineCfg.InputType.GetString()))
+	pipelineLogger.Info("Pipeline is enabled.")
 
 	bp := &builtPipeline{
+		pipelineLogger,
 		tc,
 		mc,
 		mutatesConsumedData,

service/builder/pipelines_builder_test.go

@@ -161,8 +161,7 @@ func testPipeline(t *testing.T, pipelineName string, exporterNames []string) {
 	assert.NoError(t, err)
 	require.NotNil(t, pipelineProcessors)
 
-	err = pipelineProcessors.StartProcessors(zap.NewNop(), componenttest.NewNopHost())
-	assert.NoError(t, err)
+	assert.NoError(t, pipelineProcessors.StartProcessors(context.Background(), componenttest.NewNopHost()))
 
 	processor := pipelineProcessors[cfg.Service.Pipelines[pipelineName]]
 
@@ -201,7 +200,7 @@ func testPipeline(t *testing.T, pipelineName string, exporterNames []string) {
 		assertEqualTraceData(t, generateTestTraceDataWithAttributes(), consumer.Traces[0])
 	}
 
-	err = pipelineProcessors.ShutdownProcessors(zap.NewNop())
+	err = pipelineProcessors.ShutdownProcessors(context.Background())
 	assert.NoError(t, err)
 }
 
@@ -273,17 +272,17 @@ func (b *badProcessorFactory) CreateDefaultConfig() configmodels.Processor {
 }
 
 func (b *badProcessorFactory) CreateTraceProcessor(
-	logger *zap.Logger,
-	nextConsumer consumer.TraceConsumerOld,
-	cfg configmodels.Processor,
+	_ *zap.Logger,
+	_ consumer.TraceConsumerOld,
+	_ configmodels.Processor,
 ) (component.TraceProcessorOld, error) {
 	return nil, nil
 }
 
 func (b *badProcessorFactory) CreateMetricsProcessor(
-	logger *zap.Logger,
-	consumer consumer.MetricsConsumerOld,
-	cfg configmodels.Processor,
+	_ *zap.Logger,
+	_ consumer.MetricsConsumerOld,
+	_ configmodels.Processor,
 ) (component.MetricsProcessorOld, error) {
 	return nil, nil
 }