Documentation
¶
Overview ¶
Package logs owns the write side of a task's event history: it turns a node's event batch into rows, applies the status transitions those events imply, and fans the result out to live subscribers. Everything a client ever sees about a running task comes through here.
Index ¶
- Constants
- func KindString(k podiumv1.TaskEventKind) string
- type Retries
- type RollupConfig
- type Service
- func (s *Service) Ingest(ctx context.Context, taskID string, batch []*podiumv1.TaskEvent) (uint64, error)
- func (s *Service) Note(ctx context.Context, taskID, message string) error
- func (s *Service) PruneOnce(ctx context.Context) (int64, error)
- func (s *Service) RollUpOnce(ctx context.Context) (int, error)
- func (s *Service) Run(ctx context.Context) error
- func (s *Service) RunRollUp(ctx context.Context) error
- func (s *Service) SetArchive(a *artifacts.Service, cfg RollupConfig)
- func (s *Service) SetRetries(r Retries)
- func (s *Service) SetSlots(sl Slots)
- func (s *Service) Subscribe(ctx context.Context, taskID string, fromSeq uint64) (<-chan *podiumv1.TaskEvent, error)
- type Slots
Constants ¶
const ( KindProvisioning = "provisioning" KindPulling = "pulling" KindStarted = "started" KindLog = "log" KindStep = "step" KindArtifact = "artifact" KindExited = "exited" KindFinished = "finished" KindError = "error" KindMessage = "message" )
The canonical task_events.kind strings, matching podium.v1.TaskEventKind one for one.
const ( RollupIntervalEnv = "PODIUM_LOG_ROLLUP_INTERVAL" RollupSettleEnv = "PODIUM_LOG_ROLLUP_SETTLE" PruneIntervalEnv = "PODIUM_LOG_PRUNE_INTERVAL" ChunkGraceEnv = "PODIUM_LOG_CHUNK_GRACE" )
Environment overrides for the roll-up schedule. They exist so the prune horizon can be driven to a second in a test without waiting a day, and so an operator with a small database can shorten it.
const FailureReasonOOM = "oom"
FailureReasonOOM is the tasks.failure_reason a task gets when its container was killed for exceeding its memory limit.
Variables ¶
This section is empty.
Functions ¶
func KindString ¶
func KindString(k podiumv1.TaskEventKind) string
KindString is the stored spelling of a wire event kind.
Types ¶
type Retries ¶
type Retries interface {
// Retry takes the task off the node it was on: queued again if max_attempts allows,
// terminal with reason if it does not. Either way the task stops being the node's.
Retry(ctx context.Context, taskID, reason string)
}
Retries hands a task back to whatever owns placement, after an error that ended the node's work on it without ending the task. The scheduler implements it: it owns the attempt budget, the lease and the queue, so the decision between another attempt and a terminal failure is not this package's to make.
type RollupConfig ¶
type RollupConfig struct {
// Interval is how often terminal tasks are swept for roll-up.
Interval time.Duration
// Settle is how long after a task finishes before its logs are rolled up. It keeps
// the roll-up off the heels of the last event batch.
Settle time.Duration
// PruneInterval is how often rolled-up chunks are pruned.
PruneInterval time.Duration
// Grace is how long a task's chunks survive after its logs reached the object store.
Grace time.Duration
}
RollupConfig is the log roll-up schedule.
Grace is the part that matters. A task's chunks are what store.TaskStreamOffsets and store.MaxTaskSeq are computed from, and those are what a restarting node is told to resume against, so deleting them is not a housekeeping detail. Roll-up only ever touches terminal tasks — which reconciliation never tells a node to adopt — and the grace period is the second belt: a task that finished minutes ago still has its hot rows if anything is still reading them.
func DefaultRollupConfig ¶
func DefaultRollupConfig() RollupConfig
DefaultRollupConfig is the design's schedule: roll up a minute after a task settles, prune hourly, keep the hot rows for a day.
func RollupConfigFromEnv ¶
func RollupConfigFromEnv() RollupConfig
RollupConfigFromEnv applies the four duration overrides to the defaults. An unparseable value is ignored: a typo must not silently delete a day of log retention.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service ingests node event batches and serves StreamTaskEvents.
func (*Service) Ingest ¶
func (s *Service) Ingest(ctx context.Context, taskID string, batch []*podiumv1.TaskEvent) (uint64, error)
Ingest stores one node event batch and returns the highest seq it contained, which is what the caller acks. A node numbers all of a task's events from one seq space, so the batch is split across task_events and task_log_chunks but written in a single transaction — the ack therefore covers the whole batch, not either table's high-water mark.
Re-ingesting an acked batch is a no-op: the rows conflict on (task_id, seq) and the status transitions they imply are already applied and get rejected harmlessly.
func (*Service) Note ¶
Note appends one synthetic event to a task's history: something the control plane observed rather than the node reported. Losing a node is the case that matters — a task that simply stops, with nothing in its log to say why, is the single most confusing thing an operator can be shown.
It is stored as a non-retryable error event, which is how the CLI and the UI already render "this is why it ended", but it deliberately does *not* run the status transition such an event would imply when a node sends one: the caller owns the transition, and for a lost node the right end state is lost, not failed.
The sequence number is one past everything stored. That is safe precisely because the node whose events would have collided is gone; a node that comes back is handed the new high-water mark in its HelloAck and numbers above it.
func (*Service) PruneOnce ¶
PruneOnce deletes the hot rows of tasks rolled up longer ago than the grace period.
func (*Service) RollUpOnce ¶
RollUpOnce rolls up one batch of terminal tasks and reports how many it did.
func (*Service) Run ¶
Run follows the task-event notification channel until ctx is cancelled, waking the subscribers of every task that gains rows. One subscription serves the whole process.
func (*Service) RunRollUp ¶
RunRollUp sweeps terminal tasks into the object store and prunes the hot rows of ones that have been there long enough. It returns when ctx is cancelled.
func (*Service) SetArchive ¶
func (s *Service) SetArchive(a *artifacts.Service, cfg RollupConfig)
SetArchive wires in the object store and the roll-up schedule. It is a setter for the same reason SetSlots is: the server builds the pieces in an order neither can be a constructor argument of.
func (*Service) SetRetries ¶
SetRetries wires in the scheduler, for the same reason SetSlots is a setter: the scheduler is built on the node registry, which is built on this service.
func (*Service) SetSlots ¶
SetSlots wires in the node session registry. It is a setter rather than a constructor argument because the registry is built around this service: nodes.NewService takes the Ingestor, so neither can be constructed before the other.
func (*Service) Subscribe ¶
func (s *Service) Subscribe(ctx context.Context, taskID string, fromSeq uint64) (<-chan *podiumv1.TaskEvent, error)
Subscribe replays a task's stored events from fromSeq (exclusive, so 0 replays everything) and then follows live ones. The returned channel is closed when the task is terminal and every event has been delivered, or when ctx is cancelled.
Source Files