go/extra/redisotel-native/redisotel.go

package redisotel

import (
	"errors"
	"fmt"
	"strings"
	"sync"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/metric"
	"go.opentelemetry.io/otel/semconv/v1.38.0/dbconv"

	"github.com/redis/go-redis/v9"
)

// Metric name constants
const (
	// Operation metrics
	MetricOperationDuration = "db.client.operation.duration"

	// Connection metrics
	MetricConnectionCount          = "db.client.connection.count"
	MetricConnectionCreateTime     = "db.client.connection.create_time"
	MetricConnectionWaitTime       = "db.client.connection.wait_time"
	MetricConnectionPendingReqs    = "db.client.connection.pending_requests"
	MetricConnectionRelaxedTimeout = "redis.client.connection.relaxed_timeout"
	MetricConnectionHandoff        = "redis.client.connection.handoff"
	MetricConnectionClosed         = "redis.client.connection.closed"

	// Resiliency metrics
	MetricClientErrors             = "redis.client.errors"
	MetricMaintenanceNotifications = "redis.client.maintenance.notifications"

	// Pub/Sub metrics
	MetricPubSubMessages = "redis.client.pubsub.messages"

	// Stream metrics
	MetricStreamLag = "redis.client.stream.lag"

	// Special pool names
	PoolNameMain   = "main"
	PoolNamePubSub = "pubsub"
)

var (
	// Global observability instance
	observabilityInstance     *ObservabilityInstance
	observabilityInstanceOnce sync.Once
)

// ObservabilityInstance manages the global observability singleton.
type ObservabilityInstance struct {
	mu          sync.RWMutex
	config      *Config
	recorder    *metricsRecorder
	initialized bool
}

// GetObservabilityInstance returns the global observability singleton.
func GetObservabilityInstance() *ObservabilityInstance {
	observabilityInstanceOnce.Do(func() {
		observabilityInstance = &ObservabilityInstance{}
	})
	return observabilityInstance
}

// Init initializes OpenTelemetry observability globally for all Redis clients.
// This should be called once at application startup, BEFORE creating any Redis clients.
// After initialization, all Redis clients will automatically collect and export
// metrics without needing any additional configuration.
func (o *ObservabilityInstance) Init(cfg *Config) error {
	o.mu.Lock()
	defer o.mu.Unlock()

	// If already initialized, return error
	if o.initialized {
		return errors.New("redisotel: already initialized, call Shutdown() before reinitializing")
	}

	o.config = cfg

	if !cfg.Enabled {
		return nil
	}

	// Get meter provider (use global if not provided)
	meterProvider := cfg.MeterProvider
	if meterProvider == nil {
		meterProvider = otel.GetMeterProvider()
	}

	meter := meterProvider.Meter(
		"github.com/redis/go-redis",
		metric.WithInstrumentationVersion(redis.Version()),
	)

	internalCfg := o.configToInternal(cfg)
	recorder, err := o.createRecorder(meter, internalCfg)
	if err != nil {
		return fmt.Errorf("failed to create metrics recorder: %w", err)
	}

	o.recorder = recorder
	o.initialized = true
	redis.SetOTelRecorder(recorder)

	return nil
}

// IsEnabled returns true if observability is initialized and enabled.
func (o *ObservabilityInstance) IsEnabled() bool {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.initialized && o.config != nil && o.config.Enabled
}

// Shutdown cleans up resources and flushes any pending metrics.
// This should be called at application shutdown.
func (o *ObservabilityInstance) Shutdown() error {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.shutdownLocked()
}

func (o *ObservabilityInstance) shutdownLocked() error {
	if !o.initialized {
		return nil
	}

	redis.SetOTelRecorder(nil)

	// Note: We don't shutdown the MeterProvider since it's owned by the application
	// The application should call provider.Shutdown() when appropriate

	o.recorder = nil
	o.initialized = false

	return nil
}

// configToInternal converts the public Config to internal config format
func (o *ObservabilityInstance) configToInternal(cfg *Config) config {
	enabledGroups := make(map[MetricGroup]bool)
	if cfg.MetricGroups&MetricGroupFlagCommand != 0 {
		enabledGroups[MetricGroupCommand] = true
	}
	if cfg.MetricGroups&MetricGroupFlagConnectionBasic != 0 {
		enabledGroups[MetricGroupConnectionBasic] = true
	}
	if cfg.MetricGroups&MetricGroupFlagResiliency != 0 {
		enabledGroups[MetricGroupResiliency] = true
	}
	if cfg.MetricGroups&MetricGroupFlagConnectionAdvanced != 0 {
		enabledGroups[MetricGroupConnectionAdvanced] = true
	}
	if cfg.MetricGroups&MetricGroupFlagPubSub != 0 {
		enabledGroups[MetricGroupPubSub] = true
	}
	if cfg.MetricGroups&MetricGroupFlagStream != 0 {
		enabledGroups[MetricGroupStream] = true
	}

	return config{
		meterProvider:                   cfg.MeterProvider,
		enabled:                         cfg.Enabled,
		enabledMetricGroups:             enabledGroups,
		includeCommands:                 cfg.IncludeCommands,
		excludeCommands:                 cfg.ExcludeCommands,
		hidePubSubChannelNames:          cfg.HidePubSubChannelNames,
		hideStreamNames:                 cfg.HideStreamNames,
		histAggregation:                 cfg.HistogramAggregation,
		bucketsOperationDuration:        cfg.BucketsOperationDuration,
		bucketsStreamProcessingDuration: cfg.BucketsStreamLag,
		bucketsConnectionCreateTime:     cfg.BucketsConnectionCreateTime,
		bucketsConnectionWaitTime:       cfg.BucketsConnectionWaitTime,
	}
}

