whstore

package
v1.146.2 Latest Latest
Warning

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

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

Documentation

Overview

Package whstore is the control data of inbound webhooks (#1870): which compaction windows of each source have segments, which of them are compacted and where, what retention has done to them, and the request counts and rejections a source's page reports. Events are never written here.

Index

Constants

This section is empty.

Variables

View Source 
var ErrNotFound = errors.New("webhook window not found")

ErrNotFound is returned when a write names a window no row holds.

Functions

func AckBound added in v1.145.0 

func AckBound(i int) int64

AckBound is the upper bound, in milliseconds, of acknowledgement-latency bucket i, for i below ackBuckets-1. The last bucket has no upper bound.

func AckBucket added in v1.145.0 

func AckBucket(ms int64) int

AckBucket is the bucket a latency of ms milliseconds is counted in.

Types

type AckLatency added in v1.145.0 

type AckLatency struct {
	Count int64 `json:"count"`
	P50MS int64 `json:"p50_ms"`
	P99MS int64 `json:"p99_ms"`
	MaxMS int64 `json:"max_ms"`
	// Slowest is the slowest delivery of the span, null when there was none.
	Slowest *SlowAck `json:"slowest"`
}

AckLatency is a source's acknowledgement latency over a span. P50MS and P99MS are the upper bound of the bucket the percentile falls in, capped at MaxMS, so each is the most it could be: "p99_ms 250" means 99% of deliveries were answered in 250ms or less.

type Compaction 

type Compaction struct {
	Segments   int
	Events     int64
	Duplicates int64
	Digest     string
	ResourceID string
	Location   string
}

Compaction is what one compaction of a window produced.

type CompactionState added in v1.141.0 

type CompactionState struct {
	Pending    int64
	Running    int64
	Waiting    int64
	OldestOwed time.Duration
}

CompactionState is the compactor's backlog as the database holds it, read on a metrics scrape (#1897): windows that have ended and are owed a compaction, windows a compactor holds now, windows still receiving events (owed one once they end), and how long ago the oldest owed window ended. Every replica reads the same rows, so a reader takes the max.

type Count 

type Count struct {
	Source  string
	Minute  time.Time
	Outcome string
	Count   int64
}

Count is a number of requests of one source with one outcome in one minute.

type Latency added in v1.145.0 

type Latency struct {
	Source  string
	Minute  time.Time
	Buckets [ackBuckets]int64
	Slowest SlowAck
}

Latency is the acknowledgement latency of one source's deliveries answered 202 in one minute: a count per bucket and the slowest of them.

func (*Latency) Merge added in v1.145.0 

func (l *Latency) Merge(o Latency)

Merge adds o's deliveries to l.

func (*Latency) Observe added in v1.145.0 

func (l *Latency) Observe(s SlowAck)

Observe adds one acknowledged delivery to l.

type Overview added in v1.138.2 

type Overview struct {
	// Summaries holds one entry per source that has counts or windows; a
	// source with neither has no entry.
	Summaries map[string]Summary
	// Volume is the request counts from Since, summed per bucket of Step,
	// oldest bucket first. A bucket with no requests has no point.
	Volume []VolumePoint
	// Rejections are the newest rejected requests of every source, or of the
	// source read, whose latest rejection is at or after Since.
	Rejections []Rejection
	// Since is the series' lower bound: a count is in the series when its
	// minute is at or after it, the bound LastHour and LastDay are read with.
	Since time.Time
}

Overview is what the overview of every source reads, in one pass over the control tables.

type Rejection 

type Rejection struct {
	Source string `json:"-"`
	// At is the latest rejection of the run.
	At time.Time `json:"at"`
	// FirstAt is the earliest. Equal to At for a single rejection.
	FirstAt time.Time `json:"first_at"`
	// Count is how many rejections the row stands for, at least one.
	Count   int64  `json:"count"`
	Outcome string `json:"outcome"`
	Reason  string `json:"reason"`
	// RequestHash is the SHA-256 of the refused request's body, and EventIDs
	// the ids of its first events, when its events were read; empty for a
	// request refused before that. A rejection carrying ids is one request,
	// never folded into a run.
	RequestHash string   `json:"request_hash"`
	EventIDs    []string `json:"event_ids"`
}

Rejection is a run of refused requests of one source with one outcome and reason: how many, the first and the latest. Never the body. A rejection the receiver records is a run of one.

type SlowAck added in v1.145.0 

type SlowAck struct {
	At       time.Time `json:"at"`
	TotalMS  int64     `json:"total_ms"`
	BufferMS int64     `json:"buffer_ms"`
	MarkMS   int64     `json:"mark_ms"`
	PutMS    int64     `json:"put_ms"`
}

SlowAck is the slowest acknowledged delivery of a span: when it was answered, how long the sender waited in all, and how long its events spent in the buffer, recording their window and writing their segment.

type SpoolState added in v1.146.0 

type SpoolState struct {
	Replica string
	Dir     string
	Bytes   int64
	Events  int64
	Files   int
	// Oldest is when the oldest event waiting was received; nil when the
	// spool is empty.
	Oldest   *time.Time
	MaxBytes int64
	MaxAge   time.Duration
	// Drained and DrainFailures count since the receiver started.
	Drained       int64
	DrainFailures int64
	LastDrainAt   *time.Time
	LastError     string
	StartedAt     time.Time
	UpdatedAt     time.Time
}

SpoolState is one receiver's local spool as it last reported it (#2121): what waits to be drained to object storage, and how the drain is going.

type Status 

type Status struct {
	// LastHour and LastDay are request counts by outcome.
	LastHour map[string]int64 `json:"last_hour"`
	LastDay  map[string]int64 `json:"last_day"`
	// LastSegmentAt is when a segment of accepted events was last written,
	// which is when the source last received an event. A sender that stops
	// produces no error on the receiving side; this is how that shows.
	LastSegmentAt *time.Time `json:"last_segment_at"`
	// LastCompactedWindow is the start of the newest window whose Parquet
	// file holds every event its segments do.
	LastCompactedWindow *time.Time `json:"last_compacted_window"`
	// Pending counts windows owed a compaction: not yet compacted, or
	// dirtied by a segment written after they were.
	Pending int `json:"pending"`
	// Failing counts pending windows whose last compaction attempt failed.
	Failing   int    `json:"failing"`
	LastError string `json:"last_error,omitempty"`
	// OldestWindow is the start of the oldest window still held.
	OldestWindow *time.Time `json:"oldest_window"`
	// AckLastHour and AckLastDay are how long the deliveries answered 202
	// waited for that answer.
	AckLastHour AckLatency  `json:"ack_last_hour"`
	AckLastDay  AckLatency  `json:"ack_last_day"`
	Rejections  []Rejection `json:"rejections"`
	// SegmentsOpenWindow and SegmentsPerWindowP95 are the segments recorded
	// against the open window, and the 95th percentile of that count over
	// the windows that started in the last day (#2132), as the overview of
	// every source reports them.
	SegmentsOpenWindow   int64 `json:"segments_open_window"`
	SegmentsPerWindowP95 int64 `json:"segments_per_window_p95"`
}

Status is what a source's page reports about it.

type Store 

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

Store reads and writes the webhook control tables.

func New 

func New(db *sql.DB) *Store

New creates a store over the platform database.

func (*Store) ClaimOwed 

func (s *Store) ClaimOwed(ctx context.Context, endedBy time.Time, lease time.Duration, limit int) ([]Window, error)

ClaimOwed claims up to limit windows owed a compaction that ended at or before endedBy, for lease. A window another replica holds is skipped until its lease runs out, and a window that failed is held back until the backoff recorded against it passes.

func (*Store) CompactionState added in v1.141.0 

func (s *Store) CompactionState(ctx context.Context) (CompactionState, error)

CompactionState reads the compactor's backlog.

func (*Store) DeleteExpired added in v1.137.0 

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

DeleteExpired removes the records of windows expired before before, and reports how many it removed. An expired window's partition, file and raw segments are already gone; the record is kept until then only so a pass can tell the window was handled (#1904).

func (*Store) Expirable 

func (s *Store) Expirable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)

