From e5833934d8c37a5cb17a076918a527726cb0a205 Mon Sep 17 00:00:00 2001 From: Simo Lin <25425177+slin1237@users.noreply.github.com> Date: Wed, 19 Aug 2026 11:08:14 -0700 Subject: [PATCH] [Core] Add the workload types package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First slice of the InferenceReplica runtime. This is the vocabulary the rest of it is written in: the per-Instance plan and operation kinds, Instance and Runner status shapes that mirror the CRD field-for-field, the retry-block and expectations bookkeeping, migration and termination records, event reasons, condition helpers, and the dependency and service seams the reconciler is constructed with. It carries no OME imports at all — only the standard library, k8s API machinery and controller-runtime — so it stands alone and every later slice can be reviewed against it. Nothing in the tree consumes it yet; the packages that do (query, audit, ops, and the InferenceReplica controller itself) follow. Co-authored-by: Simo Lin <25425177+slin1237@users.noreply.github.com> Co-authored-by: yunfanw Signed-off-by: Fan Yang <250624800+fanyang-real@users.noreply.github.com> --- .../v1beta1/workload/types/conditions.go | 33 ++ pkg/controller/v1beta1/workload/types/deps.go | 117 +++++ .../v1beta1/workload/types/events.go | 142 ++++++ .../v1beta1/workload/types/expectations.go | 162 +++++++ .../workload/types/expectations_clock_test.go | 50 +++ .../v1beta1/workload/types/input.go | 343 +++++++++++++++ .../v1beta1/workload/types/migration.go | 165 +++++++ pkg/controller/v1beta1/workload/types/plan.go | 178 ++++++++ .../v1beta1/workload/types/retryblock.go | 78 ++++ .../v1beta1/workload/types/retryblock_test.go | 53 +++ .../workload/types/retryblock_writer.go | 72 ++++ .../v1beta1/workload/types/service.go | 62 +++ .../v1beta1/workload/types/source.go | 207 +++++++++ .../v1beta1/workload/types/termination.go | 197 +++++++++ .../v1beta1/workload/types/types.go | 407 ++++++++++++++++++ .../v1beta1/workload/types/types_test.go | 367 ++++++++++++++++ 16 files changed, 2633 insertions(+) create mode 100644 pkg/controller/v1beta1/workload/types/conditions.go create mode 100644 pkg/controller/v1beta1/workload/types/deps.go create mode 100644 pkg/controller/v1beta1/workload/types/events.go create mode 100644 pkg/controller/v1beta1/workload/types/expectations.go create mode 100644 pkg/controller/v1beta1/workload/types/expectations_clock_test.go create mode 100644 pkg/controller/v1beta1/workload/types/input.go create mode 100644 pkg/controller/v1beta1/workload/types/migration.go create mode 100644 pkg/controller/v1beta1/workload/types/plan.go create mode 100644 pkg/controller/v1beta1/workload/types/retryblock.go create mode 100644 pkg/controller/v1beta1/workload/types/retryblock_test.go create mode 100644 pkg/controller/v1beta1/workload/types/retryblock_writer.go create mode 100644 pkg/controller/v1beta1/workload/types/service.go create mode 100644 pkg/controller/v1beta1/workload/types/source.go create mode 100644 pkg/controller/v1beta1/workload/types/termination.go create mode 100644 pkg/controller/v1beta1/workload/types/types.go create mode 100644 pkg/controller/v1beta1/workload/types/types_test.go diff --git a/pkg/controller/v1beta1/workload/types/conditions.go b/pkg/controller/v1beta1/workload/types/conditions.go new file mode 100644 index 000000000..f7d57dd2e --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/conditions.go @@ -0,0 +1,33 @@ +package types + +// ConditionType / ConditionReason are workload-internal identifiers +// stamped on metav1.Condition entries. Values match the legacy +// omenative/status strings byte-for-byte so operator dashboards keep +// matching. +type ConditionType string + +func (t ConditionType) String() string { return string(t) } + +type ConditionReason string + +func (r ConditionReason) String() string { return string(r) } + +const ( + // ConditionGangSchedulingUnavailable is True when the Component + // has at least one multi-pod Instance but the scheduler-plugins + // PodGroup CRD is missing. The reconciler still creates pods — + // gang scheduling is a soft requirement so workloads proceed + // without blocking — but partial-gang placement is possible and + // the runtime may hang. + ConditionGangSchedulingUnavailable ConditionType = "GangSchedulingUnavailable" +) + +const ( + // ReasonPodGroupCRDNotInstalled stamps the + // GangSchedulingUnavailable condition when the + // scheduler-plugins PodGroup CRD is missing. + ReasonPodGroupCRDNotInstalled ConditionReason = "PodGroupCRDNotInstalled" + // ReasonGangSchedulingAvailable stamps Status=False (CRD present, + // or Component is single-pod). + ReasonGangSchedulingAvailable ConditionReason = "GangSchedulingAvailable" +) diff --git a/pkg/controller/v1beta1/workload/types/deps.go b/pkg/controller/v1beta1/workload/types/deps.go new file mode 100644 index 000000000..c1e4324f2 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/deps.go @@ -0,0 +1,117 @@ +package types + +import ( + "context" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/client-go/tools/record" + "k8s.io/utils/clock" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// ObservationReader is a watch-backed (cached) reader; it may lag the +// API server. Safe wherever a later live read or optimistic-concurrency +// write re-validates the observation. +type ObservationReader = client.Reader + +// AuthoritativeReader reads live from the API server. Required where +// cache lag is a correctness bug: revision bookkeeping, audit ledger, +// EndpointSlice drain checks, gang topology proofs. +type AuthoritativeReader = client.Reader + +// Deps is the manager-scoped wiring the workload reconciler needs. +// Shared across every workload the same controller manages (unlike +// the per-workload ReconcileInput). +type Deps struct { + // Client is the controller-runtime cached client used for the + // vast majority of read/write operations. + Client client.Client + + // APIReader is the AuthoritativeReader role (see type docs). + // Used for revision bookkeeping and immutable topology observations where + // stale cache contents could cause a collision or split a live gang. When + // nil the cached Client is used. + APIReader client.Reader + + // Recorder emits K8s Events at lifecycle transitions. nil-safe: + // when unset, event helpers are no-ops. + Recorder record.EventRecorder + + // Expectations is the create/delete bookkeeping cache used to + // avoid re-issuing batches before the controller-runtime watch + // has confirmed prior writes. Adapters wire the same instance + // into the Pod event handler so observed create / delete events + // release expectations immediately. nil falls back to + // DefaultExpectations. + Expectations *Expectations + + // RenderHook is an optional per-pod template mutator invoked + // after composing the canonical pod (name, hostname, labels, + // controller env, owner ref) but before c.Create. The IR + // reconciler (the sole workload.Reconcile caller) wires this to + // coordination.InjectPeerEnv via core.ISVCRenderHook so + // OME__ENDPOINT vars land on every container. + // + // The hook MUST be idempotent — Render may be invoked twice in + // the same reconcile (e.g., cache-lag retry after AlreadyExists). + RenderHook RenderHook + + // EnsureGangPodGroup, when set, creates the scheduler-plugins + // PodGroup for one multi-pod surge Instance just before its pods are + // created. The gang-surge op needs PodGroup-before-pods to hold even + // in the window before the surge index lands in the plan the + // top-level EnsurePodGroups keys off (else a gang scheduler rejects + // the surge pods with "PodGroup not found"). The callback lives on + // Deps rather than as a direct call so the workload/ops package stays + // free of the workload/podgroup dependency (ops is imported by + // podgroup's test deps; a direct edge would close a cycle). The gang + // package wires it; nil callback / single-pod Instance / absent CRD + // all no-op. Idempotent. + EnsureGangPodGroup EnsureGangPodGroupFn + + // Clock supplies wall-clock time for deadlines and status + // timestamps. nil falls back to the real clock — inject a fake + // (k8s.io/utils/clock/testing) for deterministic boundary tests. + Clock clock.Clock +} + +// RenderHook is the optional adapter-specific render-time mutator. +// Receives the pod under construction; may mutate any field except +// Name/Namespace/OwnerReferences (workload-controlled). +type RenderHook func(pod *corev1.Pod, runnerName string, ordinal int32, revisionHash string) + +// EnsureGangPodGroupFn ensures the PodGroup for one multi-pod surge Instance +// and returns the effective topology key selected for its live pods. See +// Deps.EnsureGangPodGroup. +type EnsureGangPodGroupFn func(ctx context.Context, input ReconcileInput, plan ComponentPlan, inst InstancePlan) (string, error) + +// Reader returns the live API reader when APIReader is set, otherwise +// the cached Client. Use for reads that must not observe a stale +// cache. +func (d *Deps) Reader() client.Reader { + if d.APIReader != nil { + return d.APIReader + } + return d.Client +} + +// ExpectationsCache returns the per-controller cache threaded through +// Deps when set, otherwise the DefaultExpectations singleton. +func (d *Deps) ExpectationsCache() *Expectations { + if d.Expectations != nil { + return d.Expectations + } + return DefaultExpectations +} + +// Now returns the injected clock's time, or time.Now() when no clock +// is wired. Lifecycle code holding a Deps or ReconcileInput should read +// time through Now, not time.Now(); subpackages without a seam +// (audit, podreadiness, gang) still read real time. +func (d *Deps) Now() time.Time { + if d.Clock != nil { + return d.Clock.Now() + } + return time.Now() +} diff --git a/pkg/controller/v1beta1/workload/types/events.go b/pkg/controller/v1beta1/workload/types/events.go new file mode 100644 index 000000000..0748a37e8 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/events.go @@ -0,0 +1,142 @@ +package types + +// EventReason is the workload-internal event-reason identifier the ops +// state machines stamp on K8s Events. Values match the legacy +// omenative/status reason strings byte-for-byte so existing operator +// dashboards and `kubectl describe` output keep matching. +type EventReason string + +func (r EventReason) String() string { return string(r) } + +const ( + // Create / scale-up (workload/ops/create.go). + EventReasonInstanceCreated EventReason = "InstanceCreated" + EventReasonInstanceReady EventReason = "InstanceReady" + + // In-place update (workload/ops/update.go). + EventReasonInPlaceUpdateStarted EventReason = "InPlaceUpdateStarted" + EventReasonInPlaceUpdateCompleted EventReason = "InPlaceUpdateCompleted" + EventReasonInPlaceUpdateNotPossible EventReason = "InPlaceUpdateNotPossible" + + // Recreate / surge update (workload/ops/update.go). + EventReasonRecreateUpdateStarted EventReason = "RecreateUpdateStarted" + EventReasonRecreateUpdateCompleted EventReason = "RecreateUpdateCompleted" + + // Restart (workload/ops/restart.go). + EventReasonRestartTriggered EventReason = "RestartTriggered" + EventReasonRestartCompleted EventReason = "RestartCompleted" + + // EventReasonFoundOrphan fires when Restart or recreate-Update + // finds a pod under the OMENative selector but missing the + // ome.io/instance-incarnation label. The reconciler refuses to + // delete it; the operator must re-classify or remove the pod + // manually. + EventReasonFoundOrphan EventReason = "FoundOrphan" + + // EventReasonSupersededWreckageCleaned fires when the corrective- + // edit cleanup deletes pods keyed to a superseded revision — debris + // a failed rollout left behind that no revision-diff trigger could + // reach. + EventReasonSupersededWreckageCleaned EventReason = "SupersededWreckageCleaned" + + // EventReasonAutoMigrationTriggered fires when the deadline + // disposition records a relocation directive (terminal AutoRecover + // ledger entry) for a stuck Instance — its rebuild will be steered + // off the recorded node. Value matches the legacy omenative + // detector's reason string. + EventReasonAutoMigrationTriggered EventReason = "AutoMigrationTriggered" + + // EventReasonAutoMigrationCapReached fires exactly once per budget + // fill — when the disposition records the relocation directive that + // exhausts the (component, instance) AutoRecover budget + // (lifecycle.autoMigrate.maxAttempts). Subsequent over-budget + // dispositions dispose terminal silently. Operator intervention (or + // an instance reaching Ready, which prunes its records) is required + // before relocation resumes. + EventReasonAutoMigrationCapReached EventReason = "AutoMigrationCapReached" + + // EventReasonInstanceFailed fires when an escalation backstop (the + // stuck-pod fast path or the deadline disposition) stamps an + // Instance Phase=Failed. Emitted by adapters through + // ReconcileInput.WarnInstanceFailed. + EventReasonInstanceFailed EventReason = "InstanceFailed" + + // EventReasonRetryHeld fires once, at the RetryBlock transition into + // State=Held — same-target update retries exhausted; a corrected + // revision (or raised retry limits) is required. Emitted by adapters + // through ReconcileInput.WarnRetryHeld. + EventReasonRetryHeld EventReason = "RetryHeld" + + // EventReasonRetryBlockReleased fires when the operator release + // annotation (ome.io/release-held-revision) removes a Held + // RetryBlock — the manual exit from the terminal Held state. Names + // the released revision and that the removal was operator-requested. + EventReasonRetryBlockReleased EventReason = "RetryBlockReleased" + + // EventReasonRetryBlockReleaseSkipped fires when the release + // annotation names no releasable block — no RetryBlock exists for + // the requested revision, or the matched block is not State=Held. + // The annotation is still consumed; the event explains why nothing + // changed. + EventReasonRetryBlockReleaseSkipped EventReason = "RetryBlockReleaseSkipped" + + // EventReasonPodForceDeleted fires when scale-down escalation + // force-deletes (grace 0, UID-preconditioned) a Terminating pod + // overdue past its own deletion deadline on a node that provably + // cannot acknowledge the termination (gone, or unreachable-tainted / + // NotReady beyond the configured threshold). Names the pod, node, + // evidence branch, and overdue duration. + EventReasonPodForceDeleted EventReason = "PodForceDeleted" + + // EventReasonPodDeleteBlockedByFinalizer fires (once per pod UID) + // when a Terminating pod is overdue past its deletion deadline but + // pinned by foreign finalizers. Report-only: OME never strips + // another controller's finalizer, so the teardown stays blocked + // until the finalizer owner resolves it. + EventReasonPodDeleteBlockedByFinalizer EventReason = "PodDeleteBlockedByFinalizer" + + // Migration (workload/ops/migrate.go + the IR accept pass). + EventReasonMigrationRequestAccepted EventReason = "MigrationRequestAccepted" + EventReasonMigrationRequestRejected EventReason = "MigrationRequestRejected" + // EventReasonUnsupportedSchemaVersion fires when a migration-request + // annotation carries a schemaVersion the controller doesn't + // understand. Kept distinct from MigrationRequestRejected so + // dashboards can alert on requester/controller version skew. + EventReasonUnsupportedSchemaVersion EventReason = "UnsupportedSchemaVersion" + EventReasonMigrationCompleted EventReason = "MigrationCompleted" + // EventReasonMigrationExpired fires when a non-terminal Manual + // migration record passes its Deadline: the record is closed + // Failed, the pair's Migrate ops are cleared, the surge is torn + // down by the ordinary scale-down batch pipeline, and the source + // phase is restored from observation. + EventReasonMigrationExpired EventReason = "MigrationExpired" + EventReasonRateLimited EventReason = "RateLimited" + EventReasonMigrationFromNodeMismatch EventReason = "MigrationFromNodeMismatch" + EventReasonMigrationNodeAffinityConflict EventReason = "MigrationNodeAffinityConflict" + + // EventReasonMaybeNoGangScheduler is a soft Warning fired the first + // time a multi-pod Instance's PodGroup is created under a pod + // template whose `spec.schedulerName` is unset or equals the + // upstream default ("default-scheduler"). A stock kube-scheduler + // does NOT read scheduling.x-k8s.io/v1alpha1 PodGroup objects, so + // the gang contract degrades to per-pod scheduling silently. + // Operators install scheduler-plugins as a secondary scheduler + // (`scheduler-plugins-scheduler`) or as a default-scheduler plugin + // (in which case the warning is a false positive the controller + // can't detect from inside the cluster). Dedup'd per (owner, + // Component) per process. + EventReasonMaybeNoGangScheduler EventReason = "MaybeNoGangScheduler" + + // EventReasonGangSplitRisk is a soft Warning fired the first time a + // multi-node gang WORKER pod is created with no co-location + // podAffinity at all — neither an OME-injected topologyKey term nor a + // user-declared one. Such a gang may schedule across separate + // network / NVLink / TPU topology domains, which breaks the + // tightly-coupled collectives a multi-node runtime needs (NCCL/RCCL/ + // NIXL all-reduce, multi-host TPU sessions). The operator sets + // engine.topologyKey / decoder.topologyKey (e.g. a NVLink/RDMA domain + // label, or the GKE TPU topology label) or declares a worker + // podAffinity. Advisory only — never blocks the create. Dedup'd per + // (owner, Component) per process. + EventReasonGangSplitRisk EventReason = "GangSplitRisk" +) diff --git a/pkg/controller/v1beta1/workload/types/expectations.go b/pkg/controller/v1beta1/workload/types/expectations.go new file mode 100644 index 000000000..d090c3fde --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/expectations.go @@ -0,0 +1,162 @@ +package types + +import ( + "sync" + "time" + + "k8s.io/utils/clock" +) + +// expectationsTTL is how long an Instance's expectation stays in the +// cache before falling back to a fresh observation. Two minutes is +// generous; even slow watch propagation should converge well before. +const expectationsTTL = 2 * time.Minute + +// Expectations is a per-Instance counter cache. Before issuing a batch +// of pod creates or deletes, the controller calls ExpectCreates / ExpectDeletes +// so subsequent reconciles can tell when the watch event chain has caught +// up. Subsequent reconciles call Satisfied to check whether they can +// safely issue another batch (or proceed past the create step). +// +// Pattern borrowed from sigs.k8s.io/controller-runtime samples and the +// kubernetes/kubernetes ReplicaSet/StatefulSet controllers. +// +// Concurrency: all methods are safe for concurrent use; an in-process +// singleton (DefaultExpectations) is provided for the dispatch path. +type Expectations struct { + mu sync.Mutex + entries map[expectationKey]*expectationEntry + clock clock.Clock +} + +// expectationKey is the per-Instance cache key. OwnerName mirrors +// types.Key.OwnerName — the workload-side owner identifier shared by +// every adapter (ISVC.Name, IR.Name, future owners) rather than the +// CRD-specific "ISVC" name. +type expectationKey struct { + Namespace string + OwnerName string + Component ComponentType + Instance int32 +} + +type expectationEntry struct { + Adds int + Deletes int + Deadline time.Time +} + +// NewExpectations builds an empty cache. +func NewExpectations() *Expectations { + return &Expectations{ + entries: make(map[expectationKey]*expectationEntry), + clock: clock.RealClock{}, + } +} + +// NewExpectationsWithClock is NewExpectations with an injected clock, +// for TTL-boundary tests. +func NewExpectationsWithClock(c clock.Clock) *Expectations { + e := NewExpectations() + if c != nil { + e.clock = c + } + return e +} + +// DefaultExpectations is the in-process singleton used by the workload +// dispatcher. Tests construct their own via NewExpectations. +var DefaultExpectations = NewExpectations() + +// ExpectCreates records that n pod creates are in flight for the given +// (OwnerName, Component, Instance). Satisfied returns false until all +// of them are observed via ObservedCreate, or the deadline elapses. +func (e *Expectations) ExpectCreates(namespace, ownerName string, component ComponentType, instance int32, n int) { + if n <= 0 { + return + } + e.mu.Lock() + defer e.mu.Unlock() + k := expectationKey{namespace, ownerName, component, instance} + ent, ok := e.entries[k] + if !ok { + ent = &expectationEntry{} + e.entries[k] = ent + } + ent.Adds += n + ent.Deadline = e.clock.Now().Add(expectationsTTL) +} + +// ExpectDeletes records that n pod deletes are in flight. +func (e *Expectations) ExpectDeletes(namespace, ownerName string, component ComponentType, instance int32, n int) { + if n <= 0 { + return + } + e.mu.Lock() + defer e.mu.Unlock() + k := expectationKey{namespace, ownerName, component, instance} + ent, ok := e.entries[k] + if !ok { + ent = &expectationEntry{} + e.entries[k] = ent + } + ent.Deletes += n + ent.Deadline = e.clock.Now().Add(expectationsTTL) +} + +// ObservedCreate decrements the create counter, called when a pod +// belonging to (OwnerName, Component, Instance) appears in the watch +// cache. +func (e *Expectations) ObservedCreate(namespace, ownerName string, component ComponentType, instance int32) { + e.observed(namespace, ownerName, component, instance, true) +} + +// ObservedDelete decrements the delete counter, called when a pod +// belonging to (OwnerName, Component, Instance) disappears. +func (e *Expectations) ObservedDelete(namespace, ownerName string, component ComponentType, instance int32) { + e.observed(namespace, ownerName, component, instance, false) +} + +func (e *Expectations) observed(namespace, ownerName string, component ComponentType, instance int32, isAdd bool) { + e.mu.Lock() + defer e.mu.Unlock() + k := expectationKey{namespace, ownerName, component, instance} + ent, ok := e.entries[k] + if !ok { + return + } + if isAdd && ent.Adds > 0 { + ent.Adds-- + } + if !isAdd && ent.Deletes > 0 { + ent.Deletes-- + } + if ent.Adds == 0 && ent.Deletes == 0 { + delete(e.entries, k) + } +} + +// Satisfied returns true when no outstanding creates or deletes are +// expected for the Instance — either the watch cache has caught up, +// the entry expired, or no expectations were ever recorded. +func (e *Expectations) Satisfied(namespace, ownerName string, component ComponentType, instance int32) bool { + e.mu.Lock() + defer e.mu.Unlock() + k := expectationKey{namespace, ownerName, component, instance} + ent, ok := e.entries[k] + if !ok { + return true + } + if !e.clock.Now().Before(ent.Deadline) { + delete(e.entries, k) + return true + } + return ent.Adds <= 0 && ent.Deletes <= 0 +} + +// Forget clears the entry, e.g., when the Instance is deleted entirely. +func (e *Expectations) Forget(namespace, ownerName string, component ComponentType, instance int32) { + e.mu.Lock() + defer e.mu.Unlock() + delete(e.entries, expectationKey{namespace, ownerName, component, instance}) +} diff --git a/pkg/controller/v1beta1/workload/types/expectations_clock_test.go b/pkg/controller/v1beta1/workload/types/expectations_clock_test.go new file mode 100644 index 000000000..166c01e12 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/expectations_clock_test.go @@ -0,0 +1,50 @@ +package types + +// Verifies the Expectations TTL failsafe against the injected clock: an +// unobserved expectation blocks Satisfied until expectationsTTL elapses, +// then expires (treated satisfied) without any watch event. + +import ( + "testing" + "time" + + clocktesting "k8s.io/utils/clock/testing" +) + +func TestExpectations_TTLBoundary(t *testing.T) { + t0 := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + fc := clocktesting.NewFakeClock(t0) + e := NewExpectationsWithClock(fc) + + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 1) + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("outstanding create must block Satisfied") + } + + // Just inside the TTL: still blocked. + fc.SetTime(t0.Add(expectationsTTL - time.Second)) + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("1s before TTL expiry must still block Satisfied") + } + + // Exactly at the deadline: expired. The deadline is the instant the + // failsafe fires, not the last instant it is withheld. + fc.SetTime(t0.Add(expectationsTTL)) + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("an expectation at its exact deadline must be treated satisfied") + } +} + +func TestExpectations_ExpiresPastTTL(t *testing.T) { + t0 := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + fc := clocktesting.NewFakeClock(t0) + e := NewExpectationsWithClock(fc) + + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 1) + + // Past the TTL: the entry expires and Satisfied reports true. + fc.SetTime(t0.Add(expectationsTTL + time.Second)) + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("expired expectation must be treated satisfied (TTL failsafe)") + } +} diff --git a/pkg/controller/v1beta1/workload/types/input.go b/pkg/controller/v1beta1/workload/types/input.go new file mode 100644 index 000000000..26d714669 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/input.go @@ -0,0 +1,343 @@ +package types + +import ( + "context" + "errors" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + k8stypes "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/clock" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// ReconcileInput carries everything a workload reconcile needs as +// plain data. Callers populate this struct from their source-of-truth +// types; workload code reads it and never reaches back. The boundary +// lets one workload package serve both the ISVC and IR control +// planes without sharing a type system. +type ReconcileInput struct { + // OwnerObject is the K8s object whose UID/Kind/APIVersion becomes + // the controllerOwner on every emitted pod / revision / service / + // PodGroup / PDB. + OwnerObject client.Object + + // OwnerGVK is the GroupVersionKind stamped on emitted pods' + // OwnerReference. Passed alongside OwnerObject because + // controller-runtime strips TypeMeta during deserialization, so + // the workload package cannot reliably derive the GVK. + OwnerGVK schema.GroupVersionKind + + // EventTarget is the object emitted events are stamped against. + // Usually OwnerObject; the IR caller may set this to the parent + // ISVC instead so user-facing event streams stay coherent. Nil + // falls back to OwnerObject. + EventTarget client.Object + + // LedgerOwner owns the migration audit-ledger ConfigMap (load + + // persist + its controller OwnerReference). Usually OwnerObject; the + // IR caller sets it to the parent ISVC so the migration ledger lives + // on the same user-facing resource as the operator's migration-request + // annotation (the IR owns the pods; the ISVC owns the migration audit + // trail). Nil falls back to OwnerObject. LedgerOwnerGVK is its GVK. + LedgerOwner client.Object + LedgerOwnerGVK schema.GroupVersionKind + + // Key is the workload-owned identity. Adapters compose it from the + // owner-CRD shape; workload code reads it opaquely. + Key Key + + // DesiredSpec is the per-reconcile projection of the source spec + // the workload pipeline drives toward. + DesiredSpec WorkloadDesiredSpec + + // ObservedState is the per-reconcile snapshot of the source status + // subtree. Read-only from workload code; status writes go through + // MutateInstance. + ObservedState WorkloadObservedState + + // MutateInstance applies the mutate callback to the idx-th + // InstanceStatus entry. Callers wrap persistence (apiserver round- + // trip under retry.RetryOnConflict, in-memory mirror) inside the + // closure. The mutate callback returns true when a real change was + // made; false short-circuits the status write. + // + // MUST be set when ObservedState carries any Instance entries — + // workload code panics on a nil callback rather than silently + // dropping status updates. + MutateInstance func(ctx context.Context, idx int32, mutate func(*InstanceStatus) bool) error + + // ApplyInstanceMutations applies a batch of InstanceMutations in ONE + // status write: one fresh read, every mutation applied to its slot, + // one persist, retried on conflict as a whole. Only mutations whose + // Mutate reports a change are persisted; a batch with no changes + // writes nothing. Write-ahead mutations may be batched only when the + // complete batch is persisted before any corresponding external effect. + // + // Optional: nil falls back to one MutateInstance call per mutation. + ApplyInstanceMutations func(ctx context.Context, muts []InstanceMutation) error + + // ApplyInstanceMutationsWithRetryBlock atomically applies InstanceStatus + // mutations and an optional RetryBlock mutation in one owner-status write. + // The adapter re-reads once per conflict attempt, applies both mutation sets + // to that fresh snapshot, and persists either all reported changes or none. + // A nil mutateRetryBlock callback skips the RetryBlock mutation; an empty or + // all-no-op mutation set writes nothing. + // + // Optional: nil preserves the separate InstanceStatus and RetryBlock writes + // used by adapters that do not expose this stronger capability. + ApplyInstanceMutationsWithRetryBlock func(ctx context.Context, muts []InstanceMutation, targetRevision string, mutateRetryBlock func(*RetryBlock) RetryBlockDisposition) error + + // ScaleUpPodBatchSize bounds the number of missing Pods selected by one + // Create pass. Selection remains atomic at the Instance boundary, so a gang + // is either selected in full or deferred. The first eligible Instance may + // exceed a positive budget and proceeds alone. A nil pointer preserves the + // unbounded compatibility behavior. A non-nil zero value fails closed. + ScaleUpPodBatchSize *int32 + + // ScaleDownPodBatchSize bounds active delete work in Pod-equivalent units. + // Selection remains atomic at the Instance boundary, so a gang is either + // selected in full or deferred. The first eligible Instance may exceed a + // positive budget and proceeds alone. A nil pointer preserves unbounded + // candidate selection. A non-nil zero value fails closed. + ScaleDownPodBatchSize *int32 + + // ScaleDownRequeueInterval is the configured poll cadence while a delete + // wave is waiting on drain or resource disappearance. Zero disables cadence + // polling; watched resources and exact force-delete deadlines still wake it. + ScaleDownRequeueInterval time.Duration + + // AuthoritativePods is a live Component-wide Pod observation shared by + // every destructive consumer in one reconcile pass. Nil means the caller + // did not preload an observation; a non-nil snapshot is authoritative even + // when Pods and ByInstance are empty. + AuthoritativePods *ComponentPodSnapshot + + // FinalizeInstanceResources removes optional per-Instance resources after + // the authoritative Pod snapshot is empty. It reports complete only after + // those resources are authoritatively absent. + FinalizeInstanceResources func(ctx context.Context, idx int32) (complete bool, err error) + + // RemoveInstance drops the InstanceStatus entry for idx and + // returns (true, nil) when a real removal happened, (false, nil) + // when already absent. Compatibility lifecycle paths use this seam + // when they remove one status slot outside an atomic batch. + // + // It must be set whenever a caller can enter one of those paths; nil + // fails closed rather than silently leaking a status entry. + RemoveInstance func(ctx context.Context, idx int32) (bool, error) + + // WriteAggregateCondition merges cond into the owner's per-Component + // condition list so workload code can stamp top-level Component- + // scoped conditions (today only GangSchedulingUnavailable) without + // reaching into the owner-CRD typed status. + // + // MUST be set on every constructed ReconcileInput — nil panics. + // No-op adapters wire an explicit + // `func(_ context.Context, _ metav1.Condition) error { return nil }`. + WriteAggregateCondition func(ctx context.Context, cond metav1.Condition) error + + // WarnInstanceFailed emits a Warning event against EventTarget + // reporting that the (idx) Instance escalated to Phase=Failed. + // + // MUST be set on every constructed ReconcileInput — nil panics. + // No-op adapters wire `func(_ int32, _, _ string) {}`. + WarnInstanceFailed func(idx int32, podName, reason string) + + // WarnRetryHeld emits the operator-facing Warning when a revision's + // retry budget exhausts (spec: emitted at the Held transition, once). + // Nil-safe: unset means no event. + WarnRetryHeld func(targetRevision string, attempts int32, reason string) + + // MutateMigration reads-modifies-writes the owner's persisted + // MigrationRecord for requestUUID (status.migrations). mutate + // receives the existing record and returns true when a real change + // was made; false short-circuits the status write. A missing record + // (trimmed, or the owner was recreated) is a clean no-op — the + // callback is not invoked. + // + // MUST be set when ObservedState.Migrations carries any records — + // the Migrate executor errors on a nil callback rather than + // silently dropping phase advancement. + MutateMigration func(ctx context.Context, requestUUID string, mutate func(*MigrationRecord) bool) error + + // AppendMigration appends a NEW MigrationRecord to the owner's + // persisted status.migrations. Idempotent: a record with the same + // RequestUUID already present writes nothing. Deliberately separate + // from MutateMigration — mutate-on-missing stays a no-op (a stamper + // must never resurrect a trimmed record as a phantom), so record + // CREATION gets its own seam. Optional: nil disables workload-side + // record creation; callers treat the write as a best-effort mirror. + AppendMigration func(ctx context.Context, rec MigrationRecord) error + + // UpdateGate, when non-nil, is called per Instance in the Update + // pass before starting a fresh Update operation. Lets the adapter + // inject cross-Component coordination gates (ratio-balanced + // pacing, surge / unavailability budgets, sequential rollout + // ordering) that workload code MUST NOT know about. + // + // allowed=false skips this Instance for this reconcile pass; the + // dispatcher emits a short requeue. inFlightSurge / inFlightUnavail + // are the dispatcher's within-pass counters so the gate can + // project against the post-this-pass shape. + // + // Nil is treated as always-allowed. + UpdateGate func(strategy UpdateStrategyType, inFlightSurge, inFlightUnavail int32) (allowed bool, denyReason string) + + // MutateRetryBlock reads-modifies-writes the owner's persisted + // RetryBlock for targetRevision. mutate receives the existing block + // (or a zero block with TargetRevision set) and returns whether to + // persist, remove, or leave it. Nil closure disables retry-block + // WRITES only; the update-trigger gate still honors blocks present + // in ObservedState.RetryBlocks. + MutateRetryBlock func(ctx context.Context, targetRevision string, mutate func(*RetryBlock) RetryBlockDisposition) error + + // UpdateRetryPolicy bounds automatic same-target update retries. + // nil = unconfigured → fail-safe: first failure Holds. + UpdateRetryPolicy *RetryPolicy + + // ForceDelete gates the stuck-Terminating force-delete escalation. + // nil = unconfigured → the escalation is disabled entirely (does not + // exist); when non-nil both durations are > 0 — config validation + // guarantees it, consumers never re-check. + ForceDelete *ForceDeletePolicy + + // StuckPodGrace is the wait window before the terminal-failure + // escalation pass fast-fails an Instance on a pod parked in a + // terminal kubelet waiting state. Operator config + // (lifecycle.stuckPodGracePeriod); zero or negative disables fast + // escalation (the InstanceReadyTimeout backstop still fires). + StuckPodGrace time.Duration + + // Disposition carries the operator-config inputs the terminal-failure + // disposition (DisposeExpiredAttempt) branches on. The zero value + // fails safe (no relocation, terminal branch for non-workload-caused + // failures). MigrationMode is plan-derived: the escalation pass + // overlays ComponentPlan.MigrationMode, so adapters leave it zero. + Disposition DispositionDeps + + // Teardown marks this reconcile as owner-deletion teardown. When + // set, the dispatcher treats the planned index set as empty — every + // observed Instance is a scale-down extra and runs the scale-down batch + // pipeline (drain gate, graceful delete, stuck-Terminating + // force-delete escalation, audit) — and runs NOTHING else: no + // Paused gate, no Restart / Migrate / Update / Create. The caller + // owns completion detection (live component pod list) and all + // finalizer decisions. + Teardown bool + + // Clock mirrors Deps.Clock for helpers that receive only the input. + // See Deps.Now for the rule. + Clock clock.Clock +} + +// ErrStatusOwnerGone reports that the object owning a requested atomic status +// transition disappeared before the transition could commit. Callers that +// guard an external effect must treat it as an aborted write, even though +// ordinary idempotent status writers may translate it to a successful no-op. +var ErrStatusOwnerGone = errors.New("status owner is gone") + +// ErrStatusMutationPrecondition reports that an authoritative owner/status +// snapshot does not match the decision that produced a mutation batch. +// The complete batch is rejected and callers must replan before effects. +var ErrStatusMutationPrecondition = errors.New("status mutation precondition failed") + +// ComponentPodSnapshot is one authoritative Component Pod LIST represented +// both as the complete set and as per-Instance buckets. The two views share +// Pod pointers and are immutable for the lifetime of a reconcile pass. +type ComponentPodSnapshot struct { + Pods []*corev1.Pod + ByInstance map[int32][]*corev1.Pod +} + +// InstanceMutationSnapshot is the authoritative owner/status view presented +// to a batch precondition on every conflict attempt. +type InstanceMutationSnapshot struct { + OwnerUID k8stypes.UID + OwnerGeneration int64 + Instances map[int32]InstanceStatus +} + +// InstanceMutation is one buffered InstanceStatus mutation, keyed by Instance +// index. An upsert sets Mutate; a removal sets Remove and leaves Mutate nil. +// Precondition can reject a mutation against the fresh status snapshot. +// OnCommit receives isolated copies of the exact before/after values only after +// the containing status write commits; nil identifies an absent side. A +// mutation with OnCommit requires its index to appear only once in the batch. +type InstanceMutation struct { + Index int32 + Mutate func(*InstanceStatus) bool + Remove bool + Precondition func(*InstanceStatus) bool + // BatchPrecondition guards the complete mutation set. When any batch + // precondition rejects the fresh snapshot, no mutation in the set is + // applied. Callers normally attach one shared guard to the first mutation. + BatchPrecondition func(InstanceMutationSnapshot) bool + // Postcondition identifies this mutation's committed representation after + // an ambiguous status-update response. Every mutation in a confirmable + // batch supplies one; removals are confirmed by authoritative absence. + Postcondition func(*InstanceStatus) bool + OnCommit func(previous, current *InstanceStatus) +} + +// DispositionDeps carries the operator-config inputs and adapter hooks +// the disposition branches on. Resolved once per reconcile by the +// adapter. +type DispositionDeps struct { + // AutoMigrateMaxAttempts bounds the relocation branch per + // (component, instance) against the audit ledger's AutoRecover + // entry count. <= 0 (unconfigured) disables relocation entirely. + AutoMigrateMaxAttempts int32 + // MigrationMode is the effective migration disposition from the + // plan (MigrationModeOrDefault). Only Auto (and its Surge spelling + // alias) enables relocation; the zero value fails safe to the + // terminal branch. + MigrationMode MigrationMode + // PodSpec / WorkerPodSpec are the Component's desired pod templates + // (leader/single-pod and worker role). The relocation branch + // consults their REQUIRED node affinity before recording a + // directive: when excluding the suspect node (plus the instance's + // already-recorded exclusions) would leave a template unschedulable + // — e.g. a required In[node] pin on the wedged node — the attempt + // disposes terminal instead of recording an unsatisfiable exclusion + // that would leave the rebuild permanently Pending. Nil skips the + // guard. + PodSpec *corev1.PodSpec + WorkerPodSpec *corev1.PodSpec + // OnRelocationDirective, when non-nil, is invoked once per recorded + // relocation directive with the component name — the adapter's + // metrics hook (the IR adapter wires the auto-migration counter). + OnRelocationDirective func(component string) +} + +// ForceDeletePolicy configures the stuck-Terminating force-delete +// escalation: a Terminating pod overdue past its own deletion deadline +// by OverdueSlack, on a node whose unreachable evidence is at least +// NodeUnreachableThreshold old, may be force-deleted. Config-driven +// (chart values → inferenceservice-config); nil means unconfigured and +// the escalation is disabled entirely. Both durations are always > 0 +// when the policy is non-nil (config validation rejects anything else). +type ForceDeletePolicy struct { + // OverdueSlack is how long past the pod's own DeletionTimestamp + // (which already includes the pod's own grace period) a Terminating + // pod must be before it counts as wedged. + OverdueSlack time.Duration + // NodeUnreachableThreshold is the minimum age of the node's + // unreachable evidence (taint TimeAdded / NotReady + // LastTransitionTime, or the Node object gone) before the + // escalation may act. + NodeUnreachableThreshold time.Duration +} + +// Now returns the injected clock's time, or time.Now() when no clock +// is wired. Lifecycle code holding a Deps or ReconcileInput should read +// time through Now, not time.Now(); subpackages without a seam +// (audit, podreadiness, gang) still read real time. +func (r *ReconcileInput) Now() time.Time { + if r.Clock != nil { + return r.Clock.Now() + } + return time.Now() +} diff --git a/pkg/controller/v1beta1/workload/types/migration.go b/pkg/controller/v1beta1/workload/types/migration.go new file mode 100644 index 000000000..ab19b6ef6 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/migration.go @@ -0,0 +1,165 @@ +package types + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Workload-side mirror of the InferenceReplica MigrationStatus entry. +// The adapter converts field-for-field (like RetryBlock / InstanceStatus); +// workload code never sees the CRD type. status.migrations on the owner +// is the single source of truth for migration work: the dispatcher +// selects work from non-terminal Manual records and the Migrate executor +// resumes from the record's SurgeInstance + Phase. + +// MigrationTrigger identifies who initiated a migration record. +type MigrationTrigger string + +const ( + // MigrationTriggerManual marks an operator-requested migration — + // a resumable process born Accepted. + MigrationTriggerManual MigrationTrigger = "Manual" + // MigrationTriggerAuto marks a controller-initiated relocation — + // a born-terminal Relocated record, never resumable work. + MigrationTriggerAuto MigrationTrigger = "Auto" +) + +// MigrationPhase is the lifecycle phase of a migration record. Manual +// records walk Accepted -> SurgePending -> SurgeReady -> Draining -> +// Completed | Failed; Auto records are born terminal (Relocated). +type MigrationPhase string + +const ( + MigrationPhaseAccepted MigrationPhase = "Accepted" + MigrationPhaseSurgePending MigrationPhase = "SurgePending" + MigrationPhaseSurgeReady MigrationPhase = "SurgeReady" + MigrationPhaseDraining MigrationPhase = "Draining" + MigrationPhaseCompleted MigrationPhase = "Completed" + MigrationPhaseFailed MigrationPhase = "Failed" + MigrationPhaseRelocated MigrationPhase = "Relocated" +) + +// Terminal reports whether p is a terminal phase. Executors and the +// dispatcher select work on non-terminal phase only — terminal records +// are records, never work. +func (p MigrationPhase) Terminal() bool { + switch p { + case MigrationPhaseCompleted, MigrationPhaseFailed, MigrationPhaseRelocated: + return true + } + return false +} + +// migrationPhaseRank orders the Manual phase chain for forward-only +// advancement. Terminal phases rank above every transient phase. An +// unrecognized phase has no rank and reports ok=false. +func migrationPhaseRank(p MigrationPhase) (int, bool) { + switch p { + case MigrationPhaseAccepted: + return 0, true + case MigrationPhaseSurgePending: + return 1, true + case MigrationPhaseSurgeReady: + return 2, true + case MigrationPhaseDraining: + return 3, true + case MigrationPhaseCompleted, MigrationPhaseFailed, MigrationPhaseRelocated: + return 4, true + default: + return 0, false + } +} + +// MigrationPhaseAtOrPast reports whether p has already reached (or +// passed) the given phase in the Manual chain — the guard the executor's +// forward-only phase advancement uses so a stale write can never move a +// record backward. +// +// An unrecognized phase on either side reports false. Neither answer is +// knowable for a phase outside the chain, and false is the recoverable +// one: it lets the executor drive the record forward, where true would +// report every advancement as already done and wedge it permanently. +func MigrationPhaseAtOrPast(p, target MigrationPhase) bool { + pRank, pOK := migrationPhaseRank(p) + targetRank, targetOK := migrationPhaseRank(target) + if !pOK || !targetOK { + return false + } + return pRank >= targetRank +} + +// MigrationRecord mirrors v1beta1.MigrationStatus field-for-field. +type MigrationRecord struct { + // RequestUUID uniquely identifies the migration request. + RequestUUID string + + Trigger MigrationTrigger + + // SourceInstance is the Instance index being migrated away from. + SourceInstance int32 + + // SurgeInstance is the allocated surge Instance index; nil until + // the executor allocates it (0 is a valid surge index). + SurgeInstance *int32 + + // AllocatedAt is when the surge index was allocated — execution + // start. Nil while the record is queued (Accepted, not yet picked + // up). Capacity counts execution from this stamp. + AllocatedAt *metav1.Time + + // FromNode is the node the source is being moved off. + FromNode string + + // HintTargetNodes are preferred placement targets for the surge. + HintTargetNodes []string + + Phase MigrationPhase + + // Attempt is the relocation attempt ordinal (Auto records). + Attempt int32 + + // Reason is the requester-supplied reason (Manual) or disposition + // branch (Auto). + Reason string + + // Message describes the current blocker (non-terminal) or the + // terminal outcome. + Message string + + StartedAt metav1.Time + + // Deadline is when a non-terminal record expires. + Deadline metav1.Time + + CompletedAt *metav1.Time + + Succeeded *bool +} + +// FindMigrationRecord returns a pointer to the record for requestUUID +// (aliasing the slice element), or nil. +func FindMigrationRecord(records []MigrationRecord, requestUUID string) *MigrationRecord { + for i := range records { + if records[i].RequestUUID == requestUUID { + return &records[i] + } + } + return nil +} + +// NextManualMigration selects the migration the dispatcher should drive +// this pass: the oldest-StartedAt Manual record whose phase is +// non-terminal. Auto records are excluded structurally — born terminal, +// they never rank. Returns nil when no work exists. +func NextManualMigration(records []MigrationRecord) *MigrationRecord { + var picked *MigrationRecord + for i := range records { + r := &records[i] + if r.Trigger != MigrationTriggerManual || r.Phase.Terminal() { + continue + } + if picked == nil || r.StartedAt.Time.Before(picked.StartedAt.Time) { + picked = r + } + } + return picked +} diff --git a/pkg/controller/v1beta1/workload/types/plan.go b/pkg/controller/v1beta1/workload/types/plan.go new file mode 100644 index 000000000..0a2d24525 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/plan.go @@ -0,0 +1,178 @@ +package types + +import ( + "time" +) + +// ComponentPlan is the desired Component → Instance → Runner → Pod +// shape computed each reconcile from desired + observed state. Not +// persisted; policy fields hold the effective values after defaults. +type ComponentPlan struct { + // Component identifies which of router / engine / decoder this plan + // describes. Typed as the workload-side ComponentType so the + // workload package stays free of v1beta1 imports; adapters convert + // from v1beta1.ComponentType at the boundary. + Component ComponentType + + // Replicas is the desired number of Instances for this Component + // (= MinReplicas, or 1 when MinReplicas is unset). + Replicas int32 + + // Instances enumerates the desired Instances. Normally one entry + // per index 0..Replicas-1; indices may be sparse while a surge + // migration is in flight (the surge Instance carries an + // out-of-band index until promotion settles the plan). + Instances []InstancePlan + + // RestartPolicy is the effective Instance-group restart policy. + RestartPolicy RestartPolicy + + // UpdateStrategy is the effective rollout policy across Instances. + UpdateStrategy UpdateStrategy + + // ReadyPolicy is the effective Instance-level readiness aggregation. + ReadyPolicy InstanceReadyPolicy + + // InstanceReadyTimeout is the wait ceiling on a newly-created + // Instance becoming Ready. + InstanceReadyTimeout time.Duration + + // MigrationMode is the effective migration disposition (auto / surge / + // never). + MigrationMode MigrationMode + + // Paused stops the dispatcher before it starts or advances Restart, + // Migration, Update, or Create operations. Scale-down remains active so a + // reduced desired replica count can still release capacity while paused. + Paused bool + + // TopologyKey is the resolved gang co-location node-label key for + // this Component (e.g. an NVLink/RDMA fabric-domain label). Empty for + // single-pod Components or when unset. When non-empty on a multi-pod + // Component, Render auto-generates the per-Instance worker→leader + // podAffinity and the PodGroup carries the same key for topology-aware + // gang schedulers. Both constraints therefore select the same configured + // topology domain. + TopologyKey string + + // InstanceTopologyKeys holds a temporary per-Instance override while live + // pods still carry affinity rendered from an older revision. Missing pods in + // that active gang must use the same immutable topology contract; fresh + // Instances and surge indices fall back to TopologyKey. + InstanceTopologyKeys map[int32]string +} + +// TopologyKeyForInstance returns the live-safe topology contract for an +// Instance, falling back to the Component's current desired key when the group +// has no active revision-specific override. +func (p ComponentPlan) TopologyKeyForInstance(index int32) string { + if key, ok := p.InstanceTopologyKeys[index]; ok { + return key + } + return p.TopologyKey +} + +// InstancePlan is the desired state for one Instance — the atomic +// unit of gang scheduling, restart, and migration. +type InstancePlan struct { + // Index is the stable Instance ordinal. Sparse after surge + // migration. + Index int32 + + // Incarnation is the monotonic lifecycle token for this + // (Component, Index) pair. Initial create stamps 1; full recreate + // / restart-on-loss bumps it; in-place update does not. Distinguishes + // old pod materializations from current ones when stable names get + // reused after a recreate. + Incarnation int64 + + // Runners enumerates the Runners that constitute this Instance. + // Single-pod has one "default" Runner of size 1; multi-pod has + // leader + worker. + Runners []RunnerPlan + + // MigrationOverlay carries placement hints from an in-flight surge + // migration. Render injects hard anti-affinity from the source node + // and preferred affinity to hint target nodes. Nil for non-migration + // paths. + MigrationOverlay *MigrationOverlay + + // ExcludedNodes lists nodes this Instance's pods must not land on — + // the relocation-directive memory (AutoRecover ledger entries) + // projected through WorkloadObservedState.ExcludedNodesByInstance. + // Render materializes each entry as the same required NotIn + // hostname term the migration overlay uses. Empty for normal + // Instances. + ExcludedNodes []string + + // PeerHostnames is an optional render-time cache of the Instance's + // ordered peer-DNS host list (one entry per pod, flat gang-rank + // order). The list is identical for every pod in the gang, so the + // gang-render loop computes it once and stashes it here to avoid the + // O(gangsize^2) per-pod recompute. Nil when unset; Render then + // derives the list from Runners as before. Transient render scratch, + // not part of the desired state and never persisted. + PeerHostnames []string +} + +// MigrationOverlay records the per-Instance placement constraints a +// surge migration carries onto the rendered pod. Materialized as pod +// affinity in Render and intentionally absent from the canonical +// revision payload — transient operation overlay, not a steady-state +// template field. +type MigrationOverlay struct { + // FromNode surfaces as RequiredDuringScheduling anti-affinity on + // kubernetes.io/hostname. + FromNode string + // HintTargetNodes surface as PreferredDuringScheduling affinity; + // the scheduler is free to pick another node when these are + // unavailable. + HintTargetNodes []string +} + +// RunnerPlan is the desired state for one Runner within an Instance. +type RunnerPlan struct { + // Name is "leader", "worker", or "default" (single-pod). + Name string + // Size is the number of pods of this Runner within the Instance. + Size int32 +} + +// TotalPods returns the total desired pod count across all Runners — +// 1 for single-pod, 1+N for leader+worker. +func (i InstancePlan) TotalPods() int32 { + var n int32 + for _, r := range i.Runners { + n += r.Size + } + return n +} + +// AllocateSurgeIndex returns the lowest int32 not present in any of +// the InstanceStatuses — the slot the migration op picks for its +1 +// surge pod without colliding with steady-state indices or other +// in-flight surges. +func AllocateSurgeIndex(instances []InstanceStatus) int32 { + used := make(map[int32]struct{}, len(instances)*2) + for _, s := range instances { + used[s.Index] = struct{}{} + // Also exclude in-flight surge slots. A source mid-surge records its + // replacement's index in Operation.SurgeIndex BEFORE the replacement + // Instance exists (it's created a pass later, and within a single + // reconcile pass a sibling that just claimed a slot has it recorded + // here but not yet as a real Index). Without excluding these, every + // Instance surging in the same pass collides on the same lowest-free + // index; when that single shared surge becomes Ready the drain-on-ready + // logic releases ALL sharing sources at once — a full-fleet wipe. + if s.Operation != nil && s.Operation.SurgeIndex != nil { + used[*s.Operation.SurgeIndex] = struct{}{} + } + } + // Lowest-free exists at or below |used| (distinct ints), so an unbounded + // scan terminates; keep it unbounded so the 2x-entry set can't off-by-one. + for i := int32(0); ; i++ { + if _, taken := used[i]; !taken { + return i + } + } +} diff --git a/pkg/controller/v1beta1/workload/types/retryblock.go b/pkg/controller/v1beta1/workload/types/retryblock.go new file mode 100644 index 000000000..b496b2146 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/retryblock.go @@ -0,0 +1,78 @@ +package types + +import ( + "math" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// Workload-side mirror of the InferenceReplica RetryBlock status. The +// adapter converts field-for-field; workload code never sees the CRD +// type. Revision-scoped: one block per target revision. +type RetryBlockState string + +const ( + RetryBlockBackoff RetryBlockState = "Backoff" + RetryBlockHeld RetryBlockState = "Held" + RetryBlockRetryInProgress RetryBlockState = "RetryInProgress" +) + +type RetryBlock struct { + TargetRevision string + State RetryBlockState + AttemptsStarted int32 + NextRetryAt *metav1.Time + FirstFailureAt *metav1.Time + LastFailureAt *metav1.Time + Reason string +} + +// RetryBlockDisposition is the MutateRetryBlock callback's verdict. +type RetryBlockDisposition int + +const ( + RetryBlockUnchanged RetryBlockDisposition = iota // no write + RetryBlockPersist // upsert the mutated block + RetryBlockRemove // delete the block (success prune) +) + +// RetryPolicy bounds automatic same-target update retries. Config-driven +// (chart values → inferenceservice-config); nil means unconfigured and +// fails safe: Exhausted is always true, so the first failure Holds. +type RetryPolicy struct { + MaxAttempts int32 + InitialDelay time.Duration + MaxDelay time.Duration + Multiplier float64 +} + +// NextRetryDelay returns the backoff before attempt attemptsStarted+1: +// InitialDelay * Multiplier^(attemptsStarted-1), capped at MaxDelay. +func (p *RetryPolicy) NextRetryDelay(attemptsStarted int32) time.Duration { + d := time.Duration(float64(p.InitialDelay) * math.Pow(p.Multiplier, float64(attemptsStarted-1))) + if d > p.MaxDelay || d <= 0 { + return p.MaxDelay + } + return d +} + +// Exhausted reports whether no automatic attempts remain. A nil policy +// (unconfigured) is always exhausted — fail-safe Held. +func (p *RetryPolicy) Exhausted(attemptsStarted int32) bool { + if p == nil { + return true + } + return attemptsStarted >= p.MaxAttempts +} + +// FindRetryBlock returns a pointer to the block for targetRevision +// (aliasing the slice element), or nil. +func FindRetryBlock(blocks []RetryBlock, targetRevision string) *RetryBlock { + for i := range blocks { + if blocks[i].TargetRevision == targetRevision { + return &blocks[i] + } + } + return nil +} diff --git a/pkg/controller/v1beta1/workload/types/retryblock_test.go b/pkg/controller/v1beta1/workload/types/retryblock_test.go new file mode 100644 index 000000000..468708a74 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/retryblock_test.go @@ -0,0 +1,53 @@ +package types + +import ( + "testing" + "time" +) + +func TestNextRetryDelay(t *testing.T) { + p := &RetryPolicy{MaxAttempts: 3, InitialDelay: time.Minute, MaxDelay: 30 * time.Minute, Multiplier: 2.0} + cases := []struct { + attempts int32 + want time.Duration + }{ + {1, time.Minute}, // first failure → initial + {2, 2 * time.Minute}, // initial * m^1 + {3, 4 * time.Minute}, // initial * m^2 + {10, 30 * time.Minute}, // capped at MaxDelay + } + for _, c := range cases { + if got := p.NextRetryDelay(c.attempts); got != c.want { + t.Errorf("NextRetryDelay(%d) = %v, want %v", c.attempts, got, c.want) + } + } +} + +func TestRetryPolicyExhausted(t *testing.T) { + p := &RetryPolicy{MaxAttempts: 3, InitialDelay: time.Minute, MaxDelay: 30 * time.Minute, Multiplier: 2.0} + if p.Exhausted(2) { + t.Error("2 < 3 attempts must not be exhausted") + } + if !p.Exhausted(3) { + t.Error("3 >= 3 attempts must be exhausted") + } + var nilP *RetryPolicy + if !nilP.Exhausted(0) { + t.Error("nil policy (unconfigured) is always exhausted → fail-safe Held") + } +} + +func TestFindRetryBlock(t *testing.T) { + blocks := []RetryBlock{{TargetRevision: "a"}, {TargetRevision: "b"}} + if got := FindRetryBlock(blocks, "b"); got == nil || got.TargetRevision != "b" { + t.Errorf("FindRetryBlock(b) = %v", got) + } + if got := FindRetryBlock(blocks, "c"); got != nil { + t.Errorf("FindRetryBlock(missing) must be nil, got %v", got) + } + // Returned pointer aliases the slice element (callers mutate in place). + FindRetryBlock(blocks, "a").AttemptsStarted = 7 + if blocks[0].AttemptsStarted != 7 { + t.Error("FindRetryBlock must return a pointer into the slice") + } +} diff --git a/pkg/controller/v1beta1/workload/types/retryblock_writer.go b/pkg/controller/v1beta1/workload/types/retryblock_writer.go new file mode 100644 index 000000000..8ff1fb3e0 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/retryblock_writer.go @@ -0,0 +1,72 @@ +package types + +import ( + "context" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// RecordUpdateFailureInRetryBlock upserts the RetryBlock for targetRev +// after a terminal same-target attempt failure (failed update rollout, +// deadline-disposed create/update attempt). Wave counting: an existing +// Backoff block means this wave already recorded — refresh evidence +// only. Policy nil (unconfigured) or exhausted → Held; else Backoff +// with persisted NextRetryAt. No-op when the adapter did not wire +// MutateRetryBlock. Callers hold the writer-ordering invariant: the +// failed attempt's Operation is cleared in the same transition (block +// write first — a crash between the two re-enters the caller's failed +// branch, where the wave dedup refreshes without recounting). +// +// Lives in the leaf types package so both workload/ops (gang abandon) +// and the workload-root disposition share ONE implementation without +// closing the workload → workload/ops import cycle. +func RecordUpdateFailureInRetryBlock(ctx context.Context, input ReconcileInput, targetRev, reason string) error { + if input.MutateRetryBlock == nil || targetRev == "" { + return nil + } + now := metav1.NewTime(input.Now()) + // heldAttempts captures the Held transition inside the mutate; the + // warning is emitted only after the write COMMITS so RMW conflict + // retries cannot duplicate the event. + var heldAttempts int32 + err := input.MutateRetryBlock(ctx, targetRev, func(b *RetryBlock) RetryBlockDisposition { + var disposition RetryBlockDisposition + disposition, heldAttempts = ApplyUpdateFailureToRetryBlock(b, input.UpdateRetryPolicy, now, reason) + return disposition + }) + if err == nil && heldAttempts > 0 && input.WarnRetryHeld != nil { + input.WarnRetryHeld(targetRev, heldAttempts, reason) + } + return err +} + +// ApplyUpdateFailureToRetryBlock applies one terminal failure wave to a +// RetryBlock value. It is pure apart from mutating b, so callers can compose +// the transition into a larger atomic owner-status update. heldAttempts is +// non-zero only for a new transition into Held. +func ApplyUpdateFailureToRetryBlock(b *RetryBlock, policy *RetryPolicy, now metav1.Time, reason string) (RetryBlockDisposition, int32) { + if b == nil { + return RetryBlockUnchanged, 0 + } + if b.FirstFailureAt == nil { + b.FirstFailureAt = &now + } + b.LastFailureAt = &now + b.Reason = reason + switch b.State { + case RetryBlockBackoff, RetryBlockHeld: + // This wave is already recorded or terminally held; refresh only the + // failure evidence. + return RetryBlockPersist, 0 + } + b.AttemptsStarted++ + if policy.Exhausted(b.AttemptsStarted) { + b.State = RetryBlockHeld + b.NextRetryAt = nil + return RetryBlockPersist, b.AttemptsStarted + } + b.State = RetryBlockBackoff + next := metav1.NewTime(now.Add(policy.NextRetryDelay(b.AttemptsStarted))) + b.NextRetryAt = &next + return RetryBlockPersist, 0 +} diff --git a/pkg/controller/v1beta1/workload/types/service.go b/pkg/controller/v1beta1/workload/types/service.go new file mode 100644 index 000000000..9d74e36b7 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/service.go @@ -0,0 +1,62 @@ +// service.go declares the typed input the workload package consumes +// when rendering per-Component supporting Services. Splitting the spec +// type out of the renderer keeps `workload/services.go` free of any +// owner-CRD coupling: adapters (the ISVC OMENative dispatcher and the +// InferenceReplica controller) populate PerComponentServiceSpec from +// their respective owner shapes and hand it to +// `workload.ReconcileHeadlessService`. +// +// Why this lives here and not under `workload/`: the parent workload +// package's Service helpers in services.go must depend on this type, +// and several of the other typed inputs (Deps, ReconcileInput, plan, +// ...) already live in workload/types/. Co-locating the Service input +// alongside them keeps the workload package boundary clean — every +// data carrier the workload renderer reads is in the same subpackage, +// and the parent workload package's helpers reach for them via a +// single `workload/types` import. +package types + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// PerComponentServiceSpec is the typed input the workload package +// consumes to render a per-Component supporting Service (today: the +// headless Service that gives every pod a stable peer-DNS FQDN). +// Adapters populate every field from their owner-CRD shape; workload +// code reads the spec opaquely and never reaches back into the owner. +// +// One Service per (workload-key, component). The ISVC adapter +// constructs one spec per Component reconcile pass; the IR adapter +// constructs one spec per InferenceReplica reconcile. +type PerComponentServiceSpec struct { + // Name is the Service object name. The workload renderer stamps it + // verbatim but does NOT compute it — adapters pass the canonical + // name (e.g., query.HeadlessServiceName(owner.Name, component) on + // the ISVC side) so the per-Component naming convention stays in + // the adapter where the owner-CRD details live. + Name string + + // Namespace is the Service object namespace. Matches the owner CR + // namespace; adapters set it from owner.GetNamespace(). + Namespace string + + // Selector is the pod-label selector scoping which pods are members + // of this Service. Adapters set it so the Service only selects pods + // the workload owns (not RawDeployment pods that may share the bare + // Component label). The ISVC adapter wires + // `ome.io/inferenceservice + component + managed-by=OMENative`; the + // IR adapter wires the IR-side equivalent. + Selector map[string]string + + // Labels are stamped on the Service object metadata block for + // downstream tooling (`kubectl get svc -l`, dashboards). Typically + // the same key/value pairs as Selector plus the controller's + // `managed-by` label. + Labels map[string]string + + // OwnerReferences point back to the workload's owner CR. Adapters + // pass `*metav1.NewControllerRef(owner, ownerGVK)` so deletion of + // the owner cascades to the Service via the K8s garbage collector. + OwnerReferences []metav1.OwnerReference +} diff --git a/pkg/controller/v1beta1/workload/types/source.go b/pkg/controller/v1beta1/workload/types/source.go new file mode 100644 index 000000000..02d02e54d --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/source.go @@ -0,0 +1,207 @@ +package types + +import ( + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" +) + +// WorkloadDesiredSpec.Lifecycle and WorkloadAggregateStatus.Traffic +// use workload-owned mirror types. Adapters project the CRD-shape +// values into these structs at the boundary. + +// WorkloadDesiredSpec is the per-reconcile projection of the source +// spec the workload pipeline drives toward. Read-only from workload +// code. +type WorkloadDesiredSpec struct { + // Replicas is the desired Instance count. + Replicas int32 + + // MinReadySeconds is the minimum age (in seconds) for an Instance + // to count as Available after becoming Ready. + // + // Inert: the adapter that projects the InferenceReplica spec fills + // this in, but no rollout-engine path reads it. Availability comes + // from EndpointSlice membership, not a readiness-age gate, so the + // spec field it carries has no effect on rollout pacing, and the + // AvailablePodCount doc in types.go ("ready for at least + // MinReadySeconds") describes the intent rather than the behaviour. + // Honoring it means teaching availability to consult it, not + // changing what sets it. + MinReadySeconds int32 + + // Runners enumerates the Runner roles (default | leader | worker) + // and per-role pod counts that constitute one Instance. + Runners []Runner + + // PodSpec is the rendered leader / single-pod template the + // renderer drives toward. May be nil when Replicas=0 (nothing to + // render). + PodSpec *corev1.PodSpec + + // WorkerPodSpec is the rendered worker template for multi-pod + // Instances. Nil for single-pod Components. + WorkerPodSpec *corev1.PodSpec + + // PodTemplateObjectMeta is the rendered per-Component metadata + // (labels, annotations, owner refs) the renderer stamps onto each + // emitted pod. + PodTemplateObjectMeta *metav1.ObjectMeta + + // MultiPod indicates the workload uses Leader + Worker Runners + // (each Instance materializes more than one pod). + MultiPod bool + + // TopologyKey is the resolved gang co-location node-label key for + // this Component (e.g. an NVLink/RDMA fabric-domain label). When non-empty on + // a multi-pod Component, the renderer auto-generates the per-Instance + // worker→leader podAffinity and stamps the same key on the PodGroup for + // topology-aware gang schedulers. Empty means no controller-generated + // topology constraint. Adapters project it from the owner-CRD component + // spec at the boundary. + TopologyKey string + + // Lifecycle holds RestartPolicy / UpdateStrategy / ReadyPolicy / + // InstanceReadyTimeout / MigrationPolicy as workload-owned mirror + // values; adapters project the CRD-shape lifecycle into this struct + // at the boundary. + Lifecycle Lifecycle + + // Pacing is the projected rollout pacing for this reconcile. Nil + // means "no pacing constraint" (treated as allowed). + Pacing *WorkloadPacing + + // Paused, when true, stops the reconciler from starting any fresh + // Update / Restart / Create operations. In-flight operations + // resume on unpause. + Paused bool + + // GangSchedulingAvailable is true when the scheduler-plugins + // PodGroup CRD was discovered at controller startup. Drives + // whether podgroup.EnsurePodGroup runs for multi-pod Instances. + GangSchedulingAvailable bool +} + +// WorkloadObservedState is the per-reconcile snapshot of the source +// status subtree the reconciler reasons about. +type WorkloadObservedState struct { + // ObservedGeneration is the spec generation the last status flush + // reflects. + ObservedGeneration int64 + + // CollisionCount is the ControllerRevision-hash salt. nil on the + // first reconcile. + CollisionCount *int32 + + // CurrentRevision names the ControllerRevision currently serving + // traffic. + CurrentRevision string + + // UpdateRevision names the ControllerRevision being rolled out. + UpdateRevision string + + // InstanceStatuses reports per-Instance state. Read-only here; + // writes go through ReconcileInput.MutateInstance. + InstanceStatuses []InstanceStatus + + // Conditions reports component-level conditions. + Conditions []metav1.Condition + + // RetryBlocks mirrors the owner's persisted per-revision retry + // authority. Read-only from workload code; writes go through + // ReconcileInput.MutateRetryBlock. + RetryBlocks []RetryBlock + + // Migrations mirrors the owner's persisted migration records + // (status.migrations) — the single source of truth for migration + // work. Read-only from workload code; writes go through + // ReconcileInput.MutateMigration. + Migrations []MigrationRecord + + // ExcludedNodesByInstance maps Instance index → nodes its rebuild + // must avoid, projected per reconcile by the adapter from the + // audit ledger's AutoRecover relocation directives (bounded to the + // operator's autoMigrate.maxAttempts most recent entries). BuildPlan + // copies it onto InstancePlan.ExcludedNodes; Render materializes it + // as a required NodeAffinity NotIn overlay. Nil / missing index = + // no exclusion (zero change for normal instances). + ExcludedNodesByInstance map[int32][]string +} + +// WorkloadAggregateStatus is the per-reconcile flush of counters, +// conditions, and traffic that the reconciler hands off to +// Source.WriteAggregateStatus. +type WorkloadAggregateStatus struct { + ObservedGeneration int64 + Replicas int32 + ReadyReplicas int32 + ServingReplicas int32 + AvailableReplicas int32 + UpdatedReplicas int32 + UpdatedReadyReplicas int32 + CurrentRevision string + UpdateRevision string + LabelSelector string + Conditions []metav1.Condition + Traffic []ComponentTrafficTarget +} + +// WorkloadPacing is the projected rollout pacing. +// Adapters compute it once per reconcile. +// +// Inert: no rollout-engine path reads Partition, MaxUnavailable or +// Decisions. The adapter that projects the InferenceReplica spec fills +// Partition and MaxUnavailable in, but pacing actually flows through +// RollingUpdate.Partition and the UpdateGate callback, and availability +// comes from EndpointSlice membership — so the two spec fields they +// carry have no effect. Decisions is never constructed outside tests. +// The rollback-to-revision half of the same spec stanza is live; it +// takes a different path and does not pass through here. +type WorkloadPacing struct { + // Partition holds back updates for Instances whose index is less + // than Partition. 0 (the default) updates all Instances. Used + // for canary holds. + // + // NOT READ by the engine — see the WorkloadPacing type note above. + Partition *int32 + + // MaxUnavailable caps in-rollout disruption. nil falls back to + // the reconciler default (25%). + // + // NOT READ by the engine — see the WorkloadPacing type note above. + MaxUnavailable *intstr.IntOrString + + // Decisions is the projected per-gate allow/deny map for this + // reconcile. The ISVC adapter fills it from coordination gates; + // the IR adapter leaves it nil. nil means allowed. + Decisions *PacingDecisions +} + +// PacingDecisions is the workload-facing projection of cross-Component +// coordination gates. Plain data — adapters compute it outside the +// workload package. +type PacingDecisions struct { + // UpdateAllowed is false when the reconciler MUST NOT start any + // fresh Update operations this reconcile. + UpdateAllowed bool + + // UpdateDenyReason is the operator-visible reason recorded onto + // events / conditions when UpdateAllowed is false. + UpdateDenyReason string + + // SurgeBudgetRemaining is the upper bound on in-flight surge pods + // the reconciler may keep alive concurrently. The dispatcher + // tracks its own in-flight count within a pass; this is a + // per-reconcile ceiling. + SurgeBudgetRemaining int32 +} + +// Runner is a named role within an Instance with a per-Instance pod +// count. Single-pod has one "default" Runner of Size=1; multi-pod has +// "leader" (Size=1) + "worker" (Size=N). +type Runner struct { + // Name is "leader" | "worker" | "default". + Name string + // Size is the per-Instance pod count for this Runner. + Size int32 +} diff --git a/pkg/controller/v1beta1/workload/types/termination.go b/pkg/controller/v1beta1/workload/types/termination.go new file mode 100644 index 000000000..d119e4f7b --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/termination.go @@ -0,0 +1,197 @@ +package types + +import ( + "fmt" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// PodTermination extracts the most operator-relevant container failure +// diagnostics from pod into an *InstanceTermination, stamping `now` as the +// record time. Returns nil when pod is nil and when pod carries no +// failure signal at all — a healthy pod has nothing to report. +// +// Selection precedence (the failure operators most want to see first): +// +// 1. A container terminated with a non-zero exit code — the canonical +// crash signal (OOMKilled exit 137, generic Error exit 1, ...). Both +// the live State and the LastTerminationState are consulted, since a +// CrashLoopBackOff pod shows the crash in LastTerminationState while +// its live State is Waiting. +// 2. A container stuck in a terminal waiting state (CrashLoopBackOff, +// ImagePullBackOff, CreateContainerError, ...) — the wedge signal the +// stuck-pod escalator fires on. ExitCode is left nil (no process ran). +// 3. Any terminated container at all, once the pod phase is Failed +// (exit 0 but the pod is Failed — rare, e.g. a restartPolicy=Never +// sidecar exiting). The phase guard matters: an exit-0 container is +// the normal state of a completed init container, so an unguarded +// scan reports a termination for every healthy pod that has one. +// 4. Pod-level fallback: Reason="PodFailed" with the pod's status message, +// when the pod phase is Failed but no per-container detail survived. +// +// Init containers are consulted with the same precedence after regular +// containers so an init-stage crash is still captured. +func PodTermination(pod *corev1.Pod, now metav1.Time) *InstanceTermination { + if pod == nil { + return nil + } + + allStatuses := func() []corev1.ContainerStatus { + out := make([]corev1.ContainerStatus, 0, len(pod.Status.ContainerStatuses)+len(pod.Status.InitContainerStatuses)) + out = append(out, pod.Status.ContainerStatuses...) + out = append(out, pod.Status.InitContainerStatuses...) + return out + }() + + // 1. Non-zero terminated exit code (live or last-termination). + for _, cs := range allStatuses { + if t := nonZeroTerminated(cs); t != nil { + return terminationFromTerminated(pod, cs.Name, t, now) + } + } + // 2. Terminal waiting-state wedge. + for _, cs := range allStatuses { + if cs.State.Waiting != nil && isTerminalWaitingReason(cs.State.Waiting.Reason) { + return &InstanceTermination{ + PodName: pod.Name, + ContainerName: cs.Name, + Reason: cs.State.Waiting.Reason, + Message: cs.State.Waiting.Message, + Time: now, + } + } + } + // 3. Any terminated container, only once the pod itself has Failed. + // A clean exit is ordinary in a healthy pod — every completed init + // container is one — so without the phase guard a Running pod would + // report a termination it never suffered. + if pod.Status.Phase == corev1.PodFailed { + for _, cs := range allStatuses { + if cs.State.Terminated != nil { + return terminationFromTerminated(pod, cs.Name, cs.State.Terminated, now) + } + if cs.LastTerminationState.Terminated != nil { + return terminationFromTerminated(pod, cs.Name, cs.LastTerminationState.Terminated, now) + } + } + } + // 4. Pod-level fallback. + if pod.Status.Phase == corev1.PodFailed { + return &InstanceTermination{ + PodName: pod.Name, + Reason: "PodFailed", + Message: pod.Status.Message, + Time: now, + } + } + return nil +} + +// PodTerminationWithReason is PodTermination with an explicit reason +// override used by the stuck-pod escalator, which has already classified +// the wedge reason from the live waiting state. When PodTermination can't +// extract any per-container detail (e.g. the status snapshot raced the +// kubelet write), the override still gives operators the classified reason +// + pod name rather than an empty record. When PodTermination DID extract a +// record but with an empty Reason, the override fills it in. +func PodTerminationWithReason(pod *corev1.Pod, reason string, now metav1.Time) *InstanceTermination { + t := PodTermination(pod, now) + if t == nil { + name := "" + if pod != nil { + name = pod.Name + } + return &InstanceTermination{PodName: name, Reason: reason, Time: now} + } + if t.Reason == "" { + t.Reason = reason + } + return t +} + +// nonZeroTerminated returns the terminated state (live or last) carrying a +// non-zero exit code, or nil. Live State wins over LastTerminationState so +// the freshest crash is reported. +func nonZeroTerminated(cs corev1.ContainerStatus) *corev1.ContainerStateTerminated { + if cs.State.Terminated != nil && cs.State.Terminated.ExitCode != 0 { + return cs.State.Terminated + } + if cs.LastTerminationState.Terminated != nil && cs.LastTerminationState.Terminated.ExitCode != 0 { + return cs.LastTerminationState.Terminated + } + return nil +} + +// terminationFromTerminated builds an InstanceTermination from a terminated +// container state. Reason falls back to "Error" when kubelet left it blank +// (it occasionally does for OOM races) so the record is never reason-less. +func terminationFromTerminated(pod *corev1.Pod, container string, t *corev1.ContainerStateTerminated, now metav1.Time) *InstanceTermination { + reason := t.Reason + if reason == "" { + reason = "Error" + } + exit := t.ExitCode + return &InstanceTermination{ + PodName: pod.Name, + ContainerName: container, + Reason: reason, + ExitCode: &exit, + Message: t.Message, + Time: now, + } +} + +// isTerminalWaitingReason mirrors the workload escalator's +// terminalPullFailureReasons set. Duplicated here (rather than imported +// from the root workload package) to keep workload/types free of an import +// edge back to its parent — the set is small and changes rarely. +func isTerminalWaitingReason(reason string) bool { + switch reason { + case "ErrImagePull", + "ImagePullBackOff", + "InvalidImageName", + "CreateContainerConfigError", + "CreateContainerError", + "CrashLoopBackOff", + "RunContainerError": + return true + } + return false +} + +// ShortString renders an InstanceTermination as a compact, grep-friendly +// fragment for K8s event messages, e.g. `pod foo-0 container main failed +// (OOMKilled, exit 137)` or `pod foo-0 container main stuck +// (ImagePullBackOff)`. Returns "" for a nil receiver so callers can embed +// it unconditionally. +func (t *InstanceTermination) ShortString() string { + if t == nil { + return "" + } + var b string + if t.PodName != "" { + b = "pod " + t.PodName + } else { + b = "pod " + } + if t.ContainerName != "" { + b += " container " + t.ContainerName + } + switch { + case t.ExitCode != nil: + b += fmt.Sprintf(" failed (%s, exit %d)", reasonOrUnknown(t.Reason), *t.ExitCode) + case t.Reason == "PodFailed": + b += " failed (PodFailed)" + default: + b += fmt.Sprintf(" stuck (%s)", reasonOrUnknown(t.Reason)) + } + return b +} + +func reasonOrUnknown(reason string) string { + if reason == "" { + return "unknown" + } + return reason +} diff --git a/pkg/controller/v1beta1/workload/types/types.go b/pkg/controller/v1beta1/workload/types/types.go new file mode 100644 index 000000000..e96db7464 --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/types.go @@ -0,0 +1,407 @@ +// Package types holds the workload package's value types — the +// per-Instance / per-Component / per-Plan data structures every layer +// of the workload pipeline reads and writes. Lives separately from +// the parent `workload` package so `workload/ops` and +// `workload.Reconcile` can both depend on these types without closing +// an import cycle. +// +// The workload package owns its own status / operation / component / +// phase / key types and does NOT depend on the OME CRD API package. +// Every type here mirrors the corresponding v1beta1 status type +// field-for-field; the v1beta1convert helpers +// (`controller/v1beta1/v1beta1convert`) perform the conversion, and +// the InferenceReplica adapter (inferencereplica) wires those +// converters into its reconcile loop. +// +// External callers reference these types as `workload.X` — the parent +// `workload` package re-exports them as Go aliases. +package types + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" +) + +// ComponentType identifies router | engine | decoder. String-mirrors +// the CRD ComponentType so the adapter conversion is a verbatim cast; +// the distinct Go type catches accidental cross-package coupling. +type ComponentType string + +const ( + ComponentRouter ComponentType = "router" + ComponentEngine ComponentType = "engine" + ComponentDecoder ComponentType = "decoder" +) + +// InstancePhase is the per-Instance lifecycle phase. String-mirrors +// the CRD OMENativeInstancePhase. +type InstancePhase string + +const ( + // InstancePhaseEmpty is the zero value used before the first + // observation lands on a freshly-allocated InstanceStatus. + InstancePhaseEmpty InstancePhase = "" + InstancePhasePending InstancePhase = "Pending" + InstancePhaseCreating InstancePhase = "Creating" + InstancePhaseReady InstancePhase = "Ready" + InstancePhaseUpdating InstancePhase = "Updating" + InstancePhaseRestarting InstancePhase = "Restarting" + InstancePhaseMigrating InstancePhase = "Migrating" + InstancePhaseFailed InstancePhase = "Failed" + InstancePhaseDeleting InstancePhase = "Deleting" +) + +// UpdateStrategyType is the rollout-mechanism selector. String-mirrors +// the CRD UpdateStrategyType; the distinct Go type catches accidental +// drift between workload-side dispatch logic and the v1beta1 enum. +type UpdateStrategyType string + +const ( + // UpdateStrategySurgeThenDrain creates a new pod for the target + // revision, waits for it to serve, then drains + deletes the old. + UpdateStrategySurgeThenDrain UpdateStrategyType = "SurgeThenDrain" + // UpdateStrategyRecreatePod always deletes pods and recreates them + // at a bumped Incarnation. + UpdateStrategyRecreatePod UpdateStrategyType = "RecreatePod" + // UpdateStrategyInPlaceIfPossible tries image-patch in place; falls + // through to recreate on non-image-only diffs. + UpdateStrategyInPlaceIfPossible UpdateStrategyType = "InPlaceIfPossible" + // UpdateStrategyInPlaceOnly image-patches in place when eligible; + // errors otherwise. + UpdateStrategyInPlaceOnly UpdateStrategyType = "InPlaceOnly" +) + +// RestartPolicy is the Instance-group restart policy. String-mirrors +// the CRD InstanceRestartPolicy. +type RestartPolicy string + +const ( + // RestartPolicyNone restarts only the failed pod; OMENative leaves + // the Instance Phase=Ready and lets the kubelet recover in place. + RestartPolicyNone RestartPolicy = "None" + // RestartPolicyRecreateInstance drains, deletes, and recreates + // every pod in the Instance at a bumped Incarnation. + RestartPolicyRecreateInstance RestartPolicy = "RecreateInstanceOnPodRestart" +) + +// MigrationMode is the migration disposition. String-mirrors the CRD +// MigrationPolicyMode. +type MigrationMode string + +const ( + // MigrationModeAuto / MigrationModeSurge — operator may trigger + // migration via the per-UUID annotation; controller drives the + // surge cycle. Surge is a spelling alias for Auto. + MigrationModeAuto MigrationMode = "Auto" + MigrationModeSurge MigrationMode = "Surge" + // MigrationModeNever — controller refuses any migration; + // annotations are ignored. + MigrationModeNever MigrationMode = "Never" +) + +// InstanceOperationType identifies the kind of an in-flight +// destructive operation against one Instance. String-mirrors the CRD +// InstanceOperationType. +type InstanceOperationType string + +const ( + InstanceOperationCreate InstanceOperationType = "Create" + InstanceOperationUpdate InstanceOperationType = "Update" + InstanceOperationRestart InstanceOperationType = "Restart" + InstanceOperationMigrate InstanceOperationType = "Migrate" + InstanceOperationDelete InstanceOperationType = "Delete" +) + +// UpdateStepGangSurgeTarget is the InstanceOperation.Step marking the +// surge-target (replacement) Instance of an in-flight multi-pod (gang) +// SurgeThenDrain update. Lives here in the leaf types package so both the +// plan (root workload) and the ops dispatcher reference one canonical +// value. The gang SOURCE uses Step="Surge" so the surge budget counts it. +const UpdateStepGangSurgeTarget = "GangSurgeTarget" + +// UpdateStepGangSurgeTargetCleanup marks a replacement gang whose terminal +// cleanup owns its Pods and PodGroup until the marker status is removed. +const UpdateStepGangSurgeTargetCleanup = "GangSurgeTargetCleanup" + +// InstanceOperation is the durable recovery anchor written before any +// destructive action against an Instance and cleared on completion. +// Mirrors the CRD InstanceOperation field-for-field. +type InstanceOperation struct { + // ID is the idempotency key. Retries with the same ID are a no-op. + ID string + + Type InstanceOperationType + + // Step is the fine-grained resume point within the operation + // (e.g., Drain, DeletePods, WaitZero, Recreate, WaitReady). + Step string + + StartedAt metav1.Time + // LastProgressAt is when the operation last advanced. Used for + // stall detection. + LastProgressAt metav1.Time + // Deadline is the hard timeout for the current Step. + Deadline metav1.Time + // RetryCount is the per-step escalation counter. + RetryCount int32 + + // TargetRevision is the ControllerRevision the operation is + // converging toward. + TargetRevision string + + Reason string + + // Migrate-only fields. The Operation is a pin: SurgeIndex correlates + // the source/surge pair and RequestUUID keys the authoritative + // status.migrations record. + SurgeIndex *int32 + RequestUUID string + + // FromNode / HintTargetNodes are inert: migration facts live on the + // owner's status.migrations record, so nothing writes or reads these. + // They exist so the CRD InstanceOperation mirror round-trips values + // already stamped on an object unchanged. Do not add writers. + FromNode string + HintTargetNodes []string +} + +// InstanceStatus is the per-Instance status. Mirrors the CRD +// OMENativeInstanceStatus field-for-field; adapters round-trip between +// this type and the CRD type. +type InstanceStatus struct { + // Index is the stable Instance ordinal. Values may be sparse + // after surge migration. + Index int32 + + // Incarnation increments each time the Instance is recreated + // (full recreate update, restart-on-loss). In-place updates do + // not increment. + Incarnation int64 + + Phase InstancePhase + + // RunningRevision / TargetRevision are the ControllerRevision the + // live pods are running and converging toward, respectively. + RunningRevision string + TargetRevision string + + PodCount int32 + // ReadyPodCount counts pods reporting Ready=True. + ReadyPodCount int32 + // ServingPodCount counts pods that are BOTH ContainersReady AND + // have serving=True on the controller-managed readiness gate. + ServingPodCount int32 + // AvailablePodCount counts pods ready for at least MinReadySeconds. + AvailablePodCount int32 + // ScheduledPodCount counts pods with Spec.NodeName set. + ScheduledPodCount int32 + // Admitted reports that the Instance has pods and none remain behind an + // admission scheduling gate. + Admitted bool + + // NodesOccupied is the set of node names hosting pods of this + // Instance. + NodesOccupied []string + + Conditions []metav1.Condition + + // Operation is the durable record of an in-flight destructive + // action against this Instance. Set before the action starts, + // cleared after it completes. + Operation *InstanceOperation + + // ActiveOrdinal identifies which of two pod-naming slots (0 or 1) + // currently holds the canonical pod for this single-pod Instance. + ActiveOrdinal int32 + + // LastFailure preserves the container-termination diagnostics of the + // pod whose failure (or stuck-pod escalation) last triggered a + // recreate of this Instance. Survives the drain+recreate that deletes + // the failing pod. Mirrors the CRD OMENativeInstanceStatus.LastFailure. + LastFailure *InstanceTermination +} + +// InstanceTermination captures the container-termination diagnostics of a +// pod that failed or was escalated. Mirrors the CRD InstanceTermination +// field-for-field; the v1beta1convert helpers round-trip it at the adapter +// boundary. +type InstanceTermination struct { + // PodName is the name of the pod that failed or was escalated. + PodName string + + // ContainerName is the container the diagnostics were read from. + // Empty when only a pod-level signal was available. + ContainerName string + + // Reason is the kubelet terminated- or waiting-state reason + // (OOMKilled, Error, CrashLoopBackOff, ImagePullBackOff, ...), or + // "PodFailed" when only the pod phase was available. + Reason string + + // ExitCode is the container exit code when the signal came from a + // terminated state; nil for a stuck waiting-state signal. + ExitCode *int32 + + // Message is the kubelet-supplied detail message, if any. + Message string + + // Time is when OMENative recorded this termination. + Time metav1.Time +} + +// Key uniquely identifies one logical workload (a per-Component slice +// of an owner). Adapters populate every field; workload code reads +// them opaquely. +type Key struct { + Namespace string + Component ComponentType + + // OwnerName is the bare name of the workload's owner object + // (ISVC.Name or IR.Name). workload composes pod / Service names + // from this — never reaches back into OwnerObject for the field. + OwnerName string + + // OwnerLabels is the seed label set stamped on every emitted + // ControllerRevision / PodGroup / pod's metadata labels block. + // Read-only from workload code. + OwnerLabels map[string]string + + // SelectorLabels is the seed label set used in every pod LIST + // selector and per-revision Service selector. Read-only from + // workload code. + SelectorLabels map[string]string +} + +// WorkloadName is the composed identity used for ControllerRevision +// naming and as the prefix on emitted pod / PodGroup / PDB names. +// Always "-" — the pod-naming convention every +// adapter shares so existing selectors keep matching. +func (k Key) WorkloadName() string { + return k.OwnerName + "-" + string(k.Component) +} + +// Lifecycle is the workload-owned mirror of the v1beta1 LifecycleSpec. +// Adapters project the CRD-shape lifecycle into this struct so the +// workload package can read it without importing v1beta1. Fields +// match the CRD field-for-field; nil pointers signal "unset, default +// at planning time". +type Lifecycle struct { + // RestartPolicy is the per-Instance restart disposition. nil falls + // back to the per-shape default (multi-pod → RecreateInstance, + // single-pod → None) in BuildPlan. + RestartPolicy *RestartPolicy + + // UpdateStrategy controls how OMENative rolls template changes + // across the Component's Instances. nil falls back to defaults. + UpdateStrategy *UpdateStrategy + + // ReadyPolicy controls how Instance-level readiness is aggregated + // from the underlying pods. nil falls back to the per-shape default. + ReadyPolicy *InstanceReadyPolicy + + // InstanceReadyTimeout is the wait ceiling on a newly-created + // Instance becoming Ready. nil falls back to the 30m default. + InstanceReadyTimeout *metav1.Duration + + // MigrationPolicy controls whether and how OMENative honors a + // migration request annotation for this Component. nil falls back + // to the Auto default. + MigrationPolicy *MigrationPolicy +} + +// UpdateStrategy is the workload-owned mirror of the v1beta1 +// UpdateStrategy struct. Adapters project the CRD-shape strategy onto +// this struct so the workload package can read it without importing +// v1beta1. +type UpdateStrategy struct { + // Type selects the rollout mechanism. Empty defaults to + // SurgeThenDrain in BuildPlan. + Type UpdateStrategyType + + // RollingUpdate paces the rollout across Instances of the + // Component. + RollingUpdate *RollingUpdate + + // InPlaceUpdateStrategy tunes lifecycle drain timing. Its grace period + // also provides the post-unroute connection-settle window before a + // SurgeThenDrain source pod is deleted. + InPlaceUpdateStrategy *InPlaceUpdateStrategy +} + +// RollingUpdate paces rollout across Instances of one Component. +// Workload-owned mirror of v1beta1.RollingUpdate. +type RollingUpdate struct { + // Partition holds back updates for Instances whose index is < + // Partition. 0 (the default) updates all Instances. + Partition *int32 + + // MaxUnavailable is the maximum number of Instances (or percent + // expression) allowed to be in a non-Ready state during the rollout. + // nil falls back to the reconciler default. Mirrors + // v1beta1.RollingUpdate.MaxUnavailable. + MaxUnavailable *intstr.IntOrString + + // MaxSurge is the maximum number of extra Instances (or percent + // expression) the rollout may create above the Component's desired + // replica count during a rolling update. nil falls back to the + // reconciler default. Mirrors v1beta1.RollingUpdate.MaxSurge. + MaxSurge *intstr.IntOrString +} + +// InPlaceUpdateStrategy tunes the per-pod in-place update sequence. +// Workload-owned mirror of v1beta1.InPlaceUpdateStrategy. +type InPlaceUpdateStrategy struct { + // GracePeriodSeconds is the time OMENative waits between marking + // the pod not-ready and applying an in-place mutation. SurgeThenDrain + // also waits this long after EndpointSlice removal before deletion so + // persistent load-balancer connections can drain while workers live. + GracePeriodSeconds *int32 + + // MarkNotReadyDuringLifecycle, when true, flips ome.io/serving=False + // on the pod before the in-place mutation so EndpointSlice drains + // traffic first. + MarkNotReadyDuringLifecycle *bool +} + +// InstanceReadyPolicy controls how Instance-level readiness is +// aggregated from the underlying pods. Workload-owned mirror of the +// v1beta1.InstanceReadyPolicy enum. +type InstanceReadyPolicy string + +const ( + // InstanceReadyPolicyAllPodReady reports the Instance Ready only + // when every pod has Ready=True. Default for multi-pod Instances. + InstanceReadyPolicyAllPodReady InstanceReadyPolicy = "AllPodReady" + // InstanceReadyPolicyNone disables Instance-level aggregation; pods + // are reported individually. Default for single-pod Instances. + InstanceReadyPolicyNone InstanceReadyPolicy = "None" +) + +// MigrationPolicy controls whether and how OMENative honors a +// migration request annotation for this Component. Workload-owned +// mirror of v1beta1.MigrationPolicy. +type MigrationPolicy struct { + // Mode selects the migration disposition. Empty defaults to Auto. + Mode MigrationMode +} + +// ComponentTrafficTarget describes the percentage of traffic routed to +// one revision of one Component. Workload-owned mirror of +// v1beta1.ComponentTrafficTarget; serialized field-for-field by the +// adapter when projecting onto ISVC status. +type ComponentTrafficTarget struct { + // RevisionName is the per-revision Service name for this target. + RevisionName string + + // Percent is the percentage of traffic the consumer should route to + // RevisionName. Sum across all entries for one Component is 100. + Percent int32 + + // Tag is an optional short identifier for the target (e.g. + // "latest", "prev"). Cosmetic; not used by the consumer. + Tag string + + // LatestRevision is true when this entry corresponds to the + // LatestRolledoutRevision for the Component. + LatestRevision bool +} diff --git a/pkg/controller/v1beta1/workload/types/types_test.go b/pkg/controller/v1beta1/workload/types/types_test.go new file mode 100644 index 000000000..b3c4cc83b --- /dev/null +++ b/pkg/controller/v1beta1/workload/types/types_test.go @@ -0,0 +1,367 @@ +package types + +import ( + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestExpectations_SatisfiedWhenEmpty(t *testing.T) { + e := NewExpectations() + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("empty cache should be satisfied") + } +} + +func TestExpectations_BlocksUntilObserved(t *testing.T) { + e := NewExpectations() + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 2) + + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("should NOT be satisfied with 2 pending adds") + } + + e.ObservedCreate("ns", "isvc", ComponentEngine, 0) + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("should NOT be satisfied with 1 pending add still") + } + + e.ObservedCreate("ns", "isvc", ComponentEngine, 0) + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("should be satisfied after observing both creates") + } +} + +func TestExpectations_DeadlineForcesSatisfied(t *testing.T) { + e := NewExpectations() + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 1) + // Manually expire the entry. + e.mu.Lock() + for _, v := range e.entries { + v.Deadline = time.Now().Add(-1 * time.Second) + } + e.mu.Unlock() + + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("expired entry should be reported as satisfied") + } +} + +func TestExpectations_ScopedByKey(t *testing.T) { + e := NewExpectations() + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 1) + // Different instance index — should be unaffected. + if !e.Satisfied("ns", "isvc", ComponentEngine, 1) { + t.Fatal("instance 1 should still be satisfied") + } + // Different component — should be unaffected. + if !e.Satisfied("ns", "isvc", ComponentDecoder, 0) { + t.Fatal("decoder instance 0 should still be satisfied") + } +} + +func TestExpectations_Forget(t *testing.T) { + e := NewExpectations() + e.ExpectCreates("ns", "isvc", ComponentEngine, 0, 3) + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("should not be satisfied before Forget") + } + e.Forget("ns", "isvc", ComponentEngine, 0) + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("Forget should clear the entry") + } +} + +func TestExpectations_DeletesTracked(t *testing.T) { + e := NewExpectations() + e.ExpectDeletes("ns", "isvc", ComponentEngine, 0, 1) + if e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("should not be satisfied with pending delete") + } + e.ObservedDelete("ns", "isvc", ComponentEngine, 0) + if !e.Satisfied("ns", "isvc", ComponentEngine, 0) { + t.Fatal("delete observation should satisfy") + } +} + +func int32p(v int32) *int32 { return &v } + +func podWith(name string, phase corev1.PodPhase, css ...corev1.ContainerStatus) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Status: corev1.PodStatus{Phase: phase, ContainerStatuses: css}, + } +} + +func TestPodTermination_NilPod(t *testing.T) { + if got := PodTermination(nil, metav1.Now()); got != nil { + t.Fatalf("PodTermination(nil) = %+v, want nil", got) + } +} + +// A non-zero terminated exit code is the highest-precedence signal — the +// canonical crash the operator most wants to see (OOMKilled, exit 137). +func TestPodTermination_NonZeroTerminatedWins(t *testing.T) { + pod := podWith("engine-0", corev1.PodFailed, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "OOMKilled", + ExitCode: 137, + Message: "out of memory", + }}, + }) + got := PodTermination(pod, metav1.Now()) + if got == nil { + t.Fatal("PodTermination = nil, want a record") + } + if got.PodName != "engine-0" || got.ContainerName != "main" || got.Reason != "OOMKilled" { + t.Errorf("identity: %+v", got) + } + if got.ExitCode == nil || *got.ExitCode != 137 { + t.Errorf("ExitCode: got %v want 137", got.ExitCode) + } + if got.Message != "out of memory" { + t.Errorf("Message: got %q", got.Message) + } + if got.ShortString() != "pod engine-0 container main failed (OOMKilled, exit 137)" { + t.Errorf("ShortString: %q", got.ShortString()) + } +} + +// CrashLoopBackOff surfaces the crash in LastTerminationState while the +// live State is Waiting — the extractor must read LastTerminationState's +// non-zero exit code in preference to the bare waiting reason. +func TestPodTermination_CrashLoopReadsLastTerminationState(t *testing.T) { + pod := podWith("engine-0", corev1.PodRunning, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{Reason: "CrashLoopBackOff"}}, + LastTerminationState: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "Error", + ExitCode: 1, + }}, + }) + got := PodTermination(pod, metav1.Now()) + if got == nil || got.Reason != "Error" || got.ExitCode == nil || *got.ExitCode != 1 { + t.Fatalf("want Error/exit 1 from LastTerminationState, got %+v", got) + } +} + +// A terminal waiting state with no terminated history (ImagePullBackOff) +// yields a reason with a nil ExitCode and a "stuck" ShortString. +func TestPodTermination_TerminalWaitingNoExitCode(t *testing.T) { + pod := podWith("engine-0", corev1.PodPending, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{Reason: "ImagePullBackOff", Message: "back-off pulling"}}, + }) + got := PodTermination(pod, metav1.Now()) + if got == nil { + t.Fatal("want a record") + } + if got.Reason != "ImagePullBackOff" || got.ExitCode != nil { + t.Errorf("want ImagePullBackOff/nil exit, got %+v", got) + } + if got.ShortString() != "pod engine-0 container main stuck (ImagePullBackOff)" { + t.Errorf("ShortString: %q", got.ShortString()) + } +} + +// A non-terminal waiting reason (ContainerCreating) is not a failure +// signal; with no other signal and a non-Failed phase, return nil. +func TestPodTermination_TransientWaitingIgnored(t *testing.T) { + pod := podWith("engine-0", corev1.PodPending, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Waiting: &corev1.ContainerStateWaiting{Reason: "ContainerCreating"}}, + }) + if got := PodTermination(pod, metav1.Now()); got != nil { + t.Fatalf("ContainerCreating must not produce a record, got %+v", got) + } +} + +// Pod phase Failed with no per-container detail falls back to a +// pod-level PodFailed record. +func TestPodTermination_PodLevelFallback(t *testing.T) { + pod := podWith("engine-0", corev1.PodFailed) + pod.Status.Message = "node shutdown" + got := PodTermination(pod, metav1.Now()) + if got == nil || got.Reason != "PodFailed" || got.ContainerName != "" { + t.Fatalf("want pod-level PodFailed, got %+v", got) + } + if got.Message != "node shutdown" { + t.Errorf("Message: got %q", got.Message) + } + if got.ShortString() != "pod engine-0 failed (PodFailed)" { + t.Errorf("ShortString: %q", got.ShortString()) + } +} + +// A Running pod with no termination signal yields nil (no false-positive +// capture during healthy operation). +func TestPodTermination_HealthyRunningNil(t *testing.T) { + pod := podWith("engine-0", corev1.PodRunning, corev1.ContainerStatus{ + Name: "main", + Ready: true, + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + }) + if got := PodTermination(pod, metav1.Now()); got != nil { + t.Fatalf("healthy running pod must yield nil, got %+v", got) + } +} + +// A healthy Running pod whose init container completed cleanly must +// still yield nil. Exit 0 is the normal resting state of every +// completed init container, so an unguarded exit-0 scan would report a +// termination for the majority of healthy pods and stamp a bogus +// LastFailure on the Instance. +func TestPodTermination_CompletedInitContainerNotAFailure(t *testing.T) { + pod := podWith("engine-0", corev1.PodRunning, corev1.ContainerStatus{ + Name: "main", + Ready: true, + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + }) + pod.Status.InitContainerStatuses = []corev1.ContainerStatus{{ + Name: "init-model", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "Completed", + ExitCode: 0, + }}, + }} + if got := PodTermination(pod, metav1.Now()); got != nil { + t.Fatalf("a cleanly-completed init container is not a failure, got %+v", got) + } +} + +// The exit-0 scan still fires once the pod itself has Failed — that is +// the restartPolicy=Never sidecar case it exists for. +func TestPodTermination_ExitZeroCapturedOncePodFailed(t *testing.T) { + pod := podWith("engine-0", corev1.PodFailed, corev1.ContainerStatus{ + Name: "sidecar", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "Completed", + ExitCode: 0, + }}, + }) + got := PodTermination(pod, metav1.Now()) + if got == nil || got.ContainerName != "sidecar" { + t.Fatalf("want the sidecar record, got %+v", got) + } + if got.ExitCode == nil || *got.ExitCode != 0 { + t.Errorf("want exit 0, got %+v", got.ExitCode) + } +} + +// Init-container failures are captured with the same precedence after +// regular containers. +func TestPodTermination_InitContainerCrash(t *testing.T) { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "engine-0"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + InitContainerStatuses: []corev1.ContainerStatus{{ + Name: "init-model", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ + Reason: "Error", + ExitCode: 2, + }}, + }}, + }, + } + got := PodTermination(pod, metav1.Now()) + if got == nil || got.ContainerName != "init-model" || got.ExitCode == nil || *got.ExitCode != 2 { + t.Fatalf("want init-model/exit 2, got %+v", got) + } +} + +// kubelet occasionally leaves the terminated Reason blank; the extractor +// must still produce a non-empty reason ("Error") so the record is +// never reason-less. +func TestPodTermination_BlankReasonFallsBackToError(t *testing.T) { + pod := podWith("engine-0", corev1.PodFailed, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 139}}, + }) + got := PodTermination(pod, metav1.Now()) + if got == nil || got.Reason != "Error" { + t.Fatalf("want fallback reason Error, got %+v", got) + } +} + +// PodTerminationWithReason fills a missing reason from the override but +// keeps container/exit detail when the extractor found it. +func TestPodTerminationWithReason(t *testing.T) { + // No per-container detail at all → override supplies the whole record. + bare := podWith("engine-0", corev1.PodPending) + got := PodTerminationWithReason(bare, "ImagePullBackOff", metav1.Now()) + if got == nil || got.PodName != "engine-0" || got.Reason != "ImagePullBackOff" { + t.Fatalf("override path: got %+v", got) + } + + // Extractor already produced detail with a reason → override does not + // clobber it. + rich := podWith("engine-1", corev1.PodFailed, corev1.ContainerStatus{ + Name: "main", + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{Reason: "OOMKilled", ExitCode: 137}}, + }) + got = PodTerminationWithReason(rich, "ImagePullBackOff", metav1.Now()) + if got.Reason != "OOMKilled" || got.ExitCode == nil || *got.ExitCode != 137 { + t.Fatalf("override must not clobber richer record, got %+v", got) + } +} + +func TestInstanceTermination_ShortStringNil(t *testing.T) { + var t0 *InstanceTermination + if got := t0.ShortString(); got != "" { + t.Fatalf("nil ShortString = %q, want empty", got) + } +} + +func TestInstanceTermination_ShortStringUnknownPod(t *testing.T) { + tm := &InstanceTermination{Reason: "OOMKilled", ExitCode: int32p(137)} + if got := tm.ShortString(); got != "pod failed (OOMKilled, exit 137)" { + t.Fatalf("ShortString = %q", got) + } +} + +func TestMigrationPhaseAtOrPast_ManualChain(t *testing.T) { + chain := []MigrationPhase{ + MigrationPhaseAccepted, + MigrationPhaseSurgePending, + MigrationPhaseSurgeReady, + MigrationPhaseDraining, + MigrationPhaseCompleted, + } + for i, p := range chain { + for j, target := range chain { + want := i >= j + if got := MigrationPhaseAtOrPast(p, target); got != want { + t.Errorf("MigrationPhaseAtOrPast(%s, %s) = %v, want %v", p, target, got, want) + } + } + } +} + +// An unrecognized phase must not read as "past everything". Ranking it +// at the top of the chain reports every advancement as already done and +// wedges the record permanently; ranking it at the bottom lets the +// executor drive it forward. +func TestMigrationPhaseAtOrPast_UnknownPhaseIsNotPastEverything(t *testing.T) { + for _, unknown := range []MigrationPhase{"", "Bogus"} { + if MigrationPhaseAtOrPast(unknown, MigrationPhaseAccepted) { + t.Errorf("phase %q must not report as at-or-past Accepted", unknown) + } + if MigrationPhaseAtOrPast(unknown, MigrationPhaseCompleted) { + t.Errorf("phase %q must not report as at-or-past Completed", unknown) + } + // The same holds with the unknown value as the target. + if MigrationPhaseAtOrPast(MigrationPhaseAccepted, unknown) { + t.Errorf("Accepted must not report as at-or-past target %q", unknown) + } + if MigrationPhaseAtOrPast(MigrationPhaseCompleted, unknown) { + t.Errorf("Completed must not report as at-or-past target %q", unknown) + } + if MigrationPhaseAtOrPast(unknown, unknown) { + t.Errorf("phase %q must not report as at-or-past itself", unknown) + } + } +}