otelconf: add logger provider internal funcs (open-telemetry#8089)

Signed-off-by: alex boten <[email protected]>
Co-authored-by: Tyler Yahn <[email protected]>

Author: codeboten

otelconf/config_common.go

@@ -4,10 +4,32 @@
 package otelconf // import "go.opentelemetry.io/contrib/otelconf"
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"reflect"
+
+	"go.opentelemetry.io/otel/baggage"
+	sdklog "go.opentelemetry.io/otel/sdk/log"
+)
+
+const (
+	compressionGzip = "gzip"
+	compressionNone = "none"
 )
 
+type configOptions struct {
+	ctx                   context.Context
+	opentelemetryConfig   OpenTelemetryConfiguration
+	loggerProviderOptions []sdklog.LoggerProviderOption
+}
+
+type shutdownFunc func(context.Context) error
+
+func noopShutdown(context.Context) error {
+	return nil
+}
+
 type errBound struct {
 	Field string
 	Bound int
@@ -85,7 +107,7 @@ type errInvalid struct {
 }
 
 func (e *errInvalid) Error() string {
-	return "invalid " + e.Identifier
+	return "invalid config: " + e.Identifier
 }
 
 func (e *errInvalid) Is(target error) bool {
@@ -201,3 +223,25 @@ func validateSpanLimits(plain *SpanLimits) error {
 func ptr[T any](v T) *T {
 	return &v
 }
+
+// createHeadersConfig combines the two header config fields. Headers take precedence over headersList.
+func createHeadersConfig(headers []NameStringValuePair, headersList *string) (map[string]string, error) {
+	result := make(map[string]string)
+	if headersList != nil {
+		// Parsing follows https://github.com/open-telemetry/opentelemetry-configuration/blob/568e5080816d40d75792eb754fc96bde09654159/schema/type_descriptions.yaml#L584.
+		headerslist, err := baggage.Parse(*headersList)
+		if err != nil {
+			return nil, errors.Join(newErrInvalid("invalid headers_list"), err)
+		}
+		for _, kv := range headerslist.Members() {
+			result[kv.Key()] = kv.Value()
+		}
+	}
+	// Headers take precedence over HeadersList, so this has to be after HeadersList is processed.
+	for _, kv := range headers {
+		if kv.Value != nil {
+			result[kv.Name] = *kv.Value
+		}
+	}
+	return result, nil
+}

otelconf/log.go

@@ -0,0 +1,236 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package otelconf // import "go.opentelemetry.io/contrib/otelconf"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"time"
+
+	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
+	"go.opentelemetry.io/otel/exporters/stdout/stdoutlog"
+	"go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/log/noop"
+	sdklog "go.opentelemetry.io/otel/sdk/log"
+	"go.opentelemetry.io/otel/sdk/resource"
+	"google.golang.org/grpc/credentials"
+
+	"go.opentelemetry.io/contrib/otelconf/internal/tls"
+)
+
+func loggerProvider(cfg configOptions, res *resource.Resource) (log.LoggerProvider, shutdownFunc, error) {
+	if cfg.opentelemetryConfig.LoggerProvider == nil {
+		return noop.NewLoggerProvider(), noopShutdown, nil
+	}
+	provider, ok := cfg.opentelemetryConfig.LoggerProvider.(*LoggerProviderJson)
+	if !ok {
+		return noop.NewLoggerProvider(), noopShutdown, newErrInvalid("logger_provider")
+	}
+	opts := append(cfg.loggerProviderOptions, sdklog.WithResource(res))
+
+	var errs []error
+	for _, processor := range provider.Processors {
+		sp, err := logProcessor(cfg.ctx, processor)
+		if err == nil {
+			opts = append(opts, sdklog.WithProcessor(sp))
+		} else {
+			errs = append(errs, err)
+		}
+	}
+
+	if len(errs) > 0 {
+		return noop.NewLoggerProvider(), noopShutdown, errors.Join(errs...)
+	}
+
+	lp := sdklog.NewLoggerProvider(opts...)
+	return lp, lp.Shutdown, nil
+}
+
+func logProcessor(ctx context.Context, processor LogRecordProcessor) (sdklog.Processor, error) {
+	if processor.Batch != nil && processor.Simple != nil {
+		return nil, newErrInvalid("must not specify multiple log processor type")
+	}
+	if processor.Batch != nil {
+		exp, err := logExporter(ctx, processor.Batch.Exporter)
+		if err != nil {
+			return nil, err
+		}
+		return batchLogProcessor(processor.Batch, exp)
+	}
+	if processor.Simple != nil {
+		exp, err := logExporter(ctx, processor.Simple.Exporter)
+		if err != nil {
+			return nil, err
+		}
+		return sdklog.NewSimpleProcessor(exp), nil
+	}
+	return nil, newErrInvalid("unsupported log processor type, must be one of simple or batch")
+}
+
+func logExporter(ctx context.Context, exporter LogRecordExporter) (sdklog.Exporter, error) {
+	exportersConfigured := 0
+	var exportFunc func() (sdklog.Exporter, error)
+
+	if exporter.Console != nil {
+		exportersConfigured++
+		exportFunc = func() (sdklog.Exporter, error) {
+			return stdoutlog.New(
+				stdoutlog.WithPrettyPrint(),
+			)
+		}
+	}
+
+	if exporter.OTLPHttp != nil {
+		exportersConfigured++
+		exportFunc = func() (sdklog.Exporter, error) {
+			return otlpHTTPLogExporter(ctx, exporter.OTLPHttp)
+		}
+	}
+	if exporter.OTLPGrpc != nil {
+		exportersConfigured++
+		exportFunc = func() (sdklog.Exporter, error) {
+			return otlpGRPCLogExporter(ctx, exporter.OTLPGrpc)
+		}
+	}
+	if exporter.OTLPFileDevelopment != nil {
+		// TODO: implement file exporter https://github.com/open-telemetry/opentelemetry-go/issues/5408
+		return nil, newErrInvalid("otlp_file/development")
+	}
+
+	if exportersConfigured > 1 {
+		return nil, newErrInvalid("must not specify multiple exporters")
+	}
+
+	if exportFunc != nil {
+		return exportFunc()
+	}
+
+	return nil, newErrInvalid("no valid log exporter")
+}
+
+func batchLogProcessor(blp *BatchLogRecordProcessor, exp sdklog.Exporter) (*sdklog.BatchProcessor, error) {
+	var opts []sdklog.BatchProcessorOption
+	if err := validateBatchLogRecordProcessor(blp); err != nil {
+		return nil, err
+	}
+	if blp.ExportTimeout != nil {
+		opts = append(opts, sdklog.WithExportTimeout(time.Millisecond*time.Duration(*blp.ExportTimeout)))
+	}
+	if blp.MaxExportBatchSize != nil {
+		opts = append(opts, sdklog.WithExportMaxBatchSize(*blp.MaxExportBatchSize))
+	}
+	if blp.MaxQueueSize != nil {
+		opts = append(opts, sdklog.WithMaxQueueSize(*blp.MaxQueueSize))
+	}
+
+	if blp.ScheduleDelay != nil {
+		opts = append(opts, sdklog.WithExportInterval(time.Millisecond*time.Duration(*blp.ScheduleDelay)))
+	}
+
+	return sdklog.NewBatchProcessor(exp, opts...), nil
+}
+
+func otlpHTTPLogExporter(ctx context.Context, otlpConfig *OTLPHttpExporter) (sdklog.Exporter, error) {
+	var opts []otlploghttp.Option
+
+	if otlpConfig.Endpoint != nil {
+		u, err := url.ParseRequestURI(*otlpConfig.Endpoint)
+		if err != nil {
+			return nil, errors.Join(newErrInvalid("endpoint parsing failed"), err)
+		}
+		opts = append(opts, otlploghttp.WithEndpoint(u.Host))
+
+		if u.Scheme == "http" {
+			opts = append(opts, otlploghttp.WithInsecure())
+		}
+		if u.Path != "" {
+			opts = append(opts, otlploghttp.WithURLPath(u.Path))
+		}
+	}
+	if otlpConfig.Compression != nil {
+		switch *otlpConfig.Compression {
+		case compressionGzip:
+			opts = append(opts, otlploghttp.WithCompression(otlploghttp.GzipCompression))
+		case compressionNone:
+			opts = append(opts, otlploghttp.WithCompression(otlploghttp.NoCompression))
+		default:
+			return nil, newErrInvalid(fmt.Sprintf("unsupported compression %q", *otlpConfig.Compression))
+		}
+	}
+	if otlpConfig.Timeout != nil && *otlpConfig.Timeout > 0 {
+		opts = append(opts, otlploghttp.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
+	}
+	headersConfig, err := createHeadersConfig(otlpConfig.Headers, otlpConfig.HeadersList)
+	if err != nil {
+		return nil, err
+	}
+	if len(headersConfig) > 0 {
+		opts = append(opts, otlploghttp.WithHeaders(headersConfig))
+	}
+
+	tlsConfig, err := tls.CreateConfig(otlpConfig.CertificateFile, otlpConfig.ClientCertificateFile, otlpConfig.ClientKeyFile)
+	if err != nil {
+		return nil, errors.Join(newErrInvalid("tls configuration"), err)
+	}
+	opts = append(opts, otlploghttp.WithTLSClientConfig(tlsConfig))
+
+	return otlploghttp.New(ctx, opts...)
+}
+
+func otlpGRPCLogExporter(ctx context.Context, otlpConfig *OTLPGrpcExporter) (sdklog.Exporter, error) {
+	var opts []otlploggrpc.Option
+
+	if otlpConfig.Endpoint != nil {
+		u, err := url.ParseRequestURI(*otlpConfig.Endpoint)
+		if err != nil {
+			return nil, errors.Join(newErrInvalid("endpoint parsing failed"), err)
+		}
+		// ParseRequestURI leaves the Host field empty when no
+		// scheme is specified (i.e. localhost:4317). This check is
+		// here to support the case where a user may not specify a
+		// scheme. The code does its best effort here by using
+		// otlpConfig.Endpoint as-is in that case
+		if u.Host != "" {
+			opts = append(opts, otlploggrpc.WithEndpoint(u.Host))
+		} else {
+			opts = append(opts, otlploggrpc.WithEndpoint(*otlpConfig.Endpoint))
+		}
+		if u.Scheme == "http" || (u.Scheme != "https" && otlpConfig.Insecure != nil && *otlpConfig.Insecure) {
+			opts = append(opts, otlploggrpc.WithInsecure())
+		}
+	}
+	if otlpConfig.Compression != nil {
+		switch *otlpConfig.Compression {
+		case compressionGzip:
+			opts = append(opts, otlploggrpc.WithCompressor(*otlpConfig.Compression))
+		case compressionNone:
+			// none requires no options
+		default:
+			return nil, newErrInvalid(fmt.Sprintf("unsupported compression %q", *otlpConfig.Compression))
+		}
+	}
+	if otlpConfig.Timeout != nil && *otlpConfig.Timeout > 0 {
+		opts = append(opts, otlploggrpc.WithTimeout(time.Millisecond*time.Duration(*otlpConfig.Timeout)))
+	}
+	headersConfig, err := createHeadersConfig(otlpConfig.Headers, otlpConfig.HeadersList)
+	if err != nil {
+		return nil, err
+	}
+	if len(headersConfig) > 0 {
+		opts = append(opts, otlploggrpc.WithHeaders(headersConfig))
+	}
+
+	if otlpConfig.CertificateFile != nil || otlpConfig.ClientCertificateFile != nil || otlpConfig.ClientKeyFile != nil {
+		tlsConfig, err := tls.CreateConfig(otlpConfig.CertificateFile, otlpConfig.ClientCertificateFile, otlpConfig.ClientKeyFile)
+		if err != nil {
+			return nil, errors.Join(newErrInvalid("tls configuration"), err)
+		}
+		opts = append(opts, otlploggrpc.WithTLSCredentials(credentials.NewTLS(tlsConfig)))
+	}
+
+	return otlploggrpc.New(ctx, opts...)
+}