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 ¶
- Variables
- func EncodeLine(ev Event) string
- func RefineLevel(line, fallback string) string
- type DiskSink
- type Event
- type GCSConfig
- type GCSStore
- type LogWriter
- type MarkerSink
- type MemoryTailer
- type ObjectSink
- type ObjectStore
- type Pruner
- type RedisTailer
- type Ref
- type S3Config
- type S3Store
- type Sink
- type Tailer
Constants ¶
This section is empty.
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. On the unlikely marshal error it falls back to the raw message, so a line is never dropped.
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.
Types ¶
type DiskSink ¶
type DiskSink struct {
// contains filtered or unexported fields
}
DiskSink writes logs to ${root}/{tenant}/{dag}/{run}/{task}/{try}.log.
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.
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.
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.
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). 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) *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).
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.
func (*ObjectSink) Read ¶
func (o *ObjectSink) Read(ref Ref) (io.ReadCloser, error)
Read fetches the stored object for the ref. A missing object surfaces as ErrObjectNotFound.
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) 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 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) 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) (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.
Source Files