go-app/pkg/otel/otel.go

package otel

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	opentelemetry "go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/exporters/prometheus"
	"go.opentelemetry.io/otel/exporters/stdout/stdoutmetric"
	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"

	"gitea.libretechconsulting.com/rmcguire/go-app/pkg/config"

	noopMetric "go.opentelemetry.io/otel/metric/noop"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/sdk/metric"
	"go.opentelemetry.io/otel/sdk/resource"
	traceSDK "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
	trace "go.opentelemetry.io/otel/trace"
	"go.opentelemetry.io/otel/trace/noop"
)

// OTEL Options
var (
	EnableStdoutExporter     Option = enableStdoutExporter{}
	EnablePrometheusExporter Option = enablePrometheusExporter{}
	// Overide the default metric export interval
	WithMetricExportInterval = func(interval time.Duration) Option {
		return exportInterval{interval: interval}
	}
)

const defMetricInterval = 15 * time.Second

// Context must carry config.AppConfig
func Init(ctx context.Context) (context.Context, func(context.Context) error) {
	cfg := config.MustFromCtx(ctx)

	// Nothing to do here if not enabled
	if !cfg.OTEL.Enabled {
		opentelemetry.SetMeterProvider(noopMetric.NewMeterProvider())
		opentelemetry.SetTracerProvider(noop.NewTracerProvider())
		// Won't function with noop providers
		meter := opentelemetry.Meter(cfg.Name)
		ctx = AddMeterToCtx(ctx, meter)
		tracer := opentelemetry.Tracer(cfg.Name)
		ctx = AddTracerToCtx(ctx, tracer)
		return ctx, func(context.Context) error {
			return nil
		}
	}

	var (
		metricInterval = defMetricInterval
		options        = make([]Option, 0)
		s              = &settings{}
		shutdownFuncs  []func(context.Context) error
	)

	// Prepare settings for OTEL from configuration
	if cfg.OTEL.StdoutEnabled {
		options = append(options, EnableStdoutExporter)
	}
	if cfg.OTEL.PrometheusEnabled {
		options = append(options, EnablePrometheusExporter)
	}
	if cfg.OTEL.MetricIntervalSecs > 0 {
		metricInterval = time.Duration(cfg.OTEL.MetricIntervalSecs) * time.Second
	}
	options = append(options,
		WithMetricExportInterval(metricInterval))

	// Apply settings
	for _, opt := range options {
		opt.apply(s)
	}

	// shutdown calls cleanup functions registered via shutdownFuncs.
	// The errors from the calls are joined.
	// Each registered cleanup will be invoked once.
	shutdown := func(ctx context.Context) error {
		var err error
		for _, fn := range shutdownFuncs {
			err = errors.Join(err, fn(ctx))
		}
		shutdownFuncs = nil
		return err
	}

	// handleErr calls shutdown for cleanup and makes sure that all errors are returned.
	handleErr := func(inErr error) {
		if err := errors.Join(inErr, shutdown(ctx)); err != nil {
			fmt.Fprintln(os.Stderr, "OTEL Error:", err)
		}
	}

	// Set up meter provider.
	meterProvider, err := s.newMeterProvider(ctx)
	if err != nil {
		handleErr(err)
		return ctx, shutdown
	}

	shutdownFuncs = append(shutdownFuncs, meterProvider.Shutdown)
	opentelemetry.SetMeterProvider(meterProvider)

	meter := opentelemetry.Meter(cfg.Name)
	ctx = AddMeterToCtx(ctx, meter)

	// Set up tracing
	opentelemetry.SetTextMapPropagator(newPropagator())
	var tracerProvider *traceSDK.TracerProvider
	tracerProvider, err = s.newTracerProvider(ctx)
	if err != nil {
		handleErr(err)
		return ctx, shutdown
	}
	shutdownFuncs = append(shutdownFuncs, tracerProvider.Shutdown)
	opentelemetry.SetTracerProvider(tracerProvider)

	tracer := opentelemetry.Tracer(cfg.Name)
	ctx = AddTracerToCtx(ctx, tracer)

	return ctx, shutdown
}

func newPropagator() propagation.TextMapPropagator {
	return propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	)
}

func (s *settings) newTracerProvider(ctx context.Context) (traceProvider *traceSDK.TracerProvider, err error) {
	traceOpts := []traceSDK.TracerProviderOption{
		traceSDK.WithResource(newResource(ctx)),
	}

	host, set := os.LookupEnv("OTEL_EXPORTER_OTLP_ENDPOINT")
	if set && host != "" {
		exporter, err := otlptracegrpc.New(ctx)
		if err != nil {
			return nil, err
		}
		traceOpts = append(traceOpts, traceSDK.WithBatcher(exporter))
	}

	if s.EnableStdoutExporter {
		stdoutExporter, err := stdouttrace.New(stdouttrace.WithPrettyPrint())
		if err != nil {
			return nil, err
		}
		traceOpts = append(traceOpts, traceSDK.WithBatcher(stdoutExporter))
	}

	traceProvider = traceSDK.NewTracerProvider(traceOpts...)

	return
}

func newResource(ctx context.Context) *resource.Resource {
	cfg := config.MustFromCtx(ctx)
	attributes := []attribute.KeyValue{
		semconv.ServiceName(cfg.Name),
		semconv.ServiceVersion(cfg.Version),
		semconv.K8SPodName(os.Getenv("HOSTNAME")),
	}

	return resource.NewWithAttributes(semconv.SchemaURL, attributes...)
}

// Configures meter provider
// Always provides a prometheus metrics exporter
// Conditionally provides an OTLP metrics exporter
func (s *settings) newMeterProvider(ctx context.Context) (*metric.MeterProvider, error) {
	// OTEL Prometheus Exporter
	exporter, err := prometheus.New(
		prometheus.WithResourceAsConstantLabels(attribute.NewDenyKeysFilter()),
	)
	if err != nil {
		return nil, err
	}

	metricOptions := make([]metric.Option, 0, 5)
	if s.EnableStdoutExporter {
		stdoutMetricExporter, err := stdoutmetric.New()
		if err != nil {
			return nil, err
		}
		metricOptions = append(metricOptions,
			metric.WithReader(metric.NewPeriodicReader(stdoutMetricExporter)),
		)
	}

	host, set := os.LookupEnv("OTEL_EXPORTER_OTLP_ENDPOINT")
	var otlpExporter *otlpmetricgrpc.Exporter
	if set && host != "" {
		if exp, err := otlpmetricgrpc.New(ctx, otlpmetricgrpc.WithInsecure()); err != nil {
			return nil, fmt.Errorf("otlpmetricgrpc.New: %w", err)
		} else {
			otlpExporter = exp
		}
	}

	var meterProvider *metric.MeterProvider
	if otlpExporter != nil {
		metricOptions = append(metricOptions,
			metric.WithResource(newResource(ctx)),
			metric.WithReader(exporter),
			metric.WithReader(
				metric.NewPeriodicReader(
					otlpExporter,
					metric.WithInterval(s.MetricExportInterval),
				),
			),
		)
	} else {
		metricOptions = append(metricOptions,
			metric.WithResource(newResource(ctx)),
			metric.WithReader(exporter),
		)
	}

	meterProvider = metric.NewMeterProvider(metricOptions...)

	return meterProvider, nil
}

// Creates a new tracer from the global opentelemetry provider
func NewTracer(options ...trace.TracerOption) trace.Tracer {
	return opentelemetry.GetTracerProvider().Tracer(
		os.Getenv("OTEL_SERVICE_NAME"),
		options...,
	)
}