zitadel/internal/logstore/debouncer.go

package logstore

import (
	"context"
	"sync"
	"time"

	"github.com/benbjohnson/clock"
	"github.com/zitadel/logging"
)

type bulkSink interface {
	sendBulk(ctx context.Context, bulk []LogRecord) error
}

var _ bulkSink = bulkSinkFunc(nil)

type bulkSinkFunc func(ctx context.Context, items []LogRecord) error

func (s bulkSinkFunc) sendBulk(ctx context.Context, items []LogRecord) error {
	return s(ctx, items)
}

type debouncer struct {
	// Storing context.Context in a struct is generally bad practice
	// https://go.dev/blog/context-and-structs
	// However, debouncer starts a go routine that triggers side effects itself.
	// So, there is no incoming context.Context available when these events trigger.
	// The only context we can use for the side effects is the app context.
	// Because this can be cancelled by os signals, it's the better solution than creating new background contexts.
	binarySignaledCtx context.Context
	clock             clock.Clock
	ticker            *clock.Ticker
	mux               sync.Mutex
	cfg               DebouncerConfig
	storage           bulkSink
	cache             []LogRecord
	cacheLen          uint
}

type DebouncerConfig struct {
	MinFrequency time.Duration
	MaxBulkSize  uint
}

func newDebouncer(binarySignaledCtx context.Context, cfg DebouncerConfig, clock clock.Clock, ship bulkSink) *debouncer {
	a := &debouncer{
		binarySignaledCtx: binarySignaledCtx,
		clock:             clock,
		cfg:               cfg,
		storage:           ship,
	}

	if cfg.MinFrequency > 0 {
		a.ticker = clock.Ticker(cfg.MinFrequency)
		go a.shipOnTicks()
	}
	return a
}

func (d *debouncer) add(item LogRecord) {
	d.mux.Lock()
	defer d.mux.Unlock()
	d.cache = append(d.cache, item)
	d.cacheLen++
	if d.cfg.MaxBulkSize > 0 && d.cacheLen >= d.cfg.MaxBulkSize {
		// Add should not block and release the lock
		go d.ship()
	}
}

func (d *debouncer) ship() {
	if d.cacheLen == 0 {
		return
	}
	d.mux.Lock()
	defer d.mux.Unlock()
	if err := d.storage.sendBulk(d.binarySignaledCtx, d.cache); err != nil {
		logging.WithError(err).WithField("size", len(d.cache)).Error("storing bulk failed")
	}
	d.cache = nil
	d.cacheLen = 0
	if d.cfg.MinFrequency > 0 {
		d.ticker.Reset(d.cfg.MinFrequency)
	}
}

func (d *debouncer) shipOnTicks() {
	for range d.ticker.C {
		d.ship()
	}
}