Custom flushers

A destination is any type with one method. There is no registration step and no interface to embed.

// source: producer.go#L20-L38
// EventFlusher delivers a batch of events to its destination.
//
// The batch is sorted by capture time, is shared with the other flushers of
// the same Producer or Server and must be treated as read-only.
// Implementations should be safe for concurrent use.
//
// A flusher may also implement Tester, Closer and fmt.Stringer; the name
// returned by String is used in error messages.
type EventFlusher interface {
    FlushEvents(ctx context.Context, evts []*Event) error
}

// EventFlusherFunc adapts a function to the EventFlusher interface.
type EventFlusherFunc func(ctx context.Context, evts []*Event) error

// FlushEvents calls f.
func (f EventFlusherFunc) FlushEvents(ctx context.Context, evts []*Event) error {
    return f(ctx, evts)
}

patrol.EventFlusherFunc wraps a plain function, which is enough for a webhook, a metrics counter or a test spy.

The read-only batch rule

The slice is shared. Every flusher of a producer receives the same backing array at the same time, from different goroutines. Reading it is safe. Sorting it, appending to it, or changing a field on an event in it is a data race with every other flusher, and the race detector is what finds it.

If you need a different order or a subset, copy first:

package main

import (
    "context"
    "slices"

    "github.com/kataras/patrol"
)

// errorsOnly forwards only the Error events, in a slice of its own.
func errorsOnly(next patrol.EventFlusher) patrol.EventFlusherFunc {
    return func(ctx context.Context, evts []*patrol.Event) error {
        picked := slices.Collect(func(yield func(*patrol.Event) bool) {
            for _, e := range evts {
                if e.Type == patrol.Error && !yield(e) {
                    return
                }
            }
        })
        if len(picked) == 0 {
            return nil
        }
        return next.FlushEvents(ctx, picked)
    }
}

func main() {
    _ = errorsOnly(patrol.EventFlusherFunc(func(context.Context, []*patrol.Event) error { return nil }))
}

The new slice holds the same *patrol.Event pointers, which is fine: filtering and reordering your own slice changes nothing the other flushers can see. Writing to e.Message would.

The three optional interfaces

// source: producer.go#L40-L50
// Tester is implemented by flushers that can check their connectivity or
// credentials. Producer.Test runs every Tester.
type Tester interface {
    Test(ctx context.Context) error
}

// Closer is implemented by flushers that hold resources. Producer.Close
// calls Close on each of them after the final flush.
type Closer interface {
    Close(ctx context.Context) error
}

Plus fmt.Stringer. The name String() returns is what prefixes this flusher’s errors in the joined error the producer hands to OnError, and it is what an operator reads at 3 a.m. Without it the prefix is the Go type, *main.webhookFlusher, which works but says less.

By convention a failed Test wraps patrol.ErrTest, so a caller can tell a connectivity failure from anything else with errors.Is.

A whole one

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"

    "github.com/kataras/patrol"
)

// webhook posts each batch to one URL as JSON.
type webhook struct {
    url    string
    client *http.Client
}

func newWebhook(url string) (*webhook, error) {
    if url == "" {
        return nil, fmt.Errorf("webhook: URL is required")
    }
    return &webhook{url: url, client: &http.Client{Timeout: 10 * time.Second}}, nil
}

// The four interfaces, checked at compile time.
var (
    _ patrol.EventFlusher = (*webhook)(nil)
    _ patrol.Tester       = (*webhook)(nil)
    _ patrol.Closer       = (*webhook)(nil)
    _ fmt.Stringer        = (*webhook)(nil)
)

func (w *webhook) FlushEvents(ctx context.Context, evts []*patrol.Event) error {
    if len(evts) == 0 {
        return nil
    }
    body, err := json.Marshal(evts) // reads the batch, never writes to it
    if err != nil {
        return err
    }
    req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.url, bytes.NewReader(body))
    if err != nil {
        return err
    }
    req.Header.Set("Content-Type", "application/json")
    resp, err := w.client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 300 {
        return fmt.Errorf("unexpected status %s", resp.Status)
    }
    return nil
}

func (w *webhook) Test(ctx context.Context) error {
    req, err := http.NewRequestWithContext(ctx, http.MethodHead, w.url, nil)
    if err != nil {
        return err
    }
    resp, err := w.client.Do(req)
    if err != nil {
        return fmt.Errorf("%w: %w", patrol.ErrTest, err)
    }
    resp.Body.Close()
    return nil
}

