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 ¶
- Variables
- type Compaction
- type Count
- type Rejection
- type Status
- type Store
- func (s *Store) ClaimOwed(ctx context.Context, endedBy time.Time, lease time.Duration, limit int) ([]Window, error)
- func (s *Store) Expirable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)
- func (s *Store) MarkExpired(ctx context.Context, w Window) error
- func (s *Store) MarkRawDeleted(ctx context.Context, w Window) error
- func (s *Store) MarkSegment(ctx context.Context, source string, start time.Time, length time.Duration) error
- func (s *Store) MarkUnregistered(ctx context.Context, w Window) error
- func (s *Store) PruneCounts(ctx context.Context, before time.Time) error
- func (s *Store) RawDeletable(ctx context.Context, source string, cutoff time.Time) ([]Window, error)
- func (s *Store) RecordCompacted(ctx context.Context, w Window, c Compaction) error
- func (s *Store) RecordCounts(ctx context.Context, counts []Count) error
- func (s *Store) RecordFailure(ctx context.Context, w Window, reason string, hold time.Duration) error
- func (s *Store) RecordLocation(ctx context.Context, w Window, resourceID, location string) error
- func (s *Store) RecordRejections(ctx context.Context, rejections []Rejection) error
- func (s *Store) ResourceIDs(ctx context.Context, source string) ([]string, error)
- func (s *Store) SourcesForResources(ctx context.Context, ids []string) (map[string]string, error)
- func (s *Store) Status(ctx context.Context, source string, now time.Time) (Status, error)
- type Window
Constants ¶
This section is empty.
Variables ¶
var ErrNotFound = errors.New("webhook window not found")
ErrNotFound is returned when a write names a window no row holds.
Functions ¶
This section is empty.
Types ¶
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 Rejection ¶
type Rejection struct {
Source string `json:"-"`
At time.Time `json:"at"`
Outcome string `json:"outcome"`
Reason string `json:"reason"`
}
Rejection is one refused request: when, which outcome, and why. Never the body.
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"`
Rejections []Rejection `json:"rejections"`
}
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 (*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) Expirable ¶
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 ¶
MarkExpired records that a window is gone: its partition, its resource and its raw segments.
func (*Store) MarkRawDeleted ¶
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 ¶
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) PruneCounts ¶
PruneCounts deletes per-minute counts 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) RecordCompacted ¶
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 ¶
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) RecordLocation ¶
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 ¶
RecordRejections keeps rejected requests, then trims each source named to its newest maxRejections.
func (*Store) ResourceIDs ¶
ResourceIDs returns the resources a source's windows were written as.
func (*Store) SourcesForResources ¶
SourcesForResources returns, for each resource id that holds a compacted window, the source the window belongs to.
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.