[service] add logger provider configuration support (#10544)

This allows us to use the otel-go/config package to support configuring
external destinations for logs. I'm putting this in draft to gather
community feedback on whether this is a desirable feature for the
collector.

I used the following configuration with this PR to send data to an OTLP
backend:

```yaml
telemetry:
   logs:
     processors:
       - batch:
           exporter:
             otlp:
               protocol: http/protobuf
               endpoint: https://api.honeycomb.io:443
               headers:
                 "x-honeycomb-team": "${env:HONEYCOMB_API_KEY}"
```

This allowed me to see logs in my backend:

![Screenshot 2024-07-04 at 1 49
04 PM](https://github.com/open-telemetry/opentelemetry-collector/assets/223565/bba7c82a-e9f9-4b15-933a-6bb3e7a66bb8)

---------

Signed-off-by: Alex Boten <[email protected]>

.chloggen/codeboten_add-log-record-processor-config.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: service
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add support for using the otelzap bridge and emit logs using the OTel Go SDK
+
+# One or more tracking issues or pull requests related to the change
+issues: [10544]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/otelcorecol/go.mod

@@ -117,6 +117,7 @@ require (
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.112.0 // indirect
 	go.opentelemetry.io/collector/service v0.112.0 // indirect
+	go.opentelemetry.io/contrib/bridges/otelzap v0.6.0 // indirect
 	go.opentelemetry.io/contrib/config v0.10.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 // indirect

internal/e2e/go.mod

@@ -96,6 +96,7 @@ require (
 	go.opentelemetry.io/collector/processor/processortest v0.112.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.112.0 // indirect
+	go.opentelemetry.io/contrib/bridges/otelzap v0.6.0 // indirect
 	go.opentelemetry.io/contrib/config v0.10.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 // indirect

otelcol/go.mod

@@ -82,6 +82,7 @@ require (
 	go.opentelemetry.io/collector/processor/processorprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.112.0 // indirect
+	go.opentelemetry.io/contrib/bridges/otelzap v0.6.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.31.0 // indirect
 	go.opentelemetry.io/otel v1.31.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.7.0 // indirect

otelcol/otelcoltest/go.mod

@@ -80,6 +80,7 @@ require (
 	go.opentelemetry.io/collector/processor/processorprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/collector/semconv v0.112.0 // indirect
+	go.opentelemetry.io/contrib/bridges/otelzap v0.6.0 // indirect
 	go.opentelemetry.io/contrib/config v0.10.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.31.0 // indirect
 	go.opentelemetry.io/otel v1.31.0 // indirect

service/go.mod

@@ -39,13 +39,15 @@ require (
 	go.opentelemetry.io/collector/receiver v0.112.0
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0
 	go.opentelemetry.io/collector/semconv v0.112.0
+	go.opentelemetry.io/contrib/bridges/otelzap v0.6.0
 	go.opentelemetry.io/contrib/config v0.10.0
 	go.opentelemetry.io/contrib/propagators/b3 v1.31.0
 	go.opentelemetry.io/otel v1.31.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.31.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.31.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.53.0
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.31.0
+	go.opentelemetry.io/otel/log v0.7.0
 	go.opentelemetry.io/otel/metric v1.31.0
 	go.opentelemetry.io/otel/sdk v1.31.0
 	go.opentelemetry.io/otel/sdk/metric v1.31.0
@@ -107,7 +109,6 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.31.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.7.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.31.0 // indirect
-	go.opentelemetry.io/otel/log v0.7.0 // indirect
 	go.opentelemetry.io/otel/sdk/log v0.7.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	golang.org/x/net v0.30.0 // indirect

service/service.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"runtime"
 
+	"go.opentelemetry.io/otel/log"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/noop"
 	sdkresource "go.opentelemetry.io/otel/sdk/resource"
@@ -92,6 +93,7 @@ type Service struct {
 	telemetrySettings component.TelemetrySettings
 	host              *graph.Host
 	collectorConf     *confmap.Conf
+	loggerProvider    log.LoggerProvider
 }
 
 // New creates a new Service, its telemetry, and Components.
@@ -122,10 +124,11 @@ func New(ctx context.Context, set Settings, cfg Config) (*Service, error) {
 		ZapOptions: set.LoggingOptions,
 	}
 
-	logger, err := telFactory.CreateLogger(ctx, telset, &cfg.Telemetry)
+	logger, lp, err := telFactory.CreateLogger(ctx, telset, &cfg.Telemetry)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create logger: %w", err)
 	}
+	srv.loggerProvider = lp
 
 	tracerProvider, err := telFactory.CreateTracerProvider(ctx, telset, &cfg.Telemetry)
 	if err != nil {
@@ -250,6 +253,12 @@ func (srv *Service) shutdownTelemetry(ctx context.Context) error {
 			err = multierr.Append(err, fmt.Errorf("failed to shutdown tracer provider: %w", shutdownErr))
 		}
 	}
+
+	if prov, ok := srv.loggerProvider.(shutdownable); ok {
+		if shutdownErr := prov.Shutdown(ctx); shutdownErr != nil {
+			err = multierr.Append(err, fmt.Errorf("failed to shutdown logger provider: %w", shutdownErr))
+		}
+	}
 	return err
 }
 

service/telemetry/config.go

@@ -105,6 +105,10 @@ type LogsConfig struct {
 	//
 	// By default, there is no initial field.
 	InitialFields map[string]any `mapstructure:"initial_fields"`
+
+	// Processors allow configuration of log record processors to emit logs to
+	// any number of suported backends.
+	Processors []config.LogRecordProcessor `mapstructure:"processors"`
 }
 
 // LogsSamplingConfig sets a sampling strategy for the logger. Sampling caps the

service/telemetry/factory.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/contrib/config"
+	"go.opentelemetry.io/otel/log"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
@@ -50,7 +51,7 @@ type Factory interface {
 	CreateDefaultConfig() component.Config
 
 	// CreateLogger creates a logger.
-	CreateLogger(ctx context.Context, set Settings, cfg component.Config) (*zap.Logger, error)
+	CreateLogger(ctx context.Context, set Settings, cfg component.Config) (*zap.Logger, log.LoggerProvider, error)
 
 	// CreateTracerProvider creates a TracerProvider.
 	CreateTracerProvider(ctx context.Context, set Settings, cfg component.Config) (trace.TracerProvider, error)
@@ -65,9 +66,9 @@ type Factory interface {
 // NewFactory creates a new Factory.
 func NewFactory() Factory {
 	return newFactory(createDefaultConfig,
-		withLogger(func(_ context.Context, set Settings, cfg component.Config) (*zap.Logger, error) {
+		withLogger(func(ctx context.Context, set Settings, cfg component.Config) (*zap.Logger, log.LoggerProvider, error) {
 			c := *cfg.(*Config)
-			return newLogger(c.Logs, set.ZapOptions)
+			return newLogger(ctx, set, c)
 		}),
 		withTracerProvider(func(ctx context.Context, set Settings, cfg component.Config) (trace.TracerProvider, error) {
 			c := *cfg.(*Config)

service/telemetry/factory_impl.go

@@ -6,6 +6,8 @@ package telemetry // import "go.opentelemetry.io/collector/service/telemetry"
 import (
 	"context"
 
+	"go.opentelemetry.io/otel/log"
+	lognoop "go.opentelemetry.io/otel/log/noop"
 	"go.opentelemetry.io/otel/metric"
 	metricnoop "go.opentelemetry.io/otel/metric/noop"
 	"go.opentelemetry.io/otel/trace"
@@ -45,7 +47,7 @@ func (f *factory) CreateDefaultConfig() component.Config {
 }
 
 // createLoggerFunc is the equivalent of Factory.CreateLogger.
-type createLoggerFunc func(context.Context, Settings, component.Config) (*zap.Logger, error)
+type createLoggerFunc func(context.Context, Settings, component.Config) (*zap.Logger, log.LoggerProvider, error)
 
 // withLogger overrides the default no-op logger.
 func withLogger(createLogger createLoggerFunc) factoryOption {
@@ -54,9 +56,9 @@ func withLogger(createLogger createLoggerFunc) factoryOption {
 	})
 }
 
-func (f *factory) CreateLogger(ctx context.Context, set Settings, cfg component.Config) (*zap.Logger, error) {
+func (f *factory) CreateLogger(ctx context.Context, set Settings, cfg component.Config) (*zap.Logger, log.LoggerProvider, error) {
 	if f.createLoggerFunc == nil {
-		return zap.NewNop(), nil
+		return zap.NewNop(), lognoop.NewLoggerProvider(), nil
 	}
 	return f.createLoggerFunc(ctx, set, cfg)
 }

service/telemetry/factory_test.go

@@ -99,7 +99,7 @@ func TestTelemetryConfiguration(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			f := NewFactory()
 			set := Settings{ZapOptions: []zap.Option{}}
-			logger, err := f.CreateLogger(context.Background(), set, tt.cfg)
+			logger, _, err := f.CreateLogger(context.Background(), set, tt.cfg)
 			if tt.success {
 				require.NoError(t, err)
 				assert.NotNil(t, logger)
@@ -158,7 +158,7 @@ func TestSampledLogger(t *testing.T) {
 			f := NewFactory()
 			ctx := context.Background()
 			set := Settings{ZapOptions: []zap.Option{}}
-			logger, err := f.CreateLogger(ctx, set, tt.cfg)
+			logger, _, err := f.CreateLogger(ctx, set, tt.cfg)
 			require.NoError(t, err)
 			assert.NotNil(t, logger)
 		})

service/telemetry/logger.go

@@ -4,38 +4,83 @@
 package telemetry // import "go.opentelemetry.io/collector/service/telemetry"
 
 import (
+	"context"
+
+	"go.opentelemetry.io/contrib/bridges/otelzap"
+	"go.opentelemetry.io/contrib/config"
+	"go.opentelemetry.io/otel/log"
+	semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
 
-func newLogger(cfg LogsConfig, options []zap.Option) (*zap.Logger, error) {
+func newLogger(ctx context.Context, set Settings, cfg Config) (*zap.Logger, log.LoggerProvider, error) {
 	// Copied from NewProductionConfig.
 	zapCfg := &zap.Config{
-		Level:             zap.NewAtomicLevelAt(cfg.Level),
-		Development:       cfg.Development,
-		Encoding:          cfg.Encoding,
+		Level:             zap.NewAtomicLevelAt(cfg.Logs.Level),
+		Development:       cfg.Logs.Development,
+		Encoding:          cfg.Logs.Encoding,
 		EncoderConfig:     zap.NewProductionEncoderConfig(),
-		OutputPaths:       cfg.OutputPaths,
-		ErrorOutputPaths:  cfg.ErrorOutputPaths,
-		DisableCaller:     cfg.DisableCaller,
-		DisableStacktrace: cfg.DisableStacktrace,
-		InitialFields:     cfg.InitialFields,
+		OutputPaths:       cfg.Logs.OutputPaths,
+		ErrorOutputPaths:  cfg.Logs.ErrorOutputPaths,
+		DisableCaller:     cfg.Logs.DisableCaller,
+		DisableStacktrace: cfg.Logs.DisableStacktrace,
+		InitialFields:     cfg.Logs.InitialFields,
 	}
 
 	if zapCfg.Encoding == "console" {
 		// Human-readable timestamps for console format of logs.
 		zapCfg.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
 	}
 
-	logger, err := zapCfg.Build(options...)
+	logger, err := zapCfg.Build(set.ZapOptions...)
+
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	if cfg.Sampling != nil && cfg.Sampling.Enabled {
-		logger = newSampledLogger(logger, cfg.Sampling)
+
+	var lp log.LoggerProvider
+
+	if len(cfg.Logs.Processors) > 0 {
+		sch := semconv.SchemaURL
+		res := config.Resource{
+			SchemaUrl:  &sch,
+			Attributes: attributes(set, cfg),
+		}
+		sdk, err := config.NewSDK(
+			config.WithContext(ctx),
+			config.WithOpenTelemetryConfiguration(
+				config.OpenTelemetryConfiguration{
+					LoggerProvider: &config.LoggerProvider{
+						Processors: cfg.Logs.Processors,
+					},
+					Resource: &res,
+				},
+			),
+		)
+
+		if err != nil {
+			return nil, nil, err
+		}
+
+		lp = sdk.LoggerProvider()
+
+		logger = logger.WithOptions(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+			return zapcore.NewTee(
+				c,
+				otelzap.NewCore("go.opentelemetry.io/collector/service/telemetry",
+					otelzap.WithLoggerProvider(lp),
+				),
+			)
+		}))
+
+	}
+
+	if cfg.Logs.Sampling != nil && cfg.Logs.Sampling.Enabled {
+		logger = newSampledLogger(logger, cfg.Logs.Sampling)
 	}
 
-	return logger, nil
+	return logger, lp, nil
 }
 
 func newSampledLogger(logger *zap.Logger, sc *LogsSamplingConfig) *zap.Logger {