Documentation
¶
Overview ¶
Package logs ships task logs from the agent to durable storage and serves them back via the API, so logs remain available after the task pod is gone.
Index ¶
- Constants
- Variables
- func EncodeLine(ev Event) string
- func MarkReplay(line string) string
- func ReadAttempts(sink Sink, ref Ref, set TryEpochs) (io.ReadCloser, error)
- func RefineLevel(line, fallback string) string
- func SplitReplay(line string) (string, bool)
- type DiskSink
- type EpochLister
- type Event
- type GCSConfig
- type GCSStore
- type LineWriter
- type LogWriter
- type MarkerSink
- type MemoryTailer
- type ObjectLister
- type ObjectOption
- type ObjectSink
- type ObjectStore
- type Pruner
- type RedisTailer
- type Ref
- type S3Config
- type S3Store
- type Sink
- type Tailer
- type TryEpochs
Constants ¶
const ( ObjectLayoutSingle = "single" ObjectLayoutSegmented = "segmented" )
Object layouts. ObjectLayoutSingle (the default) keeps each attempt in one object at {try}.log, rewritten on every flush. ObjectLayoutSegmented writes the attempt as numbered segments under {try}.log.d/ ({try}.e{epoch}.log.d/ for a later execution of the try), sealing a segment once it reaches objectSegmentBytes, so a flush uploads the open segment instead of the whole attempt and the writer holds one segment in memory instead of the whole attempt. A server older than the segmented layout reads only {try}.log, which is why it is opt-in; every sink reads both layouts regardless of its own.
Variables ¶
var ErrObjectNotFound = errors.New("log object not found")
ErrObjectNotFound reports that an object-store sink holds no stored log for a Ref. Backends translate their own not-found signal (S3 NoSuchKey, GCS ErrObjectNotExist, a missing in-memory entry) into this sentinel so the read path maps it uniformly.
var ErrUnsafeRef = errors.New("unsafe log reference")
ErrUnsafeRef reports a Ref whose fields cannot be used as path segments.
Functions ¶
func EncodeLine ¶
EncodeLine serializes an Event to the same JSON line used for storage, so the live-tail channel carries the full event (level + stream + timestamp), not just the message text. The result is always one JSON object: a zero time is stamped with the current time, and so is a time JSON cannot encode (a year outside 0 to 9999, which time.MarshalJSON refuses). It used to hand the raw message back on that marshal error, and a raw message holding a newline is stored as two lines, so whoever controls the message text and the timestamp (the agent of an attempt, through its own token) could forge a stored log entry that way.
func MarkReplay ¶ added in v0.5.1
MarkReplay flags an encoded line as a replay for the live tail. A line that is not a JSON object (a legacy plain line) is returned unchanged.
func ReadAttempts ¶ added in v0.5.1
ReadAttempts reads the log of one try as every execution of it stored, oldest attempt epoch first (ADR 0051 amendment, #863). The Airflow-compatible log endpoint addresses a try, while an infra re-place, a reschedule poke or a repeated dispatch runs the same try again under a new epoch, each writing its own stream. ref.AttemptEpoch is ignored.
A sink that implements EpochLister names the stored executions in one call; any other sink, or a failed listing, has every candidate epoch probed. A candidate with no stored log is skipped (a claimed epoch whose pod never ran wrote nothing). A try with exactly one stored stream is returned unchanged, so a try that ran once, and every log written before the epoch existed, reads byte for byte as before. Several streams are concatenated, each preceded by one system line naming the execution, and opened one at a time as the read reaches them (without a listing, one stream ahead is open to tell a lone stream from several). When set.Max leaves older executions out, a first system line says which. When no candidate has a stored stream the sink's own absence error is returned, so the API still answers 404. A failure to open the first streams fails the call; a later one ends the body with that error, so a log is never served with a silent hole.
func RefineLevel ¶
RefineLevel returns the severity indicated by a log line's own text when it carries a clear level token, otherwise the supplied fallback. It lets the line content correct a level that was derived only from the source stream — so an "ERROR …" line printed to stdout is colored error (not info), and an "INFO …" line written to stderr is colored info (not error). When the text gives no signal, the stream-derived fallback stands, so behavior is unchanged for plain output. The returned values match the Event.Level vocabulary the UI colors by (error/warning/info/debug); callers control the contract, this only sharpens the level value.
func SplitReplay ¶ added in v0.5.1
SplitReplay reports whether a published line was flagged by MarkReplay and returns it with the flag removed, which is its stored encoding.
Types ¶
type DiskSink ¶
type DiskSink struct {
// contains filtered or unexported fields
}
DiskSink writes logs to ${root}/{tenant}/{dag}/{run}/{task}/{try}.log, or {try}.e{epoch}.log for an execution with a non-zero attempt epoch.
func NewDiskSink ¶
NewDiskSink builds a DiskSink rooted at dir.
func (*DiskSink) AppendEvent ¶
AppendEvent adds one event to the attempt's log file. The disk sink already opens with O_APPEND, so this is a plain open-write-close that preserves the streamed content — the marker lands after whatever the agent wrote (#861).
func (*DiskSink) Open ¶
Open creates the log file (appending if it exists) and returns a buffered writer.
func (*DiskSink) Prune ¶
Prune deletes log files whose last modification is older than retention, reclaiming space for completed runs. A missing root is not an error.
type EpochLister ¶ added in v0.5.1
EpochLister is implemented by sinks that can name the stored executions of one try in a single call (a directory read, or one object listing), so a try log read fetches only the streams that exist instead of probing every epoch the try can span. StoredEpochs ignores ref.AttemptEpoch. An error means the listing is unavailable, and the read falls back to probing.
type Event ¶
type Event struct {
Time time.Time `json:"ts"`
Level string `json:"level"`
Stream string `json:"stream"`
Message string `json:"msg"`
}
Event is one structured log line: when it was emitted, its severity level and source stream (stdout/stderr), and the message text. Logs are stored as JSONL (one Event per line) so the UI's drill-down viewer can color by real level instead of guessing.
func DecodeLine ¶
DecodeLine parses a stored log line into an Event. Lines that are not JSON (legacy plain-text logs written before structured storage) decode as a stdout message with a level inferred from the text, so the reader serves both formats with sensible coloring.
type GCSConfig ¶
type GCSConfig struct {
// Bucket is the target bucket. Required.
Bucket string
// CredentialsFile is a path to a service-account JSON key. Empty (recommended)
// uses Application Default Credentials — GKE Workload Identity keyless.
CredentialsFile string
}
GCSConfig configures the Google Cloud Storage object store backing an ObjectSink. Auth is keyless-first (ADR 0035): leave CredentialsFile empty to use Application Default Credentials, which resolves GKE Workload Identity (or the metadata server on a GCE node) with no key material on disk. A credentials file is a discouraged escape hatch for local development and clusters without an identity broker.
Unlike the S3 backend, GCS needs no region, endpoint, or path-style toggle — the native SDK addresses the global storage.googleapis.com API and locates the bucket by name. This is why GCS is a first-class provider here rather than an S3-interop endpoint: the interop endpoint requires HMAC keys and cannot use Workload Identity, so routing GCS through the S3 client would forfeit keyless.
type GCSStore ¶
type GCSStore struct {
// contains filtered or unexported fields
}
GCSStore is an ObjectStore backed by Google Cloud Storage.
func NewGCSStore ¶
NewGCSStore builds a GCSStore from cfg. With no CredentialsFile it constructs the client on the keyless ADC chain; with one set it detects credentials from that file via the modern auth library (not the deprecated WithCredentialsFile option). Client construction may touch the credential chain but makes no bucket call; a missing bucket or denied permission surfaces on the first Put/Get.
func (*GCSStore) Get ¶
Get reads the object under key, translating GCS's ErrObjectNotExist into ErrObjectNotFound so the read path maps a missing log to a 404, matching the S3 backend.
func (*GCSStore) List ¶ added in v0.5.1
List returns the keys under prefix, and with a delimiter each deeper key's common prefix once (see ObjectLister). It needs storage.objects.list.
type LineWriter ¶ added in v0.5.1
LineWriter is implemented by LogWriters that also accept a line already encoded by EncodeLine (without the trailing newline), so a caller that both stores and publishes a line encodes it once. DiskSink and ObjectSink writers implement it; a caller type-asserts and falls back to WriteEvent.
type MarkerSink ¶
MarkerSink appends a single event to an attempt's existing log while PRESERVING prior content: O_APPEND on disk, read-modify-write on an object store (which has no native append, so Open+Close there would overwrite the whole object). A reaper uses it to add a terminal marker to a killed task's log without clobbering the streamed output (#861). DiskSink and ObjectSink both implement it; a caller holding a Sink type-asserts to reach it.
type MemoryTailer ¶
type MemoryTailer struct {
// contains filtered or unexported fields
}
MemoryTailer is an in-process Tailer for Dexaflow Lite: it fans task log lines to live subscribers over Go channels, with no Redis. It is valid only within a single process (Lite runs the agent gRPC and the read API together); a multi-replica deployment must use RedisTailer instead (ADR 0026).
func NewMemoryTailer ¶
func NewMemoryTailer() *MemoryTailer
NewMemoryTailer builds an empty in-process tailer.
func (*MemoryTailer) HasSubscribers ¶ added in v0.5.1
HasSubscribers reports whether the attempt has a live subscriber.
type ObjectLister ¶ added in v0.5.1
type ObjectLister interface {
List(ctx context.Context, prefix, delimiter string) ([]string, error)
}
ObjectLister is implemented by object stores that can list keys under a prefix. With a delimiter, a key that continues past the delimiter is returned once as its common prefix, ending in the delimiter, so a segmented attempt lists as one entry. S3Store and GCSStore implement it.
type ObjectOption ¶ added in v0.5.1
type ObjectOption func(*ObjectSink)
ObjectOption configures an ObjectSink.
func WithObjectLayout ¶ added in v0.5.1
func WithObjectLayout(layout string) ObjectOption
WithObjectLayout selects the layout new attempts are written in. Empty means ObjectLayoutSingle; an unknown value is rejected by NewDurableSink.
type ObjectSink ¶
type ObjectSink struct {
// contains filtered or unexported fields
}
ObjectSink stores each task attempt as a single object in an ObjectStore, keyed the same way the disk sink lays out files ({prefix}/{tenant}/{dag}/{run}/{task}/{try}.log, or {try}.e{epoch}.log). Object stores have no append, so a writer accumulates the attempt's events and rewrites the object incrementally (by size and on a time cadence) with a final rewrite on Close, so a control plane killed mid-attempt leaves a partial object rather than nothing; the live-tail path (a separate Tailer) still streams lines as they arrive. It is opt-in — the disk sink stays the default (see NewDurableSink).
func NewObjectSink ¶
func NewObjectSink(ctx context.Context, store ObjectStore, prefix string, logger *slog.Logger, opts ...ObjectOption) *ObjectSink
NewObjectSink builds an ObjectSink writing to store under an optional key prefix. ctx bounds the store operations issued by the writers and readers the sink hands out (the Sink interface is context-free by design); pass the server's lifecycle context.
logger receives the warnings the write path cannot return — a failed incremental flush is retried, not surfaced to the agent, so the log line is the ONLY evidence it happened. It is injected rather than taken from slog.Default() so the sink honors the handler (format, level, destination) its owner configured without depending on that owner having reassigned a process-wide global — leoflow-server does call slog.SetDefault, which is what covers the call sites nothing injects into, but an embedder need not, and a library reaching for its caller's global is the wrong seam either way. nil falls back to slog.Default(), deliberately: making the parameter mandatory would not pin the production hand-off, it would just move the trap from "warnings go somewhere unwatched" to "nil dereference on a flush path that only runs while the process is already shutting down". What pins the hand-off is a test that reaches this constructor the way the server does, via NewDurableSink (#918).
func (*ObjectSink) AppendEvent ¶
func (o *ObjectSink) AppendEvent(ref Ref, ev Event) error
AppendEvent adds one event to the attempt's stored object WITHOUT clobbering it. Object stores have no append, so a plain Open+Close would Put a marker-only object over the agent's streamed log; instead this reads the existing object (tolerating a not-yet-written one), appends the event as a JSONL line, and Puts the combined object back (#861).
Read-modify-write with no locking, which is safe because of WHEN it runs, not because nothing else can write the key: a live attempt's writer does flush the same object incrementally. The only caller is the agent-lost reaper, and what separates the two writers is the agent-lost threshold — 90s of heartbeat silence (executor.defaultAgentLostThreshold), not "minutes" and not the life of the pod. Normally that is enough, because an agent that has stopped heartbeating for 90s has stopped streaming too.
The exception, and it IS reachable: heartbeats can be refused while the log stream survives. An attempt past max_attempt_credential_lifetime stops getting its token renewed, so its heartbeats start failing authentication, while StreamLogs — authenticated once at Open and never re-checked — keeps receiving lines and flushing them. Agent-lost then fires at the 90s mark, this function appends the marker, and the live writer's very next flush Puts its own buffer straight over it. Each side Puts the whole object it last read, so the loser loses whole flushes, not one line. Anything that shortens the gap between the two writers — a lower threshold, a caller that is not the reaper — widens this from an exception into the normal case (#918). In the segmented layout the live writer never writes {try}.log, so the marker there cannot be overwritten; Read serves it after the segments.
func (*ObjectSink) Open ¶
func (o *ObjectSink) Open(ref Ref) (LogWriter, error)
Open validates the ref and returns a writer that keeps the attempt's object current as lines arrive (see objectWriter) and performs the last flush on Close. In the segmented layout the writer starts after any segment already stored for the attempt, so a second stream opened once the previous one has closed appends instead of overwriting. Two writers live at the same time (a reconnect before the old stream ended) can still overwrite each other's segments, as they overwrite each other's object in the single layout.
When the probe for the stored segments fails, even after its retries, Open refuses the stream instead of guessing where the attempt ends. Starting at segment zero, as it once did, overwrote the first stream of an attempt on every reconnect whenever the store answered the probe with something other than not-found (S3 answers AccessDenied, not NoSuchKey, to a caller without s3:ListBucket; a 503 does the same for a moment), and writing the lines to the single object instead is no safer: the single layout overwrites too, and the Get that would seed an append fails exactly like the probe did, so the writer cannot tell an absent object from a denied one. The lines of a refused stream are not shipped, the agent is told which step failed (StreamLogs) and keeps running, and every line already stored for the attempt stays where it is. The store is checked for this answer at boot (checkSegmentedLayout), so in practice only a failure that outlasts the retries gets here.
func (*ObjectSink) Read ¶
func (o *ObjectSink) Read(ref Ref) (io.ReadCloser, error)
Read returns the attempt's stored log in either layout: its segments in order followed by the single object, which holds the whole log of an attempt written in the single layout and only the reaper's markers of a segmented one. A missing log surfaces as ErrObjectNotFound.
Each layout looks for its own shape first. The single layout GETs {try}.log exactly as it did before segments existed and probes segment zero only when that object is missing, so a default deployment pays no extra round trip and a store that answers a missing key with something other than not-found (S3 returns 403 AccessDenied to a caller without s3:ListBucket) cannot fail its reads. The price: a segmented attempt that also holds a reaper marker reads back as the marker alone once the layout is switched back to single. The segmented layout probes segment zero first, retrying a failure other than not-found once, and falls back to {try}.log when the probe still fails; when that object is missing as well, the probe's failure is what the caller gets, not an absent log, so a store outage does not read as a task that logged nothing.
func (*ObjectSink) StoredEpochs ¶ added in v0.5.1
func (o *ObjectSink) StoredEpochs(ref Ref) ([]int, error)
StoredEpochs lists the attempt epochs with a stored log for ref's try, in either layout, with one listing of the {try}. prefix (see EpochLister). A store without ObjectLister, or a failed listing (S3 without s3:ListBucket), returns an error and the caller probes instead.
type ObjectStore ¶
type ObjectStore interface {
Put(ctx context.Context, key string, r io.Reader) error
Get(ctx context.Context, key string) (io.ReadCloser, error)
}
ObjectStore is the minimal object-store API the ObjectSink needs: write a whole object under a key, and read one back. It is deliberately tiny so the sink is testable against an in-memory fake, and so any backend can satisfy it with its own native SDK — the S3 backend (AWS S3, MinIO, Ceph RGW via aws-sdk-go-v2) and the GCS backend (Google Cloud Storage via its native SDK). Each provider keeps its own keyless auth path this way. Implementations target a single, pre-configured bucket.
type Pruner ¶
Pruner is implemented by sinks that reclaim their own expired storage: the disk sink deletes old files. An object-store sink deliberately does not implement it — bucket lifecycle policy owns retention there — so the janitor skips pruning for sinks that manage their own lifecycle.
type RedisTailer ¶
type RedisTailer struct {
// contains filtered or unexported fields
}
RedisTailer fans task log lines out over Redis pub/sub so the API can stream them live to the UI without polling.
func NewRedisTailer ¶
func NewRedisTailer(client *redis.Client) *RedisTailer
NewRedisTailer builds a RedisTailer over the given go-redis client.
func (*RedisTailer) HasSubscribers ¶ added in v0.5.1
HasSubscribers reports whether any client, on any replica, may be listening on the attempt's tail channel, so the writer can skip publishing lines nobody would receive. NUMSUB counts the channel's own subscribers; NUMPAT counts pattern subscriptions, which NUMSUB cannot attribute to a channel, so any pattern subscriber (say, a forwarder on log_tail:*) keeps every attempt publishing. Both go in one round trip.
func (*RedisTailer) Subscribe ¶
func (t *RedisTailer) Subscribe(ctx context.Context, ref Ref) (lines <-chan string, cancel func())
Subscribe returns a channel of live log lines for the task and a cancel function that ends the subscription. The line channel closes when the subscription is canceled or the context is done.
type Ref ¶
type Ref struct {
TenantID string
DagID string
RunID string
TaskID string
TryNumber int
AttemptEpoch int
}
Ref identifies a task instance's log stream and maps to its storage location.
AttemptEpoch names one execution of the try (ADR 0051 amendment): an infra re-place, a reschedule poke or a repeated dispatch runs the same try again, and each execution keeps its own stream. Epoch 0 maps to the key every log had before the epoch existed, so those logs are still found.
type S3Config ¶
type S3Config struct {
// Bucket is the target bucket. Required.
Bucket string
// Region is the store region. Required by AWS S3; ignored by some stores.
Region string
// Endpoint overrides the S3 endpoint (MinIO, Ceph RGW). Empty = AWS default.
Endpoint string
// ForcePathStyle uses path-style addressing (needed by MinIO and some stores).
ForcePathStyle bool
// AccessKeyID / SecretAccessKey are static credentials. Empty = keyless.
AccessKeyID string
SecretAccessKey string
}
S3Config configures the S3-compatible object store backing an ObjectSink. It targets AWS S3 and any S3-compatible store (MinIO, Ceph RGW). Google Cloud Storage is NOT reached through here — it has its own native backend (GCSStore) because the S3 interop endpoint requires HMAC keys and cannot use Workload Identity, which would forfeit keyless auth. Keyless is the default (ADR 0035): leave AccessKeyID/SecretAccessKey empty to use the ambient credential chain (IRSA / instance profile / env). Static keys are a discouraged escape hatch for dev and stores without an identity broker.
type S3Store ¶
type S3Store struct {
// contains filtered or unexported fields
}
S3Store is an ObjectStore backed by an S3-compatible API.
func NewS3Store ¶
NewS3Store builds an S3Store from cfg. It loads the ambient AWS configuration (region + keyless credential chain) and, when set, overrides the endpoint (for MinIO / Ceph RGW), forces path-style addressing, and installs static credentials. No network call is made here; credential resolution and any failure surface on the first Put/Get.
func (*S3Store) Get ¶
Get reads the object under key, translating S3's NoSuchKey into ErrObjectNotFound so the read path maps a missing log to a 404.
func (*S3Store) List ¶ added in v0.5.1
List returns the keys under prefix, and with a delimiter each deeper key's common prefix once (see ObjectLister). It needs s3:ListBucket.
func (*S3Store) Put ¶
Put writes the object under key. Request signing needs a length-known, seekable payload; the object sink already hands a *bytes.Reader, so the common path passes it straight through with no extra copy. A non-seekable reader is materialized once as a fallback so the interface stays honest for other callers.
type Sink ¶
Sink stores and retrieves task logs.
func NewDurableSink ¶
func NewDurableSink(ctx context.Context, backend, dir string, store ObjectStore, prefix string, logger *slog.Logger, opts ...ObjectOption) (Sink, error)
NewDurableSink selects the durable log sink from configuration. The default — an empty or "disk" backend — returns a DiskSink rooted at dir, so Lite and every deployment that does not opt in keep the exact on-disk behavior. The "s3" and "gcs" backends return an ObjectSink over store (which the caller builds with the matching native SDK from the configured bucket/credentials) and require a non-nil store. An unknown backend is rejected rather than silently falling back. logger is the process's configured logger, carried to the object sink so its retry warnings honor that contract (see NewObjectSink); the disk sink ignores it, and so it does opts. An unknown object layout is rejected like an unknown backend, and so is the segmented layout on a store that does not answer a missing key with not-found (checkSegmentedLayout), so that misconfiguration fails the boot with the permission to grant instead of failing every attempt's stream.
type Tailer ¶
type Tailer interface {
Publish(ctx context.Context, ref Ref, line string) error
Subscribe(ctx context.Context, ref Ref) (<-chan string, func())
}
Tailer publishes and subscribes to a task's live log lines for the UI's live tail. RedisTailer is the production implementation.
type TryEpochs ¶ added in v0.5.1
type TryEpochs struct {
Low, High, Max int
}
TryEpochs names the executions one try log read considers: epoch 0, where every log written before the epoch existed lives, and every epoch in (Low, High]. Max bounds how many non-zero epochs are served, keeping the most recent; 0 means no bound. A reschedule-mode sensor starts a new execution on every poke, so a long-running try can span many epochs.
Source Files