Expirable returns the windows of source that ended at or before cutoff and are not yet expired, and the expired ones a segment has landed in since, whose raw objects have to be deleted again so the view does not serve them.

func (*Store) MarkExpired 

func (s *Store) MarkExpired(ctx context.Context, w Window) error

MarkExpired records that a window is gone: its partition, its resource and its raw segments.

func (*Store) MarkRawDeleted 

func (s *Store) MarkRawDeleted(ctx context.Context, w Window) error

MarkRawDeleted records that a window's raw segments were deleted.

func (*Store) MarkSegment 

func (s *Store) MarkSegment(ctx context.Context, source string, start time.Time, length time.Duration) error

MarkSegment records that a segment for the window starting at start and length long was written. A window already compacted becomes owed another compaction, because its Parquet file no longer holds every event its segments do. A window recorded with a shorter length is lengthened: a source whose compact_every was raised can land a segment in a window the old setting had already ended.

func (*Store) MarkUnregistered 

func (s *Store) MarkUnregistered(ctx context.Context, w Window) error

MarkUnregistered records that a window's compacted partition was removed from the table. It is written before the window's objects are deleted, so a pass that stops between the two finds the partition already gone and deletes what is left.

func (*Store) Overview added in v1.138.2 

func (s *Store) Overview(ctx context.Context, source string, now time.Time, span, step time.Duration) (Overview, error)

Overview reads the summary at now of every source, or of source alone when it is not empty, the request series over the span before now in buckets of step, and the newest rejections within that span. The five reads are independent and run concurrently.

func (*Store) PruneCounts 

func (s *Store) PruneCounts(ctx context.Context, before time.Time) error

PruneCounts deletes per-minute counts and acknowledgement latency older than before.

func (*Store) RawDeletable 

func (s *Store) RawDeletable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)

RawDeletable returns the windows of source whose raw segments may be deleted: compacted at the generation they now hold, so every segment's events are in the Parquet file, and whose last segment was written at or before cutoff. A window owed a compaction is never returned, which is what keeps retention from deleting an event that has not been compacted.

