Documentation
¶
Index ¶
- Variables
- func NewMessageHandler(logger *logging.Logger, batchAdder BatchAdder) consumer.MessageHandler
- func WithQueue(queueConfig *mqs.QueueConfig) func(opts *LogMQOption)
- type AlertEvaluator
- type AlertPipeline
- type BatchAdder
- type BatchProcessor
- type BatchProcessorConfig
- type DestinationDisabler
- type LogInfra
- type LogMQ
- type LogMQOption
- type LogStore
- type ReplayGate
- type SuppressionWindow
Constants ¶
This section is empty.
Variables ¶
var ErrInvalidLogEntry = errors.New("invalid log entry: both event and attempt are required")
ErrInvalidLogEntry is returned when a LogEntry is missing required fields.
Functions ¶
func NewMessageHandler ¶
func NewMessageHandler(logger *logging.Logger, batchAdder BatchAdder) consumer.MessageHandler
func WithQueue ¶
func WithQueue(queueConfig *mqs.QueueConfig) func(opts *LogMQOption)
Types ¶
type AlertEvaluator ¶ added in v1.1.0
type AlertEvaluator interface {
Evaluate(ctx context.Context, attempt alert.Attempt) (alert.Evaluation, error)
// SignalsEnabled reports whether any alert signal can ever fire. When
// false, Evaluate is a stateless no-op, so the pipeline skips the replay
// gate (there is no streak or verdict to protect from replays).
SignalsEnabled() bool
}
AlertEvaluator evaluates one delivery attempt against the destination's failure history and returns the tracker's verdict as data. Acting on the verdict (opevents, auto-disable, replay dedup) is owned by the batch processor.
type AlertPipeline ¶ added in v1.1.0
type AlertPipeline struct {
// Evaluator is the alert tracker. Required.
Evaluator AlertEvaluator
// Emitter delivers the operator events. Required.
Emitter opevents.Emitter
// Disabler auto-disables a destination when the 100% threshold is crossed.
// Nil disables auto-disable.
Disabler DestinationDisabler
// ProcessedIdemp is the per-attempt replay gate: a replay of a fully
// processed failed attempt is skipped instead of re-counting/re-alerting.
// Required.
ProcessedIdemp ReplayGate
// ExhaustedIdemp is the per-(tenant,destination) suppression window for
// exhausted-retries alerts: at most one alert per destination within the
// window, regardless of which events exhaust. Nil means no suppression
// (alert on every exhaustion).
ExhaustedIdemp SuppressionWindow
}
AlertPipeline groups the post-persist alert pipeline: evaluate the attempt, act on the verdict (disable, opevents), and dedup replays. Evaluator, Emitter, and ProcessedIdemp are always required — "alerting off" is expressed by config (signals disabled, no topics subscribed), never by absent deps.
type BatchAdder ¶ added in v0.13.0
BatchAdder is the interface for adding messages to a batch processor.
type BatchProcessor ¶ added in v0.13.0
type BatchProcessor struct {
// contains filtered or unexported fields
}
BatchProcessor batches log entries and writes them to the log store, then runs the alert pipeline per entry on its own goroutine — dispatch-and-move-on, so a slow eval or sink send never blocks persistence of the next batch.
Entries process in no particular order, including within a destination: the consecutive-failure count tolerates approximate order (its store is a set of attempt IDs, so counting is idempotent and commutative; only a success/ failure race around the reset can skew it), and per-process ordering was cosmetic anyway with multiple logmq replicas interleaving a destination's attempts. See the discussion on the parallelism RFC.
func NewBatchProcessor ¶ added in v0.13.0
func NewBatchProcessor(ctx context.Context, logger *logging.Logger, logStore LogStore, alerts AlertPipeline, cfg BatchProcessorConfig) (*BatchProcessor, error)
NewBatchProcessor creates a new batch processor for log entries.
func (*BatchProcessor) Shutdown ¶ added in v0.13.0
func (bp *BatchProcessor) Shutdown()
Shutdown gracefully shuts down the batch processor: the batcher first (flushes pending batches, which may still dispatch entry goroutines), then the in-flight entries drain. Every dispatched message reaches a terminal state before Shutdown returns, and the drain is bounded by emitTimeout. Idempotent.
type BatchProcessorConfig ¶ added in v0.13.0
type BatchProcessorConfig struct {
ItemCountThreshold int
DelayThreshold time.Duration
// EmitTimeout is a test-only override for the per-send timeout; zero means
// the emitTimeout default. Production always runs the default.
EmitTimeout time.Duration
}
BatchProcessorConfig configures the batch processor.
type DestinationDisabler ¶ added in v1.1.0
type DestinationDisabler interface {
DisableDestination(ctx context.Context, tenantID, destinationID string) error
}
DestinationDisabler disables destinations that hit the auto-disable threshold.
type LogMQ ¶
type LogMQ struct {
// contains filtered or unexported fields
}
func New ¶
func New(opts ...func(opts *LogMQOption)) *LogMQ
func (*LogMQ) Subscribe ¶
func (q *LogMQ) Subscribe(ctx context.Context, opts ...mqs.SubscribeOption) (mqs.Subscription, error)
type LogMQOption ¶
type LogMQOption struct {
QueueConfig *mqs.QueueConfig
}
type LogStore ¶ added in v0.13.0
LogStore defines the interface for persisting log entries. This is a consumer-defined interface containing only what logmq needs.
type ReplayGate ¶ added in v1.1.0
type ReplayGate interface {
Processed(ctx context.Context, key string) (bool, error)
MarkProcessed(ctx context.Context, key string) error
}
ReplayGate is the split-phase idempotence pair the pipeline uses as the per-attempt replay gate: Processed is checked before eval, MarkProcessed lands after delivery. Split-phase means no in-flight conflict detection — concurrent duplicates both run and may both emit (tolerated: opevents are at-least-once). Satisfied by idempotence.Idempotence.
type SuppressionWindow ¶ added in v1.1.0
type SuppressionWindow interface {
Exec(ctx context.Context, key string, exec func(context.Context) error) error
}
SuppressionWindow wraps one send in a keyed dedup window: within the window the send is skipped and counts as delivered. Satisfied by idempotence.Idempotence.
Source Files