telemetry

package
v1.4.1 Latest Latest
Warning

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

Go to latest
Published: Jul 26, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

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

Constants

This section is empty.

Variables

This section is empty.

Functions

func NormalizeBuildLabel added in v1.4.1

func NormalizeBuildLabel(value string) string

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 NewLog

func NewLog() *Log

func (*Log) HTTPRequest added in v1.4.1

func (l *Log) HTTPRequest(string, string, string, float64)

func (*Log) InboundProcess added in v1.4.1

func (l *Log) InboundProcess(outcome string, seconds float64)

func (*Log) JanitorRowsDeleted

func (l *Log) JanitorRowsDeleted(table string, count int)

func (*Log) NotifyMissed

func (l *Log) NotifyMissed()

func (*Log) OutboundAttempt added in v1.4.1

func (l *Log) OutboundAttempt(outcome string, seconds float64)

func (*Log) OutboundQueueWait added in v1.4.1

func (l *Log) OutboundQueueWait(float64)

func (*Log) OutboundTerminal added in v1.4.1

func (l *Log) OutboundTerminal(outcome string)

func (*Log) OutboundTerminalLatency added in v1.4.1

func (l *Log) OutboundTerminalLatency(seconds float64)

func (*Log) OutboxEventsFanOut

func (l *Log) OutboxEventsFanOut(eventType string, matched int)

func (*Log) OutboxEventsNoMatch

func (l *Log) OutboxEventsNoMatch(eventType string)

func (*Log) OutboxEventsPublished

func (l *Log) OutboxEventsPublished(eventType string)

func (*Log) OutboxFailures

func (l *Log) OutboxFailures(stage string)

func (*Log) RedeliverRequests

func (l *Log) RedeliverRequests(scope string)

func (*Log) SMTPInbound added in v1.4.1

func (l *Log) SMTPInbound(outcome string, seconds float64)

func (*Log) SetPublisherLag

func (l *Log) SetPublisherLag(seconds float64)

func (*Log) SetQueueDepth added in v1.4.1

func (l *Log) SetQueueDepth(queue, state string, n int)

func (*Log) SetQueueOldestAge added in v1.4.1

func (l *Log) SetQueueOldestAge(queue string, seconds float64)

func (*Log) SetWSActive added in v1.4.1

func (l *Log) SetWSActive(int)

func (*Log) WSConnected added in v1.4.1

func (l *Log) WSConnected()

func (*Log) WSDisconnected added in v1.4.1

func (l *Log) WSDisconnected(reason string)

func (*Log) WSDrained added in v1.4.1

func (l *Log) WSDrained(count int)

func (*Log) WSHandshakeRejected added in v1.4.1

func (l *Log) WSHandshakeRejected(reason string)

func (*Log) WSSendFailure added in v1.4.1

func (l *Log) WSSendFailure()

func (*Log) WebhookAttempt added in v1.4.1

func (l *Log) WebhookAttempt(outcome, statusClass string, seconds float64)

func (*Log) WebhookDeliveryRescued added in v1.4.1

func (l *Log) WebhookDeliveryRescued(count int)

func (*Log) WebhookExpiredPending added in v1.4.1

func (l *Log) WebhookExpiredPending(count int)

func (*Log) WebhookFanOutRescued added in v1.4.1

func (l *Log) WebhookFanOutRescued(count int)

func (*Log) WebhookFirstAttemptLatency added in v1.4.1

func (l *Log) WebhookFirstAttemptLatency(seconds float64)

func (*Log) WebhookTerminal added in v1.4.1

func (l *Log) WebhookTerminal(outcome, scope string, count int)

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) HTTPRequest added in v1.4.1

func (NoOp) HTTPRequest(string, string, string, float64)

func (NoOp) InboundProcess added in v1.4.1

func (NoOp) InboundProcess(string, float64)

func (NoOp) JanitorRowsDeleted

func (NoOp) JanitorRowsDeleted(string, int)

func (NoOp) NotifyMissed

func (NoOp) NotifyMissed()

func (NoOp) OutboundAttempt added in v1.4.1

func (NoOp) OutboundAttempt(string, float64)

func (NoOp) OutboundQueueWait added in v1.4.1

func (NoOp) OutboundQueueWait(float64)

func (NoOp) OutboundTerminal added in v1.4.1

func (NoOp) OutboundTerminal(string)

func (NoOp) OutboundTerminalLatency added in v1.4.1

func (NoOp) OutboundTerminalLatency(float64)

func (NoOp) OutboxEventsFanOut

func (NoOp) OutboxEventsFanOut(string, int)

func (NoOp) OutboxEventsNoMatch

func (NoOp) OutboxEventsNoMatch(string)

func (NoOp) OutboxEventsPublished

func (NoOp) OutboxEventsPublished(string)

func (NoOp) OutboxFailures

func (NoOp) OutboxFailures(string)

func (NoOp) RedeliverRequests

func (NoOp) RedeliverRequests(string)

func (NoOp) SMTPInbound added in v1.4.1

func (NoOp) SMTPInbound(string, float64)

func (NoOp) SetPublisherLag

func (NoOp) SetPublisherLag(float64)

func (NoOp) SetQueueDepth added in v1.4.1

func (NoOp) SetQueueDepth(string, string, int)