// createRecorder creates a metricsRecorder with all instruments based on config.
func (o *ObservabilityInstance) createRecorder(meter metric.Meter, cfg config) (*metricsRecorder, error) {
	var err error

	var operationDuration metric.Float64Histogram
	if cfg.isMetricGroupEnabled(MetricGroupCommand) {
		var operationDurationOpts []metric.Float64HistogramOption
		if cfg.histAggregation == HistogramAggregationExplicitBucket {
			operationDurationOpts = append(operationDurationOpts,
				metric.WithExplicitBucketBoundaries(cfg.bucketsOperationDuration...),
			)
		}
		var operationDurationConv dbconv.ClientOperationDuration
		operationDurationConv, err = dbconv.NewClientOperationDuration(meter, operationDurationOpts...)
		if err != nil {
			return nil, fmt.Errorf("failed to create operation duration histogram: %w", err)
		}
		operationDuration = operationDurationConv.Inst()
	}

	var connectionCount metric.Int64UpDownCounter // OTel semconv: UpDownCounter
	var connectionCreateTime metric.Float64Histogram
	var connectionRelaxedTimeout metric.Int64UpDownCounter
	var connectionHandoff metric.Int64Counter

	if cfg.isMetricGroupEnabled(MetricGroupConnectionBasic) {
		connectionCount, err = meter.Int64UpDownCounter(
			dbconv.ClientConnectionCount{}.Name(),
			metric.WithDescription(dbconv.ClientConnectionCount{}.Description()),
			metric.WithUnit(dbconv.ClientConnectionCount{}.Unit()),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection count metric: %w", err)
		}

		var connectionCreateTimeOpts []metric.Float64HistogramOption
		if cfg.histAggregation == HistogramAggregationExplicitBucket {
			connectionCreateTimeOpts = append(connectionCreateTimeOpts,
				metric.WithExplicitBucketBoundaries(cfg.bucketsConnectionCreateTime...),
			)
		}
		var connectionCreateTimeConv dbconv.ClientConnectionCreateTime
		connectionCreateTimeConv, err = dbconv.NewClientConnectionCreateTime(meter, connectionCreateTimeOpts...)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection create time histogram: %w", err)
		}
		connectionCreateTime = connectionCreateTimeConv.Inst()

		connectionRelaxedTimeout, err = meter.Int64UpDownCounter(
			MetricConnectionRelaxedTimeout,
			metric.WithDescription("How many times the connection timeout has been increased/decreased (after a server maintenance notification)"),
			metric.WithUnit("{relaxation}"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection relaxed timeout metric: %w", err)
		}

		connectionHandoff, err = meter.Int64Counter(
			MetricConnectionHandoff,
			metric.WithDescription("Connections that have been handed off to another node (e.g after a MOVING notification)"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection handoff metric: %w", err)
		}
	}

	var clientErrors metric.Int64Counter
	var maintenanceNotifications metric.Int64Counter

	if cfg.isMetricGroupEnabled(MetricGroupResiliency) {
		clientErrors, err = meter.Int64Counter(
			MetricClientErrors,
			metric.WithDescription("Number of errors handled by the Redis client"),
			metric.WithUnit("{error}"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create client errors metric: %w", err)
		}

		maintenanceNotifications, err = meter.Int64Counter(
			MetricMaintenanceNotifications,
			metric.WithDescription("Number of maintenance notifications received"),
			metric.WithUnit("{notification}"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create maintenance notifications metric: %w", err)
		}
	}

	var connectionWaitTime metric.Float64Histogram
	var connectionClosed metric.Int64Counter
	var connectionPendingReqs metric.Int64UpDownCounter // OTel semconv: UpDownCounter

	if cfg.isMetricGroupEnabled(MetricGroupConnectionAdvanced) {
		var connectionWaitTimeOpts []metric.Float64HistogramOption
		if cfg.histAggregation == HistogramAggregationExplicitBucket {
			connectionWaitTimeOpts = append(connectionWaitTimeOpts,
				metric.WithExplicitBucketBoundaries(cfg.bucketsConnectionWaitTime...),
			)
		}
		var connectionWaitTimeConv dbconv.ClientConnectionWaitTime
		connectionWaitTimeConv, err = dbconv.NewClientConnectionWaitTime(meter, connectionWaitTimeOpts...)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection wait time histogram: %w", err)
		}
		connectionWaitTime = connectionWaitTimeConv.Inst()

		connectionClosed, err = meter.Int64Counter(
			MetricConnectionClosed,
			metric.WithDescription("The number of connections that have been closed"),
			metric.WithUnit("{connection}"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection closed metric: %w", err)
		}

		connectionPendingReqs, err = meter.Int64UpDownCounter(
			dbconv.ClientConnectionPendingRequests{}.Name(),
			metric.WithDescription(dbconv.ClientConnectionPendingRequests{}.Description()),
			metric.WithUnit(dbconv.ClientConnectionPendingRequests{}.Unit()),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create connection pending requests metric: %w", err)
		}
	}

	var pubsubMessages metric.Int64Counter

	if cfg.isMetricGroupEnabled(MetricGroupPubSub) {
		pubsubMessages, err = meter.Int64Counter(
			MetricPubSubMessages,
			metric.WithDescription("The number of Pub/Sub messages sent or received"),
			metric.WithUnit("{message}"),
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create Pub/Sub messages metric: %w", err)
		}
	}

	var streamLag metric.Float64Histogram

	if cfg.isMetricGroupEnabled(MetricGroupStream) {
		var streamLagOpts []metric.Float64HistogramOption
		streamLagOpts = append(streamLagOpts,
			metric.WithDescription("The lag between message creation and consumption in a stream consumer group"),
			metric.WithUnit("s"),
		)
		if cfg.histAggregation == HistogramAggregationExplicitBucket {
			streamLagOpts = append(streamLagOpts,
				metric.WithExplicitBucketBoundaries(cfg.bucketsStreamProcessingDuration...),
			)
		}
		streamLag, err = meter.Float64Histogram(
			MetricStreamLag,
			streamLagOpts...,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to create stream lag histogram: %w", err)
		}
	}

	// Create recorder
	recorder := &metricsRecorder{
		operationDuration:        operationDuration,
		connectionCount:          connectionCount,
		connectionCreateTime:     connectionCreateTime,
		connectionRelaxedTimeout: connectionRelaxedTimeout,
		connectionHandoff:        connectionHandoff,
		clientErrors:             clientErrors,
		maintenanceNotifications: maintenanceNotifications,
		connectionWaitTime:       connectionWaitTime,
		connectionClosed:         connectionClosed,
		connectionPendingReqs:    connectionPendingReqs,
		pubsubMessages:           pubsubMessages,
		streamLag:                streamLag,
		cfg:                      &cfg,
	}

	return recorder, nil
}

// parsePoolName extracts server address, port, and database index from a pool name.
// Pool name format: "host:port/db" or "host/db" or "host:port" or "host"
// Returns: (serverAddr, serverPort, dbIndex)
func parsePoolName(poolName string) (string, string, string) {
	// Handle special pool names
	if poolName == PoolNameMain || poolName == PoolNamePubSub {
		return "", "", ""
	}

	parts := strings.Split(poolName, "/")
	addrPart := parts[0]
	dbIndex := ""
	if len(parts) > 1 {
		dbIndex = parts[1]
	}
	host, port := parseAddr(addrPart)
	return host, port, dbIndex
}