Documentation
¶
Overview ¶
Package compactor turns each window of a webhook source's raw segments into one Parquet file (#1870), and applies the source's retention.
A window is compacted once it has ended and a grace period for segments still being written has passed. The compactor reads every raw segment of the window, and the window's previous Parquet file when there is one, removes duplicate event ids keeping the earliest received, writes the result as the window's managed resource, and registers the window's partition at that resource's directory. A segment written for a window after it was compacted makes it owed another compaction, which starts again from everything the window holds, so rewriting a window is idempotent.
The worker claims windows under a Postgres lease, so any number of replicas run it and each window is compacted by one of them at a time.
Index ¶
Constants ¶
const ( ResultCompacted = "compacted" ResultFailed = "failed" )
Results a compaction is counted under.
const ( DefaultPoll = 30 * time.Second DefaultLease = 10 * time.Minute DefaultBatch = 4 DefaultGrace = 2 * time.Minute DefaultRetryBackoff = time.Minute DefaultRetentionEvery = 10 * time.Minute )
Defaults for the worker's pacing.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Deps ¶
type Deps struct {
Windows WindowStore
Sources Sources
Objects Objects
Bucket string
Tables Tables
Resources WindowResources
Metrics Metrics
Logger *slog.Logger
Now func() time.Time
}
Deps are what the worker acts through.
type Metrics ¶
type Metrics interface {
WebhookCompaction(ctx context.Context, source, result string)
WebhookDuplicatesDropped(ctx context.Context, source string, n int64)
}
Metrics is what the compactor reports. A nil Metrics reports nothing.
type Objects ¶
type Objects interface {
// ListKeys returns every key under prefix, across as many pages as the
// listing takes.
ListKeys(ctx context.Context, bucket, prefix string) ([]string, error)
GetObject(ctx context.Context, bucket, key string) ([]byte, error)
DeleteObject(ctx context.Context, bucket, key string) error
}
Objects reads, lists and deletes objects in the managed-resources bucket.
type Sources ¶
type Sources interface {
Get(ctx context.Context, name string) (whsource.Source, error)
List(ctx context.Context) ([]whsource.Source, error)
}
Sources reads source definitions. whsource.Store satisfies it.
type StoredWindow ¶
type StoredWindow struct {
ResourceID string
// Key is the object key of the file, in the managed-resources bucket.
// Its directory is what the window's partition is registered at.
Key string
}
StoredWindow is where a compacted window's resource keeps its file.
type Tables ¶
type Tables interface {
TargetFor(connection string) (whtable.Target, error)
S3Location(prefix string) string
RegisterWindow(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time, location string) error
UnregisterWindow(ctx context.Context, tg whtable.Target, src whsource.Source, window time.Time) error
SyncRaw(ctx context.Context, tg whtable.Target, src whsource.Source) error
}
Tables is what the compactor does in the query engine. whtable.Tables satisfies it.
type Tuning ¶
type Tuning struct {
Poll time.Duration
Lease time.Duration
Batch int
Grace time.Duration
RetryBackoff time.Duration
RetentionEvery time.Duration
}
Tuning paces the worker. A zero field takes its default.
type WindowResources ¶
type WindowResources interface {
// Put writes a window's Parquet file: a new resource when existingID is
// empty or names one that is gone, and a new version of it otherwise. The
// source decides whose library a new resource is in.
Put(ctx context.Context, src whsource.Source, window time.Time, existingID string, content []byte) (StoredWindow, error)
// Key returns where a resource's current file is, and false when the
// resource is gone.
Key(ctx context.Context, id string) (string, bool, error)
// Delete removes a resource and every version of its file. A resource
// already gone is not an error.
Delete(ctx context.Context, id string) error
}
WindowResources writes and deletes the managed resource each compacted window is.
type WindowStore ¶
type WindowStore interface {
ClaimOwed(ctx context.Context, endedBy time.Time, lease time.Duration, limit int) ([]whstore.Window, error)
RecordLocation(ctx context.Context, h whstore.Window, resourceID, location string) error
RecordCompacted(ctx context.Context, h whstore.Window, c whstore.Compaction) error
RecordFailure(ctx context.Context, h whstore.Window, reason string, hold time.Duration) error
RawDeletable(ctx context.Context, source string, cutoff time.Time) ([]whstore.Window, error)
Expirable(ctx context.Context, source string, cutoff time.Time) ([]whstore.Window, error)
MarkRawDeleted(ctx context.Context, h whstore.Window) error
MarkUnregistered(ctx context.Context, h whstore.Window) error
MarkExpired(ctx context.Context, h whstore.Window) error
PruneCounts(ctx context.Context, before time.Time) error
}
WindowStore is the compactor's half of the control data. whstore.Store satisfies it.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker compacts windows and applies retention until stopped.
func New ¶
New builds a worker. It returns nil when a dependency is missing, and a nil worker's Start and Stop do nothing.
func (*Worker) Pass ¶
Pass compacts one batch of owed windows, applies retention when it is due, and reports whether it found windows to compact, which is the caller's cue to run again without waiting.
func (*Worker) Retention ¶
Retention applies every source's retention, and prunes request counts no page reads any more.
Retention has two settings. A compacted window's raw segments are deleted once the last of them is older than raw retention; a window owed a compaction is never touched, so no event is deleted before it is in a Parquet file. A whole window is removed once it is older than compacted retention: its partition is unregistered first, so the view stops serving it, then its resource and any raw segments are deleted, and the window is recorded as expired. Each step is recorded in the platform database, so a pass that stops part-way is finished by the next.
Source Files