Help us shape the future of Go! Take the 2026 Go Developer Survey

observability

package
v0.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 8, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Package observability wires structured logging, Prometheus metrics, and OpenTelemetry tracing for the Dexaflow control plane (ADR 0010).

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NewLogger 

func NewLogger(level, format string) *slog.Logger

NewLogger builds a slog.Logger writing to stdout at the given level. A format of "text" uses the human-readable handler; anything else (default "json") uses the JSON handler required for production (ADR 0010).

func WithLegacyNames 

func WithLegacyNames(g prometheus.Gatherer) prometheus.Gatherer

WithLegacyNames wraps g so each dexaflow_* metric family is also exposed as leoflow_*, with identical help, type, labels and values. Dashboards, alerts and recording rules written before the rename keep working against the same scrape; other families pass through once.

Types

type Config 

type Config struct {
	ServiceName  string
	LogLevel     string
	LogFormat    string
	OTelEnabled  bool
	OTelEndpoint string
}

Config configures observability setup.

type Metrics 

type Metrics struct {
	// Scheduler
	SchedulerLoopDuration prometheus.Histogram
	SchedulerDecisions    *prometheus.CounterVec
	TasksUndispatchable   *prometheus.CounterVec
	SchedulerStepDowns    *prometheus.CounterVec // #311 leader churn observability
	SchedulerReacquire    prometheus.Histogram   // #311 step-down → re-acquire latency
	AlertsDispatched      *prometheus.CounterVec // #424 on-failure alerts, by outcome

	// Task lifecycle
	TaskStateTransitions    *prometheus.CounterVec
	TaskDuration            *prometheus.HistogramVec
	TaskRetries             *prometheus.CounterVec
	TaskPodCreationDuration prometheus.Histogram
	TaskColdStart           *prometheus.HistogramVec

	// XCom
	XComSize     *prometheus.HistogramVec
	XComPush     *prometheus.CounterVec
	XComPull     *prometheus.CounterVec
	XComRejected *prometheus.CounterVec

	// API
	HTTPRequests        *prometheus.CounterVec
	HTTPRequestDuration *prometheus.HistogramVec
	AuthFailures        *prometheus.CounterVec

	// Agent credentials: task tokens authenticated without an attempt_epoch
	// claim, accepted under the epoch-0 rule (ADR 0051 amendment).
	AgentLegacyAttemptTokens prometheus.Counter
	// Reconciler: durable SUCCESS records settled over a reaper's provisional
	// infra mark, by the mark overridden (ADR 0052 amendment).
	ReconcileInfraOverrides *prometheus.CounterVec

	// Executor (Kubernetes)
	PodsCreated        *prometheus.CounterVec
	PodPendingDuration prometheus.Histogram
	KubernetesAPICalls *prometheus.CounterVec

	// Dispatch pool (#127, ADR 0031): depth gauge, capacity-saturation counter,
	// per-dispatch latency, inner-dispatcher error counter.
	DispatchQueueDepth  prometheus.Gauge
	DispatchAtCapacity  prometheus.Counter
	DispatchLatency     prometheus.Histogram
	DispatchInnerErrors prometheus.Counter
	// UnitMisfits counts tasks let through under executor.unit.enforce=warn
	// although they do not fit their size, by stage (register, dispatch;
	// ADR 0066 §3).
	UnitMisfits *prometheus.CounterVec

	// Redis observability — port of the #311 step-down pattern for Redis (Pro
	// only; Lite uses Postgres + in-process tailer per ADR 0026, so these
	// gauges stay at 0). Operators alert on the per-reason rate, not on log
	// content: a sudden rate spike on RedisCommandFailures{reason="timeout"}
	// or RedisDialFailures{reason="tls_handshake"} catches an outage before
	// the surrounding components (XCom, log tailer) start surfacing tail
	// errors to users.
	RedisCommandFailures *prometheus.CounterVec
	RedisDialFailures    *prometheus.CounterVec
	RedisDialDuration    prometheus.Histogram
	RedisPoolActive      prometheus.Gauge
	RedisPoolIdle        prometheus.Gauge
	RedisPoolTotalConns  prometheus.Gauge
	RedisPoolTimeouts    prometheus.Counter
}

Metrics holds every Prometheus collector required by ADR 0010. Construct it once with NewMetrics and pass it to the components that emit values.

func NewMetrics 

func NewMetrics(reg prometheus.Registerer) *Metrics

NewMetrics registers every ADR 0010 collector with reg and returns the set.

func (*Metrics) ObserveRedisDialDuration 

func (m *Metrics) ObserveRedisDialDuration(d time.Duration)

ObserveRedisDialDuration records one TCP/TLS dial latency sample.

func (*Metrics) ObserveSchedulerReacquire 

func (m *Metrics) ObserveSchedulerReacquire(d time.Duration)

