observability

package module
v0.3.4 Latest Latest
Warning

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

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

Documentation

Overview

Package observability shows how an application is doing across all its instances and keeps a record of incidents (ADR-0064).

A Collector on every instance counts HTTP requests per minute, method and route: requests, client and server errors, and a latency histogram. Run as a background runner, it writes its minutes to PostgreSQL every 15 seconds through Store.WriteMinutes, so Store.Summary can report request rates, error rates and latency percentiles over any recent window for the whole deployment, per instance and per route: 

collector, err := observability.NewCollector(observability.WithInstance(tracker.InstanceID()), observability.WithSink(store))
handler := httpx.Chain(observability.RecordRoute(mux), httpx.Recover(logger), collector.Middleware(), …)

Series are keyed by the route pattern registered in code, never the requested path, so clients can't grow them; each instance holds at most a bounded number of series for a bounded number of minutes.

Incidents are opened by operators (Store.OpenIncident) or by Store.DetectIncident when the error rate crosses a threshold, and move through investigating, identified, monitoring and resolved with timeline updates. At most one automatic incident is open at a time, whichever instance runs detection.

Stability: stable (ADR-0015, ADR-0054).

Index

Constants

View Source 
const (
	// DefaultFlushInterval is how often a collector writes its windows
	// without [WithFlushInterval].
	DefaultFlushInterval = 15 * time.Second
	// DefaultMaxSeries is how many method and route pairs one minute keeps
	// without [WithMaxSeries].
	DefaultMaxSeries = 500
	// DefaultMaxPendingMinutes is how many unwritten minutes a collector
	// keeps while its sink fails, without [WithMaxPendingMinutes].
	DefaultMaxPendingMinutes = 10

	// OverflowRoute is the route of requests counted after a minute already
	// has its maximum number of series.
	OverflowRoute = "_overflow"
	// OtherMethod replaces request methods that aren't standard HTTP
	// methods, as OpenTelemetry does.
	OtherMethod = "_OTHER"
)
View Source 
const (
	MaxTitleLength   = 200
	MaxSummaryLength = 5000
	MaxMessageLength = 5000
	// MaxIncidentUpdates is how many updates one incident keeps.
	MaxIncidentUpdates = 500
	// MaxStartedAge is how long before being opened an incident may have
	// started.
	MaxStartedAge = 90 * 24 * time.Hour
)

Limits on incidents.

View Source 
const (
	// DefaultQueryTimeout bounds summary queries without
	// [WithQueryTimeout].
	DefaultQueryTimeout = 5 * time.Second
	// MaxSummaryRange is the longest time range [Store.Summary] reports.
	MaxSummaryRange = 7 * 24 * time.Hour
)
View Source 
const BucketCount = len(bucketBounds) + 1

BucketCount is the number of latency histogram buckets: one per bound in BucketBounds, and one for slower requests.

View Source 
const MaxInstanceLength = 64

MaxInstanceLength is the longest instance ID stored.

Variables

View Source 
var (
	// ErrInvalidRange reports a summary range that is empty, reversed or
	// longer than [MaxSummaryRange].
	ErrInvalidRange = errors.New("observability: invalid time range")
	// ErrQueryTimeout reports a summary that took longer than the query
	// timeout.
	ErrQueryTimeout = errors.New("observability: query timed out")
	// ErrIncidentNotFound reports an incident ID that doesn't exist.
	ErrIncidentNotFound = errors.New("observability: incident not found")
	// ErrInvalidIncident reports an incident or update with a missing or
	// out-of-bounds field; the wrapping error says which.
	ErrInvalidIncident = errors.New("observability: invalid incident")
	// ErrIncidentResolved reports a change to a resolved incident.
	ErrIncidentResolved = errors.New("observability: incident is resolved")
	// ErrTooManyUpdates reports an incident that already has
	// [MaxIncidentUpdates] updates.
	ErrTooManyUpdates = errors.New("observability: incident has too many updates")
	// ErrInvalidCursor reports a cursor [Store.Incidents] didn't return.
	ErrInvalidCursor = errors.New("observability: invalid cursor")
)

