Documentation
¶
Overview ¶
Package compact folds the funclog "raw" raw OTLP-JSON-Lines objects (ADR-0081) into columnar Parquet "compacted", time-windowed and Hive-partitioned, through the blob.Bucket port (ADR-0007). It is an always-on daemon-internal pipeline — pure-Go, no cgo, lakehouse-independent — internal substrate for the log-ingest capability, not a function-bindable provider (ADR-0082). The schema is log-specific; the timer/blob/partition skeleton is the seam F51/F52 reuse (windowing by SEAL time, not event time — see the close invariant in keys.go).
Index ¶
Constants ¶
const ( DefaultWindow = time.Hour DefaultInterval = 5 * time.Minute DefaultRetention = 720 * time.Hour // 30d )
Default window / interval / retention (overridable via Deps). Interval < Window is intentional: an open window is re-passed harmlessly (idempotently) until it closes, then compacted once.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Compactor ¶
type Compactor struct {
// contains filtered or unexported fields
}
Compactor is the daemon-internal compaction pipeline. Construct with New; Run owns the timer loop.
func New ¶
New validates deps and builds the compactor (Bucket + Clock required; Window/Interval default when zero, Retention <= 0 keeps compacted forever).
func (*Compactor) CompactOnce ¶
CompactOnce runs exactly one pass: discover raw → group into closed windows → merge into one Parquet per (ns, fn, window) → delete the consumed raw (only after the Put succeeds) → prune compacted past Retention. Deterministic and idempotent: the window's existing Parquet is read back and its rows kept, and raw it already holds is not re-added, so a re-pass rewrites the same compacted key with no loss and no duplicate. A window holding an undecodable raw object is logged and left raw; the rest of the pass goes on.
type Deps ¶
type Deps struct {
Bucket blob.Bucket
Clock clock.Clock
Logger *slog.Logger
Window time.Duration // bucket size AND closed-ness threshold; a window closes at windowStart+Window
Interval time.Duration // timer cadence between passes
Retention time.Duration // delete compacted older than this; <= 0 ⇒ keep forever
}
Deps constructs a Compactor. Bucket is the funcd-system observability bucket (the same blob the funclog sink writes). Window/Interval default when non-positive; Retention <= 0 ⇒ keep compacted forever (no pruning).
type Row ¶
type Row struct {
TimeUnixNano int64 `parquet:"ts"`
SeverityText string `parquet:"severity_text"`
SeverityNumber int32 `parquet:"severity_number"`
Body string `parquet:"body"`
Namespace string `parquet:"namespace"`
Function string `parquet:"function"`
Replica string `parquet:"replica"`
Tenant string `parquet:"tenant"`
Source string `parquet:"source"` // lifted from the funcd.source attr (console/logging/stdout/stderr)
Invocation string `parquet:"inv"` // lifted from the inv attr — the invocation id (F54 filters on it)
TraceID string `parquet:"trace_id"` // hex32, "" when absent
SpanID string `parquet:"span_id"` // hex16, "" when absent
AttrsJSON string `parquet:"attrs_json"` // remaining LogRecord attributes (minus funcd.source + inv) as JSON
}
Row is one LogRecord as a Parquet row: fixed typed columns + a single attrs_json column for the rest. The schema is STABLE (F54 reads it); new OTLP attributes land in attrs_json, never as new columns.
func DecodeJSONL ¶
DecodeJSONL decodes one raw OTLP-JSON-Lines object's bytes into Rows (one per LogRecord) — the per-line decode the compactor and the ADR-0084 reader share: split on '\n', skip blank lines, plog.JSONUnmarshaler.UnmarshalLogs each line (a bad line ⇒ fault.Invalid), appendRows. An additive export of the compactor's existing loop.
Source Files