ObserveSchedulerReacquire records the wall-clock duration of a step-down → re-acquire cycle (#311). Use the histogram's P99 in alerts: a P99 growing past a second indicates churn that is starting to delay scheduling.

func (*Metrics) RecordAlert 

func (m *Metrics) RecordAlert(dagID, channelType, result string)

RecordAlert counts one on-failure alert dispatch (#424) by dag, channel type, and outcome ("sent" or "failed"). Satisfies failurealert.Recorder.

func (*Metrics) RecordDispatchAtCapacity 

func (m *Metrics) RecordDispatchAtCapacity()

RecordDispatchAtCapacity counts one rejection because the buffered dispatch queue was full. Each rejection means the scheduler will retry on the next tick — a rising rate signals the pool needs more workers or a deeper queue.

func (*Metrics) RecordDispatchInnerError 

func (m *Metrics) RecordDispatchInnerError()

RecordDispatchInnerError counts one error returned by the inner dispatcher inside a worker — typically a Kubernetes API failure or pod-create rejection.

func (*Metrics) RecordDispatchLatencySeconds 

func (m *Metrics) RecordDispatchLatencySeconds(seconds float64)

RecordDispatchLatencySeconds observes one end-to-end dispatch latency, from enqueue to the worker's inner-dispatcher return. Reserved for future use by the BufferedDispatcher.

func (*Metrics) RecordDispatchQueueDepth 

func (m *Metrics) RecordDispatchQueueDepth(depth int)

RecordDispatchQueueDepth sets the current depth of the buffered dispatch queue (#127). Sampled on every successful enqueue.

func (*Metrics) RecordHTTPRequest 

func (m *Metrics) RecordHTTPRequest(method, path string, status int, dur time.Duration)

RecordHTTPRequest records a completed HTTP request (count + duration).

func (*Metrics) RecordInfraOverride added in v0.5.2 

func (m *Metrics) RecordInfraOverride(mark string)

RecordInfraOverride counts one durable SUCCESS settled over an infra mark. It satisfies executor.InfraOverrideRecorder.

func (*Metrics) RecordLegacyAttemptToken added in v0.5.1 

func (m *Metrics) RecordLegacyAttemptToken()

RecordLegacyAttemptToken counts one agent RPC authenticated by a task token without an attempt_epoch claim (ADR 0051 amendment). It satisfies agentrpc.LegacyTokenRecorder.

func (*Metrics) RecordRedisCommandFailure 

func (m *Metrics) RecordRedisCommandFailure(reason string)

RecordRedisCommandFailure increments the per-reason command failure counter (#312 sibling, paralleling #311's step-down reason). Caller is the go-redis hook's ProcessHook — it classifies the error before incrementing so the label cardinality stays bounded.

func (*Metrics) RecordRedisDialFailure 

func (m *Metrics) RecordRedisDialFailure(reason string)

RecordRedisDialFailure increments the per-reason dial failure counter. Separated from command failures because a dial failure means we never even reached the Redis instance — the cause is upstream (DNS, TLS, network), not Redis itself.

func (*Metrics) RecordRedisPoolTimeout 

func (m *Metrics) RecordRedisPoolTimeout()

RecordRedisPoolTimeout increments the pool-checkout timeout counter. The go-redis hook fires this when a caller waited PoolTimeout for an idle connection — the leading indicator that the pool is too small for the command rate.

func (*Metrics) RecordSchedulerDecision 

func (m *Metrics) RecordSchedulerDecision(decisionType string)

RecordSchedulerDecision records one scheduler decision by type.

func (*Metrics) RecordSchedulerStepDown 

func (m *Metrics) RecordSchedulerStepDown(reason string)

RecordSchedulerStepDown increments the leader step-down counter (#311). The reason label lets operators distinguish a transient lock_released (pgx blip) from a check_timeout (the lock-check failed repeatedly) or a clean shutdown, so they can alert on the rate of the noisy reasons without paging on expected shutdowns.

func (*Metrics) RecordTaskDuration 

func (m *Metrics) RecordTaskDuration(dagID, taskID, taskType string, seconds float64)

RecordTaskDuration records how long a task took to execute, in seconds.

func (*Metrics) RecordTaskTransition 

func (m *Metrics) RecordTaskTransition(from, to, dagID string)

RecordTaskTransition records a task instance state transition.

func (*Metrics) RecordUndispatchable 

func (m *Metrics) RecordUndispatchable(reason string)

RecordUndispatchable records a task that became queued but has no executor to launch it (e.g. pod dispatch disabled), so an operator can distinguish a resource/config gap from an actual bug.

func (*Metrics) RecordUnitMisfit added in v0.5.2 

func (m *Metrics) RecordUnitMisfit(stage string)

RecordUnitMisfit counts one task let through under executor.unit.enforce=warn although it does not fit its size. It satisfies dispatch.UnitMisfitRecorder and api.UnitMisfitRecorder.

func (*Metrics) UpdateRedisPoolStats 

func (m *Metrics) UpdateRedisPoolStats(active, idle, total uint32)

UpdateRedisPoolStats refreshes the three pool gauges. The cmd/dexaflow-server goroutine that calls this scrapes go-redis's PoolStats every N seconds.

type Telemetry 

type Telemetry struct {
	Logger   *slog.Logger
	Metrics  *Metrics
	Registry *prometheus.Registry
	Tracer   trace.Tracer
}

Telemetry bundles the configured observability primitives.

func Setup 

func Setup(ctx context.Context, cfg Config) (*Telemetry, func(), error)

Setup builds logging, metrics, and tracing from cfg and returns the telemetry bundle plus a shutdown function the caller must defer.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.