diff --git a/pkg/controller/v1beta1/workload/drain/drain.go b/pkg/controller/v1beta1/workload/drain/drain.go new file mode 100644 index 000000000..2e9721068 --- /dev/null +++ b/pkg/controller/v1beta1/workload/drain/drain.go @@ -0,0 +1,332 @@ +// Package drain owns the EndpointSlice-convergence signals OMENative +// uses to gate destructive pod operations: IsPodDrained (safe to +// delete?) and IsPodInRotation (surge eligible for new traffic?). +// +// Boundary: pure leaf. Only Kubernetes API machinery. Knows nothing +// about ReconcileParams or any OMENative-specific shape — every entry +// point takes a plain client.Reader + (namespace, serviceName, pod). +package drain + +import ( + "context" + "fmt" + + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// IsPodDrained reports whether pod is no longer routable through serviceName. +// It returns true when, across every EndpointSlice for serviceName in the +// pod's namespace, either: +// - no endpoint targets pod, or +// - the endpoint that targets pod has Conditions.Ready != true. +// +// The returned bool is the EndpointSlice-convergence signal used during +// drain: once true, kube-proxy will have removed (or marked not-ready) +// the pod's address and new connections to serviceName stop landing on +// this pod. Existing connections and runtime-level shutdown are out of +// scope — the caller layers gracePeriodSeconds on top of this signal. +// +// reader is taken as client.Reader (not client.Client) so callers may +// pass a live API reader to bypass the cached controller-runtime read +// when cache lag would make a drain falsely declare done. The cached +// client works as well and is what tests use. +func IsPodDrained(ctx context.Context, reader client.Reader, namespace, serviceName string, pod *corev1.Pod) (bool, error) { + if pod == nil { + return false, fmt.Errorf("IsPodDrained: nil pod") + } + if serviceName == "" { + return false, fmt.Errorf("IsPodDrained: empty serviceName") + } + + slices, err := EndpointSlicesForService(ctx, reader, namespace, serviceName) + if err != nil { + return false, err + } + if len(slices) == 0 { + // Distinguish "no Service at all" (drain trivially complete — + // nothing routes to anything) from "Service exists but slices + // haven't propagated yet" (cache cold-start, kube-controller- + // manager lag, no matching pods yet). Without this + // disambiguation the latter is indistinguishable from the + // former and fails open, reporting drain done while a serving + // pod is about to get an image patch or delete. + return drainedWhenSliceListEmpty(ctx, reader, namespace, serviceName) + } + + return podDrainedInSlices(slices, pod), nil +} + +// Batcher records one drain observation per Service for a whole +// gang/drain loop. Each observation lists EndpointSlices once, indexes +// routable Pod target names, and caches the empty-slice Service lookup. +// Per-Pod checks are therefore O(1) after O(E) observation construction. +// +// Semantics are identical to IsPodDrained, including Ready=nil and +// empty TargetRef.Namespace handling. Results and read errors are held +// for the Batcher lifetime so every Pod in a wave is judged against the +// same observation. +// +// Not safe for concurrent use; scoped to one reconcile pass. +type Batcher struct { + reader client.Reader + namespace string + services map[string]serviceDrainObservation +} + +type serviceDrainObservation struct { + hasEndpointSlices bool + drainedWithoutSlices bool + routableTargets routablePodTargets + err error +} + +// routablePodTargets mirrors endpointTargetsPod without retaining or scanning +// the complete EndpointSlice payload for every Pod check. A TargetRef with an +// empty Namespace matches a same-named Pod in any namespace. +type routablePodTargets struct { + exactNamespace map[client.ObjectKey]struct{} + anyNamespace map[string]struct{} +} + +// NewBatcher returns a Batcher bound to reader + namespace. Callers +// invoke IsPodDrained per pod; the underlying LIST runs at most once +// per distinct serviceName. +func NewBatcher(reader client.Reader, namespace string) *Batcher { + return &Batcher{ + reader: reader, + namespace: namespace, + services: map[string]serviceDrainObservation{}, + } +} + +func (b *Batcher) observeService(ctx context.Context, serviceName string) serviceDrainObservation { + if observation, ok := b.services[serviceName]; ok { + return observation + } + + observation := serviceDrainObservation{} + slices, err := EndpointSlicesForService(ctx, b.reader, b.namespace, serviceName) + if err != nil { + observation.err = err + b.services[serviceName] = observation + return observation + } + if len(slices) == 0 { + observation.drainedWithoutSlices, observation.err = drainedWhenSliceListEmpty(ctx, b.reader, b.namespace, serviceName) + b.services[serviceName] = observation + return observation + } + + observation.hasEndpointSlices = true + observation.routableTargets = indexRoutablePodTargets(slices) + b.services[serviceName] = observation + return observation +} + +// IsPodDrained reports whether pod is no longer routable through +// serviceName, reusing the memoized slice list. Identical semantics to +// the package-level IsPodDrained. +func (b *Batcher) IsPodDrained(ctx context.Context, serviceName string, pod *corev1.Pod) (bool, error) { + if pod == nil { + return false, fmt.Errorf("IsPodDrained: nil pod") + } + if serviceName == "" { + return false, fmt.Errorf("IsPodDrained: empty serviceName") + } + observation := b.observeService(ctx, serviceName) + if observation.err != nil { + return false, observation.err + } + if !observation.hasEndpointSlices { + return observation.drainedWithoutSlices, nil + } + return !observation.routableTargets.contains(pod), nil +} + +func indexRoutablePodTargets(slices []discoveryv1.EndpointSlice) routablePodTargets { + targets := routablePodTargets{ + exactNamespace: make(map[client.ObjectKey]struct{}), + anyNamespace: make(map[string]struct{}), + } + for i := range slices { + for j := range slices[i].Endpoints { + ep := &slices[i].Endpoints[j] + if !endpointIsReady(*ep) || ep.TargetRef == nil { + continue + } + ref := ep.TargetRef + if ref.Kind != "" && ref.Kind != "Pod" { + continue + } + if ref.Namespace == "" { + targets.anyNamespace[ref.Name] = struct{}{} + continue + } + targets.exactNamespace[client.ObjectKey{Namespace: ref.Namespace, Name: ref.Name}] = struct{}{} + } + } + return targets +} + +func (targets routablePodTargets) contains(pod *corev1.Pod) bool { + if _, ok := targets.anyNamespace[pod.Name]; ok { + return true + } + _, ok := targets.exactNamespace[client.ObjectKeyFromObject(pod)] + return ok +} + +// podDrainedInSlices is the package-level check used by IsPodDrained: a pod is +// drained when no endpoint targeting it across slices reports Ready. Assumes +// slices is non-empty; the caller owns empty-list disambiguation. Batcher builds +// the equivalent indexed representation once per Service. +func podDrainedInSlices(slices []discoveryv1.EndpointSlice, pod *corev1.Pod) bool { + for _, slice := range slices { + for _, ep := range slice.Endpoints { + if !endpointTargetsPod(ep, pod) { + continue + } + if endpointIsReady(ep) { + return false + } + } + } + return true +} + +// drainedWhenSliceListEmpty resolves the ambiguous "no slices" +// observation by checking whether the Service itself exists. Service +// absent → drained (no traffic path). Service present → slices +// haven't been materialized yet → conservative: NOT drained, caller +// requeues. +func drainedWhenSliceListEmpty(ctx context.Context, reader client.Reader, namespace, serviceName string) (bool, error) { + svc := &corev1.Service{} + err := reader.Get(ctx, client.ObjectKey{Namespace: namespace, Name: serviceName}, svc) + if apierrors.IsNotFound(err) { + // No Service → no kube-proxy routing → drain is trivially + // complete. With the services reconciler in place this branch + // fires only when the Service hasn't been created yet on this + // reconcile pass, which is a controller bug — the service + // reconciliation runs before any op that needs drain. + return true, nil + } + if err != nil { + return false, fmt.Errorf("drainedWhenSliceListEmpty: get service %s/%s: %w", namespace, serviceName, err) + } + // Service exists, slice list empty. Could mean (a) no pods match + // the selector yet, (b) kube-controller-manager hasn't created + // slices yet, (c) informer cache cold-start. Any of those is a + // transient state where we should not declare drain complete. + return false, nil +} + +// IsPodInRotation reports whether pod has at least one endpoint Ready +// AND non-terminating — eligible for NEW traffic. Migrate uses this to +// confirm the surge is receiving traffic before draining the source so +// the swap has no zero-routable-endpoint window. Terminating endpoints +// don't count: kube-proxy keeps them for in-flight requests but won't +// send new traffic, and swapping onto a terminating surge would be +// pointless. +func IsPodInRotation(ctx context.Context, reader client.Reader, namespace, serviceName string, pod *corev1.Pod) (bool, error) { + if pod == nil { + return false, fmt.Errorf("IsPodInRotation: nil pod") + } + if serviceName == "" { + return false, fmt.Errorf("IsPodInRotation: empty serviceName") + } + slices, err := EndpointSlicesForService(ctx, reader, namespace, serviceName) + if err != nil { + return false, err + } + for _, slice := range slices { + for _, ep := range slice.Endpoints { + if !endpointTargetsPod(ep, pod) { + continue + } + if EndpointAvailable(ep) { + return true, nil + } + } + } + return false, nil +} + +// EndpointSlicesForService lists EndpointSlices in namespace owned by +// serviceName via the standard kubernetes.io/service-name label that the +// in-tree endpointslice controller stamps on every slice it manages. +// +// Exported so status_aggregate's availablePodSet helper (which folds +// rotation across every pod into a single map) can reuse the same +// label query and slice walk. +func EndpointSlicesForService(ctx context.Context, reader client.Reader, namespace, serviceName string) ([]discoveryv1.EndpointSlice, error) { + list := &discoveryv1.EndpointSliceList{} + if err := reader.List(ctx, list, + client.InNamespace(namespace), + client.MatchingLabels{discoveryv1.LabelServiceName: serviceName}, + ); err != nil { + return nil, fmt.Errorf("list EndpointSlices for service %s/%s: %w", namespace, serviceName, err) + } + return list.Items, nil +} + +// endpointTargetsPod matches an EndpointSlice endpoint to pod by TargetRef. +// TargetRef is preferred over Addresses[] because IP addresses can be +// reused across pods, while TargetRef carries the stable Pod identity. +func endpointTargetsPod(ep discoveryv1.Endpoint, pod *corev1.Pod) bool { + ref := ep.TargetRef + if ref == nil { + return false + } + if ref.Kind != "" && ref.Kind != "Pod" { + return false + } + if ref.Namespace != "" && ref.Namespace != pod.Namespace { + return false + } + return ref.Name == pod.Name +} + +// endpointIsReady reports whether the endpoint is currently receiving +// new Service traffic. Drives the drain wait: a pod is considered +// drained once every slice entry targeting it reports Ready=false. +// Terminating endpoints are deliberately considered "still receiving" +// when Ready=true so the controller doesn't declare drain complete +// while kube-proxy is still routing in-flight connections. +// +// A nil Ready pointer is treated as ready per the discovery/v1 +// contract: producers SHOULD set Conditions.Ready, but if a slice +// omits it the endpoint is presumed routable, matching kube-proxy's +// behavior. +func endpointIsReady(ep discoveryv1.Endpoint) bool { + if ep.Conditions.Ready == nil { + return true + } + return *ep.Conditions.Ready +} + +// EndpointAvailable reports whether the endpoint is eligible for NEW +// traffic — Ready=true AND Terminating!=true. Drives the status +// `AvailablePodCount` counter and `IsPodInRotation`'s surge check. +// +// Distinct from endpointIsReady because of the EndpointSlice +// tri-state contract: a terminating pod whose probes still pass +// reports Ready=true + Serving=true + Terminating=true. kube-proxy +// keeps it in rotation for in-flight requests but won't send NEW +// requests. Counting it as Available would inflate AvailableReplicas; +// treating it as in-rotation would let Migrate swap onto a pod that's +// about to disappear. +// +// A nil Terminating pointer is treated as not-terminating (the +// common steady-state shape). +func EndpointAvailable(ep discoveryv1.Endpoint) bool { + if !endpointIsReady(ep) { + return false + } + if ep.Conditions.Terminating != nil && *ep.Conditions.Terminating { + return false + } + return true +} diff --git a/pkg/controller/v1beta1/workload/drain/drain_test.go b/pkg/controller/v1beta1/workload/drain/drain_test.go new file mode 100644 index 000000000..ea5608d5f --- /dev/null +++ b/pkg/controller/v1beta1/workload/drain/drain_test.go @@ -0,0 +1,903 @@ +package drain + +import ( + "context" + "errors" + "testing" + + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + 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" +) + +// newDrainTestClient builds a fake controller-runtime client with the +// scheme drain.go reads from (corev1 + discoveryv1). +func newDrainTestClient(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1: %v", err) + } + if err := discoveryv1.AddToScheme(scheme); err != nil { + t.Fatalf("add discoveryv1: %v", err) + } + return fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + Build() +} + +func testPod(namespace, name string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + UID: types.UID(name + "-uid"), + }, + } +} + +// sliceForService builds an EndpointSlice labeled for serviceName with one +// endpoint per (podName, ready) tuple supplied. Each endpoint's TargetRef +// is set to {Kind: Pod, Namespace: namespace, Name: podName}. +func sliceForService(namespace, sliceName, serviceName string, endpoints ...endpointSpec) *discoveryv1.EndpointSlice { + es := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: sliceName, + Namespace: namespace, + Labels: map[string]string{ + discoveryv1.LabelServiceName: serviceName, + }, + }, + AddressType: discoveryv1.AddressTypeIPv4, + } + for _, e := range endpoints { + ep := discoveryv1.Endpoint{ + Addresses: []string{e.address}, + Conditions: discoveryv1.EndpointConditions{ + Ready: e.ready, + }, + } + if e.podName != "" { + ep.TargetRef = &corev1.ObjectReference{ + Kind: "Pod", + Namespace: namespace, + Name: e.podName, + } + } + es.Endpoints = append(es.Endpoints, ep) + } + return es +} + +type endpointSpec struct { + podName string + address string + ready *bool +} + +func TestIsPodDrained_NilPodRejected(t *testing.T) { + c := newDrainTestClient(t) + if _, err := IsPodDrained(context.Background(), c, "ns", "svc", nil); err == nil { + t.Fatal("expected error for nil pod") + } +} + +func TestIsPodDrained_EmptyServiceNameRejected(t *testing.T) { + c := newDrainTestClient(t) + pod := testPod("ns", "p1") + if _, err := IsPodDrained(context.Background(), c, "ns", "", pod); err == nil { + t.Fatal("expected error for empty service name") + } +} + +func TestIsPodDrained_NoSlicesAtAll_ReturnsTrue(t *testing.T) { + // No EndpointSlices AND no Service in the cluster — drained by + // definition (kube-proxy has nothing to route to). PR B's + // fail-loud branch only fires when the Service exists; absence + // of both is the trivial-drain case. + c := newDrainTestClient(t) + pod := testPod("ns", "p1") + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true with no slices and no service") + } +} + +func TestIsPodDrained_ServiceExistsNoSlices_ReturnsNotDrained(t *testing.T) { + // Conservative: an empty slice list against an existing Service is + // transient (informer cold-start, no matching pods yet, KCM lag) and + // must NOT report drained. + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns"}, + } + pod := testPod("ns", "p1") + c := newDrainTestClient(t, svc) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if drained { + t.Fatalf("expected drained=false when Service exists but no slices") + } +} + +func TestIsPodDrained_SlicesForDifferentService_ReturnsTrue(t *testing.T) { + // A slice exists, but it's labeled for a different Service. Drain is + // scoped to the named Service, so this is still drained. + pod := testPod("ns", "p1") + other := sliceForService("ns", "other-1", "other-svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(true), + }) + c := newDrainTestClient(t, other) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when slice targets a different service") + } +} + +func TestIsPodDrained_PodAbsentFromSlices_ReturnsTrue(t *testing.T) { + // Slice exists for svc but lists a different pod. p1 is not in + // rotation through svc. + pod := testPod("ns", "p1") + slice := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p-other", address: "10.0.0.2", ready: ptr.To(true), + }) + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when pod is absent from slice") + } +} + +func TestIsPodDrained_PodPresentAndReady_ReturnsFalse(t *testing.T) { + pod := testPod("ns", "p1") + slice := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(true), + }) + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if drained { + t.Fatalf("expected drained=false when pod is in slice with Ready=true") + } +} + +func TestIsPodDrained_PodPresentButNotReady_ReturnsTrue(t *testing.T) { + pod := testPod("ns", "p1") + slice := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(false), + }) + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when pod endpoint has Ready=false") + } +} + +func TestIsPodDrained_PodPresentReadyNil_TreatedAsReady(t *testing.T) { + // discovery/v1 says Conditions.Ready SHOULD be set; if nil, presume + // ready (matches kube-proxy). So drained=false. + pod := testPod("ns", "p1") + slice := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: nil, + }) + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if drained { + t.Fatalf("expected drained=false when Ready is nil (presumed ready)") + } +} + +func TestIsPodDrained_AcrossMultipleSlices_FindsRoutableEndpoint(t *testing.T) { + // Slice A has p1 with Ready=false; Slice B has p1 with Ready=true. + // Any routable endpoint means not drained. + pod := testPod("ns", "p1") + a := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(false), + }) + b := sliceForService("ns", "svc-2", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(true), + }) + c := newDrainTestClient(t, a, b) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if drained { + t.Fatalf("expected drained=false when any slice still publishes pod as Ready") + } +} + +func TestIsPodDrained_AcrossMultipleSlices_AllNotReady(t *testing.T) { + pod := testPod("ns", "p1") + a := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p1", address: "10.0.0.1", ready: ptr.To(false), + }) + b := sliceForService("ns", "svc-2", "svc", endpointSpec{ + podName: "p-other", address: "10.0.0.2", ready: ptr.To(true), + }) + c := newDrainTestClient(t, a, b) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true: every endpoint targeting p1 is NotReady, other endpoints don't count") + } +} + +func TestIsPodDrained_TargetRefNamespaceMismatchIgnored(t *testing.T) { + // EndpointSlice targets a pod named p1 in a different namespace — + // must not match our pod. + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "svc-1", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + TargetRef: &corev1.ObjectReference{ + Kind: "Pod", Namespace: "other-ns", Name: "p1", + }, + }, + }, + } + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when TargetRef.Namespace differs from pod namespace") + } +} + +func TestIsPodDrained_TargetRefNilIgnored(t *testing.T) { + // An endpoint with no TargetRef (e.g., custom publisher) can't be + // matched to a pod and is ignored. + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "svc-1", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + TargetRef: nil, + }, + }, + } + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when no endpoint has TargetRef matching the pod") + } +} + +func TestIsPodDrained_TargetRefKindEmpty_TreatedAsPod(t *testing.T) { + // A publisher that omits TargetRef.Kind but supplies Namespace+Name + // should still match. Empty Kind is accepted; only an explicitly + // non-"Pod" kind is rejected. + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "svc-1", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + TargetRef: &corev1.ObjectReference{ + Kind: "", Namespace: "ns", Name: "p1", + }, + }, + }, + } + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if drained { + t.Fatalf("expected drained=false when TargetRef.Kind is empty but Name+Namespace match a ready endpoint") + } +} + +func TestIsPodDrained_TargetRefKindNotPod_Rejected(t *testing.T) { + // A TargetRef pointing at a non-Pod resource (e.g., Node) must not + // match the pod, even when the name happens to coincide. + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "svc-1", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{ + { + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + }, + TargetRef: &corev1.ObjectReference{ + Kind: "Node", Namespace: "ns", Name: "p1", + }, + }, + }, + } + c := newDrainTestClient(t, slice) + drained, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true when TargetRef.Kind is not 'Pod' (must not match)") + } +} + +// --- Batcher: single LIST per serviceName, identical semantics --- + +// countingReader records the reads that build a Service drain observation and +// can inject failures at either boundary. +type countingReader struct { + client.Reader + sliceLists int + serviceGets int + listErr error + getErr error +} + +func (r *countingReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*discoveryv1.EndpointSliceList); ok { + r.sliceLists++ + if r.listErr != nil { + return r.listErr + } + } + return r.Reader.List(ctx, list, opts...) +} + +func (r *countingReader) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.Service); ok { + r.serviceGets++ + if r.getErr != nil { + return r.getErr + } + } + return r.Reader.Get(ctx, key, obj, opts...) +} + +func TestBatcher_OneListPerServiceNameAcrossPods(t *testing.T) { + // Two pods of the same gang routed through the same per-revision + // Service — the Batcher must LIST that Service's slices exactly once. + p1 := testPod("ns", "p1") + p2 := testPod("ns", "p2") + slice := sliceForService("ns", "svc-1", "svc", + endpointSpec{podName: "p1", address: "10.0.0.1", ready: ptr.To(false)}, + endpointSpec{podName: "p2", address: "10.0.0.2", ready: ptr.To(false)}, + ) + cr := &countingReader{Reader: newDrainTestClient(t, slice)} + b := NewBatcher(cr, "ns") + + for _, pod := range []*corev1.Pod{p1, p2} { + drained, err := b.IsPodDrained(context.Background(), "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !drained { + t.Fatalf("expected drained=true for %s (endpoint Ready=false)", pod.Name) + } + } + if cr.sliceLists != 1 { + t.Fatalf("expected exactly 1 EndpointSlice LIST across 2 pods, got %d", cr.sliceLists) + } + if cr.serviceGets != 0 { + t.Fatalf("non-empty EndpointSlices must not GET the Service, got %d GETs", cr.serviceGets) + } +} + +func TestBatcher_ListsPerDistinctServiceName(t *testing.T) { + // Pods on different per-revision Services each trigger their own LIST, + // but only once per distinct serviceName. + pA := testPod("ns", "pA") + pB := testPod("ns", "pB") + sA := sliceForService("ns", "svcA-1", "svcA", endpointSpec{podName: "pA", address: "10.0.0.1", ready: ptr.To(false)}) + sB := sliceForService("ns", "svcB-1", "svcB", endpointSpec{podName: "pB", address: "10.0.0.2", ready: ptr.To(false)}) + cr := &countingReader{Reader: newDrainTestClient(t, sA, sB)} + b := NewBatcher(cr, "ns") + + if _, err := b.IsPodDrained(context.Background(), "svcA", pA); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if _, err := b.IsPodDrained(context.Background(), "svcB", pB); err != nil { + t.Fatalf("unexpected error: %v", err) + } + // Re-check pA — must reuse the memoized svcA list, no new LIST. + if _, err := b.IsPodDrained(context.Background(), "svcA", pA); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cr.sliceLists != 2 { + t.Fatalf("expected 2 LISTs (one per distinct service), got %d", cr.sliceLists) + } +} + +func TestBatcher_MatchesIsPodDrainedSemantics(t *testing.T) { + // The Batcher path must produce the same answer as the package-level + // IsPodDrained across the key cases: ready (not drained), not-ready + // (drained), Service-exists-no-slices (not drained), and no-service + // (drained). + cases := []struct { + name string + objs []client.Object + want bool + }{ + { + name: "ready endpoint -> not drained", + objs: []client.Object{sliceForService("ns", "s", "svc", endpointSpec{podName: "p1", address: "10.0.0.1", ready: ptr.To(true)})}, + want: false, + }, + { + name: "not-ready endpoint -> drained", + objs: []client.Object{sliceForService("ns", "s", "svc", endpointSpec{podName: "p1", address: "10.0.0.1", ready: ptr.To(false)})}, + want: true, + }, + { + name: "service exists, no slices -> not drained", + objs: []client.Object{&corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns"}}}, + want: false, + }, + { + name: "no service, no slices -> drained", + objs: nil, + want: true, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pod := testPod("ns", "p1") + c := newDrainTestClient(t, tc.objs...) + direct, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("IsPodDrained error: %v", err) + } + batched, err := NewBatcher(c, "ns").IsPodDrained(context.Background(), "svc", pod) + if err != nil { + t.Fatalf("Batcher.IsPodDrained error: %v", err) + } + if direct != tc.want || batched != tc.want { + t.Fatalf("want %v; IsPodDrained=%v Batcher=%v", tc.want, direct, batched) + } + }) + } +} + +func TestBatcher_NilPodAndEmptyServiceRejected(t *testing.T) { + cr := &countingReader{Reader: newDrainTestClient(t)} + b := NewBatcher(cr, "ns") + if _, err := b.IsPodDrained(context.Background(), "svc", nil); err == nil { + t.Fatal("expected error for nil pod") + } + if _, err := b.IsPodDrained(context.Background(), "", testPod("ns", "p1")); err == nil { + t.Fatal("expected error for empty service name") + } + if cr.sliceLists != 0 || cr.serviceGets != 0 { + t.Fatalf("invalid inputs must perform no reads, got LIST=%d GET=%d", cr.sliceLists, cr.serviceGets) + } +} + +func TestBatcher_EmptySlicesCachesServiceLookup(t *testing.T) { + tests := []struct { + name string + objs []client.Object + want bool + }{ + { + name: "existing service is not drained", + objs: []client.Object{&corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "svc", Namespace: "ns"}}}, + want: false, + }, + { + name: "absent service is drained", + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cr := &countingReader{Reader: newDrainTestClient(t, tt.objs...)} + batcher := NewBatcher(cr, "ns") + for _, pod := range []*corev1.Pod{testPod("ns", "p1"), testPod("ns", "p2"), testPod("ns", "p1")} { + drained, err := batcher.IsPodDrained(context.Background(), "svc", pod) + if err != nil { + t.Fatalf("IsPodDrained(%s): %v", pod.Name, err) + } + if drained != tt.want { + t.Fatalf("IsPodDrained(%s)=%v, want %v", pod.Name, drained, tt.want) + } + } + if cr.sliceLists != 1 || cr.serviceGets != 1 { + t.Fatalf("empty-slice observation reads: LIST=%d GET=%d, want 1/1", cr.sliceLists, cr.serviceGets) + } + }) + } +} + +func TestBatcher_CachesObservationErrors(t *testing.T) { + tests := []struct { + name string + listErr error + getErr error + wantList int + wantGet int + }{ + { + name: "EndpointSlice list failure", + listErr: errors.New("slice list failed"), + wantList: 1, + }, + { + name: "empty-slice Service lookup failure", + getErr: errors.New("service lookup failed"), + wantList: 1, + wantGet: 1, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cr := &countingReader{ + Reader: newDrainTestClient(t), + listErr: tt.listErr, + getErr: tt.getErr, + } + batcher := NewBatcher(cr, "ns") + for _, pod := range []*corev1.Pod{testPod("ns", "p1"), testPod("ns", "p2")} { + if _, err := batcher.IsPodDrained(context.Background(), "svc", pod); err == nil { + t.Fatalf("IsPodDrained(%s) returned no error", pod.Name) + } + } + if cr.sliceLists != tt.wantList || cr.serviceGets != tt.wantGet { + t.Fatalf("cached failure reads: LIST=%d GET=%d, want %d/%d", cr.sliceLists, cr.serviceGets, tt.wantList, tt.wantGet) + } + }) + } +} + +func TestBatcher_IndexesRoutableTargetsWithDuplicateEndpointSemantics(t *testing.T) { + tests := []struct { + name string + endpoints []endpointSpec + want bool + }{ + { + name: "duplicate false and nil Ready remains routable", + endpoints: []endpointSpec{ + {podName: "p1", address: "10.0.0.1", ready: ptr.To(false)}, + {podName: "p1", address: "10.0.0.2", ready: nil}, + }, + want: false, + }, + { + name: "duplicate false endpoints are drained", + endpoints: []endpointSpec{ + {podName: "p1", address: "10.0.0.1", ready: ptr.To(false)}, + {podName: "p1", address: "10.0.0.2", ready: ptr.To(false)}, + }, + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + slice := sliceForService("ns", "svc-1", "svc", tt.endpoints...) + cr := &countingReader{Reader: newDrainTestClient(t, slice)} + drained, err := NewBatcher(cr, "ns").IsPodDrained(context.Background(), "svc", testPod("ns", "p1")) + if err != nil { + t.Fatalf("IsPodDrained: %v", err) + } + if drained != tt.want { + t.Fatalf("IsPodDrained=%v, want %v", drained, tt.want) + } + if cr.sliceLists != 1 || cr.serviceGets != 0 { + t.Fatalf("reads: LIST=%d GET=%d, want 1/0", cr.sliceLists, cr.serviceGets) + } + }) + } +} + +func TestBatcher_TargetRefNamespaceIndexPreservesMatchingRules(t *testing.T) { + pod := testPod("other-ns", "p1") + tests := []struct { + name string + refNamespace string + refKind string + want bool + }{ + {name: "empty namespace matches", refKind: "Pod", want: false}, + {name: "exact namespace matches", refNamespace: "other-ns", refKind: "Pod", want: false}, + {name: "different namespace is ignored", refNamespace: "ns", refKind: "Pod", want: true}, + {name: "empty kind matches", refNamespace: "other-ns", want: false}, + {name: "non-Pod kind is ignored", refNamespace: "other-ns", refKind: "Node", want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + slice := sliceForService("ns", "svc-1", "svc") + slice.Endpoints = []discoveryv1.Endpoint{{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)}, + TargetRef: &corev1.ObjectReference{ + Kind: tt.refKind, + Namespace: tt.refNamespace, + Name: pod.Name, + }, + }} + c := newDrainTestClient(t, slice) + + // The Batcher index and the package-level matcher are + // separate implementations of the same TargetRef rules. + // Running the table through both fails if either drifts. + batched, err := NewBatcher(c, "ns").IsPodDrained(context.Background(), "svc", pod) + if err != nil { + t.Fatalf("Batcher.IsPodDrained: %v", err) + } + if batched != tt.want { + t.Fatalf("Batcher.IsPodDrained=%v, want %v", batched, tt.want) + } + + direct, err := IsPodDrained(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("IsPodDrained: %v", err) + } + if direct != batched { + t.Fatalf("IsPodDrained=%v diverges from Batcher.IsPodDrained=%v", direct, batched) + } + }) + } +} + +func TestBatcher_HoldsOneObservationForItsLifetime(t *testing.T) { + pod := testPod("ns", "p1") + slice := sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: pod.Name, address: "10.0.0.1", ready: ptr.To(true), + }) + base := newDrainTestClient(t, slice) + cr := &countingReader{Reader: base} + batcher := NewBatcher(cr, "ns") + + drained, err := batcher.IsPodDrained(context.Background(), "svc", pod) + if err != nil || drained { + t.Fatalf("initial observation: drained=%v err=%v, want false/nil", drained, err) + } + fresh := slice.DeepCopy() + fresh.Endpoints[0].Conditions.Ready = ptr.To(false) + if err := base.Update(context.Background(), fresh); err != nil { + t.Fatalf("update EndpointSlice: %v", err) + } + drained, err = batcher.IsPodDrained(context.Background(), "svc", pod) + if err != nil || drained { + t.Fatalf("memoized observation: drained=%v err=%v, want false/nil", drained, err) + } + if cr.sliceLists != 1 { + t.Fatalf("EndpointSlice LISTs=%d, want 1", cr.sliceLists) + } + + newBatch := NewBatcher(cr, "ns") + drained, err = newBatch.IsPodDrained(context.Background(), "svc", pod) + if err != nil || !drained { + t.Fatalf("new observation: drained=%v err=%v, want true/nil", drained, err) + } + if cr.sliceLists != 2 { + t.Fatalf("EndpointSlice LISTs=%d, want 2 after a new Batcher", cr.sliceLists) + } +} + +// --- EndpointAvailable tri-state --- + +func TestEndpointAvailable_ReadyTrueTerminatingFalseIsAvailable(t *testing.T) { + ep := discoveryv1.Endpoint{ + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + Terminating: ptr.To(false), + }, + } + if !EndpointAvailable(ep) { + t.Errorf("Ready=true, Terminating=false should be available") + } +} + +func TestEndpointAvailable_ReadyTrueTerminatingNilIsAvailable(t *testing.T) { + // nil Terminating is the steady-state shape (no termination underway). + ep := discoveryv1.Endpoint{ + Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(true)}, + } + if !EndpointAvailable(ep) { + t.Errorf("Ready=true, Terminating=nil should be available") + } +} + +func TestEndpointAvailable_ReadyTrueTerminatingTrueIsNotAvailable(t *testing.T) { + // kube-proxy keeps a terminating-but-still-Ready endpoint in rotation + // for in-flight requests; it shouldn't count as Available for status + // or surge-rotation checks. + ep := discoveryv1.Endpoint{ + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + Terminating: ptr.To(true), + }, + } + if EndpointAvailable(ep) { + t.Errorf("Ready=true, Terminating=true must NOT be available") + } +} + +func TestEndpointAvailable_ReadyFalseIsNotAvailable(t *testing.T) { + ep := discoveryv1.Endpoint{ + Conditions: discoveryv1.EndpointConditions{Ready: ptr.To(false)}, + } + if EndpointAvailable(ep) { + t.Errorf("Ready=false should never be available") + } +} + +// --- IsPodInRotation excludes terminating endpoints --- + +func TestIsPodInRotation_TerminatingEndpointReportedNotInRotation(t *testing.T) { + // Even with Ready=true, a terminating endpoint must not be + // considered "in rotation" — Migrate would otherwise swap onto a + // pod that's about to disappear. + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "s", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + Terminating: ptr.To(true), + }, + TargetRef: &corev1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: "p1"}, + }}, + } + c := newDrainTestClient(t, slice) + in, err := IsPodInRotation(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if in { + t.Errorf("terminating endpoint must not be reported as in-rotation") + } +} + +func TestIsPodInRotation_ReadyTrueNotTerminatingReportedInRotation(t *testing.T) { + pod := testPod("ns", "p1") + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Name: "s", + Namespace: "ns", + Labels: map[string]string{discoveryv1.LabelServiceName: "svc"}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: []discoveryv1.Endpoint{{ + Addresses: []string{"10.0.0.1"}, + Conditions: discoveryv1.EndpointConditions{ + Ready: ptr.To(true), + Terminating: ptr.To(false), + }, + TargetRef: &corev1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: "p1"}, + }}, + } + c := newDrainTestClient(t, slice) + in, err := IsPodInRotation(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !in { + t.Errorf("non-terminating Ready endpoint must be reported as in-rotation") + } +} + +// TestIsPodInRotation_NoRoutableEndpoint_ReportsNotInRotation covers the +// two ways the scan finds nothing. Unlike IsPodDrained, an absent +// Service is not disambiguated here: a surge pod nothing routes to is +// simply not in rotation yet. +func TestIsPodInRotation_NoRoutableEndpoint_ReportsNotInRotation(t *testing.T) { + pod := testPod("ns", "p1") + tests := []struct { + name string + objs []client.Object + }{ + {name: "no slices at all"}, + { + name: "slices exist but none target the pod", + objs: []client.Object{sliceForService("ns", "svc-1", "svc", endpointSpec{ + podName: "p2", + ready: ptr.To(true), + })}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := newDrainTestClient(t, tt.objs...) + in, err := IsPodInRotation(context.Background(), c, "ns", "svc", pod) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if in { + t.Errorf("expected not-in-rotation, got in-rotation") + } + }) + } +} + +// TestIsPodInRotation_ListErrorPropagates pins that a failed read is an +// error, not a false negative. Swallowing it would let Migrate treat an +// unreadable cluster as "surge not serving yet" and stall forever. +func TestIsPodInRotation_ListErrorPropagates(t *testing.T) { + pod := testPod("ns", "p1") + boom := errors.New("list failed") + cr := &countingReader{Reader: newDrainTestClient(t), listErr: boom} + + in, err := IsPodInRotation(context.Background(), cr, "ns", "svc", pod) + if !errors.Is(err, boom) { + t.Fatalf("expected the list error to propagate, got %v", err) + } + if in { + t.Error("a failed read must not report the pod in rotation") + } +} + +func TestIsPodInRotation_NilPodAndEmptyServiceRejected(t *testing.T) { + c := newDrainTestClient(t) + if _, err := IsPodInRotation(context.Background(), c, "ns", "svc", nil); err == nil { + t.Error("expected an error for a nil pod") + } + if _, err := IsPodInRotation(context.Background(), c, "ns", "", testPod("ns", "p1")); err == nil { + t.Error("expected an error for an empty serviceName") + } +} diff --git a/pkg/controller/v1beta1/workload/podreadiness/readiness.go b/pkg/controller/v1beta1/workload/podreadiness/readiness.go new file mode 100644 index 000000000..6a54e3931 --- /dev/null +++ b/pkg/controller/v1beta1/workload/podreadiness/readiness.go @@ -0,0 +1,582 @@ +// Package podreadiness implements the multi-writer readiness gate +// protocol for OMENative's `ome.io/serving` pod condition. +// +// Pattern ported from RBG (`sigs.k8s.io/rbgs/pkg/inplace/pod/readiness`). +// The condition's Message field carries a JSON list of {UserAgent, Key} +// entries — one per writer that wants the pod NotReady right now. +// Status=True iff the list is empty AND containers are ready (the +// latter is kubelet's responsibility via the standard readiness gate +// machinery). +// +// Why multi-writer: OMENative has at least four overlapping reasons +// to want a pod NotReady — migration source drain, in-place update +// drain, restart drain, scale-down drain — that may be in flight on +// the same pod simultaneously. A single binary True/False gate has +// the writers race to overwrite each other; the multi-writer protocol +// lets them coexist as independent message-list entries. +// +// All writes use Status().Patch with strategic merge on the condition +// list (patch-merge-key=type), so kubelet's concurrent writes to +// PodScheduled / ContainersReady / PodReady are preserved. Every patch +// pins metadata.resourceVersion from the read it was computed against, +// so a stale base 409s instead of silently overwriting another +// writer's hold; retry.RetryOnConflict re-reads and re-applies. The +// in-loop re-read goes through the caller-supplied live reader (the +// AuthoritativeReader role): a watch-backed cache lagging the +// controller's own preceding status patch would re-serve the stale +// resourceVersion on every retry and the loop could never converge. +// +// # Gang readiness equivalence (multi-pod) +// +// Multi-pod Instances require gang-readiness: each pod should remain +// NotReady until every sibling in the Instance reaches ContainersReady. +// The multi-pod path carries no separately-keyed gang writer; the +// gang-readiness contract is enforced instead by the Instance-create +// flow, which calls MarkPodServing on the Instance's pods ONLY once +// every leader and every worker reports ContainersReady=True. Before +// that gate no pod in the gang has serving=True, so none receives +// traffic. +// +// That gate is observably equivalent to a per-pod gang writer: +// - A new pod's WriterLifecycle/KeyLifecycleInstanceReady stays in +// the message list until the controller writes the MarkPodServing +// removal in the create flow's Ready promotion. (The condition +// starts non-existent on a fresh pod, which is treated as "writer +// is implicitly holding"; the first MarkPodServing creates it with +// Status=True.) +// - The gate covers the WHOLE Instance, so the flip happens +// atomically across leader and workers. There is no partial +// "leader serving, workers still warming" window. +// +// A per-pod gang hold — wanted, for instance, by a partial in-place +// update on a multi-pod gang — is a separately-keyed writer the +// multi-writer protocol already admits without re-architecting. +package podreadiness + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// ConditionType is the controller-owned readiness gate OMENative +// stamps on every managed pod. The same value Render appends to +// pod.spec.readinessGates so kubelet AND's it into PodReady. +const ConditionType corev1.PodConditionType = "ome.io/serving" + +// Message identifies one writer that wants the pod NotReady. The +// pair {UserAgent, Key} is unique per logical reason — e.g., +// {"Update-in-place", "0-3"} for an in-place update on Instance 0 +// incarnation 3. +type Message struct { + UserAgent string `json:"userAgent"` + Key string `json:"key"` +} + +// ErrPodIdentityChanged means a same-name Pod's UID differs from the caller's +// observation. The replacement must not receive the stale effect. +var ErrPodIdentityChanged = errors.New("pod identity changed") + +// AddNotReadyKey appends msg to the readiness condition's message list +// and sets Status=False. No-op if msg is already present AND the +// condition Status is already False (the writer slot is already held). +// Patches only the single condition slot so kubelet's concurrent +// status writes are preserved; wrapped in retry.RetryOnConflict. +// +// reader is the live reader the RV-pinned patch base is read through +// (nil falls back to c, acceptable only when c does not lag the API +// server — e.g. tests). +// +// Self-healing branch: when the message list already contains msg but +// Status is True we still issue a patch to flip Status back to False. +// This resyncs a divergent shape — stale {UserAgent, Key} entries left +// in the message while Status has already flipped True. Without this +// recovery, a fresh in-place update reusing the same {idx, incarnation} +// key would short-circuit as "already held", no NotReady patch fires, +// drain.IsPodDrained sees the pod still Ready in the EndpointSlice, +// and the Instance hangs at Phase=Updating forever. With this branch, +// the first Add after a controller restart resyncs Status to match +// the message-list reality. +func AddNotReadyKey(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, msg Message) error { + if pod == nil { + return fmt.Errorf("AddNotReadyKey: nil pod") + } + if reader == nil { + reader = c + } + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + fresh := &corev1.Pod{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(pod), fresh); err != nil { + return fmt.Errorf("re-read pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + if pod.UID != "" && fresh.UID != pod.UID { + return fmt.Errorf("%w: pod %s/%s", ErrPodIdentityChanged, pod.Namespace, pod.Name) + } + cond := findCondition(fresh, ConditionType) + var base string + var existingStatus corev1.ConditionStatus + if cond != nil { + base = cond.Message + existingStatus = cond.Status + } + changed, list, err := addMessage(base, msg) + if err != nil { + return fmt.Errorf("pod %s/%s: %w", fresh.Namespace, fresh.Name, err) + } + if !changed && existingStatus == corev1.ConditionFalse { + // Nothing to do — key already in the list AND Status is + // already False (writer slot already held; everything + // consistent). + return nil + } + return patchCondition(ctx, c, fresh, corev1.PodCondition{ + Type: ConditionType, + Status: corev1.ConditionFalse, + Reason: "NotReady", + Message: list.dump(), + LastTransitionTime: transitionTime(cond, corev1.ConditionFalse), + }) + }) +} + +// transitionTime is the LastTransitionTime to stamp when writing status +// onto cond. A write that only reshuffles the writer list must carry the +// old timestamp forward, or every writer that joins or leaves resets the +// "how long has this pod been NotReady" clock consumers read. +func transitionTime(cond *corev1.PodCondition, status corev1.ConditionStatus) metav1.Time { + if cond != nil && cond.Status == status { + return cond.LastTransitionTime + } + return metav1.Now() +} + +// RemoveNotReadyKey removes msg from the readiness condition's +// message list. If the list becomes empty (or the condition doesn't +// exist yet — fresh pod case), sets Status=True. Otherwise keeps +// Status=False with the shrunken list. +// +// A pod that no longer exists is an ERROR (NotFound). This is the +// promote-to-serving contract: callers flip a pod into rotation and +// then act on that promotion (e.g. drain its predecessor in the same +// pass), so a silently-vanished pod must abort the caller before it +// removes the only serving replica. Drain-hold release paths, where +// the hold genuinely dies with the pod, use +// RemoveNotReadyKeyIgnoreNotFound instead. +// +// Pod names here are slot-based, so a same-name pod carrying a +// different UID is a replacement, not the caller's pod: that is +// ErrPodIdentityChanged, for the same reason. +// +// reader is the live reader the RV-pinned patch base is read through +// (nil falls back to c). Wrapped in retry.RetryOnConflict; patches +// only the single condition slot. +func RemoveNotReadyKey(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, msg Message) error { + _, err := removeNotReadyKey(ctx, c, reader, pod, msg, false) + return err +} + +// RemoveNotReadyKeyIgnoreNotFound is RemoveNotReadyKey for drain-hold +// release: a pod that no longer exists — vanished outright, or replaced +// under the same name by a pod with a different UID — is a no-op (nil) +// because the hold dies with the pod. Never use it on a +// promote-to-serving path — +// tolerating NotFound there turns "replacement is in rotation" into +// "replacement may be gone" and lets the caller drain its predecessor +// with nothing serving. +func RemoveNotReadyKeyIgnoreNotFound(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, msg Message) error { + _, err := removeNotReadyKey(ctx, c, reader, pod, msg, true) + return err +} + +func removeNotReadyKey(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, msg Message, ignoreNotFound bool) (bool, error) { + if pod == nil { + return false, fmt.Errorf("RemoveNotReadyKey: nil pod") + } + if reader == nil { + reader = c + } + changed := false + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + changed = false + fresh := &corev1.Pod{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(pod), fresh); err != nil { + if ignoreNotFound && apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("re-read pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + if pod.UID != "" && fresh.UID != pod.UID { + // A same-name replacement occupies the slot. The caller's + // hold died with its pod, so releasing it here would clear + // a gate the replacement never earned. For the tolerant + // variant that is the same no-op as a vanished pod. + if ignoreNotFound { + return nil + } + return fmt.Errorf("%w: pod %s/%s", ErrPodIdentityChanged, pod.Namespace, pod.Name) + } + cond := findCondition(fresh, ConditionType) + if cond == nil { + // An absent condition is the implicit Lifecycle hold on a + // fresh pod. Only the promote path may resolve it: that IS + // its Ready promotion. A drain-hold release has nothing to + // release here — its hold was never recorded — and writing + // Status=True would promote a pod no writer promoted, + // slipping past the Instance-wide gang gate. + if ignoreNotFound { + return nil + } + // Removing a key that isn't present is vacuously "no + // writers hold it NotReady" → write Status=True. + if err := patchCondition(ctx, c, fresh, corev1.PodCondition{ + Type: ConditionType, + Status: corev1.ConditionTrue, + Reason: "Serving", + LastTransitionTime: metav1.Now(), + }); err != nil { + return err + } + changed = true + return nil + } + messageChanged, list, err := removeMessage(cond.Message, msg) + if err != nil { + return fmt.Errorf("pod %s/%s: %w", fresh.Namespace, fresh.Name, err) + } + if !messageChanged && cond.Status == corev1.ConditionTrue { + // Key wasn't in the list AND condition already True. + // Nothing to do. + return nil + } + status := corev1.ConditionTrue + reason := "Serving" + message := "" + if len(list) > 0 { + status = corev1.ConditionFalse + reason = "NotReady" + message = list.dump() + } + // If only the status field needs updating, still issue a + // patch — the list dedup is in the helper. + if err := patchCondition(ctx, c, fresh, corev1.PodCondition{ + Type: ConditionType, + Status: status, + Reason: reason, + Message: message, + LastTransitionTime: transitionTime(cond, status), + }); err != nil { + return err + } + changed = messageChanged || cond.Status != status + return nil + }) + return changed, err +} + +// ContainsNotReadyKey reports whether the readiness condition holds +// msg as a live hold — msg is in the message list AND Status is False. +// Read-side check used by callers that want to skip a redundant Add. +// +// The Status=False requirement makes this "is the hold in effect", not +// "is the entry present". A condition in the paradoxical Status=True +// state with a non-empty list therefore reports false. That is what an +// Add caller wants, since AddNotReadyKey repairs the paradox on its +// next write. A Remove caller must NOT use this to skip its call: the +// stale entry is exactly what needs removing, and skipping leaves it. +func ContainsNotReadyKey(pod *corev1.Pod, msg Message) bool { + if pod == nil { + return false + } + cond := findCondition(pod, ConditionType) + if cond == nil || cond.Status == corev1.ConditionTrue || cond.Message == "" { + return false + } + list, err := parseList(cond.Message) + if err != nil { + return false + } + for _, m := range list { + if m == msg { + return true + } + } + return false +} + +// IsServing reports whether the condition exists with Status=True — +// true iff no writer currently holds the pod NotReady. +func IsServing(pod *corev1.Pod) bool { + if pod == nil { + return false + } + cond := findCondition(pod, ConditionType) + return cond != nil && cond.Status == corev1.ConditionTrue +} + +// IsContainersReady reports whether the kubelet-owned ContainersReady +// condition is True — i.e., all containers in the pod have passed their +// readiness probes. +// +// Distinct from corev1.PodReady (which ANDs ContainersReady with every +// readiness gate including ome.io/serving). Gating MarkPodServing on +// PodReady would deadlock: kubelet won't flip PodReady=True until +// ome.io/serving=True, and OMENative won't write ome.io/serving=True +// until PodReady=True. ContainersReady reflects only probe-level +// readiness and is the correct signal that the runtime is healthy +// enough to receive traffic. +func IsContainersReady(pod *corev1.Pod) bool { + if pod == nil { + return false + } + for _, c := range pod.Status.Conditions { + if c.Type == corev1.ContainersReady { + return c.Status == corev1.ConditionTrue + } + } + return false +} + +// IsPodReady reports whether the kubelet-owned PodReady condition is True — +// containers ready AND every readiness gate (including ome.io/serving) +// satisfied. This is the signal that the pod is actually in Service rotation, +// distinct from IsContainersReady (probes only) and IsServing (the gate +// condition alone, before kubelet re-evaluates PodReady). Use this to confirm +// a freshly-served replacement is carrying traffic before draining its source. +func IsPodReady(pod *corev1.Pod) bool { + if pod == nil { + return false + } + for _, c := range pod.Status.Conditions { + if c.Type == corev1.PodReady { + return c.Status == corev1.ConditionTrue + } + } + return false +} + +// MarkPodServing removes the {userAgent, key} entry from the +// ome.io/serving condition's writer list. If the resulting list is +// empty (or the condition didn't exist yet — fresh pod case), the +// condition is written with Status=True and the pod becomes eligible +// for Service rotation. +// +// Idempotent: removing a key that isn't present is a no-op (no patch +// issued) when the condition is already Status=True. A deleted pod is +// an error (see RemoveNotReadyKey): success means the pod IS in +// rotation, so promote-then-drain callers may safely act on it. +func MarkPodServing(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, userAgent, key string) error { + return RemoveNotReadyKey(ctx, c, reader, pod, Message{UserAgent: userAgent, Key: key}) +} + +// MarkPodServingWithChange is MarkPodServing plus whether this call committed +// a condition change. Callers use the result when they may need to compensate +// only the serving transition they own. +func MarkPodServingWithChange(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, userAgent, key string) (bool, error) { + return removeNotReadyKey(ctx, c, reader, pod, Message{UserAgent: userAgent, Key: key}, false) +} + +// MarkPodNotServing adds the {userAgent, key} entry to the +// ome.io/serving condition's writer list. Status becomes False (because +// the list is non-empty). The pod is removed from Service rotation via +// kube-proxy's standard PodReady gate machinery. +// +// Idempotent: re-adding the same key is a no-op. +func MarkPodNotServing(ctx context.Context, c client.Client, reader client.Reader, pod *corev1.Pod, userAgent, key string) error { + return AddNotReadyKey(ctx, c, reader, pod, Message{UserAgent: userAgent, Key: key}) +} + +// Writer userAgents used across the OMENative ops. Centralized here so +// a stray drift in any one site is easy to catch in `git grep`. +const ( + // WriterLifecycle flips a fresh pod's gate to Status=True once all + // Instance siblings reach ContainersReady. The key + // (KeyLifecycleInstanceReady) is the same across all sites — the + // userAgent + key tuple is the lookup; we never need to differentiate. + WriterLifecycle = "Lifecycle" + + // WriterUpdateInPlace drains a pod for in-place container image + // update. Same key on Add (drain) and Remove (un-drain) on the + // same pod. + WriterUpdateInPlace = "Update-in-place" + + // WriterUpdateRecreateDrain drains the OLD pods for a recreate + // update. The NEW pods are fresh and go through WriterLifecycle. + WriterUpdateRecreateDrain = "Update-recreate-drain" + + // WriterRestartDrain drains the OLD pods for an Instance restart. + // The NEW pods are fresh and go through WriterLifecycle. + WriterRestartDrain = "Restart-drain" + + // WriterDeleteDrain drains a pod for scale-down deletion. The key + // is removed by virtue of the pod being deleted; no explicit + // RemoveNotReadyKey is required. + WriterDeleteDrain = "Delete-drain" + + // WriterMigrateSourceDrain drains the source pods of a surge + // migration. The key is removed by virtue of the source pods being + // deleted at the end of the migration. + WriterMigrateSourceDrain = "Migrate-source-drain" + + // WriterUpdateSurgeDrain drains the OLD pod during a SurgeThenDrain + // rollout — after the surge pod (at the other ordinal slot) reaches + // Ready, the old pod is drained via this writer then deleted. The + // surge pod is fresh and goes through WriterLifecycle. + WriterUpdateSurgeDrain = "Update-surge-drain" +) + +// KeyLifecycleInstanceReady is the universal Lifecycle key. Matches +// RBG's convention. The pair (WriterLifecycle, KeyLifecycleInstanceReady) +// always means "the gate is unheld; if no other writer holds it, status +// is True". +const KeyLifecycleInstanceReady = "InstanceReady" + +// patchCondition issues a strategic-merge patch updating only the +// single ome.io/serving condition. Kubelet's concurrent writes to +// the other Pod.Status.Conditions entries (PodScheduled, Initialized, +// ContainersReady, Ready, ...) are preserved because patch-merge-key +// on PodCondition is `type`. +// +// The patch is hand-marshaled (not via corev1.PodCondition) because +// PodCondition.Message has `json:",omitempty"` — when the controller +// removes the last writer and writes Status=True with Message="" the +// typed Marshal omits the message field entirely, so strategic-merge +// keeps the stale message list from the previous Status=False patch. +// The next AddNotReadyKey call then sees the orphaned key still in +// the list, addMessage returns changed=false, no patch fires, the +// pod stays Status=True with a stale writer in the message, and +// drain.IsPodDrained never observes the pod leaving rotation — the +// in-place update stalls forever in Phase=Updating. Forcing the +// "message" field into the patch payload (even when empty) makes +// strategic-merge clear the stale list so the next writer cycle +// starts from a clean slate. +// +// The patch pins pod's resourceVersion (the fresh read the new list +// was computed from) so a stale base gets a 409 instead of silently +// dropping a concurrent writer's entry; the callers' RetryOnConflict +// then re-reads and recomputes. +func patchCondition(ctx context.Context, c client.Client, pod *corev1.Pod, cond corev1.PodCondition) error { + condMap := map[string]any{ + "type": string(cond.Type), + "status": string(cond.Status), + "reason": cond.Reason, + "message": cond.Message, + "lastTransitionTime": cond.LastTransitionTime, + } + patch := map[string]any{ + "metadata": map[string]any{ + "resourceVersion": pod.ResourceVersion, + }, + "status": map[string]any{ + "conditions": []any{condMap}, + }, + } + raw, err := json.Marshal(patch) + if err != nil { + return fmt.Errorf("marshal condition patch for pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + if err := c.Status().Patch(ctx, pod, client.RawPatch(types.StrategicMergePatchType, raw)); err != nil { + return fmt.Errorf("patch %s on pod %s/%s: %w", ConditionType, pod.Namespace, pod.Name, err) + } + return nil +} + +// findCondition returns a pointer to the condition matching condType +// in pod.Status.Conditions, or nil. Used by the multi-writer helpers +// to extract the current message list. +func findCondition(pod *corev1.Pod, condType corev1.PodConditionType) *corev1.PodCondition { + if pod == nil { + return nil + } + for i := range pod.Status.Conditions { + if pod.Status.Conditions[i].Type == condType { + return &pod.Status.Conditions[i] + } + } + return nil +} + +// addMessage appends msg to the list parsed from base (an empty +// string is treated as an empty list). Returns changed=false if msg +// is already in the list. +func addMessage(base string, msg Message) (bool, messageList, error) { + list, err := parseList(base) + if err != nil { + return false, nil, err + } + for _, m := range list { + if m == msg { + return false, list, nil + } + } + list = append(list, msg) + return true, list, nil +} + +// removeMessage drops msg from the list parsed from base. Returns +// changed=false if msg isn't in the list. +func removeMessage(base string, msg Message) (bool, messageList, error) { + list, err := parseList(base) + if err != nil { + return false, nil, err + } + var out messageList + var removed bool + for _, m := range list { + if m == msg { + removed = true + continue + } + out = append(out, m) + } + return removed, out, nil +} + +// parseList decodes a message list from its JSON serialization. An +// empty string yields nil ("no writers hold the gate"). A malformed +// string is an error, NOT an empty list — treating corruption as +// "no writers" would let RemoveNotReadyKey flip Status=True and +// release every other writer's drain hold with zero signal. +func parseList(raw string) (messageList, error) { + if raw == "" { + return nil, nil + } + var list messageList + if err := json.Unmarshal([]byte(raw), &list); err != nil { + return nil, fmt.Errorf("malformed %s writer list %q: %w", ConditionType, raw, err) + } + return list, nil +} + +// messageList is a sortable slice of Messages. Stored in a stable +// order so the serialized form is deterministic — operators +// inspecting `kubectl get pod -o yaml` see the same list across +// reconciles when nothing changed. +type messageList []Message + +func (l messageList) Len() int { return len(l) } +func (l messageList) Swap(i, j int) { l[i], l[j] = l[j], l[i] } +func (l messageList) Less(i, j int) bool { + if l[i].UserAgent == l[j].UserAgent { + return l[i].Key < l[j].Key + } + return l[i].UserAgent < l[j].UserAgent +} + +// dump serializes the list in stable {UserAgent, Key} sort order. +func (l messageList) dump() string { + if len(l) == 0 { + return "" + } + sort.Sort(l) + raw, _ := json.Marshal(l) + return string(raw) +} diff --git a/pkg/controller/v1beta1/workload/podreadiness/readiness_test.go b/pkg/controller/v1beta1/workload/podreadiness/readiness_test.go new file mode 100644 index 000000000..ef60d3005 --- /dev/null +++ b/pkg/controller/v1beta1/workload/podreadiness/readiness_test.go @@ -0,0 +1,964 @@ +package podreadiness + +import ( + "context" + "errors" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +func newReadinessTestClient(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1: %v", err) + } + return fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithStatusSubresource(&corev1.Pod{}). + Build() +} + +func mustParseList(t *testing.T, raw string) messageList { + t.Helper() + list, err := parseList(raw) + if err != nil { + t.Fatalf("parseList(%q): %v", raw, err) + } + return list +} + +func newReadinessTestPod(name string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: "default", + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "main", Image: "x"}}, + ReadinessGates: []corev1.PodReadinessGate{{ + ConditionType: ConditionType, + }}, + }, + } +} + +func TestFreshPod_RemoveNotReadyKey_WritesStatusTrue(t *testing.T) { + // Pod has no condition. The Lifecycle writer's Remove is the Ready + // promotion — it creates the condition with Status=True. + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + if err := RemoveNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}); err != nil { + t.Fatalf("RemoveNotReadyKey: %v", err) + } + + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil { + t.Fatalf("condition not written") + } + if cond.Status != corev1.ConditionTrue { + t.Errorf("expected Status=True, got %s", cond.Status) + } + if cond.Message != "" { + t.Errorf("expected empty message, got %q", cond.Message) + } +} + +func TestMarkPodServingWithChangeReportsOnlyOwnedTransition(t *testing.T) { + t.Run("fresh condition", func(t *testing.T) { + pod := newReadinessTestPod("p-change-fresh") + c := newReadinessTestClient(t, pod) + + changed, err := MarkPodServingWithChange(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if err != nil { + t.Fatalf("first MarkPodServingWithChange: %v", err) + } + if !changed { + t.Fatal("fresh Pod serving promotion must report a committed change") + } + changed, err = MarkPodServingWithChange(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if err != nil { + t.Fatalf("idempotent MarkPodServingWithChange: %v", err) + } + if changed { + t.Fatal("already-serving Pod must not report a change owned by this call") + } + }) + + t.Run("held lifecycle key", func(t *testing.T) { + pod := newReadinessTestPod("p-change-held") + c := newReadinessTestClient(t, pod) + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + + changed, err := MarkPodServingWithChange(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if err != nil { + t.Fatalf("MarkPodServingWithChange: %v", err) + } + if !changed { + t.Fatal("removing the held Lifecycle key must report a committed change") + } + }) + + t.Run("unrelated writer key", func(t *testing.T) { + pod := newReadinessTestPod("p-change-unrelated") + c := newReadinessTestClient(t, pod) + other := Message{UserAgent: WriterDeleteDrain, Key: "delete-0"} + if err := AddNotReadyKey(context.Background(), c, c, pod, other); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + + changed, err := MarkPodServingWithChange(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if err != nil { + t.Fatalf("MarkPodServingWithChange: %v", err) + } + if changed { + t.Fatal("an unrelated writer's hold must not be reported as a Lifecycle-owned transition") + } + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get Pod: %v", err) + } + if !ContainsNotReadyKey(got, other) { + t.Fatal("unrelated writer hold was not preserved") + } + }) + + t.Run("inconsistent false condition", func(t *testing.T) { + pod := newReadinessTestPod("p-change-inconsistent") + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionFalse, + }} + c := newReadinessTestClient(t, pod) + + changed, err := MarkPodServingWithChange(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if err != nil { + t.Fatalf("MarkPodServingWithChange: %v", err) + } + if !changed { + t.Fatal("self-healing False to True must report an owned serving transition") + } + }) +} + +func TestAddNotReadyKey_SetsStatusFalse(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "Delete-drain", Key: "0"}); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil || cond.Status != corev1.ConditionFalse { + t.Fatalf("expected Status=False, got %+v", cond) + } + list := mustParseList(t, cond.Message) + if len(list) != 1 || list[0].UserAgent != "Delete-drain" || list[0].Key != "0" { + t.Errorf("unexpected message list: %v", list) + } +} + +func TestMultiWriter_TwoKeysHoldNotReady(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + got := &corev1.Pod{} + reread := func(stage string) { + t.Helper() + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get pod after %s: %v", stage, err) + } + } + condition := func(stage string) corev1.PodCondition { + t.Helper() + cond := findCondition(got, ConditionType) + if cond == nil { + t.Fatalf("condition missing after %s", stage) + } + return *cond + } + + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Add A: %v", err) + } + reread("Add A") + if err := AddNotReadyKey(context.Background(), c, c, got, Message{UserAgent: "B", Key: "2"}); err != nil { + t.Fatalf("Add B: %v", err) + } + + // Removing only A — pod must stay NotReady because B still holds it. + reread("Add B") + if err := RemoveNotReadyKey(context.Background(), c, c, got, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Remove A: %v", err) + } + reread("Remove A") + cond := condition("Remove A") + if cond.Status != corev1.ConditionFalse { + t.Errorf("expected Status=False after removing one of two keys, got %s", cond.Status) + } + list := mustParseList(t, cond.Message) + if len(list) != 1 || list[0].UserAgent != "B" { + t.Errorf("expected only B remaining, got %v", list) + } + + // Now remove B — pod must flip Ready. + if err := RemoveNotReadyKey(context.Background(), c, c, got, Message{UserAgent: "B", Key: "2"}); err != nil { + t.Fatalf("Remove B: %v", err) + } + reread("Remove B") + if cond := condition("Remove B"); cond.Status != corev1.ConditionTrue { + t.Errorf("expected Status=True after removing last key, got %s", cond.Status) + } +} + +func TestAddNotReadyKey_Idempotent(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("first Add: %v", err) + } + first := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), first) + firstRV := first.ResourceVersion + + // Second Add of the same key must be a no-op (no patch issued). + if err := AddNotReadyKey(context.Background(), c, c, first, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("second Add: %v", err) + } + second := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), second) + if second.ResourceVersion != firstRV { + t.Errorf("ResourceVersion bumped on idempotent Add: %s -> %s", firstRV, second.ResourceVersion) + } +} + +func TestRemoveNotReadyKey_IdempotentOnUnknownKey(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + // Seed Status=True via the Lifecycle promotion. + if err := RemoveNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}); err != nil { + t.Fatalf("seed Remove: %v", err) + } + first := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), first) + firstRV := first.ResourceVersion + + // Removing a key that was never added must be a no-op — condition + // already True, key isn't in the list. + if err := RemoveNotReadyKey(context.Background(), c, c, first, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("second Remove: %v", err) + } + second := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), second) + if second.ResourceVersion != firstRV { + t.Errorf("ResourceVersion bumped on idempotent Remove: %s -> %s", firstRV, second.ResourceVersion) + } +} + +func TestContainsNotReadyKey(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + if got := ContainsNotReadyKey(pod, Message{UserAgent: "A", Key: "1"}); got { + t.Errorf("fresh pod has no keys; ContainsNotReadyKey should be false") + } + + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Add: %v", err) + } + got := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), got) + if !ContainsNotReadyKey(got, Message{UserAgent: "A", Key: "1"}) { + t.Errorf("expected ContainsNotReadyKey=true after Add") + } + if ContainsNotReadyKey(got, Message{UserAgent: "B", Key: "2"}) { + t.Errorf("expected ContainsNotReadyKey=false for unrelated key") + } +} + +func TestIsServing(t *testing.T) { + if IsServing(nil) { + t.Errorf("nil pod should not be serving") + } + pod := newReadinessTestPod("p") + if IsServing(pod) { + t.Errorf("fresh pod should not be serving (no condition)") + } + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionFalse, + }} + if IsServing(pod) { + t.Errorf("Status=False should not be serving") + } + pod.Status.Conditions[0].Status = corev1.ConditionTrue + if !IsServing(pod) { + t.Errorf("Status=True should be serving") + } +} + +func TestAddRemove_PreservesKubeletConditions(t *testing.T) { + // Patches must NOT clobber kubelet-managed conditions like + // ContainersReady — the strategic-merge patch keyed by `type` leaves + // other condition slots alone. + pod := newReadinessTestPod("p") + pod.Status.Conditions = []corev1.PodCondition{ + {Type: corev1.PodScheduled, Status: corev1.ConditionTrue}, + {Type: corev1.ContainersReady, Status: corev1.ConditionTrue}, + } + c := newReadinessTestClient(t, pod) + + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "Delete-drain", Key: "0"}); err != nil { + t.Fatalf("Add: %v", err) + } + got := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), got) + + if findCondition(got, corev1.PodScheduled) == nil { + t.Errorf("kubelet's PodScheduled condition was clobbered") + } + if findCondition(got, corev1.ContainersReady) == nil { + t.Errorf("kubelet's ContainersReady condition was clobbered") + } + if findCondition(got, ConditionType) == nil { + t.Errorf("our condition was not written") + } +} + +func TestAddNotReadyKey_NilPodRejected(t *testing.T) { + c := newReadinessTestClient(t) + if err := AddNotReadyKey(context.Background(), c, c, nil, Message{UserAgent: "x", Key: "y"}); err == nil { + t.Fatalf("expected error on nil pod") + } +} + +func TestRemoveNotReadyKey_NilPodRejected(t *testing.T) { + c := newReadinessTestClient(t) + if err := RemoveNotReadyKey(context.Background(), c, c, nil, Message{UserAgent: "x", Key: "y"}); err == nil { + t.Fatalf("expected error on nil pod") + } +} + +// TestRemoveLastKey_ClearsMessageField pins the bug where in-place +// updates stalled forever after a second annotation-only patch. +// +// Background: PodCondition.Message has `json:",omitempty"`, so the +// typed corev1.PodCondition.MarshalJSON drops the field when empty. +// When the controller removed the last writer and re-issued the patch +// with Status=True + Message="" the strategic-merge payload contained +// no `message` key, so the apiserver preserved the prior Status=False +// message list. The pod ended up Status=True with a stale writer +// still in the message — the next in-place update's AddNotReadyKey +// found its (UserAgent, Key) tuple already present, short-circuited +// without issuing a Status=False patch, and drain.IsPodDrained +// observed the pod still in rotation. The Instance state machine +// stalled at Phase=Updating; readyReplicas read 0 for the entire +// post-patch window. +// +// Assertion: after the last writer is removed the message field is +// actually empty on the persisted pod. Without the patchCondition +// hand-marshal fix the assertion fails — the message survives. +func TestRemoveLastKey_ClearsMessageField(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + // Add a writer to populate Status=False + non-empty message. + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "Update-in-place", Key: "0-1"}); err != nil { + t.Fatalf("Add: %v", err) + } + mid := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), mid) + if cond := findCondition(mid, ConditionType); cond == nil || cond.Status != corev1.ConditionFalse || cond.Message == "" { + t.Fatalf("seed: expected Status=False with non-empty message, got %+v", cond) + } + + // Remove the writer — list becomes empty, Status flips True. + if err := RemoveNotReadyKey(context.Background(), c, c, mid, Message{UserAgent: "Update-in-place", Key: "0-1"}); err != nil { + t.Fatalf("Remove: %v", err) + } + got := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), got) + cond := findCondition(got, ConditionType) + if cond == nil { + t.Fatalf("condition not found after Remove") + } + if cond.Status != corev1.ConditionTrue { + t.Errorf("expected Status=True after removing last writer, got %s", cond.Status) + } + if cond.Message != "" { + t.Errorf("expected empty message after removing last writer, got %q (stale entry would deadlock the next in-place update)", cond.Message) + } +} + +// TestSecondInPlaceCycle_CanReDrain pins the end-to-end shape of the +// reported regression: two back-to-back in-place updates on the same +// (Instance, Incarnation) must each be able to flip the pod to +// Status=False. The first cycle drains, ContainersReady stays True, +// we re-mark Serving; the second cycle issues the same {UserAgent, +// Key} tuple — it MUST take effect, not short-circuit on a stale +// message list. +func TestSecondInPlaceCycle_CanReDrain(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + key := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + + // Cycle 1: drain → un-drain. + if err := AddNotReadyKey(context.Background(), c, c, pod, key); err != nil { + t.Fatalf("cycle 1 Add: %v", err) + } + mid := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), mid) + if err := RemoveNotReadyKey(context.Background(), c, c, mid, key); err != nil { + t.Fatalf("cycle 1 Remove: %v", err) + } + + // Cycle 2: drain again — MUST flip Status to False. + cycle2Start := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), cycle2Start) + if !IsServing(cycle2Start) { + t.Fatalf("cycle 2 precondition: expected Serving=True after cycle 1 finished") + } + if err := AddNotReadyKey(context.Background(), c, c, cycle2Start, key); err != nil { + t.Fatalf("cycle 2 Add: %v", err) + } + after := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), after) + cond := findCondition(after, ConditionType) + if cond == nil { + t.Fatalf("cycle 2: condition missing") + } + if cond.Status != corev1.ConditionFalse { + t.Errorf("cycle 2 expected Status=False after Add, got %s — second-cycle drain would never converge", cond.Status) + } + list := mustParseList(t, cond.Message) + if len(list) != 1 || list[0] != key { + t.Errorf("cycle 2 expected single {%s, %s} entry, got %v", key.UserAgent, key.Key, list) + } +} + +// TestAddNotReadyKey_RecoversInconsistentStatusTrueWithStaleMessage +// pins the self-healing branch in AddNotReadyKey. Existing pods that +// ran the pre-fix controller may be sitting at the paradoxical +// {Status=True, Message=[stale-writer]} state because the +// last-writer-removed patch omitted Message via json:",omitempty". +// The next in-place update cycle MUST be able to drain those pods — +// otherwise the only recovery path is operator-triggered pod restarts. +// +// Setup: synthesize a pod in the paradoxical state. Call +// AddNotReadyKey for the same {UserAgent, Key} tuple that's already +// in the list (mirroring a follow-up update reusing the same +// {Instance, Incarnation}). Assert Status flips to False so drain +// can proceed. +func TestAddNotReadyKey_RecoversInconsistentStatusTrueWithStaleMessage(t *testing.T) { + staleMsg := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + staleList := messageList{staleMsg} + pod := newReadinessTestPod("p") + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionTrue, // paradox: True but list non-empty + Reason: "Serving", + Message: staleList.dump(), + }} + c := newReadinessTestClient(t, pod) + + if err := AddNotReadyKey(context.Background(), c, c, pod, staleMsg); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + + got := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), got) + cond := findCondition(got, ConditionType) + if cond == nil { + t.Fatalf("condition missing after Add") + } + if cond.Status != corev1.ConditionFalse { + t.Errorf("expected Status=False after self-heal Add on paradoxical condition, got %s — drain would never converge for an in-place rollout reusing the same key", cond.Status) + } + list := mustParseList(t, cond.Message) + if len(list) != 1 || list[0] != staleMsg { + t.Errorf("expected single {%s, %s} entry after self-heal, got %v", staleMsg.UserAgent, staleMsg.Key, list) + } +} + +// TestAddNotReadyKey_ConflictRetryPreservesConcurrentWriter pins the +// RV-pinned patch: a competing writer landing between our re-read and +// our patch must force a 409 so RetryOnConflict recomputes against the +// new base. Without the resourceVersion in the patch body the stale +// patch applies cleanly and silently drops the competing writer's hold. +func TestAddNotReadyKey_ConflictRetryPreservesConcurrentWriter(t *testing.T) { + pod := newReadinessTestPod("p") + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1: %v", err) + } + competitor := Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"} + raced := false + c := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(pod). + WithStatusSubresource(&corev1.Pod{}). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourcePatch: func(ctx context.Context, cl client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if !raced { + raced = true + live := &corev1.Pod{} + if err := cl.Get(ctx, client.ObjectKeyFromObject(pod), live); err != nil { + t.Fatalf("interceptor get: %v", err) + } + if err := AddNotReadyKey(ctx, cl, cl, live, competitor); err != nil { + t.Fatalf("interceptor competing Add: %v", err) + } + } + return cl.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + ours := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + if err := AddNotReadyKey(context.Background(), c, c, pod, ours); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil || cond.Status != corev1.ConditionFalse { + t.Fatalf("expected Status=False, got %+v", cond) + } + list := mustParseList(t, cond.Message) + if len(list) != 2 { + t.Fatalf("expected both writers' holds to survive the race, got %v", list) + } + seen := map[Message]bool{} + for _, m := range list { + seen[m] = true + } + if !seen[ours] || !seen[competitor] { + t.Errorf("stale-base patch dropped a concurrent writer's hold: %v", list) + } +} + +func TestPatchCondition_StaleBaseRejected(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + stale := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), stale); err != nil { + t.Fatalf("get: %v", err) + } + // Bump the stored pod so the copy above goes stale. + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Add: %v", err) + } + + err := patchCondition(context.Background(), c, stale, corev1.PodCondition{ + Type: ConditionType, + Status: corev1.ConditionTrue, + Reason: "Serving", + LastTransitionTime: metav1.Now(), + }) + if !apierrors.IsConflict(err) { + t.Fatalf("expected conflict from stale-base patch, got %v", err) + } +} + +// TestRemoveNotReadyKey_PodGone_Errors pins the promote-to-serving +// contract: a deleted pod must surface NotFound, not silently succeed. +// A surge pod evicted between the reconcile-start snapshot and the +// promote patch would otherwise let the caller "promote" nothing and +// drain the old pod in the same pass — an availability outage (a full +// outage at replicas=1) with zero signal. +func TestRemoveNotReadyKey_PodGone_Errors(t *testing.T) { + pod := newReadinessTestPod("gone") + c := newReadinessTestClient(t) + err := RemoveNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}) + if !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound for a deleted pod on the promote path, got %v", err) + } +} + +// TestMarkPodServing_PodGone_Errors pins the same contract through the +// wrapper every promote caller uses. +func TestMarkPodServing_PodGone_Errors(t *testing.T) { + pod := newReadinessTestPod("gone") + c := newReadinessTestClient(t) + err := MarkPodServing(context.Background(), c, c, pod, WriterLifecycle, KeyLifecycleInstanceReady) + if !apierrors.IsNotFound(err) { + t.Fatalf("expected NotFound from MarkPodServing on a deleted pod, got %v", err) + } +} + +// TestRemoveNotReadyKeyIgnoreNotFound_PodGone_ReturnsNil pins the +// drain-hold release contract: the hold dies with the pod, so a +// deleted pod is a clean no-op (migration-expiry un-drain of source +// pods that may already be gone). +func TestRemoveNotReadyKeyIgnoreNotFound_PodGone_ReturnsNil(t *testing.T) { + pod := newReadinessTestPod("gone") + c := newReadinessTestClient(t) + if err := RemoveNotReadyKeyIgnoreNotFound(context.Background(), c, c, pod, Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"}); err != nil { + t.Fatalf("expected nil for a deleted pod on drain-hold release, got %v", err) + } +} + +// TestLastTransitionTime_MovesOnlyOnStatusChange pins the condition +// timestamp contract. Writers join and leave the message list +// constantly while Status stays False; restamping on every one of +// those writes resets the "how long has this pod been NotReady" clock +// consumers measure against. +func TestLastTransitionTime_MovesOnlyOnStatusChange(t *testing.T) { + held := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + joining := Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"} + original := metav1.NewTime(time.Now().Add(-time.Hour).Truncate(time.Second)) + + pod := newReadinessTestPod("p") + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionFalse, + Reason: "NotReady", + Message: messageList{held}.dump(), + LastTransitionTime: original, + }} + c := newReadinessTestClient(t, pod) + + readCondition := func(stage string) corev1.PodCondition { + t.Helper() + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get pod after %s: %v", stage, err) + } + cond := findCondition(got, ConditionType) + if cond == nil { + t.Fatalf("condition missing after %s", stage) + } + return *cond + } + + // A second writer joins: list grows, Status stays False. + if err := AddNotReadyKey(context.Background(), c, c, pod, joining); err != nil { + t.Fatalf("AddNotReadyKey: %v", err) + } + if cond := readCondition("add"); !cond.LastTransitionTime.Equal(&original) { + t.Errorf("Status stayed False, so LastTransitionTime must hold at %v, got %v", original, cond.LastTransitionTime) + } + + // One writer leaves: list shrinks, Status still False. + if err := RemoveNotReadyKey(context.Background(), c, c, pod, held); err != nil { + t.Fatalf("RemoveNotReadyKey: %v", err) + } + if cond := readCondition("partial remove"); !cond.LastTransitionTime.Equal(&original) { + t.Errorf("Status stayed False, so LastTransitionTime must hold at %v, got %v", original, cond.LastTransitionTime) + } + + // Last writer leaves: Status transitions to True. + if err := RemoveNotReadyKey(context.Background(), c, c, pod, joining); err != nil { + t.Fatalf("RemoveNotReadyKey: %v", err) + } + cond := readCondition("final remove") + if cond.Status != corev1.ConditionTrue { + t.Fatalf("expected Status=True once the list emptied, got %s", cond.Status) + } + if cond.LastTransitionTime.Equal(&original) { + t.Errorf("Status changed to True, so LastTransitionTime must advance past %v", original) + } +} + +// TestRemoveNotReadyKey_ReplacementPod_Errors pins the promote-path +// identity contract. Pod names are slot-based, so a same-name pod +// carrying a different UID is a replacement that never held the +// caller's key. Releasing it would create the condition with +// Status=True and hand the replacement a promotion it did not earn. +func TestRemoveNotReadyKey_ReplacementPod_Errors(t *testing.T) { + stored := newReadinessTestPod("slot-0") + stored.UID = "replacement" + observed := newReadinessTestPod("slot-0") + observed.UID = "original" + + c := newReadinessTestClient(t, stored) + err := RemoveNotReadyKey(context.Background(), c, c, observed, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}) + if !errors.Is(err, ErrPodIdentityChanged) { + t.Fatalf("expected ErrPodIdentityChanged for a same-name replacement, got %v", err) + } + + fresh := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(stored), fresh); err != nil { + t.Fatalf("get replacement: %v", err) + } + if cond := findCondition(fresh, ConditionType); cond != nil { + t.Fatalf("replacement condition must be untouched, got %+v", cond) + } +} + +// TestMarkPodServingWithChange_ReplacementPod_ReportsNoChange pins the +// same contract through the wrapper promote callers use. A change +// report here would drain the predecessor on a promotion the +// replacement never earned. +func TestMarkPodServingWithChange_ReplacementPod_ReportsNoChange(t *testing.T) { + stored := newReadinessTestPod("slot-0") + stored.UID = "replacement" + observed := newReadinessTestPod("slot-0") + observed.UID = "original" + + c := newReadinessTestClient(t, stored) + changed, err := MarkPodServingWithChange(context.Background(), c, c, observed, WriterLifecycle, KeyLifecycleInstanceReady) + if !errors.Is(err, ErrPodIdentityChanged) { + t.Fatalf("expected ErrPodIdentityChanged, got %v", err) + } + if changed { + t.Fatal("a replacement pod must never report a serving transition") + } +} + +// TestRemoveNotReadyKeyIgnoreNotFound_ReplacementPod_ReturnsNil pins the +// drain-hold release side: the hold died with the caller's pod, so a +// same-name replacement is the same clean no-op as a vanished pod — and +// must not have its condition written. +func TestRemoveNotReadyKeyIgnoreNotFound_ReplacementPod_ReturnsNil(t *testing.T) { + stored := newReadinessTestPod("slot-0") + stored.UID = "replacement" + observed := newReadinessTestPod("slot-0") + observed.UID = "original" + + c := newReadinessTestClient(t, stored) + if err := RemoveNotReadyKeyIgnoreNotFound(context.Background(), c, c, observed, Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"}); err != nil { + t.Fatalf("expected nil for a same-name replacement on drain-hold release, got %v", err) + } + + fresh := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(stored), fresh); err != nil { + t.Fatalf("get replacement: %v", err) + } + if cond := findCondition(fresh, ConditionType); cond != nil { + t.Fatalf("replacement condition must be untouched, got %+v", cond) + } +} + +// TestRemoveNotReadyKeyIgnoreNotFound_AbsentCondition_LeavesPodUnpromoted +// pins the asymmetry between the two variants on a pod with no +// condition. An absent condition is the implicit Lifecycle hold, and +// only the promote path may resolve it. A drain-hold release that wrote +// Status=True there would promote a pod no writer promoted and slip +// past the Instance-wide gang gate — reachable whenever the caller's +// pod reference carries no UID and the identity guard is skipped. +func TestRemoveNotReadyKeyIgnoreNotFound_AbsentCondition_LeavesPodUnpromoted(t *testing.T) { + pod := newReadinessTestPod("p") + c := newReadinessTestClient(t, pod) + + if err := RemoveNotReadyKeyIgnoreNotFound(context.Background(), c, c, pod, Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"}); err != nil { + t.Fatalf("RemoveNotReadyKeyIgnoreNotFound: %v", err) + } + + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get pod: %v", err) + } + if cond := findCondition(got, ConditionType); cond != nil { + t.Fatalf("drain-hold release must not create the gate, got %+v", cond) + } + + // The promote path still owns fresh-pod promotion. + if err := RemoveNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}); err != nil { + t.Fatalf("RemoveNotReadyKey: %v", err) + } + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get pod after promote: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil || cond.Status != corev1.ConditionTrue { + t.Fatalf("promote path must still create the gate as True, got %+v", cond) + } +} + +// TestContainsNotReadyKey_ParadoxicalStatusTrueReportsFalse pins the +// documented precondition: the check answers "is this hold in effect", +// so an entry stranded under Status=True reports false. +func TestContainsNotReadyKey_ParadoxicalStatusTrueReportsFalse(t *testing.T) { + msg := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + pod := newReadinessTestPod("p") + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionTrue, + Reason: "Serving", + Message: messageList{msg}.dump(), + }} + if ContainsNotReadyKey(pod, msg) { + t.Error("an entry stranded under Status=True is not a hold in effect") + } +} + +// TestRemoveNotReadyKey_UnsetCallerUID_SkipsIdentityCheck pins that an +// observation carrying no UID still releases. Callers that synthesize a +// pod reference from a slot name have nothing to compare against. +func TestRemoveNotReadyKey_UnsetCallerUID_SkipsIdentityCheck(t *testing.T) { + stored := newReadinessTestPod("slot-0") + stored.UID = "replacement" + observed := newReadinessTestPod("slot-0") + + c := newReadinessTestClient(t, stored) + if err := RemoveNotReadyKey(context.Background(), c, c, observed, Message{UserAgent: WriterLifecycle, Key: KeyLifecycleInstanceReady}); err != nil { + t.Fatalf("expected release to proceed without a caller UID, got %v", err) + } +} + +// TestAddNotReadyKey_LaggingCacheConvergesViaLiveReader pins the +// live-reader re-read: the RV-pinned patch base must come from the +// reader, not the (possibly lagging) cached client. A cache that keeps +// serving the resourceVersion from before another writer's patch would +// otherwise feed the same stale base to every RetryOnConflict attempt — +// the loop 409s until the budget is exhausted and the whole pass fails +// instead of converging. +func TestAddNotReadyKey_LaggingCacheConvergesViaLiveReader(t *testing.T) { + pod := newReadinessTestPod("p") + scheme := runtime.NewScheme() + if err := corev1.AddToScheme(scheme); err != nil { + t.Fatalf("add corev1: %v", err) + } + live := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(pod). + WithStatusSubresource(&corev1.Pod{}). + Build() + + // Freeze a "cache" snapshot, then land a competing writer so the + // stored pod's resourceVersion moves past it. + staleSnapshot := &corev1.Pod{} + if err := live.Get(context.Background(), client.ObjectKeyFromObject(pod), staleSnapshot); err != nil { + t.Fatalf("snapshot get: %v", err) + } + competitor := Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"} + if err := AddNotReadyKey(context.Background(), live, live, pod, competitor); err != nil { + t.Fatalf("competing Add: %v", err) + } + + // lagging serves the frozen snapshot on every Get; writes pass through. + lagging := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(pod). + WithStatusSubresource(&corev1.Pod{}). + WithInterceptorFuncs(interceptor.Funcs{ + Get: func(ctx context.Context, cl client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + staleSnapshot.DeepCopyInto(obj.(*corev1.Pod)) + return nil + }, + SubResourcePatch: func(ctx context.Context, cl client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + return live.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + }). + Build() + + ours := Message{UserAgent: WriterUpdateInPlace, Key: "0-1"} + if err := AddNotReadyKey(context.Background(), lagging, live, pod, ours); err != nil { + t.Fatalf("AddNotReadyKey against a lagging cache must converge via the live reader, got %v", err) + } + + got := &corev1.Pod{} + if err := live.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil || cond.Status != corev1.ConditionFalse { + t.Fatalf("expected Status=False, got %+v", cond) + } + list := mustParseList(t, cond.Message) + seen := map[Message]bool{} + for _, m := range list { + seen[m] = true + } + if len(list) != 2 || !seen[ours] || !seen[competitor] { + t.Errorf("expected both writers' holds after converging, got %v", list) + } +} + +// TestMalformedWriterList_FailsSafe pins the corrupt-state behavior: +// a writer list that doesn't parse must surface an error, not decode +// to "no writers hold the gate" and release other writers' drain holds. +func TestMalformedWriterList_FailsSafe(t *testing.T) { + const garbage = "{not-json" + pod := newReadinessTestPod("p") + pod.Status.Conditions = []corev1.PodCondition{{ + Type: ConditionType, + Status: corev1.ConditionFalse, + Reason: "NotReady", + Message: garbage, + }} + c := newReadinessTestClient(t, pod) + msg := Message{UserAgent: WriterMigrateSourceDrain, Key: "uuid-1"} + + if err := RemoveNotReadyKey(context.Background(), c, c, pod, msg); err == nil { + t.Errorf("Remove on malformed list must error, not release the gate") + } + if err := AddNotReadyKey(context.Background(), c, c, pod, msg); err == nil { + t.Errorf("Add on malformed list must error, not rebuild the list") + } + + got := &corev1.Pod{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(pod), got); err != nil { + t.Fatalf("get: %v", err) + } + cond := findCondition(got, ConditionType) + if cond == nil || cond.Status != corev1.ConditionFalse || cond.Message != garbage { + t.Errorf("malformed condition must be left untouched, got %+v", cond) + } + if ContainsNotReadyKey(got, msg) { + t.Errorf("ContainsNotReadyKey on malformed list must be false") + } +} + +func TestMessageListSerialization_Deterministic(t *testing.T) { + // Two writers adding in opposite orders should produce the same + // serialized list, so operators inspecting the pod see stable + // output between reconciles when nothing changed. + pod := newReadinessTestPod("p1") + c := newReadinessTestClient(t, pod) + if err := AddNotReadyKey(context.Background(), c, c, pod, Message{UserAgent: "B", Key: "2"}); err != nil { + t.Fatalf("Add B: %v", err) + } + p1 := &corev1.Pod{} + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), p1) + if err := AddNotReadyKey(context.Background(), c, c, p1, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Add A: %v", err) + } + _ = c.Get(context.Background(), client.ObjectKeyFromObject(pod), p1) + cond1 := findCondition(p1, ConditionType) + + pod2 := newReadinessTestPod("p2") + c2 := newReadinessTestClient(t, pod2) + if err := AddNotReadyKey(context.Background(), c2, c2, pod2, Message{UserAgent: "A", Key: "1"}); err != nil { + t.Fatalf("Add A: %v", err) + } + p2 := &corev1.Pod{} + _ = c2.Get(context.Background(), client.ObjectKeyFromObject(pod2), p2) + if err := AddNotReadyKey(context.Background(), c2, c2, p2, Message{UserAgent: "B", Key: "2"}); err != nil { + t.Fatalf("Add B: %v", err) + } + _ = c2.Get(context.Background(), client.ObjectKeyFromObject(pod2), p2) + cond2 := findCondition(p2, ConditionType) + + if cond1.Message != cond2.Message { + t.Errorf("Message ordering not stable:\n p1: %s\n p2: %s", cond1.Message, cond2.Message) + } +}