func (w *webhook) Close(context.Context) error {
    w.client.CloseIdleConnections()
    return nil
}

func (w *webhook) String() string { return "webhook" }

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

    flusher, err := newWebhook("https://hooks.example.com/patrol")
    if err != nil {
        panic(err)
    }

    producer := patrol.NewProducer(patrol.ProducerOptions{ProjectName: "checkout-api"}, flusher)
    defer producer.Close(ctx)
}

Four things that make it behave like a built-in flusher: the constructor validates and returns an error, FlushEvents returns early on an empty batch, Test wraps patrol.ErrTest, and String gives the errors a readable prefix.

Running your own server

Three exported pieces let a program accept exactly what a patrol.Client sends, without copying anything out of the library:

Symbol	What it is for
patrol.DecodeEvents(w, r, maxBytes)	Wraps the body in http.MaxBytesReader and decodes the JSON array. A body over maxBytes comes back as an *http.MaxBytesError, which is the 413 case
patrol.WriteServerError(w, status, code, message)	Writes the {"http_error_code","message"} body a Client decodes into a patrol.ServerError
patrol.FlushAll(ctx, flushers, evts)	The fan-out itself

// source: producer.go#L166-L175
// FlushAll delivers one batch to every flusher concurrently and waits for all
// of them. A failing or slow flusher never stops the others; the errors are
// joined into one, each prefixed with its flusher's name (String() or the Go
// type), so the caller can tell which destination failed. Producer and Server
// deliver through it, and a program that runs its own server can call it with
// the flushers it built.
//
// The batch is shared by every flusher at once: sort it before the call and
// treat it as read-only inside FlushEvents.
func FlushAll(ctx context.Context, flushers []EventFlusher, evts []*Event) error {

Put together, with a smaller body cap than the library’s own and a token check of your own:

package main

import (
    "cmp"
    "context"
    "errors"
    "log"
    "net/http"
    "slices"
    "time"

    "github.com/kataras/patrol"
)

// This server meters its senders, so it refuses earlier than the library's
// own patrol.MaxRequestBody of 32 MiB.
const maxBatchBytes = 4 << 20

func main() {
    flushers := []patrol.EventFlusher{patrol.EventFlusherFunc(store)}

    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
        if r.Method != http.MethodPost {
            w.Header().Set("Allow", "POST")
            patrol.WriteServerError(w, http.StatusMethodNotAllowed, patrol.ErrorCodeMethodNotAllowed, "method not allowed")
            return
        }
        if !allowed(r) {
            w.Header().Set("WWW-Authenticate", `Bearer realm="patrol"`)
            patrol.WriteServerError(w, http.StatusUnauthorized, patrol.ErrorCodeUnauthenticated, "invalid credentials")
            return
        }

        evts, err := patrol.DecodeEvents(w, r, maxBatchBytes)
        if err != nil {
            if _, tooLarge := errors.AsType[*http.MaxBytesError](err); tooLarge {
                patrol.WriteServerError(w, http.StatusRequestEntityTooLarge, patrol.ErrorCodePayloadTooLarge, err.Error())
                return
            }
            patrol.WriteServerError(w, http.StatusBadRequest, patrol.ErrorCodeInvalidArgument, err.Error())
            return
        }
        if len(evts) == 0 {
            return
        }

        received := time.Now().Unix()
        for _, e := range evts {
            e.ReceivedAt = received
        }
        slices.SortStableFunc(evts, func(a, b *patrol.Event) int {
            return cmp.Compare(a.CapturedAt, b.CapturedAt)
        })

        // Answer now, deliver after, on a context the request cannot cancel.
        go deliver(context.WithoutCancel(r.Context()), flushers, evts)
    })

    log.Fatal(http.ListenAndServe(":8080", nil))
}

func deliver(ctx context.Context, flushers []patrol.EventFlusher, evts []*patrol.Event) {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()
    if err := patrol.FlushAll(ctx, flushers, evts); err != nil {
        log.Println("patrol:", err)
    }
}

func allowed(r *http.Request) bool { return r.Header.Get("Authorization") != "" }

func store(context.Context, []*patrol.Event) error { return nil }

Two details that are easy to miss, both of which patrol.Server does for you. The batch has to be sorted before FlushAll, because the flushers assume it; the library sorts by CapturedAt with a stable sort, which is the three lines above. And delivery has to run on a context the request cannot cancel, which is what context.WithoutCancel is there for: without it the first sender that hangs up kills a Slack post that is already in flight.