func (NoOp) SetQueueOldestAge added in v1.4.1

func (NoOp) SetQueueOldestAge(string, float64)

func (NoOp) SetWSActive added in v1.4.1

func (NoOp) SetWSActive(int)

func (NoOp) WSConnected added in v1.4.1

func (NoOp) WSConnected()

func (NoOp) WSDisconnected added in v1.4.1

func (NoOp) WSDisconnected(string)

func (NoOp) WSDrained added in v1.4.1

func (NoOp) WSDrained(int)

func (NoOp) WSHandshakeRejected added in v1.4.1

func (NoOp) WSHandshakeRejected(string)

func (NoOp) WSSendFailure added in v1.4.1

func (NoOp) WSSendFailure()

func (NoOp) WebhookAttempt added in v1.4.1

func (NoOp) WebhookAttempt(string, string, float64)

func (NoOp) WebhookDeliveryRescued added in v1.4.1

func (NoOp) WebhookDeliveryRescued(int)

func (NoOp) WebhookExpiredPending added in v1.4.1

func (NoOp) WebhookExpiredPending(int)

func (NoOp) WebhookFanOutRescued added in v1.4.1

func (NoOp) WebhookFanOutRescued(int)

func (NoOp) WebhookFirstAttemptLatency added in v1.4.1

func (NoOp) WebhookFirstAttemptLatency(float64)

func (NoOp) WebhookTerminal added in v1.4.1

func (NoOp) WebhookTerminal(string, string, int)

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 NewProm added in v1.4.1

func NewProm(build string) *Prom

func (*Prom) HTTPRequest added in v1.4.1

func (p *Prom) HTTPRequest(method, route, statusClass string, seconds float64)

func (*Prom) Handler added in v1.4.1

func (p *Prom) Handler() http.Handler

Handler returns the exposition endpoint for this backend's registry.

func (*Prom) InboundProcess added in v1.4.1

func (p *Prom) InboundProcess(outcome string, seconds float64)

func (*Prom) JanitorRowsDeleted added in v1.4.1

func (p *Prom) JanitorRowsDeleted(table string, count int)

func (*Prom) NotifyMissed added in v1.4.1

func (p *Prom) NotifyMissed()

func (*Prom) OutboundAttempt added in v1.4.1

func (p *Prom) OutboundAttempt(outcome string, seconds float64)

func (*Prom) OutboundQueueWait added in v1.4.1

func (p *Prom) OutboundQueueWait(seconds float64)

func (*Prom) OutboundTerminal added in v1.4.1

func (p *Prom) OutboundTerminal(outcome string)

func (*Prom) OutboundTerminalLatency added in v1.4.1

func (p *Prom) OutboundTerminalLatency(seconds float64)

func (*Prom) OutboxEventsFanOut added in v1.4.1

func (p *Prom) OutboxEventsFanOut(eventType string, matched int)

func (*Prom) OutboxEventsNoMatch added in v1.4.1

func (p *Prom) OutboxEventsNoMatch(eventType string)

func (*Prom) OutboxEventsPublished added in v1.4.1

func (p *Prom) OutboxEventsPublished(eventType string)

func (*Prom) OutboxFailures added in v1.4.1

func (p *Prom) OutboxFailures(stage string)

func (*Prom) RedeliverRequests added in v1.4.1

func (p *Prom) RedeliverRequests(scope string)

func (*Prom) SMTPInbound added in v1.4.1

func (p *Prom) SMTPInbound(outcome string, seconds float64)

func (*Prom) SetPublisherLag added in v1.4.1

func (p *Prom) SetPublisherLag(sec float64)

func (*Prom) SetQueueDepth added in v1.4.1

func (p *Prom) SetQueueDepth(queue, state string, n int)

func (*Prom) SetQueueOldestAge added in v1.4.1

func (p *Prom) SetQueueOldestAge(queue string, seconds float64)

func (*Prom) SetWSActive added in v1.4.1

func (p *Prom) SetWSActive(n int)

func (*Prom) WSConnected added in v1.4.1

func (p *Prom) WSConnected()

func (*Prom) WSDisconnected added in v1.4.1

func (p *Prom) WSDisconnected(reason string)

func (*Prom) WSDrained added in v1.4.1

func (p *Prom) WSDrained(count int)

func (*Prom) WSHandshakeRejected added in v1.4.1

func (p *Prom) WSHandshakeRejected(reason string)

func (*Prom) WSSendFailure added in v1.4.1

func (p *Prom) WSSendFailure()

func (*Prom) WebhookAttempt added in v1.4.1

func (p *Prom) WebhookAttempt(outcome, statusClass string, seconds float64)

func (*Prom) WebhookDeliveryRescued added in v1.4.1

func (p *Prom) WebhookDeliveryRescued(count int)

func (*Prom) WebhookExpiredPending added in v1.4.1

func (p *Prom) WebhookExpiredPending(count int)

func (*Prom) WebhookFanOutRescued added in v1.4.1

func (p *Prom) WebhookFanOutRescued(count int)

func (*Prom) WebhookFirstAttemptLatency added in v1.4.1

func (p *Prom) WebhookFirstAttemptLatency(seconds float64)

func (*Prom) WebhookTerminal added in v1.4.1

func (p *Prom) WebhookTerminal(outcome, scope string, count int)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL