Merge pull request #143291 from Jefftree/service-decode-decorator

apiserver: decorate Service and PVC objects on storage decode

pkg/registry/core/persistentvolumeclaim/storage/storage.go

@@ -57,7 +57,14 @@ func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST, error) {
 
 		TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)},
 	}
-	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: persistentvolumeclaim.GetAttrs}
+
+	rest := &REST{store}
+
+	options := &generic.StoreOptions{
+		RESTOptions:     optsGetter,
+		AttrFunc:        persistentvolumeclaim.GetAttrs,
+		DecodeDecorator: rest.defaultOnRead, // plumb the decorator down into storage
+	}
 	if err := store.CompleteWithOptions(options); err != nil {
 		return nil, nil, err
 	}
@@ -66,9 +73,6 @@ func NewREST(optsGetter generic.RESTOptionsGetter) (*REST, *StatusREST, error) {
 	statusStore.UpdateStrategy = persistentvolumeclaim.StatusStrategy
 	statusStore.ResetFieldsStrategy = persistentvolumeclaim.StatusStrategy
 
-	rest := &REST{store}
-	store.Decorator = rest.defaultOnRead
-
 	return rest, &StatusREST{store: &statusStore}, nil
 }
 

pkg/registry/core/service/storage/storage.go

@@ -96,17 +96,6 @@ func NewREST(
 
 		TableConvertor: printerstorage.TableConvertor{TableGenerator: printers.NewTableGenerator().With(printersinternal.AddHandlers)},
 	}
-	options := &generic.StoreOptions{
-		RESTOptions: optsGetter,
-		AttrFunc:    svcreg.GetAttrs,
-	}
-	if err := store.CompleteWithOptions(options); err != nil {
-		return nil, nil, nil, err
-	}
-
-	statusStore := *store
-	statusStore.UpdateStrategy = svcreg.StatusStrategy
-	statusStore.ResetFieldsStrategy = svcreg.StatusStrategy
 
 	var primaryIPFamily api.IPFamily = serviceIPFamily
 	var secondaryIPFamily api.IPFamily = "" // sentinel value
@@ -122,7 +111,20 @@ func NewREST(
 		pods:              pods,
 		proxyTransport:    proxyTransport,
 	}
-	store.Decorator = genericStore.defaultOnRead
+
+	options := &generic.StoreOptions{
+		RESTOptions:     optsGetter,
+		AttrFunc:        svcreg.GetAttrs,
+		DecodeDecorator: genericStore.defaultOnRead, // plumb the decorator down into storage
+	}
+	if err := store.CompleteWithOptions(options); err != nil {
+		return nil, nil, nil, err
+	}
+
+	statusStore := *store
+	statusStore.UpdateStrategy = svcreg.StatusStrategy
+	statusStore.ResetFieldsStrategy = svcreg.StatusStrategy
+
 	store.AfterDelete = genericStore.afterDelete
 	store.BeginCreate = genericStore.beginCreate
 	store.BeginUpdate = genericStore.beginUpdate

pkg/registry/core/service/storage/storage_test.go

@@ -11945,3 +11945,77 @@ func TestUpdateServiceLoadBalancerStatus(t *testing.T) {
 		})
 	}
 }
+
+func TestStatusDefaultOnRead(t *testing.T) {
+	storage, statusStorage, cleanup := newStorage(t, []api.IPFamily{api.IPv4Protocol})
+	defer cleanup()
+
+	ctx := genericregistrytest.NewNamespaceScopeContext(storage.Store, metav1.NamespaceDefault)
+
+	// Store a legacy Service directly in underlying storage without ClusterIPs, IPFamilies, or IPFamilyPolicy.
+	legacySvc := &api.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "legacy-svc",
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: api.ServiceSpec{
+			ClusterIP:       "10.0.0.10",
+			SessionAffinity: api.ServiceAffinityNone,
+			Type:            api.ServiceTypeLoadBalancer,
+			Ports: []api.ServicePort{{
+				Port:       80,
+				Protocol:   api.ProtocolTCP,
+				TargetPort: intstr.FromInt32(80),
+			}},
+		},
+	}
+	key, err := storage.KeyFunc(ctx, legacySvc.Name)
+	if err != nil {
+		t.Fatalf("KeyFunc failed: %v", err)
+	}
+	out := &api.Service{}
+	if err := storage.Storage.Create(ctx, key, legacySvc, out, 0, false); err != nil {
+		t.Fatalf("Storage.Create failed: %v", err)
+	}
+
+	// Verify StatusREST.Get decorates the legacy Service.
+	gotObj, err := statusStorage.Get(ctx, legacySvc.Name, &metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("statusStorage.Get failed: %v", err)
+	}
+	gotSvc := gotObj.(*api.Service)
+	if want, got := []string{"10.0.0.10"}, gotSvc.Spec.ClusterIPs; !reflect.DeepEqual(want, got) {
+		t.Errorf("statusStorage.Get clusterIPs: expected %v, got %v", want, got)
+	}
+	if want, got := []api.IPFamily{api.IPv4Protocol}, gotSvc.Spec.IPFamilies; !reflect.DeepEqual(want, got) {
+		t.Errorf("statusStorage.Get ipFamilies: expected %v, got %v", want, got)
+	}
+	if gotSvc.Spec.IPFamilyPolicy == nil || *gotSvc.Spec.IPFamilyPolicy != api.IPFamilyPolicySingleStack {
+		t.Errorf("statusStorage.Get ipFamilyPolicy: expected %v, got %v", api.IPFamilyPolicySingleStack, gotSvc.Spec.IPFamilyPolicy)
+	}
+
+	// Verify StatusREST.Update decorates the returned legacy Service.
+	toUpdate := out.DeepCopy()
+	toUpdate.Status = api.ServiceStatus{
+		LoadBalancer: api.LoadBalancerStatus{
+			Ingress: []api.LoadBalancerIngress{{
+				IP:     "1.2.3.4",
+				IPMode: ptr.To(api.LoadBalancerIPModeVIP),
+			}},
+		},
+	}
+	updatedObj, _, err := statusStorage.Update(ctx, toUpdate.Name, rest.DefaultUpdatedObjectInfo(toUpdate), rest.ValidateAllObjectFunc, rest.ValidateAllObjectUpdateFunc, false, &metav1.UpdateOptions{})
+	if err != nil {
+		t.Fatalf("statusStorage.Update failed: %v", err)
+	}
+	updatedSvc := updatedObj.(*api.Service)
+	if want, got := []string{"10.0.0.10"}, updatedSvc.Spec.ClusterIPs; !reflect.DeepEqual(want, got) {
+		t.Errorf("statusStorage.Update clusterIPs: expected %v, got %v", want, got)
+	}
+	if want, got := []api.IPFamily{api.IPv4Protocol}, updatedSvc.Spec.IPFamilies; !reflect.DeepEqual(want, got) {
+		t.Errorf("statusStorage.Update ipFamilies: expected %v, got %v", want, got)
+	}
+	if updatedSvc.Spec.IPFamilyPolicy == nil || *updatedSvc.Spec.IPFamilyPolicy != api.IPFamilyPolicySingleStack {
+		t.Errorf("statusStorage.Update ipFamilyPolicy: expected %v, got %v", api.IPFamilyPolicySingleStack, updatedSvc.Spec.IPFamilyPolicy)
+	}
+}

staging/src/k8s.io/apiserver/pkg/registry/generic/options.go

@@ -57,4 +57,6 @@ type StoreOptions struct {
 	TriggerFunc storage.IndexerFuncs
 	AttrFunc    storage.AttrFunc
 	Indexers    *cache.Indexers
+
+	DecodeDecorator func(runtime.Object)
 }

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/decorated_decoder.go

@@ -0,0 +1,36 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package registry
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+type decoratedDecoder struct {
+	runtime.Encoder
+	decoder   runtime.Decoder
+	decorator func(runtime.Object)
+}
+
+func (d *decoratedDecoder) Decode(data []byte, defaults *schema.GroupVersionKind, into runtime.Object) (runtime.Object, *schema.GroupVersionKind, error) {
+	obj, gvk, err := d.decoder.Decode(data, defaults, into)
+	if obj != nil {
+		d.decorator(obj)
+	}
+	return obj, gvk, err
+}

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store.go

@@ -1772,10 +1772,20 @@ func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
 	}
 
 	if e.Storage.Storage == nil {
-		e.Storage.Codec = opts.StorageConfig.Codec
+		storageConfig := opts.StorageConfig
+
+		if options.DecodeDecorator != nil {
+			// make a shallow copy of the storageConfig and wrap the codec to decorate on Decode
+			storageConfigCopy := *storageConfig
+			storageConfigCopy.Codec = &decoratedDecoder{Encoder: storageConfig.Codec, decoder: storageConfig.Codec, decorator: options.DecodeDecorator}
+			storageConfig = &storageConfigCopy
+		}
+
+		e.Storage.Codec = storageConfig.Codec
+
 		var err error
 		e.Storage.Storage, e.DestroyFunc, err = opts.Decorator(
-			opts.StorageConfig,
+			storageConfig,
 			prefix,
 			keyFuncs.objectKeyFunc,
 			keyFuncs.reverseKeyFunc,

staging/src/k8s.io/apiserver/pkg/registry/generic/registry/store_test.go

@@ -46,6 +46,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/apis/example"
 	examplev1 "k8s.io/apiserver/pkg/apis/example/v1"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
@@ -3680,3 +3681,98 @@ func TestStoreInvalidShardSelector(t *testing.T) {
 		t.Errorf("Watch with invalid ShardSelector: expected BadRequest (400), got %v", err)
 	}
 }
+
+func TestStoreDecodeDecorator(t *testing.T) {
+	server, sc := etcd3testing.NewUnsecuredEtcd3TestClientServer(t)
+	defer server.Terminate(t)
+
+	sc.Codec = apitesting.TestStorageCodec(codecs, examplev1.SchemeGroupVersion)
+	sc.EventsHistoryWindow = cacherstorage.DefaultEventFreshDuration
+
+	strategy := &testRESTStrategy{scheme, names.SimpleNameGenerator, true, false, true}
+	resource := example.Resource("pods")
+	store := &Store{
+		NewFunc:                   func() runtime.Object { return &example.Pod{} },
+		NewListFunc:               func() runtime.Object { return &example.PodList{} },
+		DefaultQualifiedResource:  resource,
+		SingularQualifiedResource: example.Resource("pod"),
+		CreateStrategy:            strategy,
+		UpdateStrategy:            strategy,
+		DeleteStrategy:            strategy,
+		TableConvertor:            rest.NewDefaultTableConvertor(resource),
+	}
+
+	err := store.CompleteWithOptions(&generic.StoreOptions{
+		RESTOptions: generic.RESTOptions{
+			StorageConfig:  sc.ForResource(resource),
+			Decorator:      StorageWithCacher(),
+			ResourcePrefix: "pods",
+		},
+		DecodeDecorator: func(obj runtime.Object) {
+			if pod, ok := obj.(*example.Pod); ok {
+				pod.Labels = map[string]string{"decorated": "true"}
+			}
+		},
+	})
+	if err != nil {
+		t.Fatalf("CompleteWithOptions failed: %v", err)
+	}
+	defer store.DestroyFunc()
+
+	ctx := genericapirequest.WithNamespace(genericapirequest.NewContext(), "test")
+	if err := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) {
+		return store.ReadinessCheck() == nil, nil
+	}); err != nil {
+		t.Fatalf("storage did not become ready: %v", err)
+	}
+	w, err := store.Watch(ctx, &metainternalversion.ListOptions{ResourceVersion: "0"})
+	if err != nil {
+		t.Fatalf("Watch failed: %v", err)
+	}
+	defer w.Stop()
+
+	pod := &example.Pod{ObjectMeta: metav1.ObjectMeta{Name: "pod1", Namespace: "test"}}
+	key, err := store.KeyFunc(ctx, pod.Name)
+	if err != nil {
+		t.Fatalf("KeyFunc failed: %v", err)
+	}
+	out := &example.Pod{}
+	if err := store.Storage.Create(ctx, key, pod, out, 0, false); err != nil {
+		t.Fatalf("Storage.Create failed: %v", err)
+	}
+	if out.Labels["decorated"] != "true" {
+		t.Errorf("Create: expected decorated pod, got labels %v", out.Labels)
+	}
+
+	select {
+	case event, ok := <-w.ResultChan():
+		if !ok || event.Type != watch.Added {
+			t.Fatalf("Watch: expected Added event, got ok=%v event=%+v", ok, event)
+		}
+		cacheable, ok := event.Object.(runtime.CacheableObject)
+		if !ok {
+			t.Fatalf("Watch: expected runtime.CacheableObject, got %T", event.Object)
+		}
+		if watchedPod := cacheable.GetObject().(*example.Pod); watchedPod.Labels["decorated"] != "true" {
+			t.Errorf("Watch: expected decorated pod, got labels %v", watchedPod.Labels)
+		}
+	case <-time.After(wait.ForeverTestTimeout):
+		t.Fatal("timed out waiting for watch event")
+	}
+
+	gotObj, err := store.Get(ctx, pod.Name, &metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Get failed: %v", err)
+	}
+	if gotPod := gotObj.(*example.Pod); gotPod.Labels["decorated"] != "true" {
+		t.Errorf("Get: expected decorated pod, got labels %v", gotPod.Labels)
+	}
+
+	listObj, err := store.List(ctx, &metainternalversion.ListOptions{})
+	if err != nil {
+		t.Fatalf("List failed: %v", err)
+	}
+	if items := listObj.(*example.PodList).Items; len(items) != 1 || items[0].Labels["decorated"] != "true" {
+		t.Errorf("List: expected 1 decorated pod, got %+v", items)
+	}
+}

