89 lines
2.7 KiB
Go
89 lines
2.7 KiB
Go
// Package telemetry holds shared OpenTelemetry setup helpers used by the
|
||
// service main packages (cart, checkout, profile, …).
|
||
package telemetry
|
||
|
||
import (
|
||
"context"
|
||
"math/rand/v2"
|
||
"os"
|
||
"strconv"
|
||
|
||
"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
|
||
"go.opentelemetry.io/otel/sdk/log"
|
||
)
|
||
|
||
// NewLoggerProvider builds the OTLP log provider from the environment, or returns
|
||
// (nil, nil) when logs are disabled — in which case the caller MUST skip
|
||
// global.SetLoggerProvider and the shutdown registration.
|
||
//
|
||
// Env:
|
||
//
|
||
// OTEL_LOGS_ENABLED "1"/"true"/"yes"/"on" to enable the OTLP log pipeline.
|
||
// Default OFF: the batch processor clones its record ring
|
||
// on every export, which profiling showed to be the
|
||
// dominant heap allocator in these services. Traces and
|
||
// metrics are always on regardless of this flag.
|
||
// OTEL_LOGS_SAMPLE_RATIO fraction of records to export, 0.0–1.0 (default 1.0).
|
||
// e.g. 0.1 keeps ~10% of logs. Dropped records never reach
|
||
// the batch ring, so allocation churn falls ~proportionally.
|
||
func NewLoggerProvider(ctx context.Context) (*log.LoggerProvider, error) {
|
||
if !logsEnabled() {
|
||
return nil, nil
|
||
}
|
||
exporter, err := otlploggrpc.New(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var proc log.Processor = log.NewBatchProcessor(exporter)
|
||
if ratio := sampleRatio(); ratio < 1.0 {
|
||
proc = &samplingProcessor{next: proc, ratio: ratio}
|
||
}
|
||
return log.NewLoggerProvider(log.WithProcessor(proc)), nil
|
||
}
|
||
|
||
func logsEnabled() bool {
|
||
switch os.Getenv("OTEL_LOGS_ENABLED") {
|
||
case "1", "true", "TRUE", "True", "yes", "on":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func sampleRatio() float64 {
|
||
v := os.Getenv("OTEL_LOGS_SAMPLE_RATIO")
|
||
if v == "" {
|
||
return 1.0
|
||
}
|
||
r, err := strconv.ParseFloat(v, 64)
|
||
if err != nil || r < 0 {
|
||
return 1.0
|
||
}
|
||
if r > 1 {
|
||
return 1.0
|
||
}
|
||
return r
|
||
}
|
||
|
||
// samplingProcessor forwards a random `ratio` fraction of records to the wrapped
|
||
// processor and drops the rest before they reach the (allocation-heavy) batch
|
||
// ring. This is head sampling: cheap, stateless, and per-record independent.
|
||
type samplingProcessor struct {
|
||
next log.Processor
|
||
ratio float64
|
||
}
|
||
|
||
func (s *samplingProcessor) Enabled(ctx context.Context, param log.EnabledParameters) bool {
|
||
return s.next.Enabled(ctx, param)
|
||
}
|
||
|
||
func (s *samplingProcessor) OnEmit(ctx context.Context, record *log.Record) error {
|
||
if rand.Float64() >= s.ratio {
|
||
return nil // dropped by sampling
|
||
}
|
||
return s.next.OnEmit(ctx, record)
|
||
}
|
||
|
||
func (s *samplingProcessor) Shutdown(ctx context.Context) error { return s.next.Shutdown(ctx) }
|
||
func (s *samplingProcessor) ForceFlush(ctx context.Context) error { return s.next.ForceFlush(ctx) }
|