Documentation
¶
Overview ¶
Package receiver serves POST /hooks/{source} (#1870): it authenticates each request, turns its body into events, buffers them per source, writes them to object storage as gzipped JSON-lines segments, and answers 202 only once the segment holding them is written.
A 2xx therefore means the events are in object storage. A replica that dies holding a buffer answered nobody for those events, so their senders retry, and backpressure is the sender's own retry queue: a full buffer or a failed write is answered 503 with Retry-After, and the platform keeps no backlog of its own beyond each source's buffer limit.
Index ¶
Constants ¶
const ( OutcomeAccepted = "accepted" OutcomeTooLarge = "too_large" OutcomeRateLimited = "rate_limited" OutcomeBufferFull = "buffer_full" OutcomeWriteFailed = "write_failed" OutcomeUnknownSource = "unknown_source" OutcomeInvalidBody = "invalid_body" )
Outcomes a request is counted under.
const ( DefaultWriteTimeout = 30 * time.Second DefaultRefresh = 5 * time.Second DefaultStatsFlush = 10 * time.Second )
Defaults for the receiver's own timing.
const PathPrefix = "/hooks/"
PathPrefix is where the receiver is mounted.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
Sources SourceLister
Objects ObjectWriter
Bucket string
Recorder Recorder
RawWindows RawWindows
Metrics Metrics
// Replica is this process's identity, the value of X-Platform-Instance.
Replica string
Logger *slog.Logger
// WriteTimeout bounds one segment write, including its retries by the
// S3 client. A write that has not finished by then is answered 503.
WriteTimeout time.Duration
// Refresh is how often the source list is re-read, which is how a
// change made on another replica reaches this one.
Refresh time.Duration
StatsFlush time.Duration
Now func() time.Time
}
Config is what a Receiver is built from.
type Metrics ¶
type Metrics interface {
WebhookRequest(ctx context.Context, source, outcome string)
WebhookEvents(ctx context.Context, source string, n int)
WebhookSegmentWritten(ctx context.Context, source string)
WebhookAck(ctx context.Context, source string, d time.Duration)
// WebhookBuffer reports how many events a source holds in memory: those
// waiting to be written and those in a write under way.
WebhookBuffer(ctx context.Context, source string, events int)
}
Metrics is what the receiver reports to the platform's metrics registry. A nil Metrics reports nothing.
type ObjectWriter ¶
type ObjectWriter interface {
PutObject(ctx context.Context, bucket, key string, data []byte, contentType string) error
}
ObjectWriter writes one object. The managed-resources S3 client satisfies it.
type RawWindows ¶
type RawWindows interface {
EnsureRawWindow(ctx context.Context, src whsource.Source, start time.Time) error
}
RawWindows registers a new window's raw partition, so an event is queryable as soon as it is acknowledged. A nil RawWindows leaves new windows to the compactor's sync.
type Receiver ¶
type Receiver struct {
// contains filtered or unexported fields
}
Receiver serves the webhook routes.
func (*Receiver) Reload ¶
func (r *Receiver) Reload()
Reload re-reads the sources now rather than at the next refresh: it is what an administrator's change on this replica calls.
func (*Receiver) ServeHTTP ¶
func (r *Receiver) ServeHTTP(w http.ResponseWriter, req *http.Request)
ServeHTTP handles /hooks/{source} and /hooks/{source}/{token}.
type Recorder ¶
type Recorder interface {
MarkSegment(ctx context.Context, source string, start time.Time, length time.Duration) error
RecordCounts(ctx context.Context, counts []whstore.Count) error
RecordRejections(ctx context.Context, rejections []whstore.Rejection) error
}
Recorder is the control data the receiver writes. whstore.Store satisfies it.
Source Files