From 1f9b1b85db5d8db0f798f5b3677c426657f0b5a1 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 29 Sep 2026 17:20:33 +0200 Subject: [PATCH] fix(kubernetes): reconcile KCCM EndpointSlices when tenant node VMIs change The kubevirteps controller derives the infra EndpointSlices of a LoadBalancer Service from the VMIs backing the tenant nodes, but it watches no VMI and its informers had no resync period. A VMI appearing, disappearing or changing its address therefore triggered no reconciliation, and no periodic relist ever caught up: the EndpointSlices kept pointing at VMIs that no longer existed and the LoadBalancer blackholed the share of the inbound connections hashed onto those dead addresses. Services backed by a tenant EndpointSlice that churns with its pods recovered on their own. Cilium Gateway API Services did not: their tenant slice is a static sentinel, so nothing ever forced a recompute. Observed in production on a cluster whose GPU MachineDeployment is rolled by the cluster-autoscaler: 5 of the 9 endpoints of every gateway pointed at destroyed VMs. Add an informer on the infra VirtualMachineInstances that requeues the LoadBalancer Services of this cluster, ignoring the updates that cannot change an endpoint, and give the informers a 5 minute resync as a safety net. Services of other tenant clusters sharing the namespace are not requeued, since reconcile would skip them. Also scope the all-VMIs fallback to the VMIs labelled with the name of the cluster being reconciled: an infra namespace can hold several tenant clusters and standalone VirtualMachines, whose VMIs do not serve the Service NodePort. The fallback widens to every VMI only when none of them carries the label, so that environments that do not label them keep their endpoints. Once any VMI is labelled, an empty selection means this cluster has no node right now, and widening it would publish the addresses of another cluster. The patch is named after what it does rather than after an upstream pull request, because none carries this change yet. Signed-off-by: Mattia Eleuteri Assisted-by: LLM --- .../patches/vmi-informer.diff | 429 ++++++++++++++++++ 1 file changed, 429 insertions(+) create mode 100644 packages/apps/kubernetes/images/kubevirt-cloud-provider/patches/vmi-informer.diff diff --git a/packages/apps/kubernetes/images/kubevirt-cloud-provider/patches/vmi-informer.diff b/packages/apps/kubernetes/images/kubevirt-cloud-provider/patches/vmi-informer.diff new file mode 100644 index 000000000..e300bf458 --- /dev/null +++ b/packages/apps/kubernetes/images/kubevirt-cloud-provider/patches/vmi-informer.diff @@ -0,0 +1,429 @@ +diff --git a/pkg/controller/kubevirteps/kubevirteps_controller.go b/pkg/controller/kubevirteps/kubevirteps_controller.go +index 8af51fbe8..e8f1a4488 100644 +--- a/pkg/controller/kubevirteps/kubevirteps_controller.go ++++ b/pkg/controller/kubevirteps/kubevirteps_controller.go +@@ -12,12 +12,15 @@ import ( + apiequality "k8s.io/apimachinery/pkg/api/equality" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ++ "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" ++ "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/dynamic" ++ "k8s.io/client-go/dynamic/dynamicinformer" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" +@@ -31,8 +34,19 @@ import ( + + const ( + ControllerName = controllerName("kubevirt_eps_controller") ++ ++ // A periodic resync re-reconciles every LoadBalancer Service even if an event was ++ // missed, so infra EndpointSlices cannot stay stale indefinitely. ++ defaultResyncPeriod = 5 * time.Minute ++ ++ // clusterNameLabel is set by CAPK on the VMIs backing a tenant cluster. It is used to ++ // avoid selecting VMIs that do not belong to this cluster when falling back to all VMIs. ++ clusterNameLabel = "cluster.x-k8s.io/cluster-name" + ) + ++var vmiGVR = kubevirtv1.VirtualMachineInstanceGroupVersionKind.GroupVersion(). ++ WithResource("virtualmachineinstances") ++ + type controllerName string + + func (c controllerName) dashed() string { +@@ -49,9 +63,10 @@ type Controller struct { + tenantFactory informers.SharedInformerFactory + tenantEPSTracker tenantEPSTracker + +- infraClient kubernetes.Interface +- infraDynamic dynamic.Interface +- infraFactory informers.SharedInformerFactory ++ infraClient kubernetes.Interface ++ infraDynamic dynamic.Interface ++ infraFactory informers.SharedInformerFactory ++ infraDynamicFactory dynamicinformer.DynamicSharedInformerFactory + + infraNamespace string + clusterName string +@@ -67,8 +82,11 @@ func NewKubevirtEPSController( + infraNamespace string, + clusterName string, + ) *Controller { +- tenantFactory := informers.NewSharedInformerFactory(tenantClient, 0) +- infraFactory := informers.NewSharedInformerFactoryWithOptions(infraClient, 0, informers.WithNamespace(infraNamespace)) ++ tenantFactory := informers.NewSharedInformerFactory(tenantClient, defaultResyncPeriod) ++ infraFactory := informers.NewSharedInformerFactoryWithOptions(infraClient, defaultResyncPeriod, ++ informers.WithNamespace(infraNamespace)) ++ infraDynamicFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(infraDynamic, ++ defaultResyncPeriod, infraNamespace, nil) + queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()) + + return &Controller{ +@@ -78,6 +96,7 @@ func NewKubevirtEPSController( + infraClient: infraClient, + infraDynamic: infraDynamic, + infraFactory: infraFactory, ++ infraDynamicFactory: infraDynamicFactory, + infraNamespace: infraNamespace, + clusterName: clusterName, + queue: queue, +@@ -236,9 +255,84 @@ func (c *Controller) Init() error { + return err + } + ++ // The desired infra endpoints of a LoadBalancer Service are derived from the VMIs backing the ++ // tenant nodes, so a VMI appearing, disappearing or changing its address must trigger a ++ // reconciliation. Without this, EndpointSlices keep pointing at VMIs that no longer exist ++ // (for example after a MachineDeployment rollout or a cluster-autoscaler scale down) and the ++ // LoadBalancer blackholes the share of the traffic hashed onto those dead addresses. ++ _, err = c.infraDynamicFactory.ForResource(vmiGVR).Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ ++ AddFunc: func(obj interface{}) { ++ c.enqueueLoadBalancerServices(fmt.Sprintf("VMI added: %s", vmiKey(obj))) ++ }, ++ UpdateFunc: func(oldObj, newObj interface{}) { ++ if vmiEndpointFingerprint(oldObj) == vmiEndpointFingerprint(newObj) { ++ return ++ } ++ c.enqueueLoadBalancerServices(fmt.Sprintf("VMI updated: %s", vmiKey(newObj))) ++ }, ++ DeleteFunc: func(obj interface{}) { ++ if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { ++ obj = tombstone.Obj ++ } ++ c.enqueueLoadBalancerServices(fmt.Sprintf("VMI deleted: %s", vmiKey(obj))) ++ }, ++ }) ++ if err != nil { ++ return err ++ } ++ + return nil + } + ++func vmiKey(obj interface{}) string { ++ u, ok := obj.(*unstructured.Unstructured) ++ if !ok { ++ return "" ++ } ++ return u.GetNamespace() + "/" + u.GetName() ++} ++ ++// vmiEndpointFingerprint returns the parts of a VMI that influence the endpoints derived from it: ++// the cluster it belongs to, the infra node it runs on, its phase and the address of its "default" ++// interface. Any other update, and a running VMI sees many, cannot change an EndpointSlice. ++func vmiEndpointFingerprint(obj interface{}) string { ++ u, ok := obj.(*unstructured.Unstructured) ++ if !ok { ++ return "" ++ } ++ vmi := &kubevirtv1.VirtualMachineInstance{} ++ if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, vmi); err != nil { ++ // Fall back to the resource version so an unparsable update is never silently dropped. ++ return u.GetResourceVersion() ++ } ++ ip := "" ++ for _, i := range vmi.Status.Interfaces { ++ if i.Name == "default" { ++ ip = i.IP ++ break ++ } ++ } ++ return fmt.Sprintf("%s|%s|%s|%s", u.GetLabels()[clusterNameLabel], vmi.Status.NodeName, vmi.Status.Phase, ip) ++} ++ ++// enqueueLoadBalancerServices requeues the LoadBalancer Services of this cluster only: the infra ++// namespace can hold the Services of other tenant clusters, which reconcile would skip anyway. ++func (c *Controller) enqueueLoadBalancerServices(reason string) { ++ svcs, err := c.infraFactory.Core().V1().Services().Lister().Services(c.infraNamespace). ++ List(labels.SelectorFromSet(labels.Set{kubevirt.TenantClusterNameLabelKey: c.clusterName})) ++ if err != nil { ++ klog.Errorf("Failed to list Services in namespace %s: %v", c.infraNamespace, err) ++ return ++ } ++ for _, svc := range svcs { ++ if svc.Spec.Type != v1.ServiceTypeLoadBalancer { ++ continue ++ } ++ klog.Infof("%s, requeuing Service: %v/%v", reason, svc.Namespace, svc.Name) ++ c.queue.Add(newRequest(UpdateReq, svc, nil)) ++ } ++} ++ + // getInfraServiceForEPS returns the Service in the infra cluster associated with the given EndpointSlice. + // It does this by reading the "kubernetes.io/service-name" label from the EndpointSlice, which should correspond + // to the Service name. If not found or if the Service doesn't exist, it returns nil. +@@ -276,10 +370,12 @@ func (c *Controller) Run(numWorkers int, stopCh <-chan struct{}, controllerManag + + c.tenantFactory.Start(stopCh) + c.infraFactory.Start(stopCh) ++ c.infraDynamicFactory.Start(stopCh) + + if !cache.WaitForNamedCacheSync(ControllerName.String(), stopCh, + c.infraFactory.Core().V1().Services().Informer().HasSynced, +- c.tenantFactory.Discovery().V1().EndpointSlices().Informer().HasSynced) { ++ c.tenantFactory.Discovery().V1().EndpointSlices().Informer().HasSynced, ++ c.infraDynamicFactory.ForResource(vmiGVR).Informer().HasSynced) { + return + } + +@@ -868,17 +964,45 @@ func (c *Controller) getDesiredEndpoints(service *v1.Service, tenantSlices []*di + func (c *Controller) getAllVMIEndpoints() []*discovery.Endpoint { + var endpoints []*discovery.Endpoint + +- // List all VMIs in the infra namespace +- vmiList, err := c.infraDynamic. +- Resource(kubevirtv1.VirtualMachineInstanceGroupVersionKind.GroupVersion().WithResource("virtualmachineinstances")). +- Namespace(c.infraNamespace). ++ // List every VMI of the infra namespace, then keep only the ones of this cluster. An infra ++ // namespace can hold more than one tenant cluster, as well as plain VirtualMachines, and those ++ // VMIs do not serve this Service's NodePort: selecting them would add endpoints that blackhole ++ // the traffic, or route it to another tenant. ++ vmiList, err := c.infraDynamic.Resource(vmiGVR).Namespace(c.infraNamespace). + List(context.TODO(), metav1.ListOptions{}) + if err != nil { + klog.Errorf("Failed to list VMIs in namespace %q: %v", c.infraNamespace, err) + return endpoints + } + +- for _, obj := range vmiList.Items { ++ items := vmiList.Items ++ if c.clusterName != "" { ++ mine := make([]unstructured.Unstructured, 0, len(items)) ++ anyLabelled := false ++ for i := range items { ++ name, found := items[i].GetLabels()[clusterNameLabel] ++ if !found { ++ continue ++ } ++ anyLabelled = true ++ if name == c.clusterName { ++ mine = append(mine, items[i]) ++ } ++ } ++ // Widen to every VMI only when NONE of them carries the label: the VMIs are then not ++ // managed by CAPK (or by an older version of it) and there is no way to tell the clusters ++ // apart, so returning nothing would take the LoadBalancer down. As soon as some VMIs are ++ // labelled, an empty selection is a legitimate answer -- this cluster currently has no ++ // node -- and must not be widened to the VMIs of another cluster. ++ if anyLabelled { ++ items = mine ++ } else { ++ klog.Warningf("No VMI carries %q in namespace %s, selecting every VMI of the namespace", ++ clusterNameLabel, c.infraNamespace) ++ } ++ } ++ ++ for _, obj := range items { + vmi := &kubevirtv1.VirtualMachineInstance{} + err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.Object, vmi) + if err != nil { +@@ -916,7 +1040,8 @@ func (c *Controller) getAllVMIEndpoints() []*discovery.Endpoint { + } + } + +- klog.Infof("Fallback: created %d endpoints from all VMIs in namespace %s", len(endpoints), c.infraNamespace) ++ klog.Infof("Fallback: created %d endpoints from the VMIs of cluster %q in namespace %s", ++ len(endpoints), c.clusterName, c.infraNamespace) + return endpoints + } + +diff --git a/pkg/controller/kubevirteps/kubevirteps_controller_test.go b/pkg/controller/kubevirteps/kubevirteps_controller_test.go +index 17608fa10..8fa76678b 100644 +--- a/pkg/controller/kubevirteps/kubevirteps_controller_test.go ++++ b/pkg/controller/kubevirteps/kubevirteps_controller_test.go +@@ -965,3 +965,188 @@ var _ = g.Describe("getDesiredEndpoints", func() { + Expect(endpoints).To(HaveLen(0), "Expected no endpoints due to missing NodeName or IP") + }) + }) ++ ++func createUnstructuredVMIForCluster(name, nodeName, ip, clusterName string) *unstructured.Unstructured { ++ vmi := createUnstructuredVMINode(name, nodeName, ip) ++ if clusterName != "" { ++ vmi.SetLabels(map[string]string{clusterNameLabel: clusterName}) ++ } ++ return vmi ++} ++ ++var _ = g.Describe("KubevirtEPSController VMI tracking", func() { ++ ++ createVMI := func(t *testKubevirtEPSController, vmi *unstructured.Unstructured) { ++ _, err := t.infraDynamic.Resource(vmiGVR).Namespace(infraNamespace). ++ Create(context.TODO(), vmi, metav1.CreateOptions{}) ++ Expect(err).To(BeNil()) ++ } ++ ++ g.It("Should only select the VMIs of its own cluster when falling back to all VMIs", func() { ++ t := setupTestKubevirtEPSController() ++ // "test-cluster" is the cluster name the controller is created with. ++ createVMI(t, createUnstructuredVMIForCluster("mine-1", "infra-1", "10.0.0.1", "test-cluster")) ++ createVMI(t, createUnstructuredVMIForCluster("mine-2", "infra-2", "10.0.0.2", "test-cluster")) ++ createVMI(t, createUnstructuredVMIForCluster("other", "infra-3", "10.0.0.3", "another-cluster")) ++ ++ endpoints := t.controller.getAllVMIEndpoints() ++ ++ Expect(endpoints).To(HaveLen(2), "Only the VMIs of test-cluster should be selected") ++ var addresses []string ++ for _, e := range endpoints { ++ addresses = append(addresses, e.Addresses[0]) ++ } ++ Expect(addresses).To(ConsistOf("10.0.0.1", "10.0.0.2")) ++ Expect(addresses).ToNot(ContainElement("10.0.0.3"), "A VMI of another cluster must never be an endpoint") ++ }) ++ ++ g.It("Should keep selecting every VMI when none carries the cluster label", func() { ++ t := setupTestKubevirtEPSController() ++ createVMI(t, createUnstructuredVMIForCluster("unlabelled-1", "infra-1", "10.0.0.1", "")) ++ createVMI(t, createUnstructuredVMIForCluster("unlabelled-2", "infra-2", "10.0.0.2", "")) ++ ++ endpoints := t.controller.getAllVMIEndpoints() ++ ++ Expect(endpoints).To(HaveLen(2), "Unlabelled VMIs must still be selected, so the LB does not lose every endpoint") ++ }) ++ ++ g.It("Should return no endpoint when only another cluster's VMIs are present", func() { ++ t := setupTestKubevirtEPSController() ++ // The controller's cluster is "test-cluster" and has no VMI right now (scaled to zero, or ++ // every VMI being replaced by a rollout), while another tenant cluster of the same infra ++ // namespace does have some. Widening to every VMI here would publish the other cluster's ++ // addresses as this Service's endpoints. ++ createVMI(t, createUnstructuredVMIForCluster("other-1", "infra-1", "10.0.0.1", "another-cluster")) ++ createVMI(t, createUnstructuredVMIForCluster("other-2", "infra-2", "10.0.0.2", "another-cluster")) ++ ++ endpoints := t.controller.getAllVMIEndpoints() ++ ++ Expect(endpoints).To(BeEmpty(), ++ "An empty selection is legitimate and must not be widened to another cluster's VMIs") ++ }) ++ ++ g.It("Should ignore the unlabelled VMIs when some VMIs carry the cluster label", func() { ++ t := setupTestKubevirtEPSController() ++ createVMI(t, createUnstructuredVMIForCluster("mine", "infra-1", "10.0.0.1", "test-cluster")) ++ // A standalone VirtualMachine of the same namespace, not part of any tenant cluster. ++ createVMI(t, createUnstructuredVMIForCluster("standalone", "infra-2", "10.0.0.2", "")) ++ ++ endpoints := t.controller.getAllVMIEndpoints() ++ ++ Expect(endpoints).To(HaveLen(1)) ++ Expect(endpoints[0].Addresses).To(ConsistOf("10.0.0.1")) ++ }) ++ ++ // startTracking starts the informers the VMI event handlers rely on, with a LoadBalancer Service ++ // of this cluster and one of another cluster, and returns once the initial events, want of them, ++ // have been drained from the queue. ++ startTracking := func(t *testKubevirtEPSController, stop chan struct{}, want int) { ++ for _, svc := range []*v1.Service{ ++ createInfraServiceLB("lb-svc", "tenant-svc", "test-cluster", ++ v1.ServicePort{Name: "http", Port: 80, Protocol: v1.ProtocolTCP, TargetPort: intstr.FromInt32(30000)}, ++ v1.ServiceExternalTrafficPolicyCluster), ++ createInfraServiceLB("other-lb-svc", "tenant-svc", "another-cluster", ++ v1.ServicePort{Name: "http", Port: 80, Protocol: v1.ProtocolTCP, TargetPort: intstr.FromInt32(30001)}, ++ v1.ServiceExternalTrafficPolicyCluster), ++ } { ++ _, err := t.infraClient.CoreV1().Services(infraNamespace).Create(context.TODO(), svc, metav1.CreateOptions{}) ++ Expect(err).To(BeNil()) ++ } ++ ++ // Services first, so that the add event of a VMI already present finds them in the lister. ++ t.controller.infraFactory.Start(stop) ++ Expect(cache.WaitForCacheSync(stop, ++ t.controller.infraFactory.Core().V1().Services().Informer().HasSynced)).To(BeTrue()) ++ t.controller.infraDynamicFactory.Start(stop) ++ Expect(cache.WaitForCacheSync(stop, ++ t.controller.infraDynamicFactory.ForResource(vmiGVR).Informer().HasSynced)).To(BeTrue()) ++ ++ Eventually(func() int { return t.controller.queue.Len() }, "10s", "20ms").Should(Equal(want)) ++ drainQueue(t) ++ } ++ ++ updateVMI := func(t *testKubevirtEPSController, name string, mutate func(*unstructured.Unstructured)) { ++ vmi, err := t.infraDynamic.Resource(vmiGVR).Namespace(infraNamespace). ++ Get(context.TODO(), name, metav1.GetOptions{}) ++ Expect(err).To(BeNil()) ++ mutate(vmi) ++ _, err = t.infraDynamic.Resource(vmiGVR).Namespace(infraNamespace). ++ Update(context.TODO(), vmi, metav1.UpdateOptions{}) ++ Expect(err).To(BeNil()) ++ } ++ ++ expectRequeueOf := func(t *testKubevirtEPSController, why string) { ++ Eventually(func() int { return t.controller.queue.Len() }, "10s", "20ms").Should(Equal(1), why) ++ item, _ := t.controller.queue.Get() ++ t.controller.queue.Done(item) ++ Expect(item.(*Request).Obj.(*v1.Service).Name).To(Equal("lb-svc"), ++ "Only the Service of this cluster should be requeued") ++ Consistently(func() int { return t.controller.queue.Len() }, "200ms", "20ms").Should(BeZero()) ++ } ++ ++ expectNoRequeue := func(t *testKubevirtEPSController, why string) { ++ Consistently(func() int { return t.controller.queue.Len() }, "500ms", "20ms").Should(BeZero(), why) ++ } ++ ++ g.It("Should requeue the LoadBalancer Services when a VMI appears or disappears", func() { ++ t := setupTestKubevirtEPSController() ++ stop := make(chan struct{}) ++ defer close(stop) ++ startTracking(t, stop, 2) ++ ++ // A new VMI must trigger a reconciliation, otherwise a node added by a scale up or a ++ // rollout never becomes an endpoint. ++ createVMI(t, createUnstructuredVMIForCluster("vmi-1", "infra-1", "10.0.0.1", "test-cluster")) ++ expectRequeueOf(t, "Creating a VMI should requeue the LoadBalancer Service") ++ ++ // A deleted VMI must trigger a reconciliation as well, otherwise its address stays in the ++ // EndpointSlice and the LoadBalancer blackholes the traffic hashed onto it. ++ err := t.infraDynamic.Resource(vmiGVR).Namespace(infraNamespace). ++ Delete(context.TODO(), "vmi-1", metav1.DeleteOptions{}) ++ Expect(err).To(BeNil()) ++ expectRequeueOf(t, "Deleting a VMI should requeue the LoadBalancer Service") ++ }) ++ ++ g.It("Should requeue the LoadBalancer Services only on the VMI updates that can change an endpoint", func() { ++ t := setupTestKubevirtEPSController() ++ stop := make(chan struct{}) ++ defer close(stop) ++ createVMI(t, createUnstructuredVMIForCluster("vmi-1", "infra-1", "10.0.0.1", "test-cluster")) ++ // Both Service adds, plus the add of the VMI already present. ++ startTracking(t, stop, 3) ++ ++ updateVMI(t, "vmi-1", func(u *unstructured.Unstructured) { ++ u.SetAnnotations(map[string]string{"unrelated": "change"}) ++ }) ++ expectNoRequeue(t, "An update that touches no endpoint field must not requeue") ++ ++ updateVMI(t, "vmi-1", func(u *unstructured.Unstructured) { ++ Expect(unstructured.SetNestedField(u.Object, "Failed", "status", "phase")).To(Succeed()) ++ }) ++ expectRequeueOf(t, "A phase change should requeue the LoadBalancer Service") ++ ++ updateVMI(t, "vmi-1", func(u *unstructured.Unstructured) { ++ Expect(unstructured.SetNestedField(u.Object, "infra-2", "status", "nodeName")).To(Succeed()) ++ }) ++ expectRequeueOf(t, "A node change should requeue the LoadBalancer Service") ++ ++ updateVMI(t, "vmi-1", func(u *unstructured.Unstructured) { ++ Expect(unstructured.SetNestedSlice(u.Object, []interface{}{ ++ map[string]interface{}{"name": "default", "ipAddress": "10.0.0.9"}, ++ }, "status", "interfaces")).To(Succeed()) ++ }) ++ expectRequeueOf(t, "An address change should requeue the LoadBalancer Service") ++ ++ updateVMI(t, "vmi-1", func(u *unstructured.Unstructured) { ++ u.SetLabels(map[string]string{clusterNameLabel: "another-cluster"}) ++ }) ++ expectRequeueOf(t, "Moving a VMI to another cluster should requeue the LoadBalancer Service") ++ }) ++}) ++ ++func drainQueue(t *testKubevirtEPSController) { ++ for t.controller.queue.Len() > 0 { ++ item, _ := t.controller.queue.Get() ++ t.controller.queue.Done(item) ++ } ++}