Documentation
¶
Overview ¶
Package funclog captures function telemetry over a freeze-safe side channel and persists it OTel-native through the blob.Bucket port (ADR-0081). THIS package builds LOGS (Path A/B). The PIPELINE — Reader → batch-per-Resource → Sink → blob.Bucket Put, plus the Resource/identity tagging — is signal-generic: traces (F51) and metrics (F52) reuse this seam. The per-signal RECORD type (Entry) and the NDJSON wire are log-SPECIFIC; F51/F52 add their own record + builder behind the same Reader/Sink (a Signal-discriminated set, not a reused Entry).
The freeze-safe property: nothing buffers durably inside the function — the host reads each instance's channel continuously, so a frozen/killed instance pauses the host Pump (a blocking Read) rather than losing an already-emitted backlog.
Index ¶
- func Pump(ctx context.Context, r Reader, s Sink, res Resource, log *slog.Logger) error
- func Route(ctx context.Context, r io.Reader, sinks Sinks, res Resource, log *slog.Logger) error
- func RoutePool(ctx context.Context, r io.Reader, sinks Sinks, pool Resource, ...) error
- type BlobSink
- type BlobTraceSink
- type Deps
- type Entry
- type Reader
- type Resource
- type Severity
- type Signal
- type Sink
- type Sinks
- type Source
- type Span
- type SpanKind
- type SpanStatus
- type TraceSink
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Pump ¶
Pump drains one instance's Reader into the Sink until the channel closes (io.EOF) or fails, then seals the remaining segment. It is loss-safe: a frozen/suspended instance stops sending, so Read blocks and the pump pauses — no already-emitted record is dropped. A single bad record (fault.Invalid) is skipped (logged), never killing the stream; ctx cancellation stops the pump.
func Route ¶
Route drains one instance's telemetry channel, decoding each NDJSON line and dispatching by its "funcd.signal" tag: "traces" → Sinks.Traces.AppendSpan(Span); absent/other → Sinks.Logs.Append(Entry). It replaces the single-Sink Pump at the composition-root wiring; Pump/NewNDJSONReader remain for the logs-only contract tests. Loss-safe: a frozen instance stops sending, so the blocking Scan pauses the drain — no already-emitted record is dropped. A single malformed line is logged and skipped, never killing the stream; on EOF (channel closed) or ctx cancellation both sinks' remaining segments are sealed. Every record is stored under res: a solo worker's "funcd.member" key is ignored.
func RoutePool ¶ added in v0.5.0
func RoutePool(ctx context.Context, r io.Reader, sinks Sinks, pool Resource, isMember func(name string) bool, log *slog.Logger) error
RoutePool drains a pool worker's channel like Route, but stores each record under the pooled Function its "funcd.member" key names: pool's Namespace, Replica and Tenant with Function set to the member, when isMember(member) holds. A record naming no member, or one outside the set, is dropped and counted, with a warning at most once a minute; nothing is stored under the pool worker's own name. On EOF or ctx cancellation it seals the segments of every member it stored records for.
Types ¶
type BlobSink ¶
type BlobSink struct {
// contains filtered or unexported fields
}
BlobSink is the blob-backed Sink (the only implementation; blob.Bucket's drivers provide memory/file/s3). Segments seal at SegmentMaxBytes or SegmentMaxAge.
func NewBlobSink ¶
NewBlobSink builds the blob-backed sink over the funcd-system observability bucket. Bucket and Clock are required; segment caps default when zero.
func (*BlobSink) Append ¶
Append adds e to res's open segment, sealing+Putting it if it crosses the size/age cap.
type BlobTraceSink ¶
type BlobTraceSink struct {
// contains filtered or unexported fields
}
BlobTraceSink is the blob-backed TraceSink (the only implementation; blob.Bucket's drivers provide memory/file/s3). Segments seal at SegmentMaxBytes or SegmentMaxAge. It reuses funclog.Deps.
func NewBlobTraceSink ¶
func NewBlobTraceSink(d Deps) (*BlobTraceSink, error)
NewBlobTraceSink builds the blob-backed trace sink over the funcd-system observability bucket. Bucket and Clock are required; segment caps default when zero. Keys are partitioned as traces/<ns>/<fn>/<date>/<unixnano>-<replica>.otlp.jsonl.
func (*BlobTraceSink) AppendSpan ¶
AppendSpan adds sp to res's open segment, sealing+Putting it if it crosses the size/age cap.
func (*BlobTraceSink) Close ¶
func (s *BlobTraceSink) Close() error
Close stops the age-flusher, waits for any Flush still Putting a segment, then seals and Puts every open segment (the loss-safe shutdown boundary).
type Deps ¶
type Deps struct {
Bucket blob.Bucket
SegmentMaxBytes int // seal+Put at this many buffered bytes (default 8 MiB)
SegmentMaxAge time.Duration // ...or after this long (default 10s)
Clock clock.Clock
Logger *slog.Logger
}
Deps constructs a BlobSink. Bucket is the funcd-system observability bucket (blob.Bucket).
type Entry ¶
type Entry struct {
Time time.Time // real event time (epoch nanos on the wire)
Severity Severity // exact for Path B; coarse for Path A
Body string // the message (Path B keeps console args before util.format)
Attrs map[string]string // structured attributes
Invocation string // per-invocation id (may be empty for pre-invocation Path A lines)
TraceID string // hex32, may be empty
SpanID string // hex16, may be empty
Source Source
}
Entry is one decoded LOG record from a function instance, pre-OTLP (log-specific: Body + Severity). Traces and metrics (F51/F52) define sibling record types behind the same Reader/Sink seam — Entry is not reused.
type Reader ¶
type Reader interface {
// Read returns the next Entry, or io.EOF when the channel closes (instance gone).
Read(ctx context.Context) (Entry, error)
}
Reader decodes one instance's telemetry channel into Entries. NDJSON (Path B) or raw fd lines (Path A). A frozen instance stops sending, so Read blocks — which pauses the Pump and prevents in-function backlog loss.
func NewNDJSONReader ¶
NewNDJSONReader decodes Path B (one JSON record per line) from the fd3/UDS channel.
type Resource ¶
Resource is the OTLP Resource identity stamped on every record — the F54 query-time scoping key. Tenant = Namespace today; explicit for later multi-tenant scoping.
type Severity ¶
type Severity string
Severity is the typed OTel severity (ADR-0002: typed over magic strings).
type Signal ¶
type Signal string
Signal is the OTel signal kind, discriminating the record/wire a Reader+Sink carry. Logs now; F51/F52 add SignalTraces/SignalMetrics (each with its own record type — Entry does NOT generalize to spans/data-points).
const SignalLogs Signal = "logs"
SignalLogs is the logs signal (this package).
const SignalTraces Signal = "traces"
SignalTraces is the traces signal (ADR-0101). It joins SignalLogs on the wire's "funcd.signal" discriminator: a line tagged "traces" decodes to a Span, anything else (including an absent tag) to an Entry. The transport, per-instance drain, Resource identity, and segment-seal machinery are reused from ADR-0081; only the record type (Span), the marshaler (ptrace), and the blob prefix (traces/) differ.
type Sink ¶
type Sink interface {
// Append adds one Entry to the open segment for res.
Append(ctx context.Context, res Resource, e Entry) error
// Flush seals res's current segment (also triggered by size/age) and Puts it; returns the blob key.
Flush(ctx context.Context, res Resource) (key string, err error)
io.Closer
}
Sink accumulates per-Resource segments, marshals each sealed segment to OTLP-JSON-Lines, and persists it as ONE blob.Bucket object (whose own drivers give memory/file/s3). The seam is signal-generic (this logs Sink takes Entry; F51/F52 add sibling Sinks for their record types). CONCURRENCY-SAFE: per-Resource segments are independently locked, so concurrent Append/Flush from the many per-instance Pumps is supported.
type Sinks ¶
Sinks bundles the per-signal sinks the demux routes into. A nil sink drops that signal (e.g. traces disabled ⇒ Traces == nil ⇒ span lines are read off the channel and discarded).
type Source ¶
type Source string
Source tags which capture path produced an Entry. The NDJSON wire key "funcd.source" decodes to this typed Source.
type Span ¶
type Span struct {
TraceID string // hex32; adopted from traceparent or minted (root)
SpanID string // hex16; minted per invocation
ParentID string // hex16; the traceparent span-id, empty for a root
Name string // the function name (FUNCD_FUNCTION) or "invoke"
Kind SpanKind // SERVER for the auto invocation span
Start time.Time // invocation start (epoch nanos on the wire)
End time.Time // invocation end
Status SpanStatus // OK on return, ERROR on throw/output-mismatch
StatusMsg string // the error message when Status == ERROR
Attrs map[string]string // structured attributes
Invocation string // the per-invocation id (== the log lines' inv for correlation)
Links []string // hex16 span-ids linked in the SAME trace — fan-in edges (ADR-0105)
}
Span is one decoded TRACE record from a function instance, pre-OTLP (trace-specific: start/end/ parent/kind/status, NOT body/severity). The sibling to Entry behind the same Reader/Sink seam — neither reuses the other (ADR-0081's seam: "Entry does NOT generalize to spans").
type SpanKind ¶
type SpanKind string
SpanKind is the typed OTel span kind carried on the wire (ADR-0002: typed over magic strings).
type SpanStatus ¶
type SpanStatus string
SpanStatus is the typed OTel span status.
const ( StatusUnset SpanStatus = "UNSET" StatusOk SpanStatus = "OK" StatusError SpanStatus = "ERROR" )
type TraceSink ¶
type TraceSink interface {
// AppendSpan adds one Span to the open segment for res.
AppendSpan(ctx context.Context, res Resource, s Span) error
// Flush seals res's current segment (also triggered by size/age) and Puts it; returns the blob key.
Flush(ctx context.Context, res Resource) (key string, err error)
// Close stops the age-flusher, then seals and Puts every open segment.
Close() error
}
TraceSink is the traces analog of Sink (ADR-0101): it accumulates per-Resource span segments, marshals each sealed segment to OTLP-trace-JSON-Lines (ptrace), and persists it as ONE blob.Bucket object under a traces/ prefix. It mirrors BlobSink's seal machinery rather than touching the Implemented logs sink. CONCURRENCY-SAFE: per-Resource segments are independently locked, so concurrent AppendSpan/Flush from the many per-instance drains is supported.
Directories
¶
| Path | Synopsis |
|---|---|
|
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).
|
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). |
|
Package logread is the thin pure-Go function-log reader (ADR-0084): it merges the compacted Parquet (ADR-0083) with the recent raw OTLP-JSONL tail (ADR-0081) over the blob.Bucket port, filters by time/severity/limit, and returns a tenant's function logs time-ordered.
|
Package logread is the thin pure-Go function-log reader (ADR-0084): it merges the compacted Parquet (ADR-0083) with the recent raw OTLP-JSONL tail (ADR-0081) over the blob.Bucket port, filters by time/severity/limit, and returns a tenant's function logs time-ordered. |