test/integration/service/upgrade_test.go

@@ -72,6 +72,18 @@ func Test_UpgradeService(t *testing.T) {
 	}
 	t.Logf("Service stored in etcd %v", string(svcJSON))
 
+	// Try to update the service status first (before spec is upgraded in storage)
+	statusUpdated, err := kubeclient.CoreV1().Services(ns).UpdateStatus(context.TODO(), svc, metav1.UpdateOptions{})
+	if err != nil {
+		t.Fatalf("Failed updating service status: %v", err)
+	}
+	if len(statusUpdated.Spec.ClusterIPs) == 0 || statusUpdated.Spec.ClusterIPs[0] != "10.0.0.1" {
+		t.Errorf("Expected status update response to have ClusterIPs=[10.0.0.1], got %v", statusUpdated.Spec.ClusterIPs)
+	}
+	if len(statusUpdated.Spec.IPFamilies) == 0 || statusUpdated.Spec.IPFamilies[0] != v1.IPv4Protocol {
+		t.Errorf("Expected status update response to have IPFamilies=[IPv4], got %v", statusUpdated.Spec.IPFamilies)
+	}
+
 	// Try to update the service
 	_, err = kubeclient.CoreV1().Services(ns).Update(context.TODO(), svc, metav1.UpdateOptions{})
 	if err != nil {

test/integration/servicecidr/allocator_test.go

@@ -197,7 +197,7 @@ func TestServiceAllocIPAddressLargeCIDR(t *testing.T) {
 
 func TestMigrateService(t *testing.T) {
 	etcdOptions := framework.SharedEtcd()
-	apiServerOptions := kubeapiservertesting.NewDefaultTestServerOptions()
+	apiServerOptions := &kubeapiservertesting.TestServerInstanceOptions{}
 	s := kubeapiservertesting.StartTestServerOrDie(t,
 		apiServerOptions,
 		[]string{
@@ -244,13 +244,6 @@ func TestMigrateService(t *testing.T) {
 	ns := framework.CreateNamespaceOrDie(kubeclient, namespace, t)
 	defer framework.DeleteNamespaceOrDie(kubeclient, ns, t)
 
-	// TODO: Understand why the Service can not be obtained with a List, it only works if we trigger an event
-	// by updating the Service.
-	_, err = kubeclient.CoreV1().Services(namespace).Update(context.Background(), svc, metav1.UpdateOptions{})
-	if err != nil {
-		t.Error(err)
-	}
-
 	err = wait.PollImmediate(1*time.Second, 10*time.Second, func() (bool, error) {
 		// The repair loop must create the IP address associated
 		_, err = kubeclient.NetworkingV1().IPAddresses().Get(context.TODO(), svc.Spec.ClusterIP, metav1.GetOptions{})
@@ -381,3 +374,198 @@ func TestInvalidService(t *testing.T) {
 	}
 
 }
+
+func TestLegacyServiceStatusUpdateDuplicateIP(t *testing.T) {
+	serviceCIDR := "192.168.0.0/29" // 6 usable IPs: .1 (kubernetes) and .2-.6
+	etcdOptions := framework.SharedEtcd()
+	apiServerOptions := &kubeapiservertesting.TestServerInstanceOptions{}
+	s := kubeapiservertesting.StartTestServerOrDie(t,
+		apiServerOptions,
+		[]string{
+			"--service-cluster-ip-range=" + serviceCIDR,
+			"--disable-admission-plugins=ServiceAccount",
+		},
+		etcdOptions)
+	defer s.TearDownFn()
+
+	client, err := clientset.NewForConfig(s.ClientConfig)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	ctx := context.Background()
+
+	// Wait until the default "kubernetes" service (192.168.0.1) is created.
+	if err := wait.PollUntilContextTimeout(ctx, 250*time.Millisecond, 15*time.Second, true, func(context.Context) (bool, error) {
+		_, err := client.CoreV1().Services(metav1.NamespaceDefault).Get(ctx, "kubernetes", metav1.GetOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return false, err
+		}
+		return !apierrors.IsNotFound(err), nil
+	}); err != nil {
+		t.Fatalf("creating kubernetes service timed out: %v", err)
+	}
+
+	// Store a legacy pre-1.20 Service directly in etcd with only spec.clusterIP set
+	// (spec.clusterIPs and spec.ipFamilies unset).
+	legacyServiceName := "legacy-lb-service"
+	legacyIP := "192.168.0.2"
+	legacySvc := &v1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:              legacyServiceName,
+			Namespace:         metav1.NamespaceDefault,
+			CreationTimestamp: metav1.NewTime(time.Now().Add(1 * time.Hour)),
+			UID:               "08675309-9376-9376-9376-086753099999",
+		},
+		Spec: v1.ServiceSpec{
+			ClusterIP: legacyIP,
+			Type:      v1.ServiceTypeLoadBalancer,
+			Ports: []v1.ServicePort{
+				{
+					Name:     "http",
+					Port:     80,
+					NodePort: 30080,
+					Protocol: v1.ProtocolTCP,
+				},
+			},
+		},
+	}
+	svcJSON, err := runtime.Encode(legacyscheme.Codecs.LegacyCodec(v1.SchemeGroupVersion), legacySvc)
+	if err != nil {
+		t.Fatalf("Failed creating service JSON: %v", err)
+	}
+	key := "/" + etcdOptions.Prefix + "/services/specs/" + metav1.NamespaceDefault + "/" + legacyServiceName
+	if _, err := s.EtcdClient.Put(ctx, key, string(svcJSON)); err != nil {
+		t.Fatalf("Failed writing legacy service to etcd: %v", err)
+	}
+
+	// Wait for the repair controller to ensure IPAddress/192.168.0.2 exists for legacy-lb-service.
+	if err := wait.PollUntilContextTimeout(ctx, 250*time.Millisecond, 15*time.Second, true, func(ctx context.Context) (bool, error) {
+		ipAddr, err := client.NetworkingV1().IPAddresses().Get(ctx, legacyIP, metav1.GetOptions{})
+		if err != nil {
+			return false, nil
+		}
+		return ipAddr.Spec.ParentRef != nil && ipAddr.Spec.ParentRef.Name == legacyServiceName, nil
+	}); err != nil {
+		t.Fatalf("timed out waiting for IPAddress %s to be created by repair controller: %v", legacyIP, err)
+	}
+
+	// Allocate the remaining 4 IPs in 192.168.0.0/29 (.3, .4, .5, .6).
+	for i := range 4 {
+		svcObj := &v1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("filler-svc-%d", i),
+			},
+			Spec: v1.ServiceSpec{
+				Type: v1.ServiceTypeClusterIP,
+				Ports: []v1.ServicePort{
+					{Port: 80},
+				},
+			},
+		}
+		if _, err := client.CoreV1().Services(metav1.NamespaceDefault).Create(ctx, svcObj, metav1.CreateOptions{}); err != nil {
+			t.Fatalf("failed to create filler-svc-%d: %v", i, err)
+		}
+	}
+
+	// Confirm the /29 range is now completely full.
+	_, err = client.CoreV1().Services(metav1.NamespaceDefault).Create(ctx, makeService("should-fail-full"), metav1.CreateOptions{})
+	if err == nil || !strings.Contains(err.Error(), "range is full") {
+		t.Fatalf("expected 'range is full' error when CIDR is exhausted, got: %v", err)
+	}
+
+	// Start a watch on Services to observe the watch.Modified event emitted by UpdateStatus.
+	svcList, err := client.CoreV1().Services(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		t.Fatalf("failed to LIST services: %v", err)
+	}
+	gotLegacySvc, err := client.CoreV1().Services(metav1.NamespaceDefault).Get(ctx, legacyServiceName, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("failed to GET legacy service: %v", err)
+	}
+	watchCtx, watchCancel := context.WithCancel(ctx)
+	defer watchCancel()
+	w, err := client.CoreV1().Services(metav1.NamespaceDefault).Watch(watchCtx, metav1.ListOptions{
+		ResourceVersion: svcList.ResourceVersion,
+	})
+	if err != nil {
+		t.Fatalf("failed to start service watch: %v", err)
+	}
+	defer w.Stop()
+
+	// Patch/Update the legacy Service's status (simulating cloud-controller-manager service-controller).
+	gotLegacySvc.Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{IP: "192.0.2.100"}}
+	if _, err := client.CoreV1().Services(metav1.NamespaceDefault).UpdateStatus(ctx, gotLegacySvc, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("UpdateStatus on legacy service failed: %v", err)
+	}
+
+	// Inspect the watched Service from the live watch event.
+	for {
+		select {
+		case event := <-w.ResultChan():
+			watchedSvc, ok := event.Object.(*v1.Service)
+			if !ok {
+				t.Fatalf("unexpected watch event (%v) object: %#v", event.Type, event.Object)
+			}
+			if watchedSvc.Name != legacyServiceName {
+				continue
+			}
+			if len(watchedSvc.Spec.ClusterIPs) == 0 {
+				t.Errorf("watched Service %s is missing Spec.ClusterIPs (got %v) even though Spec.ClusterIP=%q",
+					watchedSvc.Name, watchedSvc.Spec.ClusterIPs, watchedSvc.Spec.ClusterIP)
+			}
+			if len(watchedSvc.Spec.IPFamilies) == 0 {
+				t.Errorf("watched Service %s is missing Spec.IPFamilies (got %v)", watchedSvc.Name, watchedSvc.Spec.IPFamilies)
+			}
+		case <-time.After(10 * time.Second):
+			t.Fatalf("timed out waiting for watch event after UpdateStatus")
+		}
+		break
+	}
+	w.Stop()
+	watchCancel()
+
+	// Trigger ipallocator-repair-controller's IPAddress reconciliation (simulating the periodic resync).
+	ipAddr, err := client.NetworkingV1().IPAddresses().Get(ctx, legacyIP, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("failed to get IPAddress %s: %v", legacyIP, err)
+	}
+	if ipAddr.Annotations == nil {
+		ipAddr.Annotations = map[string]string{}
+	}
+	ipAddr.Annotations["resync-trigger"] = "true"
+	if _, err := client.NetworkingV1().IPAddresses().Update(ctx, ipAddr, metav1.UpdateOptions{}); err != nil {
+		t.Fatalf("failed to update IPAddress %s annotation: %v", legacyIP, err)
+	}
+
+	// Verify that the repair controller does not delete IPAddress/192.168.0.2 or allow
+	// a new Service to steal the IP address.
+	err = wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 2*time.Second, true, func(ctx context.Context) (bool, error) {
+		_, getErr := client.NetworkingV1().IPAddresses().Get(ctx, legacyIP, metav1.GetOptions{})
+		if apierrors.IsNotFound(getErr) {
+			return true, nil
+		}
+		return false, nil
+	})
+	if err == nil {
+		// IPAddress was erroneously deleted by the repair controller!
+		// Attempt to create a new Service to show it steals 192.168.0.2.
+		var duplicateSvc *v1.Service
+		_ = wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, 3*time.Second, true, func(ctx context.Context) (bool, error) {
+			var createErr error
+			duplicateSvc, createErr = client.CoreV1().Services(metav1.NamespaceDefault).Create(ctx, makeService("new-service"), metav1.CreateOptions{})
+			return createErr == nil, nil
+		})
+		if duplicateSvc != nil {
+			t.Fatalf("IPAddress %s was erroneously deleted by repair controller and re-allocated to %s (both %s and %s now have ClusterIP %s!)",
+				legacyIP, duplicateSvc.Name, legacyServiceName, duplicateSvc.Name, duplicateSvc.Spec.ClusterIP)
+		}
+		t.Fatalf("IPAddress %s was erroneously deleted by repair controller after UpdateStatus on legacy service", legacyIP)
+	}
+
+	// Confirm the /29 range is still completely full and IPAddress/192.168.0.2 still belongs to legacy-lb-service.
+	_, err = client.CoreV1().Services(metav1.NamespaceDefault).Create(ctx, makeService("new-service"), metav1.CreateOptions{})
+	if err == nil || !strings.Contains(err.Error(), "range is full") {
+		t.Fatalf("expected 'range is full' error when CIDR is exhausted, got: %v", err)
+	}
+}