Move extension creation into a builder class (#830)

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

Author: bogdandrutu

service/builder/builder.go

@@ -28,6 +28,7 @@ const (
 	kindLogReceiver  = "receiver"
 	kindLogProcessor = "processor"
 	kindLogExporter  = "exporter"
+	kindLogExtension = "extension"
 	typeLogKey       = "component_type"
 	nameLogKey       = "component_name"
 )

service/builder/extensions_builder.go

@@ -0,0 +1,181 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package builder
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector/component"
+	"github.com/open-telemetry/opentelemetry-collector/component/componenterror"
+	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+)
+
+// 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 builtExtension struct {
+	logger    *zap.Logger
+	extension component.ServiceExtension
+}
+
+// Start the receiver.
+func (ext *builtExtension) Start(ctx context.Context, host component.Host) error {
+	return ext.extension.Start(ctx, host)
+}
+
+// Stop the receiver.
+func (ext *builtExtension) Shutdown(ctx context.Context) error {
+	return ext.extension.Shutdown(ctx)
+}
+
+var _ component.ServiceExtension = (*builtExtension)(nil)
+
+// Exporters is a map of exporters created from exporter configs.
+type Extensions map[configmodels.Extension]*builtExtension
+
+// StartAll starts all exporters.
+func (exts Extensions) StartAll(ctx context.Context, host component.Host) error {
+	for _, ext := range exts {
+		ext.logger.Info("Extension is starting...")
+
+		if err := ext.Start(ctx, host); err != nil {
+			return err
+		}
+
+		ext.logger.Info("Extension started.")
+	}
+	return nil
+}
+
+// ShutdownAll stops all exporters.
+func (exts Extensions) ShutdownAll(ctx context.Context) error {
+	var errs []error
+	for _, ext := range exts {
+		err := ext.Shutdown(ctx)
+		if err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	if len(errs) != 0 {
+		return componenterror.CombineErrors(errs)
+	}
+	return nil
+}
+
+func (exts Extensions) NotifyPipelineReady() error {
+	for _, ext := range exts {
+		if pw, ok := ext.extension.(component.PipelineWatcher); ok {
+			if err := pw.Ready(); err != nil {
+				ext.logger.Error("Error notifying extension that the pipeline was started.")
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (exts Extensions) NotifyPipelineNotReady() error {
+	// Notify extensions in reverse order.
+	var errs []error
+	for _, ext := range exts {
+		if pw, ok := ext.extension.(component.PipelineWatcher); ok {
+			if err := pw.NotReady(); err != nil {
+				ext.logger.Error("Error notifying extension that the pipeline was shutdown.")
+				errs = append(errs, err)
+			}
+		}
+	}
+
+	if len(errs) != 0 {
+		return componenterror.CombineErrors(errs)
+	}
+
+	return nil
+}
+
+func (exts Extensions) GetServiceExtensions() map[configmodels.Extension]component.ServiceExtension {
+	result := make(map[configmodels.Extension]component.ServiceExtension, len(exts))
+	for k, v := range exts {
+		result[k] = v.extension
+	}
+	return result
+}
+
+// ExportersBuilder builds exporters from config.
+type ExtensionsBuilder struct {
+	logger    *zap.Logger
+	config    *configmodels.Config
+	factories map[string]component.ExtensionFactory
+}
+
+// NewExportersBuilder creates a new ExportersBuilder. Call BuildExporters() on the returned value.
+func NewExtensionsBuilder(
+	logger *zap.Logger,
+	config *configmodels.Config,
+	factories map[string]component.ExtensionFactory,
+) *ExtensionsBuilder {
+	return &ExtensionsBuilder{logger.With(zap.String(kindLogKey, kindLogExtension)), config, factories}
+}
+
+// Build extensions from config.
+func (eb *ExtensionsBuilder) Build() (Extensions, error) {
+	extensions := make(Extensions)
+
+	for _, extName := range eb.config.Service.Extensions {
+		extCfg, exists := eb.config.Extensions[extName]
+		if !exists {
+			return nil, errors.Errorf("extension %q is not configured", extName)
+		}
+
+		componentLogger := eb.logger.With(zap.String(typeLogKey, extCfg.Type()), zap.String(nameLogKey, extCfg.Name()))
+		ext, err := eb.buildExtension(componentLogger, extCfg)
+		if err != nil {
+			return nil, err
+		}
+
+		extensions[extCfg] = ext
+	}
+
+	return extensions, nil
+}
+
+func (eb *ExtensionsBuilder) buildExtension(logger *zap.Logger, cfg configmodels.Extension) (*builtExtension, error) {
+	factory := eb.factories[cfg.Type()]
+	if factory == nil {
+		return nil, errors.Errorf("extension factory for type %q is not configured", cfg.Type())
+	}
+
+	ext := &builtExtension{
+		logger: logger,
+	}
+
+	ex, err := factory.CreateExtension(context.Background(), component.ExtensionCreateParams{Logger: eb.logger}, cfg)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to create extension %q", cfg.Name())
+	}
+
+	// Check if the factory really created the extension.
+	if ex == nil {
+		return nil, errors.Errorf("factory for %q produced a nil extension", cfg.Name())
+	}
+
+	ext.extension = ex
+
+	return ext, nil
+}

service/service.go

@@ -40,20 +40,18 @@ import (
 
 // Application represents a collector application
 type Application struct {
-	info           ApplicationStartInfo
-	rootCmd        *cobra.Command
-	v              *viper.Viper
-	logger         *zap.Logger
-	exporters      builder.Exporters
-	builtReceivers builder.Receivers
-	builtPipelines builder.BuiltPipelines
+	info            ApplicationStartInfo
+	rootCmd         *cobra.Command
+	v               *viper.Viper
+	logger          *zap.Logger
+	builtExporters  builder.Exporters
+	builtReceivers  builder.Receivers
+	builtPipelines  builder.BuiltPipelines
+	builtExtensions builder.Extensions
 
 	factories config.Factories
 	config    *configmodels.Config
 
-	extensionsList []component.ServiceExtension
-	extensionsMap  map[configmodels.Extension]component.ServiceExtension
-
 	// stopTestChan is used to terminate the application in end to end tests.
 	stopTestChan chan struct{}
 	// readyChan is used in tests to indicate that the application is ready.
@@ -184,7 +182,7 @@ func (app *Application) GetFactory(kind component.Kind, componentType string) co
 }
 
 func (app *Application) GetExtensions() map[configmodels.Extension]component.ServiceExtension {
-	return app.extensionsMap
+	return app.builtExtensions.GetServiceExtensions()
 }
 
 func (app *Application) init() error {
@@ -262,37 +260,13 @@ func (app *Application) setupConfigurationComponents(ctx context.Context, factor
 }
 
 func (app *Application) setupExtensions(ctx context.Context) error {
-	app.extensionsMap = make(map[configmodels.Extension]component.ServiceExtension)
-	for _, extName := range app.config.Service.Extensions {
-		extCfg, exists := app.config.Extensions[extName]
-		if !exists {
-			return errors.Errorf("extension %q is not configured", extName)
-		}
-
-		factory, exists := app.factories.Extensions[extCfg.Type()]
-		if !exists {
-			return errors.Errorf("extension factory for type %q is not configured", extCfg.Type())
-		}
-
-		ext, err := factory.CreateExtension(ctx, component.ExtensionCreateParams{Logger: app.logger}, extCfg)
-		if err != nil {
-			return errors.Wrapf(err, "failed to create extension %q", extName)
-		}
-
-		// Check if the factory really created the extension.
-		if ext == nil {
-			return errors.Errorf("factory for %q produced a nil extension", extName)
-		}
-
-		if err := ext.Start(ctx, app); err != nil {
-			return errors.Wrapf(err, "error starting extension %q", extName)
-		}
-
-		app.extensionsList = append(app.extensionsList, ext)
-		app.extensionsMap[extCfg] = ext
+	var err error
+	app.builtExtensions, err = builder.NewExtensionsBuilder(app.logger, app.config, app.factories.Extensions).Build()
+	if err != nil {
+		return errors.Wrap(err, "cannot build builtExtensions")
 	}
-
-	return nil
+	app.logger.Info("Starting extensions...")
+	return app.builtExtensions.StartAll(ctx, app)
 }
 
 func (app *Application) setupPipelines(ctx context.Context) error {
@@ -301,19 +275,20 @@ func (app *Application) setupPipelines(ctx context.Context) error {
 
 	// First create exporters.
 	var err error
-	app.exporters, err = builder.NewExportersBuilder(app.logger, app.config, app.factories.Exporters).Build()
+	app.builtExporters, err = builder.NewExportersBuilder(app.logger, app.config, app.factories.Exporters).Build()
 	if err != nil {
-		return errors.Wrap(err, "cannot build exporters")
+		return errors.Wrap(err, "cannot build builtExporters")
 	}
 	app.logger.Info("Starting exporters...")
-	err = app.exporters.StartAll(ctx, app)
+
+	err = app.builtExporters.StartAll(ctx, app)
 	if err != nil {
-		return errors.Wrap(err, "cannot start exporters")
+		return errors.Wrap(err, "cannot start builtExporters")
 	}
 
 	// Create pipelines and their processors and plug exporters to the
 	// end of the pipelines.
-	app.builtPipelines, err = builder.NewPipelinesBuilder(app.logger, app.config, app.exporters, app.factories.Processors).Build()
+	app.builtPipelines, err = builder.NewPipelinesBuilder(app.logger, app.config, app.builtExporters, app.factories.Processors).Build()
 	if err != nil {
 		return errors.Wrap(err, "cannot build pipelines")
 	}
@@ -339,43 +314,6 @@ func (app *Application) setupPipelines(ctx context.Context) error {
 	return nil
 }
 
-func (app *Application) notifyPipelineReady() error {
-	for i, ext := range app.extensionsList {
-		if pw, ok := ext.(component.PipelineWatcher); ok {
-			if err := pw.Ready(); err != nil {
-				return errors.Wrapf(
-					err,
-					"error notifying extension %q that the pipeline was started",
-					app.config.Service.Extensions[i],
-				)
-			}
-		}
-	}
-
-	return nil
-}
-
-func (app *Application) notifyPipelineNotReady() error {
-	// Notify extensions in reverse order.
-	var errs []error
-	for i := len(app.extensionsList) - 1; i >= 0; i-- {
-		ext := app.extensionsList[i]
-		if pw, ok := ext.(component.PipelineWatcher); ok {
-			if err := pw.NotReady(); err != nil {
-				errs = append(errs, errors.Wrapf(err,
-					"error notifying extension %q that the pipeline was shutdown",
-					app.config.Service.Extensions[i]))
-			}
-		}
-	}
-
-	if len(errs) != 0 {
-		return componenterror.CombineErrors(errs)
-	}
-
-	return nil
-}
-
 func (app *Application) shutdownPipelines(ctx context.Context) error {
 	// Shutdown order is the reverse of building: first receivers, then flushing pipelines
 	// giving senders a chance to send all their data. This may take time, the allowed
@@ -395,8 +333,8 @@ func (app *Application) shutdownPipelines(ctx context.Context) error {
 		errs = append(errs, errors.Wrap(err, "failed to shutdown processors"))
 	}
 
-	app.logger.Info("Shutting down exporters...")
-	err = app.exporters.ShutdownAll(ctx)
+	app.logger.Info("Stopping exporters...")
+	err = app.builtExporters.ShutdownAll(ctx)
 	if err != nil {
 		errs = append(errs, errors.Wrap(err, "failed to shutdown exporters"))
 	}
@@ -409,21 +347,11 @@ func (app *Application) shutdownPipelines(ctx context.Context) error {
 }
 
 func (app *Application) shutdownExtensions(ctx context.Context) error {
-	// Shutdown extensions in reverse order.
-	var errs []error
-	for i := len(app.extensionsList) - 1; i >= 0; i-- {
-		ext := app.extensionsList[i]
-		if err := ext.Shutdown(ctx); err != nil {
-			errs = append(errs, errors.Wrapf(err,
-				"error shutting down extension %q",
-				app.config.Service.Extensions[i]))
-		}
-	}
-
-	if len(errs) != 0 {
-		return componenterror.CombineErrors(errs)
+	app.logger.Info("Stopping extensions...")
+	err := app.builtExtensions.ShutdownAll(ctx)
+	if err != nil {
+		return errors.Wrap(err, "failed to shutdown extensions")
 	}
-
 	return nil
 }
 
@@ -450,7 +378,7 @@ func (app *Application) execute(ctx context.Context, factory ConfigFactory) erro
 		return err
 	}
 
-	err = app.notifyPipelineReady()
+	err = app.builtExtensions.NotifyPipelineReady()
 	if err != nil {
 		return err
 	}
@@ -465,7 +393,7 @@ func (app *Application) execute(ctx context.Context, factory ConfigFactory) erro
 	runtime.KeepAlive(ballast)
 	app.logger.Info("Starting shutdown...")
 
-	err = app.notifyPipelineNotReady()
+	err = app.builtExtensions.NotifyPipelineNotReady()
 	if err != nil {
 		errs = append(errs, errors.Wrap(err, "failed to notify that pipeline is not ready"))
 	}