Documentation
¶
Overview ¶
Package telemetry defines the metrics interface for the e2a backend. Implementations:
- NoOp — production default until an operator wires a real backend.
- Log — structured log emitter; cheap; aggregator-friendly.
- (future) Prometheus, OTLP, statsd, etc.
The interface is small by design. Call sites should depend on telemetry.Metrics, not on a concrete backend. To swap backends, change the constructor at the cmd/e2a/main.go wiring; nothing else moves.
Index ¶
- func NormalizeBuildLabel(value string) string
- type Log
- func (l *Log) HTTPRequest(string, string, string, float64)
- func (l *Log) InboundProcess(outcome string, seconds float64)
- func (l *Log) JanitorRowsDeleted(table string, count int)
- func (l *Log) NotifyMissed()
- func (l *Log) OutboundAttempt(outcome string, seconds float64)
- func (l *Log) OutboundQueueWait(float64)
- func (l *Log) OutboundTerminal(outcome string)
- func (l *Log) OutboundTerminalLatency(seconds float64)
- func (l *Log) OutboxEventsFanOut(eventType string, matched int)
- func (l *Log) OutboxEventsNoMatch(eventType string)
- func (l *Log) OutboxEventsPublished(eventType string)
- func (l *Log) OutboxFailures(stage string)
- func (l *Log) RedeliverRequests(scope string)
- func (l *Log) SMTPInbound(outcome string, seconds float64)
- func (l *Log) SetPublisherLag(seconds float64)
- func (l *Log) SetQueueDepth(queue, state string, n int)
- func (l *Log) SetQueueOldestAge(queue string, seconds float64)
- func (l *Log) SetWSActive(int)
- func (l *Log) WSConnected()
- func (l *Log) WSDisconnected(reason string)
- func (l *Log) WSDrained(count int)
- func (l *Log) WSHandshakeRejected(reason string)
- func (l *Log) WSSendFailure()
- func (l *Log) WebhookAttempt(outcome, statusClass string, seconds float64)
- func (l *Log) WebhookDeliveryRescued(count int)
- func (l *Log) WebhookExpiredPending(count int)
- func (l *Log) WebhookFanOutRescued(count int)
- func (l *Log) WebhookFirstAttemptLatency(seconds float64)
- func (l *Log) WebhookTerminal(outcome, scope string, count int)
- type Metrics
- type NoOp
- func (NoOp) HTTPRequest(string, string, string, float64)
- func (NoOp) InboundProcess(string, float64)
- func (NoOp) JanitorRowsDeleted(string, int)
- func (NoOp) NotifyMissed()
- func (NoOp) OutboundAttempt(string, float64)
- func (NoOp) OutboundQueueWait(float64)
- func (NoOp) OutboundTerminal(string)
- func (NoOp) OutboundTerminalLatency(float64)
- func (NoOp) OutboxEventsFanOut(string, int)
- func (NoOp) OutboxEventsNoMatch(string)
- func (NoOp) OutboxEventsPublished(string)
- func (NoOp) OutboxFailures(string)
- func (NoOp) RedeliverRequests(string)
- func (NoOp) SMTPInbound(string, float64)
- func (NoOp) SetPublisherLag(float64)
- func (NoOp) SetQueueDepth(string, string, int)
- func (NoOp) SetQueueOldestAge(string, float64)
- func (NoOp) SetWSActive(int)
- func (NoOp) WSConnected()
- func (NoOp) WSDisconnected(string)
- func (NoOp) WSDrained(int)
- func (NoOp) WSHandshakeRejected(string)
- func (NoOp) WSSendFailure()
- func (NoOp) WebhookAttempt(string, string, float64)
- func (NoOp) WebhookDeliveryRescued(int)
- func (NoOp) WebhookExpiredPending(int)
- func (NoOp) WebhookFanOutRescued(int)
- func (NoOp) WebhookFirstAttemptLatency(float64)
- func (NoOp) WebhookTerminal(string, string, int)
- type Prom
- func (p *Prom) HTTPRequest(method, route, statusClass string, seconds float64)
- func (p *Prom) Handler() http.Handler
- func (p *Prom) InboundProcess(outcome string, seconds float64)
- func (p *Prom) JanitorRowsDeleted(table string, count int)
- func (p *Prom) NotifyMissed()
- func (p *Prom) OutboundAttempt(outcome string, seconds float64)
- func (p *Prom) OutboundQueueWait(seconds float64)
- func (p *Prom) OutboundTerminal(outcome string)
- func (p *Prom) OutboundTerminalLatency(seconds float64)
- func (p *Prom) OutboxEventsFanOut(eventType string, matched int)
- func (p *Prom) OutboxEventsNoMatch(eventType string)
- func (p *Prom) OutboxEventsPublished(eventType string)
- func (p *Prom) OutboxFailures(stage string)
- func (p *Prom) RedeliverRequests(scope string)
- func (p *Prom) SMTPInbound(outcome string, seconds float64)
- func (p *Prom) SetPublisherLag(sec float64)
- func (p *Prom) SetQueueDepth(queue, state string, n int)
- func (p *Prom) SetQueueOldestAge(queue string, seconds float64)
- func (p *Prom) SetWSActive(n int)
- func (p *Prom) WSConnected()
- func (p *Prom) WSDisconnected(reason string)
- func (p *Prom) WSDrained(count int)
- func (p *Prom) WSHandshakeRejected(reason string)
- func (p *Prom) WSSendFailure()
- func (p *Prom) WebhookAttempt(outcome, statusClass string, seconds float64)
- func (p *Prom) WebhookDeliveryRescued(count int)
- func (p *Prom) WebhookExpiredPending(count int)
- func (p *Prom) WebhookFanOutRescued(count int)
- func (p *Prom) WebhookFirstAttemptLatency(seconds float64)
- func (p *Prom) WebhookTerminal(outcome, scope string, count int)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NormalizeBuildLabel ¶ added in v1.4.1
NormalizeBuildLabel keeps operator-provided release identifiers safe and bounded for use as a Prometheus label.
Types ¶
type Log ¶
type Log struct {
// contains filtered or unexported fields
}
Log emits a structured log line for every metric call. Cheap and portable; production aggregators (Loki, CloudWatch, Datadog) can build counters and gauges from these directly.
All lines share the [metrics] prefix so they're easy to filter. Format is key=value space-separated, which both jq and Splunk parse natively.
func (*Log) HTTPRequest ¶ added in v1.4.1
func (*Log) InboundProcess ¶ added in v1.4.1
func (*Log) JanitorRowsDeleted ¶
func (*Log) NotifyMissed ¶
func (l *Log) NotifyMissed()
func (*Log) OutboundAttempt ¶ added in v1.4.1
func (*Log) OutboundQueueWait ¶ added in v1.4.1
func (*Log) OutboundTerminal ¶ added in v1.4.1
func (*Log) OutboundTerminalLatency ¶ added in v1.4.1
func (*Log) OutboxEventsFanOut ¶
func (*Log) OutboxEventsNoMatch ¶
func (*Log) OutboxEventsPublished ¶
func (*Log) OutboxFailures ¶
func (*Log) RedeliverRequests ¶
func (*Log) SMTPInbound ¶ added in v1.4.1
func (*Log) SetPublisherLag ¶
func (*Log) SetQueueDepth ¶ added in v1.4.1
func (*Log) SetQueueOldestAge ¶ added in v1.4.1
func (*Log) SetWSActive ¶ added in v1.4.1
func (*Log) WSConnected ¶ added in v1.4.1
func (l *Log) WSConnected()
func (*Log) WSDisconnected ¶ added in v1.4.1
func (*Log) WSHandshakeRejected ¶ added in v1.4.1
func (*Log) WSSendFailure ¶ added in v1.4.1
func (l *Log) WSSendFailure()
func (*Log) WebhookAttempt ¶ added in v1.4.1
func (*Log) WebhookDeliveryRescued ¶ added in v1.4.1
func (*Log) WebhookExpiredPending ¶ added in v1.4.1
func (*Log) WebhookFanOutRescued ¶ added in v1.4.1
func (*Log) WebhookFirstAttemptLatency ¶ added in v1.4.1
func (*Log) WebhookTerminal ¶ added in v1.4.1
type Metrics ¶
type Metrics interface {
// OutboxEventsPublished is incremented each time PublishTx or
// PublishBestEffortTx successfully writes a webhook_events row.
OutboxEventsPublished(eventType string)
// OutboxEventsFanOut is incremented each time the worker finishes
// fanning an event out to its matched webhooks. matched is the
// number of webhook_subscriber_deliveries rows written.
OutboxEventsFanOut(eventType string, matched int)
// OutboxEventsNoMatch is incremented each time the worker
// transitions an event to status='no_match' because zero
// subscribers matched. Useful for spotting "why didn't my
// webhook fire?"
OutboxEventsNoMatch(eventType string)
// OutboxFailures is incremented on any outbox failure — worker-side
// (stage in {"lease", "list_webhooks", "insert_delivery", "update_status"})
// or emit-side when a fire-and-forget producer's PublishTx fails and the
// event is DROPPED (stage "publish" — today only the outbound
// email.blocked producer; other producers either surface the error to
// the caller or log via PublishBestEffortTx). A non-zero "publish" rate
// means contract events are silently missing from the log.
OutboxFailures(stage string)
// RedeliverRequests is incremented on each customer-driven replay.
// scope in {"single", "since"}.
RedeliverRequests(scope string)
// JanitorRowsDeleted is incremented by the cleanup tick.
// table in {"webhook_events", "webhook_subscriber_deliveries", "webhook_deliveries", "messages", "user_sessions", "oauth"}.
JanitorRowsDeleted(table string, count int)
// NotifyMissed is incremented when the 1-second fallback poll
// finds work that LISTEN/NOTIFY didn't wake us for. A non-zero
// rate indicates reconnect churn or a dropped notification.
NotifyMissed()
// SetPublisherLag is a gauge: the age in seconds of the oldest
// pending webhook_events row. Should be set on every Tick. Alert
// if it stays > 30s.
SetPublisherLag(seconds float64)
// HTTPRequest records one served HTTP request. route is the chi
// route pattern (e.g. "/v1/agents/{email}"), NEVER the raw path.
// statusClass is "1xx".."5xx". A negative seconds means "count the
// request but record no duration sample" — used for hijacked
// (WebSocket) connections, whose handler runtime is the connection
// lifetime, not a request latency.
HTTPRequest(method, route, statusClass string, seconds float64)
// SMTPInbound records one SMTP intake decision. outcome ∈
// {accepted, accepted_dedup, tempfail, rejected_unknown_recipient,
// rejected_unverified_domain, rejected_quota}. Units differ by
// stage: accepted/accepted_dedup/tempfail are per DATA transaction;
// rejected_* are per rejected RCPT command (one transaction can
// emit several rejections and still accept). seconds is DATA
// processing time (0 for RCPT-stage rejections).
SMTPInbound(outcome string, seconds float64)
// OutboundQueueWait records due→pickup latency for one outbound
// send attempt (River attempted_at − scheduled_at; created_at would
// count each retry's full backoff as queue wait).
OutboundQueueWait(seconds float64)
// OutboundTerminal records a terminal outcome for an outbound
// message. outcome ∈ {sent, failed_suppressed, failed_provider,
// failed_local_retries, failed_cancelled}. Exactly one per message:
// a deferred final attempt is counted by the terminal reconciler
// when it settles.
OutboundTerminal(outcome string)
// OutboundTerminalLatency records acceptance→terminal latency for
// one outbound message (the terminal write's occurred_at −
// messages.created_at). Observed exactly once per message,
// co-located with the OutboundTerminal emission so the two share
// their exactly-once contract (the SNS-feedback settle path stays
// uninstrumented for both — see the guard comment at the worker's
// MarkSent emit in internal/outboundsend).
OutboundTerminalLatency(seconds float64)
// OutboundAttempt records one submission attempt to the upstream
// relay. outcome ∈ {success, temporary_failure, permanent_failure}.
// seconds is the submission duration.
OutboundAttempt(outcome string, seconds float64)
// WebhookAttempt records one webhook delivery attempt. outcome ∈
// {delivered, retryable_failure, exhausted, webhook_deleted,
// skipped_disabled}. statusClass is the HTTP status class of the
// endpoint's response, or "none" when no response was received
// (connect/DNS/SSRF-blocked).
WebhookAttempt(outcome, statusClass string, seconds float64)
// WebhookTerminal records terminal delivery outcomes after the terminal
// database transition succeeds. outcome ∈ {delivered, e2a_failure,
// endpoint_failure, excluded}; scope ∈ {initial, replay, test, unknown}.
// The hosted SLO uses initial + unknown and excludes endpoint_failure:
// customer endpoint behavior must not burn e2a's error budget.
WebhookTerminal(outcome, scope string, count int)
// WebhookExpiredPending counts delivery rows that reached their
// retention TTL while still 'pending' and were marked terminally
// failed ("expired before delivery") by the janitor instead of being
// silently deleted. With the dead-job reconciler rescuing strands,
// a sustained non-zero rate means deliveries are aging out
// un-attempted — in practice rows snoozing behind a webhook disabled
// for longer than the TTL.
WebhookExpiredPending(count int)
// WebhookFanOutRescued / WebhookDeliveryRescued count rows the
// reconcilers' dead-job arm re-drove: a pending webhook_events /
// webhook_subscriber_deliveries row whose stamped River job was
// terminal or pruned, given a fresh job. Occasional blips are normal
// (crash windows, lost terminal writes); a monotonically climbing
// rate is the poison-row signal — a deterministically failing row
// burning a fresh job envelope per rescue, forever. These counters
// are the observability half of the deliberate
// retry-forever-with-observability design.
WebhookFanOutRescued(count int)
WebhookDeliveryRescued(count int)
// WebhookFirstAttemptLatency records event→first-attempt latency
// for one subscriber delivery (attempt start − the webhook_events
// row's created_at). Observed only on a delivery's FIRST HTTP
// attempt — retries and the no-POST outcomes (webhook_deleted,
// skipped_disabled) never observe.
WebhookFirstAttemptLatency(seconds float64)
// WSConnected / WSDisconnected count WebSocket connection
// lifecycle events. reason ∈ {replaced, ping_timeout,
// client_close, error, shutdown}.
WSConnected()
WSDisconnected(reason string)
// WSHandshakeRejected counts pre-upgrade WebSocket handshake
// rejections. reason ∈ {unauthorized, not_found, forbidden,
// upgrade_failed, internal_error}. Never label with emails or tokens.
WSHandshakeRejected(reason string)
// WSDrained counts unread messages pushed during connect-drain.
WSDrained(count int)
// WSSendFailure counts failed pushes to a registered connection.
WSSendFailure()
// SetWSActive is a gauge: current registered WS connections.
SetWSActive(n int)
// InboundProcess records an async inbound-worker outcome. outcome
// ∈ {processed, noop, failed_recipient_gone, failed_exhausted,
// retryable}.
InboundProcess(outcome string, seconds float64)
// SetQueueDepth / SetQueueOldestAge are gauges sampled by the
// queue-stats maintenance job. queue ∈ jobs.Queue* names; state ∈
// {available, running, retryable, scheduled}. Oldest age is for
// runnable (available) jobs only — a growing value means workers
// are not keeping up.
SetQueueDepth(queue, state string, n int)
SetQueueOldestAge(queue string, seconds float64)
}
Metrics is the observability surface for the slice 10 design. Counter-style methods record a discrete event; SetPublisherLag is a gauge that should be set on each tick.
Stable across implementations. Adding a new metric is additive (add a method, default it to a no-op on existing implementations).
type NoOp ¶
type NoOp struct{}
NoOp swallows every call. Default for tests that don't care.
func (NoOp) InboundProcess ¶ added in v1.4.1
func (NoOp) JanitorRowsDeleted ¶
func (NoOp) NotifyMissed ¶
func (NoOp) NotifyMissed()
func (NoOp) OutboundAttempt ¶ added in v1.4.1
func (NoOp) OutboundQueueWait ¶ added in v1.4.1
func (NoOp) OutboundTerminal ¶ added in v1.4.1
func (NoOp) OutboundTerminalLatency ¶ added in v1.4.1
func (NoOp) OutboxEventsFanOut ¶
func (NoOp) OutboxEventsNoMatch ¶
func (NoOp) OutboxEventsPublished ¶
func (NoOp) OutboxFailures ¶
func (NoOp) RedeliverRequests ¶
func (NoOp) SMTPInbound ¶ added in v1.4.1
func (NoOp) SetPublisherLag ¶
func (NoOp) SetQueueOldestAge ¶ added in v1.4.1
func (NoOp) SetWSActive ¶ added in v1.4.1
func (NoOp) WSConnected ¶ added in v1.4.1
func (NoOp) WSConnected()
func (NoOp) WSDisconnected ¶ added in v1.4.1
func (NoOp) WSHandshakeRejected ¶ added in v1.4.1
func (NoOp) WSSendFailure ¶ added in v1.4.1
func (NoOp) WSSendFailure()
func (NoOp) WebhookDeliveryRescued ¶ added in v1.4.1
func (NoOp) WebhookExpiredPending ¶ added in v1.4.1
func (NoOp) WebhookFanOutRescued ¶ added in v1.4.1
func (NoOp) WebhookFirstAttemptLatency ¶ added in v1.4.1
type Prom ¶ added in v1.4.1
type Prom struct {
// contains filtered or unexported fields
}
Prom is the Prometheus backend. It owns a private registry (no global state — tests construct as many as they like) exposed via Handler().
Label hygiene is enforced here, not at call sites: every label value passes through an enum allowlist (unknown → "other") or a hard series cap (route, event type). Metric labels never carry message content, addresses, URLs, or credentials — see docs/observability.md for the full catalog and the cardinality contract.
func (*Prom) HTTPRequest ¶ added in v1.4.1
func (*Prom) Handler ¶ added in v1.4.1
Handler returns the exposition endpoint for this backend's registry.
func (*Prom) InboundProcess ¶ added in v1.4.1
func (*Prom) JanitorRowsDeleted ¶ added in v1.4.1
func (*Prom) NotifyMissed ¶ added in v1.4.1
func (p *Prom) NotifyMissed()
func (*Prom) OutboundAttempt ¶ added in v1.4.1
func (*Prom) OutboundQueueWait ¶ added in v1.4.1
func (*Prom) OutboundTerminal ¶ added in v1.4.1
func (*Prom) OutboundTerminalLatency ¶ added in v1.4.1
func (*Prom) OutboxEventsFanOut ¶ added in v1.4.1
func (*Prom) OutboxEventsNoMatch ¶ added in v1.4.1
func (*Prom) OutboxEventsPublished ¶ added in v1.4.1
func (*Prom) OutboxFailures ¶ added in v1.4.1
func (*Prom) RedeliverRequests ¶ added in v1.4.1
func (*Prom) SMTPInbound ¶ added in v1.4.1
func (*Prom) SetPublisherLag ¶ added in v1.4.1
func (*Prom) SetQueueDepth ¶ added in v1.4.1
func (*Prom) SetQueueOldestAge ¶ added in v1.4.1
func (*Prom) SetWSActive ¶ added in v1.4.1
func (*Prom) WSConnected ¶ added in v1.4.1
func (p *Prom) WSConnected()
func (*Prom) WSDisconnected ¶ added in v1.4.1
func (*Prom) WSHandshakeRejected ¶ added in v1.4.1
func (*Prom) WSSendFailure ¶ added in v1.4.1
func (p *Prom) WSSendFailure()