Documentation
¶
Overview ¶
Package observability wires structured logging, Prometheus metrics, and OpenTelemetry tracing for the Dexaflow control plane (ADR 0010).
Index ¶
- func NewLogger(level, format string) *slog.Logger
- func WithLegacyNames(g prometheus.Gatherer) prometheus.Gatherer
- type Config
- type Metrics
- func (m *Metrics) ObservePodCreation(dagID, result string, d time.Duration)
- func (m *Metrics) ObserveRedisDialDuration(d time.Duration)
- func (m *Metrics) ObserveRetentionCycle(d time.Duration)
- func (m *Metrics) ObserveSchedulerLoop(d time.Duration)
- func (m *Metrics) ObserveSchedulerReacquire(d time.Duration)
- func (m *Metrics) RecordAgentTokenRejection(reason string)
- func (m *Metrics) RecordAlert(dagID, channelType, result string)
- func (m *Metrics) RecordCIOIDCExchange(issuer, outcome string)
- func (m *Metrics) RecordDispatchAtCapacity()
- func (m *Metrics) RecordDispatchInnerError()
- func (m *Metrics) RecordDispatchLatencySeconds(seconds float64)
- func (m *Metrics) RecordDispatchQueueDepth(depth int)
- func (m *Metrics) RecordHTTPRequest(method, path string, status int, dur time.Duration)
- func (m *Metrics) RecordInfraOverride(mark string)
- func (m *Metrics) RecordLegacyAttemptToken()
- func (m *Metrics) RecordRedisCommandFailure(reason string)
- func (m *Metrics) RecordRedisDialFailure(reason string)
- func (m *Metrics) RecordRedisPoolTimeout()
- func (m *Metrics) RecordRetentionDeleted(table string, n int64)
- func (m *Metrics) RecordRetentionEligible(table string, n int64)
- func (m *Metrics) RecordSchedulerDecision(decisionType string)
- func (m *Metrics) RecordSchedulerStepDown(reason string)
- func (m *Metrics) RecordSchedulerWokenTick()
- func (m *Metrics) RecordTLSReloadFailure(listener string)
- func (m *Metrics) RecordTaskDuration(dagID, taskID, taskType string, seconds float64)
- func (m *Metrics) RecordTaskRetry(dagID, taskID string)
- func (m *Metrics) RecordTaskTransition(from, to, dagID string)
- func (m *Metrics) RecordUndispatchable(reason string)
- func (m *Metrics) RecordUnitMisfit(stage string)
- func (m *Metrics) RecordUnpinnedImage(stage, outcome string)
- func (m *Metrics) UpdateRedisPoolStats(active, idle, total uint32)
- type Telemetry
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewLogger ¶
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
// SampleRatio is the share of new root traces kept, in [0, 1].
SampleRatio float64
// SkipProbeSpans drops spans for /healthz, /readyz and /static/*.
SkipProbeSpans bool
}
Config configures observability setup.
type Metrics ¶
type Metrics struct {
// Scheduler
SchedulerLoopDuration prometheus.Histogram
SchedulerWokenTicks prometheus.Counter
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
// AgentTokenRejections counts agent RPCs refused for their bearer token, by
// reason (#1125).
AgentTokenRejections *prometheus.CounterVec
// 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
// UnpinnedImages counts images not pinned by digest that
// deploy.require_digest warned about or refused, by stage (register,
// set_current) and outcome (warned, refused) (#1703).
UnpinnedImages *prometheus.CounterVec
// CIOIDCExchanges counts CI OIDC token exchanges by issuer entry
// ("unknown" when the token named none) and outcome (ADR 0092).
CIOIDCExchanges *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
// Retention janitor (D5): rows deleted per table, rows a dry run found
// eligible per table, and the duration of each cycle.
RetentionRowsDeleted *prometheus.CounterVec
RetentionRowsEligible *prometheus.GaugeVec
RetentionCycleDuration prometheus.Histogram
// TLSReloadFailures counts certificate or key files that changed on disk
// but failed to load, by listener (api, agent_grpc). The listener keeps
// serving the last good pair, so a non-zero rate means a rotation is stuck
// and the served certificate will expire unless the files are fixed (#637).
TLSReloadFailures *prometheus.CounterVec
}
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) ObservePodCreation ¶ added in v0.5.4
ObservePodCreation counts one task pod create by dag and result ("success" or "error", ADR 0010) and observes the API call's latency. It satisfies executor.PodCreationRecorder.
func (*Metrics) ObserveRedisDialDuration ¶
ObserveRedisDialDuration records one TCP/TLS dial latency sample.
func (*Metrics) ObserveRetentionCycle ¶ added in v0.5.4
ObserveRetentionCycle observes the duration of one retention cycle.
func (*Metrics) ObserveSchedulerLoop ¶ added in v0.5.4
ObserveSchedulerLoop records the wall time of one leader scheduler tick, the SLI dexaflow_scheduler_loop_duration_seconds exists for (#1186). It satisfies scheduler.Recorder.
func (*Metrics) ObserveSchedulerReacquire ¶
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) RecordAgentTokenRejection ¶ added in v0.5.5
RecordAgentTokenRejection counts one agent RPC refused for its bearer token. It satisfies agentrpc.TokenRejectionRecorder.
func (*Metrics) RecordAlert ¶
RecordAlert counts one on-failure alert dispatch (#424) by dag, channel type, and outcome ("sent" or "failed"). Satisfies failurealert.Recorder.
func (*Metrics) RecordCIOIDCExchange ¶ added in v0.5.5
RecordCIOIDCExchange counts one CI OIDC token exchange. It satisfies api.CIOIDCExchangeRecorder.
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 ¶
RecordDispatchLatencySeconds observes one end-to-end dispatch latency, from enqueue to the worker's inner-dispatcher return, for every buffered dispatch (successful or not).
func (*Metrics) RecordDispatchQueueDepth ¶
RecordDispatchQueueDepth sets the current depth of the buffered dispatch queue (#127). Sampled on every successful enqueue.
func (*Metrics) RecordHTTPRequest ¶
RecordHTTPRequest records a completed HTTP request (count + duration).
func (*Metrics) RecordInfraOverride ¶ added in v0.5.2
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 ¶
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 ¶
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) RecordRetentionDeleted ¶ added in v0.5.4
RecordRetentionDeleted counts n rows the retention janitor deleted from table.
func (*Metrics) RecordRetentionEligible ¶ added in v0.5.4
RecordRetentionEligible sets the rows a retention dry run found eligible in table.
func (*Metrics) RecordSchedulerDecision ¶
RecordSchedulerDecision records one scheduler decision by type.
func (*Metrics) RecordSchedulerStepDown ¶
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) RecordSchedulerWokenTick ¶ added in v0.5.4
func (m *Metrics) RecordSchedulerWokenTick()
RecordSchedulerWokenTick counts one scheduler tick started early by a task completion (scheduler.eager_promotion).
func (*Metrics) RecordTLSReloadFailure ¶ added in v0.5.5
RecordTLSReloadFailure counts one rotated certificate pair that failed to load on the given listener. It is a no-op on a nil receiver so a caller without metrics wired needs no guard.
func (*Metrics) RecordTaskDuration ¶
RecordTaskDuration records how long a task took to execute, in seconds.
func (*Metrics) RecordTaskRetry ¶ added in v0.5.4
RecordTaskRetry counts one retry the scheduler released for a task (its up_for_retry reset to none with a new try number). It satisfies scheduler.Recorder.
func (*Metrics) RecordTaskTransition ¶
RecordTaskTransition records a task instance state transition.
func (*Metrics) RecordUndispatchable ¶
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
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) RecordUnpinnedImage ¶ added in v0.5.4
RecordUnpinnedImage counts one image not pinned by digest that deploy.require_digest warned about or refused at stage. It satisfies api.UnpinnedImageRecorder.
func (*Metrics) UpdateRedisPoolStats ¶
UpdateRedisPoolStats refreshes the three pool gauges. The cmd/dexaflow-server goroutine that calls this scrapes go-redis's PoolStats every N seconds.