ingest

type Handler ¶

type Handler struct {
	// contains filtered or unexported fields
}

Handler wires the OTLP ingest HTTP endpoints to the Writer.

func NewHandler(w *Writer, log *slog.Logger) *Handler

func (h *Handler) Mount(mux *http.ServeMux)

func (h *Handler) WithTee(t *Tee) *Handler

type Tee ¶ added in v0.10.0

type Tee struct {
	// contains filtered or unexported fields
}

Tee mirrors incoming log Events to an io.Writer in a human-readable format so a user can `tail -f` / `less +F` them from a shell alongside the Waggle UI. It is strictly a passthrough view of the ingest stream; the Store is still authoritative, and Tee never blocks or errors out the ingest path — write failures are logged once and the rest of the batch continues.

Not used on spans or metrics. Only log records (signal_type='log') flow through here; the handler calls us from handleLogs only.

func NewTee(cfg TeeConfig) (*Tee, error)

func (t *Tee) Close() error

func (t *Tee) WriteBatch(b store.Batch)

type TeeConfig ¶ added in v0.10.0

type TeeConfig struct {
	Path     string   // file path, "-" for stdout, empty for disabled
	Services []string // allow-list of service.name; empty = all
	MinSev   int32    // SeverityNumber floor (inclusive); 0 = no floor
	Format   string   // "console" | "logfmt" | "json"
	Color    string   // "auto" (default) | "always" | "never"
}

TeeConfig controls log-mirror output. Zero value = disabled.

type Writer ¶

type Writer struct {
	// contains filtered or unexported fields
}

Writer drains Batches from a buffered channel and writes them to the Store inside a single transaction each. One Writer should exist per process.

func NewWriter(s store.Store, log *slog.Logger, cfg WriterConfig) *Writer

func (w *Writer) DroppedCount() int64

func (w *Writer) Enqueue(b store.Batch) bool

func (w *Writer) Start(ctx context.Context)

func (w *Writer) Stop(ctx context.Context) error

type WriterConfig ¶

type WriterConfig struct {
	BufferSize  int           // how many batches we buffer before the HTTP handler blocks
	FlushEvents int           // commit when accumulated events reach this
	FlushEvery  time.Duration // or when this time has elapsed since first item
}