diff --git a/api/v1alpha1/inferenceservice_types.go b/api/v1alpha1/inferenceservice_types.go index d873668f..626c5e28 100644 --- a/api/v1alpha1/inferenceservice_types.go +++ b/api/v1alpha1/inferenceservice_types.go @@ -561,18 +561,32 @@ type InferenceServiceSpec struct { // +optional SLO *SLOSpec `json:"slo,omitempty"` - // MaxPodLifetimeSeconds sets the maximum lifetime (in seconds) for - // inference pods. When set, the operator copies this value to - // PodSpec.ActiveDeadlineSeconds on the generated Deployment's pod - // template, causing Kubernetes to terminate the pod after the - // specified duration even if it remains healthy. This is useful for - // workloads that need periodic process recycling to release driver - // memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). - // When omitted, pods run indefinitely until manually restarted or - // the Deployment is updated. + // MaxPodLifetimeSeconds requests best-effort periodic recycling of + // deployment-backed inference pods. The controller evicts one expired pod + // at a time, using its status start time as the age reference; it is not a + // strict deadline and does not set PodSpec.ActiveDeadlineSeconds. This is + // useful for workloads that need periodic process recycling to release + // driver memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). + // Eviction respects PodDisruptionBudgets, and when rolloutPolicy.waitForIdle + // is set it waits for the backend to go idle first. With a single replica + // recycling is a restart, not a rolling replacement: expect a downtime + // window while the model reloads. When omitted, pods run indefinitely until + // manually restarted or the Deployment is updated. // +kubebuilder:validation:Minimum=1 // +optional MaxPodLifetimeSeconds *int64 `json:"maxPodLifetimeSeconds,omitempty"` + + // MaxPodLifetimeIdleTimeoutSeconds bounds how long recycling will wait for + // an idle backend before evicting anyway, measured from the moment the pod + // exceeded maxPodLifetimeSeconds. It only applies when + // rolloutPolicy.waitForIdle is set, which otherwise makes recycling wait + // indefinitely — safe for in-flight requests, but a saturated service then + // never recycles, which is exactly when leaked driver memory hurts most. + // Set 0 to recycle without waiting for idle at all. When omitted, recycling + // waits indefinitely. + // +kubebuilder:validation:Minimum=0 + // +optional + MaxPodLifetimeIdleTimeoutSeconds *int64 `json:"maxPodLifetimeIdleTimeoutSeconds,omitempty"` } // RolloutPolicySpec defines how deployment updates should be gated on backend idleness. diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index c947e4f4..08667a14 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -841,6 +841,11 @@ func (in *InferenceServiceSpec) DeepCopyInto(out *InferenceServiceSpec) { *out = new(int64) **out = **in } + if in.MaxPodLifetimeIdleTimeoutSeconds != nil { + in, out := &in.MaxPodLifetimeIdleTimeoutSeconds, &out.MaxPodLifetimeIdleTimeoutSeconds + *out = new(int64) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InferenceServiceSpec. diff --git a/charts/llmkube/templates/clusterrole.yaml b/charts/llmkube/templates/clusterrole.yaml index 4ef96f7b..9bed69ac 100644 --- a/charts/llmkube/templates/clusterrole.yaml +++ b/charts/llmkube/templates/clusterrole.yaml @@ -57,6 +57,12 @@ rules: - patch - update - watch +- apiGroups: + - "" + resources: + - pods/eviction + verbs: + - create - apiGroups: - apps resources: diff --git a/charts/llmkube/templates/crds/inferenceservices.yaml b/charts/llmkube/templates/crds/inferenceservices.yaml index 46f6c8ea..bfd4814b 100644 --- a/charts/llmkube/templates/crds/inferenceservices.yaml +++ b/charts/llmkube/templates/crds/inferenceservices.yaml @@ -3447,17 +3447,32 @@ spec: Jinja enables Jinja2 chat template rendering for tool/function calling support. Required when using the OpenAI-compatible API with tools. Maps to llama.cpp --jinja flag. type: boolean + maxPodLifetimeIdleTimeoutSeconds: + description: |- + MaxPodLifetimeIdleTimeoutSeconds bounds how long recycling will wait for + an idle backend before evicting anyway, measured from the moment the pod + exceeded maxPodLifetimeSeconds. It only applies when + rolloutPolicy.waitForIdle is set, which otherwise makes recycling wait + indefinitely — safe for in-flight requests, but a saturated service then + never recycles, which is exactly when leaked driver memory hurts most. + Set 0 to recycle without waiting for idle at all. When omitted, recycling + waits indefinitely. + format: int64 + minimum: 0 + type: integer maxPodLifetimeSeconds: description: |- - MaxPodLifetimeSeconds sets the maximum lifetime (in seconds) for - inference pods. When set, the operator copies this value to - PodSpec.ActiveDeadlineSeconds on the generated Deployment's pod - template, causing Kubernetes to terminate the pod after the - specified duration even if it remains healthy. This is useful for - workloads that need periodic process recycling to release driver - memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). - When omitted, pods run indefinitely until manually restarted or - the Deployment is updated. + MaxPodLifetimeSeconds requests best-effort periodic recycling of + deployment-backed inference pods. The controller evicts one expired pod + at a time, using its status start time as the age reference; it is not a + strict deadline and does not set PodSpec.ActiveDeadlineSeconds. This is + useful for workloads that need periodic process recycling to release + driver memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). + Eviction respects PodDisruptionBudgets, and when rolloutPolicy.waitForIdle + is set it waits for the backend to go idle first. With a single replica + recycling is a restart, not a rolling replacement: expect a downtime + window while the model reloads. When omitted, pods run indefinitely until + manually restarted or the Deployment is updated. format: int64 minimum: 1 type: integer diff --git a/config/crd/bases/inference.llmkube.dev_inferenceservices.yaml b/config/crd/bases/inference.llmkube.dev_inferenceservices.yaml index 2cbf9e2b..44686420 100644 --- a/config/crd/bases/inference.llmkube.dev_inferenceservices.yaml +++ b/config/crd/bases/inference.llmkube.dev_inferenceservices.yaml @@ -3443,17 +3443,32 @@ spec: Jinja enables Jinja2 chat template rendering for tool/function calling support. Required when using the OpenAI-compatible API with tools. Maps to llama.cpp --jinja flag. type: boolean + maxPodLifetimeIdleTimeoutSeconds: + description: |- + MaxPodLifetimeIdleTimeoutSeconds bounds how long recycling will wait for + an idle backend before evicting anyway, measured from the moment the pod + exceeded maxPodLifetimeSeconds. It only applies when + rolloutPolicy.waitForIdle is set, which otherwise makes recycling wait + indefinitely — safe for in-flight requests, but a saturated service then + never recycles, which is exactly when leaked driver memory hurts most. + Set 0 to recycle without waiting for idle at all. When omitted, recycling + waits indefinitely. + format: int64 + minimum: 0 + type: integer maxPodLifetimeSeconds: description: |- - MaxPodLifetimeSeconds sets the maximum lifetime (in seconds) for - inference pods. When set, the operator copies this value to - PodSpec.ActiveDeadlineSeconds on the generated Deployment's pod - template, causing Kubernetes to terminate the pod after the - specified duration even if it remains healthy. This is useful for - workloads that need periodic process recycling to release driver - memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). - When omitted, pods run indefinitely until manually restarted or - the Deployment is updated. + MaxPodLifetimeSeconds requests best-effort periodic recycling of + deployment-backed inference pods. The controller evicts one expired pod + at a time, using its status start time as the age reference; it is not a + strict deadline and does not set PodSpec.ActiveDeadlineSeconds. This is + useful for workloads that need periodic process recycling to release + driver memory (e.g. llama.cpp on AMD Vulkan with pinned GTT memory). + Eviction respects PodDisruptionBudgets, and when rolloutPolicy.waitForIdle + is set it waits for the backend to go idle first. With a single replica + recycling is a restart, not a rolling replacement: expect a downtime + window while the model reloads. When omitted, pods run indefinitely until + manually restarted or the Deployment is updated. format: int64 minimum: 1 type: integer diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index cb0af67b..333c7c48 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -39,6 +39,12 @@ rules: - patch - update - watch +- apiGroups: + - "" + resources: + - pods/eviction + verbs: + - create - apiGroups: - "" - events.k8s.io diff --git a/internal/controller/deployment_builder.go b/internal/controller/deployment_builder.go index 70f7187b..ecf43e38 100644 --- a/internal/controller/deployment_builder.go +++ b/internal/controller/deployment_builder.go @@ -432,16 +432,15 @@ func (r *InferenceServiceReconciler) constructDeployment( Annotations: buildPodAnnotations(isvc), }, Spec: corev1.PodSpec{ - SecurityContext: inferPodSecurityContext(isvc, r.DefaultFSGroup), - InitContainers: storageConfig.initContainers, - Containers: []corev1.Container{container}, - Volumes: storageConfig.volumes, - PriorityClassName: r.resolvePriorityClassName(isvc), - RuntimeClassName: isvc.Spec.RuntimeClassName, - ImagePullSecrets: isvc.Spec.ImagePullSecrets, - EnableServiceLinks: resolveEnableServiceLinks(backend), - ResourceClaims: modelResourceClaims(model), - ActiveDeadlineSeconds: isvc.Spec.MaxPodLifetimeSeconds, + SecurityContext: inferPodSecurityContext(isvc, r.DefaultFSGroup), + InitContainers: storageConfig.initContainers, + Containers: []corev1.Container{container}, + Volumes: storageConfig.volumes, + PriorityClassName: r.resolvePriorityClassName(isvc), + RuntimeClassName: isvc.Spec.RuntimeClassName, + ImagePullSecrets: isvc.Spec.ImagePullSecrets, + EnableServiceLinks: resolveEnableServiceLinks(backend), + ResourceClaims: modelResourceClaims(model), }, }, }, diff --git a/internal/controller/inferenceservice_controller.go b/internal/controller/inferenceservice_controller.go index be240f0f..3c7bbdfc 100644 --- a/internal/controller/inferenceservice_controller.go +++ b/internal/controller/inferenceservice_controller.go @@ -147,6 +147,7 @@ func initContainerSecurityContext(isvc *inferencev1alpha1.InferenceService) *cor // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch // +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch +// +kubebuilder:rbac:groups=core,resources=pods/eviction,verbs=create // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch;create;update;patch // +kubebuilder:rbac:groups=scheduling.k8s.io,resources=priorityclasses,verbs=get;list;watch // +kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs=get;list;watch;create;update;patch;delete @@ -248,26 +249,44 @@ func (r *InferenceServiceReconciler) Reconcile(ctx context.Context, req ctrl.Req return finalResult, statusErr } + lifetimeRequeue, err := r.reconcilePodLifetime(ctx, inferenceService, isMetal, time.Now()) + if err != nil { + return finalResult, err + } + finalResult.RequeueAfter = earliestPositive(finalResult.RequeueAfter, lifetimeRequeue) + // When a rollout is deferred pending idle, reconcileRolloutPolicy set // RolloutDeferred=True (persisted by the status update above). Drive a // recheck so the controller notices when the backend goes idle or the // idleTimeoutSeconds budget is spent. Metal-backed services never defer // (no Deployment), so this does not conflict with the metal requeue below. if cond := meta.FindStatusCondition(inferenceService.Status.Conditions, ConditionRolloutDeferred); cond != nil && cond.Status == metav1.ConditionTrue { - finalResult.RequeueAfter = inferencev1alpha1.DefaultIdleCheckInterval + finalResult.RequeueAfter = earliestPositive(finalResult.RequeueAfter, inferencev1alpha1.DefaultIdleCheckInterval) } // On the metal path a heartbeat going stale generates no watch event, so // force a periodic requeue when the Endpoints carry the annotation. if isMetal { if requeue := metalHeartbeatRequeueDuration(metalSnap); requeue > 0 { - finalResult.RequeueAfter = requeue + finalResult.RequeueAfter = earliestPositive(finalResult.RequeueAfter, requeue) } } return finalResult, nil } +// earliestPositive merges the requeue timers Reconcile collects without +// allowing a zero timer (which means "no requeue") to win a plain minimum. +func earliestPositive(values ...time.Duration) time.Duration { + var earliest time.Duration + for _, value := range values { + if value > 0 && (earliest == 0 || value < earliest) { + earliest = value + } + } + return earliest +} + func (r *InferenceServiceReconciler) getModelForInferenceService(ctx context.Context, isvc *inferencev1alpha1.InferenceService) (*inferencev1alpha1.Model, bool, *ctrl.Result, error) { log := logf.FromContext(ctx) diff --git a/internal/controller/inferenceservice_deployment_test.go b/internal/controller/inferenceservice_deployment_test.go index fff3518c..dd2298ab 100644 --- a/internal/controller/inferenceservice_deployment_test.go +++ b/internal/controller/inferenceservice_deployment_test.go @@ -594,22 +594,9 @@ var _ = Describe("Multi-GPU Deployment Construction", func() { } }) - It("should leave ActiveDeadlineSeconds nil when maxPodLifetimeSeconds is unset", func() { - isvc := &inferencev1alpha1.InferenceService{ - ObjectMeta: metav1.ObjectMeta{ - Name: "lifetime-service", - Namespace: "default", - }, - Spec: inferencev1alpha1.InferenceServiceSpec{ - ModelRef: "lifetime-model", - Image: "ghcr.io/ggml-org/llama.cpp:server", - }, - } - deployment := reconciler.constructDeployment(isvc, model, 1) - Expect(deployment.Spec.Template.Spec.ActiveDeadlineSeconds).To(BeNil()) - }) - - It("should set ActiveDeadlineSeconds on the pod template when maxPodLifetimeSeconds is set", func() { + // Recycling is driven by the controller, not the pod template: the + // field must stay nil so the Deployment remains updatable. + It("should leave ActiveDeadlineSeconds nil when maxPodLifetimeSeconds is set", func() { lifetime := int64(3600) isvc := &inferencev1alpha1.InferenceService{ ObjectMeta: metav1.ObjectMeta{ @@ -623,8 +610,7 @@ var _ = Describe("Multi-GPU Deployment Construction", func() { }, } deployment := reconciler.constructDeployment(isvc, model, 1) - Expect(deployment.Spec.Template.Spec.ActiveDeadlineSeconds).NotTo(BeNil()) - Expect(*deployment.Spec.Template.Spec.ActiveDeadlineSeconds).To(Equal(int64(3600))) + Expect(deployment.Spec.Template.Spec.ActiveDeadlineSeconds).To(BeNil()) }) }) diff --git a/internal/controller/pod_lifetime.go b/internal/controller/pod_lifetime.go new file mode 100644 index 00000000..cc48b766 --- /dev/null +++ b/internal/controller/pod_lifetime.go @@ -0,0 +1,212 @@ +package controller + +import ( + "context" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" +) + +const ( + maxPodLifetimeSeconds int64 = 9223372036 + podLifetimeRetry = 30 * time.Second +) + +// reconcilePodLifetime recycles at most one healthy, expired pod. All +// workload checks are deliberately conservative: a watch will cause another +// attempt after the Deployment has settled or a replacement pod is ready. +// now is injected so tests can drive the deadline arithmetic. +func (r *InferenceServiceReconciler) reconcilePodLifetime(ctx context.Context, isvc *inferencev1alpha1.InferenceService, isMetal bool, now time.Time) (time.Duration, error) { + if isMetal || isvc.Spec.MaxPodLifetimeSeconds == nil { + return 0, nil + } + + deployment, err := r.getStableDeployment(ctx, isvc) + if err != nil || deployment == nil { + return 0, err + } + active, err := r.activePods(ctx, deployment) + if err != nil { + return 0, err + } + // stableDeployment already proved Status.Replicas is the desired count, and + // it only counts pods the Deployment owns. A foreign pod sharing the + // selector therefore fails this check and holds recycling rather than + // risking an eviction the operator does not own. + if len(active) != int(deployment.Status.Replicas) || !allPodsReady(active) { + return 0, nil + } + return r.recycleExpiredPod(ctx, isvc, active, boundedSeconds(*isvc.Spec.MaxPodLifetimeSeconds), now) +} + +func (r *InferenceServiceReconciler) getStableDeployment(ctx context.Context, isvc *inferencev1alpha1.InferenceService) (*appsv1.Deployment, error) { + deployment := &appsv1.Deployment{} + err := r.Get(ctx, types.NamespacedName{Name: isvc.Name, Namespace: isvc.Namespace}, deployment) + if apierrors.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, err + } + if !stableDeployment(deployment, isvc) { + return nil, nil + } + return deployment, nil +} + +// recycleExpiredPod evicts the pod that expired first, or reports how long +// until the next one does. Ties break on name so repeated reconciles of the +// same state pick the same pod. +func (r *InferenceServiceReconciler) recycleExpiredPod(ctx context.Context, isvc *inferencev1alpha1.InferenceService, active []*corev1.Pod, lifetime time.Duration, now time.Time) (time.Duration, error) { + var expired *corev1.Pod + var expiredDeadline time.Time + var earliest time.Duration + for _, pod := range active { + deadline := pod.Status.StartTime.Time.Add(lifetime) + if deadline.After(now) { + earliest = earliestPositive(earliest, deadline.Sub(now)) + continue + } + if expired == nil || deadline.Before(expiredDeadline) || + (deadline.Equal(expiredDeadline) && pod.Name < expired.Name) { + expired, expiredDeadline = pod, deadline + } + } + if expired == nil { + return earliest, nil + } + // waitForIdle promises never to drop in-flight generations, so recycling + // waits for an idle window rather than killing a busy pod on a timer. + if isvc.RolloutPolicyEnabled() && !idleWaitExhausted(isvc, expiredDeadline, now) { + idle, err := r.checkServiceIdle(ctx, isvc, &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: sanitizeDNSName(isvc.Name), Namespace: isvc.Namespace}}) + if err != nil || !idle { + // Fail closed, matching reconcileRolloutPolicy: an idle probe that + // cannot answer must not authorise killing a possibly-busy pod. + logf.FromContext(ctx).Info("Backend not idle, holding pod recycle", "pod", expired.Name, "error", err) + return inferencev1alpha1.DefaultIdleCheckInterval, nil + } + } + return r.evictPod(ctx, isvc, expired) +} + +// idleWaitExhausted reports whether maxPodLifetimeIdleTimeoutSeconds says to +// stop waiting for idle and recycle a still-busy pod. The budget runs from the +// pod's own expiry, so the deadline is derivable from the pod and nothing has +// to be persisted across reconciles. Omitted means wait indefinitely; 0 means +// never wait, since the caller only reaches here once deadline has passed. +func idleWaitExhausted(isvc *inferencev1alpha1.InferenceService, deadline time.Time, now time.Time) bool { + timeout := isvc.Spec.MaxPodLifetimeIdleTimeoutSeconds + if timeout == nil { + return false + } + return !now.Before(deadline.Add(boundedSeconds(*timeout))) +} + +func (r *InferenceServiceReconciler) evictPod(ctx context.Context, isvc *inferencev1alpha1.InferenceService, pod *corev1.Pod) (time.Duration, error) { + eviction := &policyv1.Eviction{ObjectMeta: metav1.ObjectMeta{Name: pod.Name, Namespace: pod.Namespace}, DeleteOptions: &metav1.DeleteOptions{Preconditions: &metav1.Preconditions{UID: &pod.UID}}} + err := r.SubResource("eviction").Create(ctx, pod, eviction) + switch { + case apierrors.IsNotFound(err): + return 0, nil + case apierrors.IsTooManyRequests(err): + // A PodDisruptionBudget rejected the eviction; nothing watches that, so + // the retry has to be a timer. + logf.FromContext(ctx).Info("Eviction blocked by a PodDisruptionBudget, retrying later", "pod", pod.Name, "retryAfter", podLifetimeRetry) + return podLifetimeRetry, nil + case apierrors.IsForbidden(err): + // Missing pods/eviction RBAC — seen for real when the operator image is + // upgraded without the chart that owns its ClusterRole. Returning an + // error here would fail the whole InferenceService reconcile and retry + // hot forever, since no amount of retrying grants a permission. Surface + // it where an operator will actually look, and back off. + logf.FromContext(ctx).Error(err, "Not permitted to evict pods; recycling is disabled until the operator is granted create on pods/eviction", "pod", pod.Name) + if r.Recorder != nil { + r.Recorder.Eventf(isvc, nil, corev1.EventTypeWarning, "PodRecycleForbidden", "Reconcile", + "Cannot recycle pod %s: %v; the operator needs create on pods/eviction", pod.Name, err) + } + return podLifetimeRetry, nil + case err != nil: + return 0, err + } + // A pod disappearing on a timer is invisible in `kubectl describe isvc` + // without this. + if r.Recorder != nil { + r.Recorder.Eventf(isvc, nil, corev1.EventTypeNormal, "PodRecycled", "Reconcile", + "Evicted pod %s after exceeding maxPodLifetimeSeconds; the Deployment will create a replacement", pod.Name) + } + return 0, nil +} + +func stableDeployment(deployment *appsv1.Deployment, isvc *inferencev1alpha1.InferenceService) bool { + if !metav1.IsControlledBy(deployment, isvc) || deployment.Generation == 0 || deployment.Status.ObservedGeneration != deployment.Generation { + return false + } + replicas := int32(1) + if deployment.Spec.Replicas != nil { + replicas = *deployment.Spec.Replicas + } + return deployment.Status.Replicas == replicas && deployment.Status.UpdatedReplicas == replicas && deployment.Status.ReadyReplicas == replicas && deployment.Status.AvailableReplicas == replicas +} + +// activePods lists the non-terminal pods matching the Deployment's selector. +// The selector labels are unique per InferenceService, and the caller +// cross-checks the count against Status.Replicas, so this deliberately does not +// walk ReplicaSets to prove ownership pod by pod. +func (r *InferenceServiceReconciler) activePods(ctx context.Context, deployment *appsv1.Deployment) ([]*corev1.Pod, error) { + listOpts := []client.ListOption{client.InNamespace(deployment.Namespace)} + if deployment.Spec.Selector != nil { + listOpts = append(listOpts, client.MatchingLabels(deployment.Spec.Selector.MatchLabels)) + } + pods := &corev1.PodList{} + if err := r.List(ctx, pods, listOpts...); err != nil { + return nil, err + } + active := make([]*corev1.Pod, 0, len(pods.Items)) + for i := range pods.Items { + pod := &pods.Items[i] + if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodFailed { + continue + } + active = append(active, pod) + } + return active, nil +} + +func allPodsReady(pods []*corev1.Pod) bool { + for _, pod := range pods { + if pod.DeletionTimestamp != nil || pod.Status.StartTime == nil || !podReady(pod) { + return false + } + } + return true +} + +// boundedSeconds converts a spec field to a Duration, clamping it so an +// unbounded value cannot overflow into the past. +func boundedSeconds(seconds int64) time.Duration { + if seconds <= 0 { + return 0 + } + if seconds > maxPodLifetimeSeconds { + seconds = maxPodLifetimeSeconds + } + return time.Duration(seconds) * time.Second +} + +func podReady(pod *corev1.Pod) bool { + for _, condition := range pod.Status.Conditions { + if condition.Type == corev1.PodReady { + return condition.Status == corev1.ConditionTrue + } + } + return false +} diff --git a/internal/controller/pod_lifetime_test.go b/internal/controller/pod_lifetime_test.go new file mode 100644 index 00000000..ba2f55f0 --- /dev/null +++ b/internal/controller/pod_lifetime_test.go @@ -0,0 +1,362 @@ +package controller + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + inferencev1alpha1 "github.com/defilantech/llmkube/api/v1alpha1" +) + +func TestReconcilePodLifetimeRequeuesUnexpiredPod(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + start := now.Add(-5 * time.Second) + r, isvc, pod := lifetimeReconciler(t, start, 30*time.Second) + + requeue, err := r.reconcilePodLifetime(context.Background(), isvc, false, now) + if err != nil { + t.Fatal(err) + } + if want := 25 * time.Second; requeue != want { + t.Fatalf("requeue = %s, want %s", requeue, want) + } + if err := r.Get(context.Background(), client.ObjectKeyFromObject(pod), &corev1.Pod{}); err != nil { + t.Fatalf("unexpired pod was deleted: %v", err) + } +} + +func TestReconcilePodLifetimeDeletesOldestExpiredPod(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, first := lifetimeReconcilerWith(t, now.Add(-2*time.Minute), time.Minute, 2, nil) + second := first.DeepCopy() + second.Name = "second" + second.UID = "pod-second" + second.ResourceVersion = "" + second.Status.StartTime = &metav1.Time{Time: now.Add(-90 * time.Second)} + if err := r.Create(context.Background(), second); err != nil { + t.Fatal(err) + } + recorder := &recordingClient{Client: r.Client} + r.Client = recorder + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } + if len(recorder.evictions) != 1 { + t.Fatalf("eviction calls = %d", len(recorder.evictions)) + } + if err := r.Get(context.Background(), client.ObjectKeyFromObject(second), &corev1.Pod{}); err != nil { + t.Fatalf("second pod was deleted too: %v", err) + } +} + +func TestReconcilePodLifetimeIgnoresTerminalPods(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, active := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + terminal := active.DeepCopy() + terminal.Name = "succeeded" + terminal.UID = "pod-succeeded" + terminal.ResourceVersion = "" + terminal.Status.Phase = corev1.PodSucceeded + terminal.Status.StartTime = nil + if err := r.Create(context.Background(), terminal); err != nil { + t.Fatal(err) + } + recorder := &recordingClient{Client: r.Client} + r.Client = recorder + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } + if len(recorder.evictions) != 1 { + t.Fatalf("eviction calls = %d, want 1", len(recorder.evictions)) + } +} + +func TestReconcilePodLifetimeNoopWhenUnsetOrMetal(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + isvc.Spec.MaxPodLifetimeSeconds = nil + if got, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil || got != 0 { + t.Fatalf("unset lifetime returned %s, %v", got, err) + } + seconds := int64(60) + isvc.Spec.MaxPodLifetimeSeconds = &seconds + if got, err := r.reconcilePodLifetime(context.Background(), isvc, true, now); err != nil || got != 0 { + t.Fatalf("metal lifetime returned %s, %v", got, err) + } +} + +func TestReconcilePodLifetimeBlocksUnstablePodsAndDeployment(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + for name, mutate := range map[string]func(*appsv1.Deployment, *corev1.Pod){ + "terminating": func(_ *appsv1.Deployment, p *corev1.Pod) { p.DeletionTimestamp = &metav1.Time{Time: now} }, + "unready": func(_ *appsv1.Deployment, p *corev1.Pod) { p.Status.Conditions[0].Status = corev1.ConditionFalse }, + "no start": func(_ *appsv1.Deployment, p *corev1.Pod) { p.Status.StartTime = nil }, + "unobserved": func(d *appsv1.Deployment, _ *corev1.Pod) { d.Status.ObservedGeneration = d.Generation - 1 }, + "replicas": func(d *appsv1.Deployment, _ *corev1.Pod) { d.Status.Replicas = 0 }, + "updated": func(d *appsv1.Deployment, _ *corev1.Pod) { d.Status.UpdatedReplicas = 0 }, + "ready": func(d *appsv1.Deployment, _ *corev1.Pod) { d.Status.ReadyReplicas = 0 }, + "available": func(d *appsv1.Deployment, _ *corev1.Pod) { d.Status.AvailableReplicas = 0 }, + } { + t.Run(name, func(t *testing.T) { + var podMutator func(*corev1.Pod) + if name == "terminating" { + podMutator = func(p *corev1.Pod) { + p.DeletionTimestamp = &metav1.Time{Time: now} + p.Finalizers = []string{"test/finalizer"} + } + } + r, isvc, pod := lifetimeReconcilerWith(t, now.Add(-2*time.Minute), time.Minute, 1, podMutator) + deployment := &appsv1.Deployment{} + if err := r.Get(context.Background(), types.NamespacedName{Name: "svc", Namespace: "ns"}, deployment); err != nil { + t.Fatal(err) + } + mutate(deployment, pod) + if err := r.Status().Update(context.Background(), deployment); err != nil { + t.Fatal(err) + } + if err := r.Status().Update(context.Background(), pod); err != nil { + t.Fatal(err) + } + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } + if err := r.Get(context.Background(), client.ObjectKeyFromObject(pod), &corev1.Pod{}); err != nil { + t.Fatalf("pod was deleted: %v", err) + } + }) + } +} + +// A pod the Deployment does not own but which shares its selector makes the +// active count disagree with Status.Replicas. Recycling must hold rather than +// evict something the operator cannot account for. +func TestReconcilePodLifetimeHoldsOnForeignPod(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, pod := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + foreign := pod.DeepCopy() + foreign.Name = "foreign" + foreign.UID = "foreign-uid" + foreign.ResourceVersion = "" + foreign.OwnerReferences[0].UID = "other-rs" + if err := r.Create(context.Background(), foreign); err != nil { + t.Fatal(err) + } + recorder := &recordingClient{Client: r.Client} + r.Client = recorder + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } + if len(recorder.evictions) != 0 { + t.Fatalf("eviction calls = %d, want 0", len(recorder.evictions)) + } +} + +// The pod starts 2 minutes ago with a 1 minute lifetime, so by now it has been +// overdue for a minute: a 30s idle timeout is spent, a 300s one is not. +func TestReconcilePodLifetimeWaitsForIdle(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + for name, tc := range map[string]struct { + busy bool + idleTimeout *int64 + wantEvicted bool + }{ + "idle backend recycles": {busy: false, wantEvicted: true}, + "busy backend waits indefinitely": {busy: true, wantEvicted: false}, + "busy backend waits within budget": {busy: true, idleTimeout: ptr.To(int64(300)), wantEvicted: false}, + "busy backend recycles once overdue": {busy: true, idleTimeout: ptr.To(int64(30)), wantEvicted: true}, + "zero timeout recycles without asking": {busy: true, idleTimeout: ptr.To(int64(0)), wantEvicted: true}, + } { + t.Run(name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.URL.Path != "/slots" { + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `[{"id":0,"is_processing":%t}]`, tc.busy) + })) + defer server.Close() + + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + isvc.Spec.RolloutPolicy = &inferencev1alpha1.RolloutPolicySpec{WaitForIdle: true} + isvc.Spec.MaxPodLifetimeIdleTimeoutSeconds = tc.idleTimeout + r.RolloutIdleBaseURL = server.URL + recorder := &recordingClient{Client: r.Client} + r.Client = recorder + + requeue, err := r.reconcilePodLifetime(context.Background(), isvc, false, now) + if err != nil { + t.Fatal(err) + } + want := 0 + if tc.wantEvicted { + want = 1 + } + if len(recorder.evictions) != want { + t.Fatalf("eviction calls = %d, want %d", len(recorder.evictions), want) + } + if !tc.wantEvicted && requeue != inferencev1alpha1.DefaultIdleCheckInterval { + t.Fatalf("requeue = %s, want %s", requeue, inferencev1alpha1.DefaultIdleCheckInterval) + } + }) + } +} + +func TestEarliestPositive(t *testing.T) { + if got := earliestPositive(0, 5*time.Second, 2*time.Second, -time.Second); got != 2*time.Second { + t.Fatalf("got %s", got) + } + if got := earliestPositive(0, -time.Second); got != 0 { + t.Fatalf("got %s", got) + } +} + +// recordingClient captures the evictions the controller submits and can fail +// them, so tests can assert on the request the apiserver would have seen. +type recordingClient struct { + client.Client + evictions []*policyv1.Eviction + evictErr error +} + +func (c *recordingClient) SubResource(name string) client.SubResourceClient { + return &recordingSubResource{SubResourceClient: c.Client.SubResource(name), parent: c} +} + +type recordingSubResource struct { + client.SubResourceClient + parent *recordingClient +} + +func (s *recordingSubResource) Create(ctx context.Context, obj client.Object, subResource client.Object, opts ...client.SubResourceCreateOption) error { + if eviction, ok := subResource.(*policyv1.Eviction); ok { + s.parent.evictions = append(s.parent.evictions, eviction) + if s.parent.evictErr != nil { + return s.parent.evictErr + } + } + return s.SubResourceClient.Create(ctx, obj, subResource, opts...) +} + +func TestReconcilePodLifetimeDeleteOptions(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + recorder := &recordingClient{Client: r.Client} + r.Client = recorder + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } + if len(recorder.evictions) != 1 { + t.Fatalf("eviction calls = %d", len(recorder.evictions)) + } + got := recorder.evictions[0].DeleteOptions + if got == nil || got.GracePeriodSeconds != nil { + t.Fatalf("delete options did not preserve graceful termination: %#v", got) + } + if got.Preconditions == nil || got.Preconditions.UID == nil || *got.Preconditions.UID != "pod-first" { + t.Fatalf("missing UID precondition: %#v", got.Preconditions) + } +} + +func TestReconcilePodLifetimeNotFoundDeleteIsSuccess(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + recorder := &recordingClient{Client: r.Client, evictErr: apierrors.NewNotFound(corev1.Resource("pods"), "first")} + r.Client = recorder + if _, err := r.reconcilePodLifetime(context.Background(), isvc, false, now); err != nil { + t.Fatal(err) + } +} + +// Found on a live cluster: swapping the operator image without the chart that +// owns its ClusterRole leaves pods/eviction ungranted. That must back off, not +// fail the reconcile and retry hot forever. +func TestReconcilePodLifetimeForbiddenBacksOff(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + forbidden := apierrors.NewForbidden(corev1.Resource("pods"), "first", + fmt.Errorf(`User "system:serviceaccount:ai:llmkube-controller-manager" cannot create resource "pods/eviction"`)) + r.Client = &recordingClient{Client: r.Client, evictErr: forbidden} + + got, err := r.reconcilePodLifetime(context.Background(), isvc, false, now) + if err != nil { + t.Fatalf("forbidden eviction failed the reconcile: %v", err) + } + if got != podLifetimeRetry { + t.Fatalf("requeue = %s, want %s", got, podLifetimeRetry) + } +} + +func TestReconcilePodLifetimePDBRetry(t *testing.T) { + now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC) + r, isvc, _ := lifetimeReconciler(t, now.Add(-2*time.Minute), time.Minute) + recorder := &recordingClient{Client: r.Client, evictErr: apierrors.NewTooManyRequests("pdb", 1)} + r.Client = recorder + got, err := r.reconcilePodLifetime(context.Background(), isvc, false, now) + if err != nil { + t.Fatal(err) + } + if got != podLifetimeRetry { + t.Fatalf("requeue = %s, want %s", got, podLifetimeRetry) + } +} + +func TestBoundedSecondsClampsOverflow(t *testing.T) { + if got := boundedSeconds(maxPodLifetimeSeconds); got != time.Duration(maxPodLifetimeSeconds)*time.Second { + t.Fatalf("duration = %s", got) + } + if got := boundedSeconds(maxPodLifetimeSeconds + 1); got != time.Duration(maxPodLifetimeSeconds)*time.Second { + t.Fatalf("overflow duration = %s", got) + } +} + +func lifetimeReconciler(t *testing.T, start time.Time, lifetime time.Duration) (*InferenceServiceReconciler, *inferencev1alpha1.InferenceService, *corev1.Pod) { + return lifetimeReconcilerWith(t, start, lifetime, 1, nil) +} + +func lifetimeReconcilerWith(t *testing.T, start time.Time, lifetime time.Duration, replicaCount int32, podMutator func(*corev1.Pod)) (*InferenceServiceReconciler, *inferencev1alpha1.InferenceService, *corev1.Pod) { + t.Helper() + const ( + isvcUID = types.UID("isvc-uid") + depUID = types.UID("deployment-uid") + rsUID = types.UID("rs-uid") + ) + controller := true + replicas := replicaCount + seconds := int64(lifetime / time.Second) + isvc := &inferencev1alpha1.InferenceService{ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns", UID: isvcUID}, Spec: inferencev1alpha1.InferenceServiceSpec{MaxPodLifetimeSeconds: &seconds}} + deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns", UID: depUID, Generation: 2, OwnerReferences: []metav1.OwnerReference{{UID: isvcUID, Controller: &controller}}}, Spec: appsv1.DeploymentSpec{Replicas: &replicas, Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "svc"}}}, Status: appsv1.DeploymentStatus{ObservedGeneration: 2, Replicas: replicaCount, UpdatedReplicas: replicaCount, ReadyReplicas: replicaCount, AvailableReplicas: replicaCount}} + ready := corev1.ConditionTrue + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "first", Namespace: "ns", UID: "pod-first", Labels: map[string]string{"app": "svc"}, OwnerReferences: []metav1.OwnerReference{{UID: rsUID, Controller: &controller}}}, Status: corev1.PodStatus{StartTime: &metav1.Time{Time: start}, Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: ready}}}} + if podMutator != nil { + podMutator(pod) + } + scheme := runtime.NewScheme() + if err := inferencev1alpha1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := appsv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + if err := policyv1.AddToScheme(scheme); err != nil { + t.Fatal(err) + } + return &InferenceServiceReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(deployment, pod).WithObjects(isvc, deployment, pod).Build()}, isvc, pod +}