Errors returned by Store methods. Check them with errors.Is.

View Source 
var (
	// ErrTooManyStreams reports a stream refused because the instance, or
	// the subject, already has the most streams allowed.
	ErrTooManyStreams = errors.New("observability: too many streams")
	// ErrStreamExpired is the cause of a stream context that reached its
	// maximum duration.
	ErrStreamExpired = errors.New("observability: stream reached its maximum duration")
	// ErrStreamsClosed is the cause of stream contexts ended by
	// [Streams.Close], and the error of Open afterwards.
	ErrStreamsClosed = errors.New("observability: streams closed")
)

Errors of Streams.

View Source 
var Migrations fs.FS = mustSub(migrationFiles, "migrations")

Migrations holds the module's goose migrations: the observability_minutes, incidents and incident_updates tables. Apps copy them into db/migrations; tests can apply them directly with pgtest.

Functions

func BucketBounds 

func BucketBounds() []time.Duration

BucketBounds returns the upper bounds of the latency histogram's buckets, fastest first. Bucket i counts requests that took at most bound i and longer than bound i-1; the last bucket counts requests slower than every bound.

func RecordRoute 

func RecordRoute(router http.Handler) http.Handler

RecordRoute wraps the application's router, normally the http.ServeMux at the end of the middleware chain, so Collector.Middleware learns the route pattern the router matched even when middleware between them replaced the request. Without the middleware, it does nothing.

Types

type Collector 

type Collector struct {
	// contains filtered or unexported fields
}

Collector counts an instance's HTTP requests in one-minute windows, per method and route, and writes them to its Sink periodically when run. Its memory is bounded: at most WithMaxSeries series per minute, each a fixed-size histogram, for at most WithMaxPendingMinutes minutes. It is safe for concurrent use.

func NewCollector 

func NewCollector(opts ...Option) (*Collector, error)

NewCollector returns a collector.

func (*Collector) Flush 

func (c *Collector) Flush(ctx context.Context) error

Flush writes every minute the collector holds to the sink, then forgets minutes that ended before the write started. The current minute is written again at the next flush, with its new totals.

func (*Collector) Instance 

func (c *Collector) Instance() string

Instance returns the instance ID the collector writes minutes under.

func (*Collector) Lost 

func (c *Collector) Lost() int64

Lost returns how many requests were dropped without being written because the sink failed for longer than the pending minutes last.

func (*Collector) Middleware 

func (c *Collector) Middleware() func(http.Handler) http.Handler

Middleware records every request in the collector: its method, the route pattern the router matched, its status and duration. Install it early in the chain, after panic recovery, so responses written by other middleware (rate limits, maintenance mode, authentication failures) are counted too; a handler that panics counts as a 500.

Paths, query strings, headers and other values clients choose never become part of a series: the route is the pattern registered in code, or empty when none matched. http.ServeMux sets the pattern on the request it routes, which is a copy when a middleware in between calls r.WithContext, so wrap the router with RecordRoute: 

handler := httpx.Chain(observability.RecordRoute(mux), httpx.Recover(logger), collector.Middleware(), …)

func (*Collector) Minutes 

func (c *Collector) Minutes() []Minute

Minutes returns what the collector holds now: the current minute and finished minutes not yet written for the last time, oldest first.

func (*Collector) Record 

func (c *Collector) Record(r Request)

Record counts a finished request. Collector.Middleware calls it; call it directly only for requests served another way.

func (*Collector) Run 

func (c *Collector) Run(ctx context.Context) error

Run writes minutes to the sink every flush interval until ctx ends, then writes once more. Failed writes are logged and retried at the next interval; Run always returns nil, so observability never stops the app.

func (*Collector) Subscribe 

func (c *Collector) Subscribe(fn func(Request)) (unsubscribe func())

Subscribe calls fn with every request recorded from now on, until unsubscribe is called. fn runs on the request's goroutine before the response is complete, so it must return quickly and never block, such as by appending to a ring buffer. Requests carry their path, request ID and trace ID only for subscribers.

type Detection 

