executor

package

Go to main page

Versions in this module

v0

v0.5.8

Nov 1, 2026

Changes in this version

+ const EventResilienceBackstopDisabled

+ func ResilienceBackstopKeys() []string

type LadderWarning

+ Event string

+ type Teardown struct

+ Preserved int

v0.5.8-rc.3

Nov 1, 2026

v0.5.8-rc.2

Nov 1, 2026

v0.5.8-rc.1

Oct 29, 2026

v0.5.7

Oct 27, 2026

Changes in this version

type KubernetesExecutor

+ func (e *KubernetesExecutor) SetDecisionRecorder(r DecisionRecorder)

+ func (e *KubernetesExecutor) SetLogger(l *slog.Logger)

v0.5.7-rc.1

Oct 27, 2026

v0.5.6

Oct 23, 2026

Changes in this version

type Disposition

+ const Unavailable

v0.5.6-rc.2

Oct 23, 2026

v0.5.6-rc.1

Oct 23, 2026

v0.5.5

Oct 22, 2026

Changes in this version

+ const StartCredentialMargin

type DispatchLostReapStore

+ MarkTaskStartExpired func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

type KubernetesExecutor

+ func (e *KubernetesExecutor) TaskPodPending(ctx context.Context, a Attempt) (bool, error)

type KubernetesWarmPods

+ func (k *KubernetesWarmPods) LastTerminationReason(podName string) string

type PodManager

+ TaskPodPending func(ctx context.Context, a Attempt) (bool, error)

type Reaper

+ func (r *Reaper) SetAttemptTokenTTL(ttl time.Duration)

type ResilienceLadder

+ DispatchLostThreshold time.Duration

+ type ServiceAccounts struct

+ func NewServiceAccounts(clientset kubernetes.Interface, namespace string) *ServiceAccounts

+ func (s *ServiceAccounts) ImagePullSecrets(ctx context.Context, name string) (count int, found bool, err error)

type StaleQueuedCandidate

+ TenantID string

type WarmBoundTI

+ DagID string

+ TenantID string

type WarmPodInfo

+ TerminationReason string

type WarmPodInformer

+ func (w *WarmPodInformer) RecordTerminationsTo(k *KubernetesWarmPods)

+ type WarmTerminationLookup interface

+ LastTerminationReason func(podName string) string

type WarmWorkerLostReapStore

+ MarkTaskPodLostOOMKilled func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

v0.5.5-rc.3

Oct 22, 2026

v0.5.5-rc.2

Oct 22, 2026

v0.5.5-rc.1

Oct 21, 2026

v0.5.4

Oct 16, 2026

Changes in this version

+ const ReapPageSize

type KubernetesExecutor

+ func (e *KubernetesExecutor) SetPodMetrics(r PodCreationRecorder)

type OutcomeReporter

+ FailTaskInfra func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int, ...) error

+ type PodCreationRecorder interface

+ ObservePodCreation func(dagID, result string, d time.Duration)

+ type ReapCursor struct

+ At time.Time

+ ID string

+ func (c ReapCursor) IsZero() bool

type Request

+ DefaultServiceAccount string

+ ExecutionPolicy *domain.ExecutionPolicy

+ type WarmPodCache interface

+ BeginCreate func(t WarmTarget)

+ CachedWarmPods func() ([]WarmPodInfo, bool)

+ Covers func(podNames []string) bool

+ CreatesPending func(dagVersionID string) bool

+ EndCreate func(t WarmTarget, podName string)

+ ExpectationExpired func() bool

+ PendingCreatesByTenant func() map[string]int

+ type WarmPodInformer struct

+ func NewWarmPodInformer(clientset kubernetes.Interface, namespace string) (*WarmPodInformer, error)

+ func (w *WarmPodInformer) BeginCreate(t WarmTarget)

+ func (w *WarmPodInformer) CachedWarmPods() ([]WarmPodInfo, bool)

+ func (w *WarmPodInformer) Changes() <-chan struct{}

+ func (w *WarmPodInformer) Covers(names []string) bool

+ func (w *WarmPodInformer) CreatesPending(dagVersionID string) bool

+ func (w *WarmPodInformer) EndCreate(t WarmTarget, podName string)

+ func (w *WarmPodInformer) ExpectationExpired() bool

+ func (w *WarmPodInformer) Kick()