func (*Store) RawRegistrable added in v1.146.1 

func (s *Store) RawRegistrable(ctx context.Context, source string, since time.Time) ([]Window, error)

RawRegistrable returns the windows of source that hold raw segments, not expired, whose last segment was written at or after since: the windows a retention pass makes sure the raw table lists (#2133). A window whose segments were deleted is included again once a segment lands in it after the deletion.

func (*Store) RecordCompacted 

func (s *Store) RecordCompacted(ctx context.Context, w Window, c Compaction) error

RecordCompacted records a compaction of w at the generation it was claimed at, and releases the claim. If a segment was recorded since the claim, the window's generation is ahead of the one recorded here and it is claimed again.

func (*Store) RecordCounts 

func (s *Store) RecordCounts(ctx context.Context, counts []Count) error

RecordCounts adds counts to the per-minute totals. Counts naming the same source, minute and outcome are summed before they are written, because one upsert cannot touch a row twice. A count for a source that no longer exists is dropped with the rest of that source's data rather than failing the batch.

func (*Store) RecordFailure 

func (s *Store) RecordFailure(ctx context.Context, w Window, reason string, hold time.Duration) error

RecordFailure keeps why a compaction of w failed and holds the window back for hold.

func (*Store) RecordLatency added in v1.145.0 

func (s *Store) RecordLatency(ctx context.Context, latencies []Latency) error

RecordLatency adds each minute's latency to the stored one: bucket counts summed, the slower of the two slowest kept. A minute of a source that no longer exists is dropped with the rest of that source's data.

func (*Store) RecordLocation 

func (s *Store) RecordLocation(ctx context.Context, w Window, resourceID, location string) error

RecordLocation records the resource and partition location of a window before its compaction is recorded, so a failure after the partition moved leaves the record describing where the partition is.

func (*Store) RecordRejections 

func (s *Store) RecordRejections(ctx context.Context, rejections []Rejection) error

RecordRejections keeps rejected requests, then trims each source and outcome named to its newest maxRejections rows.

The batch is first folded into runs (collapseRejections). Each run is added to the newest row of its source and outcome when that row has the same reason and its latest rejection is within rejectionRunGap of the run's first, and is written as a new row otherwise. A burst therefore costs one row, written once, when the database is least able to absorb many.

func (*Store) RecordSpool added in v1.146.0 

func (s *Store) RecordSpool(ctx context.Context, st SpoolState) error

RecordSpool writes a receiver's spool state, replacing what it wrote last, and removes the state of replicas that stopped reporting a week ago.

func (*Store) ResourceIDs 

func (s *Store) ResourceIDs(ctx context.Context, source string) ([]string, error)

ResourceIDs returns the resources a source's windows were written as.

func (*Store) SourcesForResources 

func (s *Store) SourcesForResources(ctx context.Context, ids []string) (map[string]string, error)

SourcesForResources returns, for each resource id that holds a compacted window, the source the window belongs to.

func (*Store) SpoolStates added in v1.146.0 

func (s *Store) SpoolStates(ctx context.Context) ([]SpoolState, error)

SpoolStates reads every receiver's spool state, by replica.

func (*Store) Status 

func (s *Store) Status(ctx context.Context, source string, now time.Time) (Status, error)

Status reads a source's status at now.

type Summary added in v1.138.2 

type Summary struct {
	// LastHour and LastDay are request counts by outcome.
	LastHour map[string]int64
	LastDay  map[string]int64
	// LastSegmentAt is when the source last received an event; nil when no
	// window it still holds records one.
	LastSegmentAt *time.Time
	// Pending and Failing count windows owed a compaction, and those of them
	// whose last attempt failed.
	Pending   int
	Failing   int
	LastError string
	// AckLastHour and AckLastDay are how long the deliveries answered 202
	// waited for that answer.
	AckLastHour AckLatency
	AckLastDay  AckLatency
	// SegmentsOpenWindow is how many segments were recorded against the
	// source's open window, and SegmentsPerWindowP95 the 95th percentile of
	// that count over the windows that started in the last day (#2132): the
	// objects a reader of a window reads before a rollup merges them. A
	// receiver records its first write of a window twice, so each count can
	// exceed the segments written by one per replica.
	SegmentsOpenWindow   int64
	SegmentsPerWindowP95 int64
}

Summary is one source's line on the overview of every source: the counts and window aggregates Status reports, without the per-source rejections and the window positions only the source's own page shows.

type VolumePoint added in v1.138.2 

type VolumePoint struct {
	Source  string
	At      time.Time
	Outcome string
	Count   int64
}

VolumePoint is the number of requests of one source with one outcome in one bucket of the overview's series. At is the bucket's start.

type Window 

type Window struct {
	Source string
	Start  time.Time
	Length time.Duration
	// Generation is the window's segment counter when it was read. Compacting
	// it records this number, so a segment recorded while the compaction ran
	// leaves the window owed another one.
	Generation int64
	// ResourceID and Location are the last compaction's resource and the
	// directory its partition is registered at, empty before the first.
	ResourceID string
	Location   string
	Attempts   int
}

Window is one compaction window of one source.

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.