type Detection struct {
	// Window is how far back requests count, 1 minute to 1 hour.
	Window time.Duration
	// Threshold is the error rate above which an incident opens, from 0 to
	// 1: 0.05 for 5% of requests being server errors.
	Threshold float64
	// MinRequests is how many requests the window needs before its error
	// rate counts, at least 1: a few failures of a quiet app don't open
	// incidents, and no traffic never looks recovered.
	MinRequests int64
	// Severity of incidents detection opens. Default: sev2.
	Severity Severity
	// Actor is who detection acts as in incidents' timelines, such as
	// actor.System("incidents_detect").
	Actor actor.Actor
}

Detection configures Store.DetectIncident.

type DetectionAction 

type DetectionAction string

DetectionAction is what a detection changed. 

const (
	// DetectionNone: nothing changed.
	DetectionNone DetectionAction = ""
	// DetectionOpened: an automatic incident was opened.
	DetectionOpened DetectionAction = "opened"
	// DetectionRecovered: the open automatic incident's error rate
	// recovered; an update says so. The incident stays open for operators
	// to resolve.
	DetectionRecovered DetectionAction = "recovered"
	// DetectionBreaching: the error rate is high again after recovering.
	DetectionBreaching DetectionAction = "breaching"
)

Detection actions.

type DetectionResult 

type DetectionResult struct {
	// From and To are the minutes counted: From inclusive, To exclusive.
	From, To     time.Time
	Requests     int64
	ServerErrors int64
	// Breaching reports an error rate above the threshold with at least
	// MinRequests requests; Healthy one at or below it with at least
	// MinRequests requests. Without enough requests, both are false.
	Breaching bool
	Healthy   bool
	Action    DetectionAction
	// Incident is the open automatic incident after detection, if any, and
	// Update the timeline update detection added.
	Incident *Incident
	Update   *IncidentUpdate
}

DetectionResult is what Store.DetectIncident saw and did.

func (DetectionResult) ErrorRate 

func (r DetectionResult) ErrorRate() float64

ErrorRate returns the share of requests that were server errors.

type Incident 

type Incident struct {
	ID       int64
	Title    string
	Summary  string
	Severity Severity
	Status   Status
	Source   Source
	// StartedAt is when the incident began, which can be before it was
	// opened; ResolvedAt is set once it is resolved.
	StartedAt  time.Time
	ResolvedAt *time.Time
	// RecoveredAt is set on an open automatic incident while detection sees
	// the error rate recovered.
	RecoveredAt *time.Time
	// CreatedByKind and CreatedByID identify who opened it: a user, or the
	// system for automatic incidents.
	CreatedByKind actor.Kind
	CreatedByID   string
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

Incident is a period during which the application didn't work as it should.

type IncidentChange 

type IncidentChange struct {
	Message string
	// Status and Severity are left unchanged when empty. Status can't be
	// resolved: use [Store.ResolveIncident].
	Status   Status
	Severity Severity
}

IncidentChange is a timeline update, optionally changing the incident's status and severity.

type IncidentFilter 

type IncidentFilter struct {
	// Open selects incidents that aren't resolved; Status one status.
	Open     bool
	Status   Status
	Severity Severity
	Source   Source
	// StartedFrom and StartedTo bound StartedAt: From inclusive, To
	// exclusive.
	StartedFrom time.Time
	StartedTo   time.Time
	// Limit is clamped to 1–100; 0 means 50.
	Limit int
	// Cursor is IncidentPage.NextCursor from the previous page.
	Cursor string
}

IncidentFilter selects incidents. Empty fields match everything.

type IncidentPage 

type IncidentPage struct {
	Incidents []Incident
	// NextCursor fetches the next page; empty on the last page.
	NextCursor string
}

IncidentPage is a page of incidents, most recently opened first.

type IncidentUpdate 

type IncidentUpdate struct {
	ID         int64
	IncidentID int64
	Kind       UpdateKind
	Message    string
	// Status and Severity are the incident's after the update.
	Status    Status
	Severity  Severity
	ActorKind actor.Kind
	ActorID   string
	CreatedAt time.Time
}

IncidentUpdate is one entry of an incident's timeline.

type InstanceStats 

type InstanceStats struct {
	Instance string
	// LastMinute is the latest minute with requests, and LastWrite when the
	// instance last wrote its minutes: an instance that stopped writing
	// has stopped or can't reach the database.
	LastMinute time.Time
	LastWrite  time.Time
	Stats
}

InstanceStats are one instance's requests.

type Minute 

type Minute struct {
	// Start is the minute's first instant, in UTC.
	Start    time.Time
	Instance string
	Method   string
	Route    string
	Stats
}

Minute is one instance's requests for one method and route during one minute. Written again while the minute is in progress, it carries the totals so far.

type MinuteStats 

type MinuteStats struct {
	Start time.Time
	Stats
}

MinuteStats are one minute's requests across instances.

type NewIncident 

type NewIncident struct {
	Title   string
	Summary string
	// Severity is required.
	Severity Severity
	// Status defaults to investigating; it can't be resolved.
	Status Status
	// StartedAt defaults to now; it can be up to [MaxStartedAge] earlier.
	StartedAt time.Time
	// Message is the first timeline update; default "Incident opened.".
	Message string
}

NewIncident is an incident an operator opens.

type Option 

type Option func(*Collector)

An Option configures a Collector.

func WithClock 

func WithClock(now func() time.Time) Option

WithClock uses now instead of time.Now to place requests in minutes, for tests.

func WithFlushInterval 

func WithFlushInterval(d time.Duration) Option

WithFlushInterval sets how often Run writes minutes: 1 second to 1 minute. Default: DefaultFlushInterval.

func WithInstance 

func WithInstance(id string) Option

WithInstance names the instance in the stored minutes, such as the release tracker's instance ID. Default: a random ID.

func WithLogger 

func WithLogger(logger *slog.Logger) Option

WithLogger sets the logger for failed writes. Default: discard.

func WithMaxPendingMinutes 

func WithMaxPendingMinutes(n int) Option

WithMaxPendingMinutes sets how many finished minutes wait for a failing sink before the oldest is dropped. Default: DefaultMaxPendingMinutes.

func WithMaxSeries 

func WithMaxSeries(n int) Option

WithMaxSeries sets how many method and route pairs one minute keeps; later pairs are counted under OverflowRoute. Default: DefaultMaxSeries.

func WithSink 

func WithSink(sink Sink) Option

WithSink sets where Collector.Run writes minutes. Without a sink, Run only discards finished minutes.

type Request 

type Request struct {
	// Time is when the request finished.
	Time time.Time
	// Method is the request method: a standard method, or [OtherMethod].
	Method string
	// Route is the path of the pattern the router matched, such as
	// "/v1/projects/{id}"; empty when no route matched, such as a request
	// answered by middleware.
	Route    string
	Status   int
	Duration time.Duration

	// Path (without the query), RequestID and TraceID are set only for
	// subscribers ([Collector.Subscribe]). The collector never stores or
	// aggregates them.
	Path      string
	RequestID string
	TraceID   string
}

Request is one finished HTTP request, as Collector.Middleware recorded it.

type RouteStats 

type RouteStats struct {
	Method string
	// Route is the route pattern's path; empty for requests no route
	// matched, and [OverflowRoute] for requests beyond the series limit.
	Route string
	Stats
}

RouteStats are one method and route's requests across instances.

type Severity 

type Severity string

Severity is how bad an incident is, from sev1 (most severe) to sev4. 

const (
	SeveritySev1 Severity = "sev1"
	SeveritySev2 Severity = "sev2"
	SeveritySev3 Severity = "sev3"
	SeveritySev4 Severity = "sev4"
)

Severities.

type Sink 

type Sink interface {
	WriteMinutes(ctx context.Context, minutes []Minute) error
}

A Sink stores a collector's minutes. Store implements it.

type Source 

type Source string

Source is who opened an incident. 

const (
	SourceManual    Source = "manual"
	SourceAutomatic Source = "automatic"
)

Sources.

type Stats 

type Stats struct {
	Requests int64
	// ClientErrors are 4xx responses, ServerErrors 5xx responses; a
	// handler that panicked counts as a server error.
	ClientErrors int64
	ServerErrors int64
	// DurationSum is the total time spent, DurationMax the slowest request.
	DurationSum time.Duration
	DurationMax time.Duration
	// Buckets holds [BucketCount] request counts, by [BucketBounds]. Nil
	// when there are no requests.
	Buckets []int64
}

Stats are request counts and latencies over some set of requests: a route, an instance, a minute, or everything in a time range.

func (Stats) ErrorRate 

func (s Stats) ErrorRate() float64

ErrorRate returns the share of requests that were server errors, from 0 to 1; 0 without requests.

func (Stats) Mean 

func (s Stats) Mean() time.Duration

Mean returns the average request duration; 0 without requests.

func (Stats) Quantile 

func (s Stats) Quantile(q float64) time.Duration

Quantile estimates the duration q (0 < q ≤ 1) of requests took at most, such as 0.95 for the 95th percentile; 0 without requests.

The estimate finds the bucket holding the q-th request and interpolates linearly inside it, assuming requests are spread evenly across the bucket; the last bucket is interpolated up to Stats.DurationMax, and no estimate exceeds it. The exact value lies in the same bucket, so the error is at most the bucket's width: under 0.5 ms for requests faster than 1 ms, at most 50% of the exact value between 1 ms and 10 s, and at most 100% above 10 s. Real latency distributions change little inside a bucket, so errors are usually a few percent (ADR-0064 records measured errors).

type Status 

type Status string

Status is where an incident is in its lifecycle. 

const (
	StatusInvestigating Status = "investigating"
	StatusIdentified    Status = "identified"
	StatusMonitoring    Status = "monitoring"
	StatusResolved      Status = "resolved"
)

Statuses. Open incidents move between the first three in any order; resolved is final.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store keeps request minutes and incidents in PostgreSQL. It is safe for concurrent use.

func NewStore 

func NewStore(pool *pgxpool.Pool, opts ...StoreOption) (*Store, error)

NewStore returns a store on pool, which must have the module's migrations applied. Store.DetectIncident counts the OpenTelemetry metric incidents.detections by action.

func (*Store) DeleteBefore 

func (s *Store) DeleteBefore(ctx context.Context, before time.Time, limit int) (int64, error)

DeleteBefore deletes up to limit minutes that started before before, and returns how many it deleted. Apps run it from a periodic job until it deletes fewer than limit.

func (*Store) DetectIncident 

func (s *Store) DetectIncident(ctx context.Context, d Detection) (DetectionResult, error)

DetectIncident compares every instance's server error rate over the last Window, up to and including the current minute, with the threshold:

  • above it, with no automatic incident open: it opens one;
  • above it again after recovering: it adds a "breaching" update;
  • at or below it while an automatic incident is open and not yet seen recovered: it adds a "recovered" update, leaving the incident open for operators to resolve.

Windows with fewer than MinRequests requests change nothing. Detection runs in one transaction holding an advisory lock, and a unique index allows one open automatic incident, so instances running it at the same time never open two. It counts incidents.detections by action.

func (*Store) Incident 

func (s *Store) Incident(ctx context.Context, id int64) (Incident, error)

Incident returns an incident, or ErrIncidentNotFound.

func (*Store) IncidentUpdates 

func (s *Store) IncidentUpdates(ctx context.Context, id int64) ([]IncidentUpdate, error)

IncidentUpdates returns an incident's timeline, oldest first: at most MaxIncidentUpdates updates. An unknown incident has none.

func (*Store) Incidents 

func (s *Store) Incidents(ctx context.Context, f IncidentFilter) (IncidentPage, error)

Incidents returns incidents matching f, most recently opened first. It returns ErrInvalidCursor for a cursor it didn't return, and an error wrapping ErrInvalidIncident for an unknown status, severity or source.

func (*Store) Oldest 

func (s *Store) Oldest(ctx context.Context) (oldest time.Time, ok bool, err error)

Oldest returns the start of the oldest stored minute; ok is false when there are none.

func (*Store) OpenIncident 

func (s *Store) OpenIncident(ctx context.Context, n NewIncident) (Incident, IncidentUpdate, error)

OpenIncident opens an incident by the context's actor, with a first timeline update. It returns an error wrapping ErrInvalidIncident for a missing or out-of-bounds field.

func (*Store) ResolveIncident 

func (s *Store) ResolveIncident(ctx context.Context, id int64, message string) (Incident, IncidentUpdate, error)

ResolveIncident resolves an open incident now, by the context's actor, with a final timeline update. It returns ErrIncidentNotFound, ErrIncidentResolved, ErrTooManyUpdates, or an error wrapping ErrInvalidIncident for a missing message.

func (*Store) Summary 

func (s *Store) Summary(ctx context.Context, from, to time.Time) (Summary, error)

Summary adds up the minutes starting in [from, to), both truncated to the minute. It returns ErrInvalidRange for an empty or reversed range or one longer than MaxSummaryRange, and ErrQueryTimeout when the query takes longer than the query timeout.

func (*Store) UpdateIncident 

func (s *Store) UpdateIncident(ctx context.Context, id int64, c IncidentChange) (Incident, IncidentUpdate, error)

UpdateIncident adds a timeline update to an open incident by the context's actor, changing its status and severity when the change sets them. It returns ErrIncidentNotFound, ErrIncidentResolved, ErrTooManyUpdates, or an error wrapping ErrInvalidIncident.

func (*Store) WriteMinutes 

func (s *Store) WriteMinutes(ctx context.Context, minutes []Minute) error

WriteMinutes implements Sink: it stores minutes, replacing earlier writes of the same instance, minute, method and route.

type StoreOption 

type StoreOption func(*Store)

A StoreOption configures a Store.

func WithQueryTimeout 

func WithQueryTimeout(d time.Duration) StoreOption

WithQueryTimeout bounds Store.Summary. Default: DefaultQueryTimeout.

func WithStoreClock 

func WithStoreClock(now func() time.Time) StoreOption

WithStoreClock uses now instead of time.Now, for tests.

type Streams 

type Streams struct {
	// contains filtered or unexported fields
}

Streams bounds long-lived responses such as Server-Sent Events on one instance: how many run at once, how many one subject (such as a user) may hold, and how long each lasts. It is safe for concurrent use.

func NewStreams 

func NewStreams(maxTotal, maxPerSubject int, maxDuration time.Duration) (*Streams, error)

NewStreams returns limits of maxTotal streams, maxPerSubject per subject, each lasting at most maxDuration.

func (*Streams) Close 

func (s *Streams) Close()

Close ends every open stream and refuses new ones, for a shutting-down instance: load balancers and HTTP servers wait for long-lived responses.

func (*Streams) Count 

func (s *Streams) Count() int

Count returns how many streams are open.

func (*Streams) MaxDuration 

func (s *Streams) MaxDuration() time.Duration

MaxDuration returns how long a stream lasts at most.

func (*Streams) Open 

func (s *Streams) Open(ctx context.Context, subject string) (streamCtx context.Context, done func(), err error)

Open starts a stream for subject. The returned context ends with ctx, after the maximum duration (cause ErrStreamExpired) or when the streams are closed (cause ErrStreamsClosed); call done when the stream ends. Open returns ErrTooManyStreams over a limit.

type Summary 

type Summary struct {
	// From and To are the range's minutes: From inclusive, To exclusive.
	From, To time.Time
	Total    Stats
	// Minutes has one entry per minute with requests, oldest first.
	Minutes []MinuteStats
	// Instances has one entry per instance with requests, by instance ID.
	Instances []InstanceStats
	// Routes has one entry per method and route with requests, by route
	// and method.
	Routes []RouteStats
}

Summary is every instance's requests in a time range.

type UpdateKind 

type UpdateKind string

UpdateKind classifies a timeline update. 

const (
	UpdateOpened   UpdateKind = "opened"
	UpdateNote     UpdateKind = "update"
	UpdateResolved UpdateKind = "resolved"
	// UpdateRecovered and UpdateBreaching are added by detection when an
	// automatic incident's error rate recovers and when it is high again.
	UpdateRecovered UpdateKind = "recovered"
	UpdateBreaching UpdateKind = "breaching"
)

Update kinds.

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.