Producer and batching

A Producer owns a buffered channel and a goroutine. SendEvent puts an event on the channel and returns. The goroutine collects events into a batch and flushes the batch when it is full or when the interval elapses, whichever happens first.

NewProducer returns a *Producer and no error. It has no required field, and it starts that goroutine before it returns, so the value is usable immediately.

The options

// source: producer.go#L52-L65
// ProducerOptions holds configuration options for a Producer.
type ProducerOptions struct {
    // ProjectName is stamped on every event that has no project name of its own.
    ProjectName string `json:"project_name" yaml:"ProjectName" toml:"ProjectName"`
    // BufferInterval is the maximum duration between flushes. Defaults to 5s when zero or negative.
    BufferInterval time.Duration `json:"buffer_interval" yaml:"BufferInterval" toml:"BufferInterval"`
    // MaxBatchSize is the number of buffered events that triggers an immediate flush. Defaults to 100.
    MaxBatchSize int `json:"max_batch_size" yaml:"MaxBatchSize" toml:"MaxBatchSize"`
    // MergeWindow, when positive, merges events with the same type and message
    // captured within this window into one event annotated with the count.
    MergeWindow time.Duration `json:"merge_window,omitempty" yaml:"MergeWindow,omitempty" toml:"MergeWindow,omitempty"`
    // OnError receives flush and close errors. Defaults to logging through log/slog.
    OnError func(err error) `json:"-" yaml:"-" toml:"-"`
}

Field	Type	Default	What it does
ProjectName	string	empty	Stamped on every event that arrives without one. Groups messages: each flusher sends one message per project in a batch
BufferInterval	time.Duration	5s	The longest a buffered event waits. A value below a second is honoured; zero or negative means 5s
MaxBatchSize	int	100	The batch size that triggers an immediate flush. Also sets the channel capacity, which is twice this number
MergeWindow	time.Duration	0, off	Collapses repeats inside one batch. See below
OnError	func(error)	log/slog at error level	Receives every flush and close error

The config-file keys are project_name, buffer_interval, max_batch_size and merge_window in JSON, and ProjectName, BufferInterval, MaxBatchSize, MergeWindow in YAML and TOML. OnError is a function, so it is tagged - in all three and is set in code. The full table for every options struct is on Configuration file keys.

When a batch flushes

Three things end a batch:

1. It reached MaxBatchSize. The flush happens on the same turn as the event that filled it.
2. BufferInterval elapsed and the batch is not empty. An empty tick does nothing.
3. Close was called. The channel closes, the loop drains what is left and flushes it once.

Before the flush, the producer sorts the batch by capture time, merges it when MergeWindow is set, and stamps SentAt on every event. Then it hands the same slice to every flusher at once through patrol.FlushAll.

That slice is shared. Flushers read it concurrently, so a flusher must not sort it, append to it or change an event in it. The race detector is what catches a flusher that does.

Independent flushers

Every flusher gets the batch, every flusher runs to completion, and a failure in one does not cancel another. The errors come back joined into one error whose parts are prefixed with the flusher’s name, from its String() method or its Go type:

slack: invalid_auth
inbox: dial tcp 10.0.0.5:587: connect: connection refused

That joined error is what OnError receives. Nothing is retried and nothing is written to disk: a destination that needs retries wraps itself.

Merging repeats

MergeWindow is for the loop that throws the same error two hundred times in four seconds. Inside one batch, a run of events with the same project, type and message, captured within the window of the first one, becomes a single event whose message gains a count:

redis: connection refused (x214)

The survivor keeps the union of the fields and files of the run. Two limits are worth knowing before you turn it on. It only merges inside one batch, so a run split across a flush boundary produces two messages. And capture times are Unix seconds, so a window under a second behaves like a window of one second.

Sending

package main

import (
    "context"
    "log"
    "time"

    "github.com/kataras/patrol"
)

func main() {
    ctx := context.Background()

    producer := patrol.NewProducer(patrol.ProducerOptions{
        ProjectName:    "checkout-api",
        BufferInterval: 2 * time.Second,
        MaxBatchSize:   50,
        MergeWindow:    10 * time.Second,
        OnError:        func(err error) { log.Println("patrol:", err) },
    })
    defer producer.Close(ctx)

    if err := producer.SendEvent(ctx, patrol.NewEvent(patrol.Info).WithMessage("cache warmed")); err != nil {
        log.Println(err)
    }

    if err := producer.CaptureException(ctx, context.DeadlineExceeded); err != nil {
        log.Println(err)
    }
}

SendEvent never panics and never blocks forever. It returns ctx.Err() when the buffer is full and the context ends first, patrol.ErrClosed after Close, and an error when the event is nil. A Debug event while debug mode is off is dropped and returns nil.

CaptureException(ctx, err) is SendEvent(ctx, patrol.NewException(err)), which is the call most services make from a recover block.

Checking the destinations

producer.Test(ctx) calls Test on every flusher that implements Tester, concurrently, and joins the errors with the flusher names. Slack checks its token, Discord reads every configured channel, Twilio lists an account, the email flusher opens an SMTP session and quits, SQL pings, and the HTTP client does a GET / against the server. Each of those is a real round trip, so it belongs at startup, not on a request path.

Next: what goes into an event.