+ func (w *WarmPodInformer) PendingCreatesByTenant() map[string]int

+ func (w *WarmPodInformer) Start(ctx context.Context)

+ func (w *WarmPodInformer) WaitForCacheSync(ctx context.Context) bool

type WarmPoolReconciler

+ func (r *WarmPoolReconciler) SetCache(c WarmPodCache, parallelism int)

type WarmTarget

+ DagID string

+ Prewarm bool

v0.5.4-rc.1

Oct 15, 2026

v0.5.3

Oct 12, 2026

Changes in this version

+ const SourceAnnotation

+ const SourceMountPath

+ type AttemptStopper interface

+ StopAttempt func(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

type KubernetesWarmPods

+ func (k *KubernetesWarmPods) ListWarmAnchors(ctx context.Context) ([]string, error)

type Request

+ SourceMode bool

type SubprocessExecutor

+ func (e *SubprocessExecutor) StopAttempt(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

type WarmPodClient

+ ListWarmAnchors func(ctx context.Context) ([]string, error)

v0.5.3-rc.1

Oct 10, 2026

v0.5.2

Oct 8, 2026

Changes in this version

+ func OutlivedCredentialCeiling(c AgentLostCandidate, ceiling time.Duration) bool

type AgentLostCandidate

+ StartedAt time.Time

type Disposition

+ const Refused

type HeartbeatReapStore

+ MarkTaskCredentialCeiling func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

+ type InfraOverride struct

+ DagID string

+ DagRunID string

+ Mark string

+ TaskID string

+ TenantID string

+ type InfraOverrideRecorder interface

+ RecordInfraOverride func(mark string)

+ type InfraOverrideReporter interface

+ SucceedTaskIfActive func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

+ SucceedTaskOverInfraMark func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (InfraOverride, bool, error)

type KubernetesExecutor

+ func (e *KubernetesExecutor) SetTeardownRecorder(r DecisionRecorder)

type Reaper

+ func (r *Reaper) SetAttemptLifetimeCeiling(d time.Duration)

type Reconciler

+ func (r *Reconciler) SetInfraOverrideObservers(rec InfraOverrideRecorder, sink logSink)

type WarmPodSpec

+ Resources *domain.Resources

v0.5.2-rc.1

Oct 8, 2026

v0.5.1

Oct 6, 2026

Changes in this version

type AgentLostCandidate

+ AttemptEpoch int

+ type Attempt struct

+ AttemptEpoch int

+ RunID string

+ TaskID string

+ TryNumber int

type Disposition

+ const Deferred

+ type InfraConfirmer interface

+ ConfirmInfraFailure func(ctx context.Context, taskInstanceID string, tryNumber, attemptEpoch int) (bool, error)

+ ListProvisionalInfraFailures func(ctx context.Context) ([]ProvisionalInfraFailure, error)

+ type OrphanStopper interface

+ StopOrphanedTask func(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

type PodIdentity

+ AttemptEpoch *int

type PodLostCandidate

+ AttemptEpoch int

+ Heartbeated bool

+ TenantID string

+ type ProcessLiveness interface

+ AttemptProcessAlive func(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

+ type ProvisionalInfraFailure struct

+ AttemptEpoch int

+ DagRunID string

+ TaskID string

+ TaskInstanceID string

+ TryNumber int

type Reaper

+ func (r *Reaper) SetProcessLiveness(p ProcessLiveness)

type Reconciler

+ func (r *Reconciler) SetInfraConfirmer(c InfraConfirmer)

+ func (r *Reconciler) SetSettledRunCollection(c SettledRunChecker)

type Request

+ AttemptEpoch int

type ResilienceLadder

+ InfraConfirmMaxWait time.Duration

+ TaskTerminationGrace time.Duration

+ type RunRef struct

+ Run string

+ Tenant string

+ type SettledRunChecker interface

+ SettledRuns func(ctx context.Context, refs []RunRef) (map[RunRef]bool, error)

type StaleQueuedCandidate

+ AttemptEpoch int

type SubprocessExecutor

+ func (e *SubprocessExecutor) AttemptProcessAlive(_ context.Context, runID, taskID string, tryNumber int) (bool, error)

+ func (e *SubprocessExecutor) SetPIDDir(dir string)

+ func (e *SubprocessExecutor) StopOrphanedTask(ctx context.Context, runID, taskID string, tryNumber int) (bool, error)

type WarmBoundTI

+ AttemptEpoch int

type WarmPodSpec

+ ReadOnlyRootFilesystem bool

v0.5.1-rc.1

Oct 5, 2026

v0.5.0

Oct 3, 2026

Changes in this version

+ const AgentIdentityAnnotation

+ const DefaultAgentTokenAudience

+ const WarmDagVersionLabel

+ const WarmTenantLabel

+ const WarmWorkerLabel

+ const WarmWorkerLabelValue

+ func BuildPod(req Request) *corev1.Pod

+ func BuildWarmPod(spec WarmPodSpec) *corev1.Pod

+ func IsAgentLost(c AgentLostCandidate, threshold time.Duration, now time.Time) bool

+ func IsDispatchLost(c StaleQueuedCandidate, threshold time.Duration, now time.Time) bool

+ func IsOrphaned(c ReapCandidate, threshold time.Duration, now time.Time) bool

+ func IsPodLostCandidate(c PodLostCandidate, grace time.Duration, now time.Time) bool

+ func StagingClaimName(dagID, runID string) string

+ func ValidateResilienceLadder(l ResilienceLadder) error

+ type AgentLostCandidate struct

+ DagID string

+ DagRunID string

+ LastHeartbeat time.Time

+ TaskID string

+ TaskInstanceID string

+ TenantID string

+ TryNumber int

+ type BusyWarmWorkerSource interface

+ ListBusyWarmWorkerPods func(ctx context.Context) (map[string]bool, error)

+ type DecisionRecorder interface

+ RecordSchedulerDecision func(decisionType string)

+ type DispatchLostReapStore interface

+ ListStaleQueuedCandidates func(ctx context.Context) ([]StaleQueuedCandidate, error)

+ MarkTaskDispatchLost func(ctx context.Context, taskInstanceID string) error

+ type Disposition int

+ const Backpressure

+ const Dispatched

+ const Rejected

+ func (d Disposition) String() string

+ type Executor interface

+ Execute func(ctx context.Context, req Request) (Disposition, error)

+ type HeartbeatReapStore interface

+ ListAgentLostCandidates func(ctx context.Context) ([]AgentLostCandidate, error)

+ MarkTaskAgentLost func(ctx context.Context, taskInstanceID string) (bool, error)

+ type KubernetesExecutor struct

+ func NewKubernetesExecutor(clientset kubernetes.Interface, namespace string) *KubernetesExecutor

+ func (e *KubernetesExecutor) DeleteRunPods(ctx context.Context, runID string) error

+ func (e *KubernetesExecutor) DeleteTaskPod(ctx context.Context, runID, taskID string, tryNumber int) error

+ func (e *KubernetesExecutor) Execute(ctx context.Context, req Request) (Disposition, error)

+ func (e *KubernetesExecutor) GCStagingClaims(ctx context.Context, ttl time.Duration) error

+ func (e *KubernetesExecutor) SetStagingStore(s StagingStore)

+ func (e *KubernetesExecutor) TaskPodPresence(ctx context.Context, runID, taskID string, tryNumber int) (PodPresence, error)

+ type KubernetesWarmPods struct

+ func NewKubernetesWarmPods(cs kubernetes.Interface, namespace string, newSpec WarmPodSpecFunc) *KubernetesWarmPods

+ func (k *KubernetesWarmPods) CreateWarmPod(ctx context.Context, t WarmTarget, anchorName, anchorUID string) error

+ func (k *KubernetesWarmPods) DeleteWarmAnchor(ctx context.Context, dagVersionID string) error

+ func (k *KubernetesWarmPods) DeleteWarmPod(ctx context.Context, name string) error

+ func (k *KubernetesWarmPods) EnsureWarmAnchor(ctx context.Context, dagVersionID string) (string, error)

+ func (k *KubernetesWarmPods) ListWarmPods(ctx context.Context) ([]WarmPodInfo, error)

+ type LadderWarning struct

+ Key string

+ Msg string

+ Value time.Duration

+ func ResilienceLadderWarnings(l ResilienceLadder) []LadderWarning

+ type OutcomeReporter interface

+ FailTask func(ctx context.Context, taskInstanceID string, tryNumber int, reason string) error

+ RescheduleTask func(ctx context.Context, taskInstanceID string, tryNumber int, at time.Time) error

+ SucceedTask func(ctx context.Context, taskInstanceID string, tryNumber int) error

+ type PodIdentity struct

+ DagID string

+ RunID string

+ TaskID string

+ TaskInstanceID string

+ TenantID string

+ TryNumber int

+ func ParseAgentIdentity(raw string) (PodIdentity, error)

+ type PodInformer struct

+ func NewPodInformer(clientset kubernetes.Interface, namespace string) *PodInformer

+ func (p *PodInformer) CachedPodActive(runID, taskID string, tryNumber int) bool

+ func (p *PodInformer) HasSynced() bool

+ func (p *PodInformer) Shutdown()

+ func (p *PodInformer) SnapshotTaskPods() ([]*corev1.Pod, error)

+ func (p *PodInformer) Start(ctx context.Context)

+ func (p *PodInformer) WaitForCacheSync(ctx context.Context) bool

+ type PodLostCandidate struct

+ DagID string

+ DagRunID string

+ RunningSince time.Time

+ TaskID string

+ TaskInstanceID string

+ TryNumber int

+ type PodLostReapStore interface

+ ListRunningTasks func(ctx context.Context, grace time.Duration) ([]PodLostCandidate, error)

+ MarkTaskPodLost func(ctx context.Context, taskInstanceID string) (bool, error)

+ type PodManager interface

+ DeleteRunPods func(ctx context.Context, runID string) error

+ DeleteTaskPod func(ctx context.Context, runID, taskID string, tryNumber int) error

+ TaskPodPresence func(ctx context.Context, runID, taskID string, tryNumber int) (PodPresence, error)

+ type PodPresence int

+ const PodPresenceAbsent

+ const PodPresenceLive

+ const PodPresenceTerminal

+ func (p PodPresence) String() string

+ type PodPresenceCache interface

+ CachedPodActive func(runID, taskID string, tryNumber int) bool

+ type PodSecurity struct

+ ReadOnlyRootFilesystem bool

+ RunAsNonRoot bool

+ type PodSnapshotter interface

+ SnapshotTaskPods func() ([]*corev1.Pod, error)

+ type ReapCandidate struct

+ DagID string

+ LastActivity time.Time

+ RunID string

+ type ReapStore interface

+ ListReapCandidates func(ctx context.Context) ([]ReapCandidate, error)

+ ReapRun func(ctx context.Context, runID string) error

+ type Reaper struct

+ func NewReaper(store ReaperStore, pods PodManager, cache PodPresenceCache, ...) *Reaper

+ func (r *Reaper) ReapOnce(ctx context.Context) error

+ func (r *Reaper) SetInformerSynced(fn func() bool)

+ func (r *Reaper) SetLastSweepCompleted(fn func() time.Time)

+ func (r *Reaper) SetLeaderSince(fn func() time.Time)

+ func (r *Reaper) SetLeading(fn func() bool)

+ func (r *Reaper) SetLogSink(s logSink)

+ type ReaperConfig struct

+ AgentLostThreshold time.Duration

+ DispatchLostThreshold time.Duration

+ OrphanThreshold time.Duration

+ PodLostGrace time.Duration

+ SettlingGrace time.Duration

+ func DefaultReaperConfig() ReaperConfig

+ type ReaperStore interface

+ type Reconciler struct

+ func NewReconciler(clientset kubernetes.Interface, namespace string, reporter OutcomeReporter) *Reconciler

+ func (r *Reconciler) LastSweepCompletedAt() time.Time

+ func (r *Reconciler) Reconcile(ctx context.Context) error

+ func (r *Reconciler) SetPodSnapshotter(s PodSnapshotter)

+ type Request struct

+ AgentTLSCAConfigMap string

+ AgentToken string

+ AgentTokenAudience string

+ AgentTokenExpirationSeconds int64

+ AgentTokenSecretKey string

+ AgentTokenSecretName string

+ AgentTokenTransport string

+ AttemptLifetimeCeilingSeconds int64

+ ControlPlaneAddr string

+ DagID string

+ Entrypoint string

+ Env map[string]string

+ Execution domain.Execution

+ Image string

+ ImagePullPolicy string

+ Operator string

+ PodSecurity PodSecurity

+ Resources domain.Resources

+ RunID string

+ SecretsBackend string

+ SecretsBackendKwargs string

+ Source string

+ StagingAccessMode string

+ StagingClaim string

+ StagingSize string

+ StagingStorageClass string

+ TaskID string

+ TaskInstanceID string

+ TaskSecretMountPath string

+ TaskSecretName string

+ TenantID string

+ TimeoutSeconds int

+ TryNumber int

+ type ResilienceLadder struct

+ AgentLostThreshold time.Duration

+ AttemptTokenTTL time.Duration

+ HeartbeatInterval time.Duration

+ InfraReplaceMaxDelay time.Duration

+ MaxAttemptCredentialLifetime time.Duration

+ OrphanThreshold time.Duration

+ ReconcileInterval time.Duration

+ SettlingGrace time.Duration

+ type StagingStore interface

+ ListActiveStagingVolumes func(ctx context.Context) ([]domain.StagingVolumeState, error)

+ MarkStagingDeleted func(ctx context.Context, pvcName, reason string) error

+ RecordStagingVolume func(ctx context.Context, tenantID, dagID, runID, pvcName, size string) error

+ type StaleQueuedCandidate struct

+ DagID string

+ DagRunID string

+ QueuedAt time.Time

+ TaskID string

+ TaskInstanceID string

+ TryNumber int

+ WarmWorkerID string

+ type SubprocessExecutor struct

+ func NewSubprocessExecutor(agentPath string, logger *slog.Logger) *SubprocessExecutor

+ func (e *SubprocessExecutor) Execute(ctx context.Context, req Request) (Disposition, error)

+ func (e *SubprocessExecutor) SetWorkDir(dir string)

+ type WarmBoundTI struct

+ DagRunID string

+ TaskID string

+ TaskInstanceID string

+ TryNumber int

+ WarmWorkerID string

+ type WarmPodClient interface

+ CreateWarmPod func(ctx context.Context, t WarmTarget, anchorName, anchorUID string) error

+ DeleteWarmAnchor func(ctx context.Context, dagVersionID string) error

+ DeleteWarmPod func(ctx context.Context, name string) error

+ EnsureWarmAnchor func(ctx context.Context, dagVersionID string) (uid string, err error)

+ ListWarmPods func(ctx context.Context) ([]WarmPodInfo, error)

+ type WarmPodInfo struct

+ DagVersionID string

+ Name string

+ TenantID string

+ Terminal bool

+ type WarmPodLister interface

+ ListWarmPods func(ctx context.Context) ([]WarmPodInfo, error)

+ type WarmPodSpec struct

+ AgentTLSCAConfigMap string

+ AgentTokenAudience string

+ AgentTokenExpirationSeconds int64

+ AgentTokenSecretKey string

+ AgentTokenSecretName string

+ AgentTokenTransport string

+ AnchorName string

+ AnchorUID types.UID

+ Annotations map[string]string

+ AttemptWatchdogSeconds int64

+ BootstrapToken string

+ ControlPlaneAddr string

+ DagVersionID string

+ Image string

+ ImagePullPolicy string

+ Labels map[string]string

+ MaxAttemptsPerWorker int

+ MaxWorkerLifetimeSeconds int64

+ Namespace string

+ PodSecurity PodSecurity

+ ServiceAccount string

+ TenantID string

+ WorkerIdleTTLSeconds int64

+ type WarmPodSpecFunc func(t WarmTarget) (WarmPodSpec, error)

+ type WarmPoolReconciler struct

+ func NewWarmPoolReconciler(targets WarmTargetSource, pods WarmPodClient, busy BusyWarmWorkerSource, ...) *WarmPoolReconciler

+ func (r *WarmPoolReconciler) Reconcile(ctx context.Context) error

+ type WarmTarget struct

+ DagVersionID string

+ EffectiveMinIdle int

+ Image string

+ MaxPoolSize int

+ TenantID string

+ type WarmTargetSource interface

+ ActiveWarmTargets func(ctx context.Context) ([]WarmTarget, error)

+ type WarmWorkerLostReapStore interface

+ ListWarmBoundRunningTIs func(ctx context.Context) ([]WarmBoundTI, error)

+ MarkTaskPodLost func(ctx context.Context, taskInstanceID string) (bool, error)

v0.5.0-rc.1

Oct 2, 2026

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL