From 3a4d93f5fb267dd0a8df36273434c37135f7778e Mon Sep 17 00:00:00 2001 From: Enrico Candino Date: Wed, 15 Jul 2026 17:02:30 +0200 Subject: [PATCH] Fix dangling Pods (#1021) * Add tests asserting GetPods is scoped to the node's own Pods Adds a unit test (fake host + virtual clients, Pods across multiple nodes plus a dangling one) and a multi-node e2e test (a Deployment with one nginx Pod per node via required anti-affinity; restart every k3k-kubelet agent Pod; assert no workload Pod is deleted from the host or virtual cluster). Both pin the intended behavior and fail against the current code; the fix follows in the next commit. * Scope Provider.GetPods to the node's own Pods to prevent cross-node dangling-pod deletion GetPods() listed host Pods cluster-wide (by the k3k.io/clusterName label only). The vendored virtual-kubelet library's deleteDanglingPods reconciliation deletes any Pod returned here that is missing from this instance's virtual Pod lister, which is scoped to spec.nodeName == agentHostname. On a multi-node host cluster, every restarting k3k-kubelet instance therefore saw Pods owned by other nodes as 'dangling' and deleted them from the host (and, in turn, the virtual cluster). Scope GetPods by the *virtual* Pod's spec.nodeName -- the same ownership signal the framework uses -- excluding Pods owned by other nodes while still returning own-node Pods and genuinely dangling ones. The host Pod's physical node is not a reliable owner (it is scheduled with only a soft, sometimes-absent affinity), so it must not be used. Virtual Pods are read live to avoid a startup cache-sync race. Makes the previous commit's tests pass. * Refactor GetPods to scope to the cluster namespace and update tests accordingly * Add AgentNameLabel to track Pods synced by the k3k-kubelet agent and update tests accordingly * Add failing test for updatePod --- k3k-kubelet/provider/provider.go | 61 +++++--- k3k-kubelet/provider/provider_test.go | 105 +++++++++++++ k3k-kubelet/translate/host.go | 11 +- pkg/controller/cluster/cluster.go | 1 + tests/cli/tests_suite_test.go | 2 - tests/e2e/cluster_kubelet_restart_test.go | 172 ++++++++++++++++++++++ 6 files changed, 328 insertions(+), 24 deletions(-) diff --git a/k3k-kubelet/provider/provider.go b/k3k-kubelet/provider/provider.go index 893ab137..c685807e 100644 --- a/k3k-kubelet/provider/provider.go +++ b/k3k-kubelet/provider/provider.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "maps" "net/http" "strconv" "strings" @@ -15,8 +16,6 @@ import ( "github.com/google/go-cmp/cmp" "github.com/virtual-kubelet/virtual-kubelet/node/api" "github.com/virtual-kubelet/virtual-kubelet/node/nodeutil" - "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/selection" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/kubernetes/scheme" @@ -411,6 +410,9 @@ func (p *Provider) createPod(ctx context.Context, pod *corev1.Pod) error { hostPod := virtualPod.DeepCopy() p.Translator.TranslateTo(hostPod) + // record which k3k-kubelet agent synced this Pod, so GetPods can scope to this agent's own Pods + hostPod.Labels[translate.AgentNameLabel] = p.agentHostname + logger = logger.WithValues("pod", hostPod.Name) // Clear the NodeName to allow scheduling, and set affinity to prefer scheduling the Pod on the same host node as the virtual kubelet, @@ -673,11 +675,33 @@ func updatePod(dst, src *corev1.Pod) { updateContainerImages(dst.Spec.Containers, src.Spec.Containers) updateContainerImages(dst.Spec.InitContainers, src.Spec.InitContainers) + updateMetadata(dst, src) + dst.Spec.ActiveDeadlineSeconds = src.Spec.ActiveDeadlineSeconds dst.Spec.Tolerations = src.Spec.Tolerations +} - dst.Annotations = src.Annotations - dst.Labels = src.Labels +// updateMetadata copies the labels and annotations from src (the virtual Pod) onto dst (the host Pod), +// while preserving the k3k metadata (labels and annotations with the "k3k.io/*" prefix) +func updateMetadata(dst, src *corev1.Pod) { + dst.Labels = mergeManagedMetadata(dst.Labels, src.Labels) + dst.Annotations = mergeManagedMetadata(dst.Annotations, src.Annotations) +} + +// mergeManagedMetadata returns a new map with all the entries from src (the virtual object), +// plus the k3k metadata (keys with the translate.MetadataPrefix) carried over from dst (the host object). +func mergeManagedMetadata(dst, src map[string]string) map[string]string { + merged := make(map[string]string, len(src)) + + maps.Copy(merged, src) + + for key, value := range dst { + if strings.HasPrefix(key, translate.MetadataPrefix) { + merged[key] = value + } + } + + return merged } // updateContainerImages will update the images of the original container images with the same name @@ -782,32 +806,31 @@ func (p *Provider) getPodFromHostCluster(ctx context.Context, hostPodName string // The Pods returned are expected to be immutable, and may be accessed // concurrently outside of the calling goroutine. Therefore it is recommended // to return a version after DeepCopy. +// +// It returns only the Pods synced by this k3k-kubelet agent, identified by the AgentNameLabel. func (p *Provider) GetPods(ctx context.Context) ([]*corev1.Pod, error) { p.logger.V(1).Info("GetPods") - selector := labels.NewSelector() + var hostPods corev1.PodList - requirement, err := labels.NewRequirement(translate.ClusterNameLabel, selection.Equals, []string{p.ClusterName}) - if err != nil { - p.logger.Error(err, "Error creating label selector for GetPods") - return nil, err + listOpts := []client.ListOption{ + client.InNamespace(p.ClusterNamespace), + client.MatchingLabels{ + translate.ClusterNameLabel: p.ClusterName, + translate.AgentNameLabel: p.agentHostname, + }, } - selector = selector.Add(*requirement) - - var podList corev1.PodList - - err = p.Host.Client.List(ctx, &podList, &client.ListOptions{LabelSelector: selector}) - if err != nil { + if err := p.Host.Client.List(ctx, &hostPods, listOpts...); err != nil { p.logger.Error(err, "Error listing pods from host cluster") return nil, err } - retPods := []*corev1.Pod{} + retPods := make([]*corev1.Pod, 0, len(hostPods.Items)) - for _, pod := range podList.DeepCopy().Items { - p.Translator.TranslateFrom(&pod) - retPods = append(retPods, &pod) + for _, hostPod := range hostPods.DeepCopy().Items { + p.Translator.TranslateFrom(&hostPod) + retPods = append(retPods, &hostPod) } return retPods, nil diff --git a/k3k-kubelet/provider/provider_test.go b/k3k-kubelet/provider/provider_test.go index 8054fe06..ece998c7 100644 --- a/k3k-kubelet/provider/provider_test.go +++ b/k3k-kubelet/provider/provider_test.go @@ -1,10 +1,15 @@ package provider import ( + "context" "reflect" "testing" + "github.com/go-logr/logr" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -300,3 +305,103 @@ func Test_configureEnv(t *testing.T) { }) } } + +// TestGetPods_ScopedToAgent pins the behavior that GetPods returns only the Pods synced by this +// k3k-kubelet agent, identified by the AgentNameLabel and scoped to this cluster's host namespace. +// Pods synced by another agent, or living in another namespace, are excluded. Pods are returned +// regardless of whether their virtual counterpart still exists, so the virtual-kubelet startup +// reconciliation can still clean up genuine orphans. +func TestGetPods_ScopedToAgent(t *testing.T) { + const ( + clusterName = "c-test" + clusterNamespace = "ns-test" + agentName = "node-a" + ) + + // host Pods carry the tracking metadata TranslateFrom reads to recover the virtual identity, + // plus the AgentNameLabel recording which agent synced them. + newHostPod := func(name, agent, namespace string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{ + translate.ClusterNameLabel: clusterName, + translate.AgentNameLabel: agent, + }, + Annotations: map[string]string{ + translate.ResourceNameAnnotation: name, + translate.ResourceNamespaceAnnotation: "default", + }, + }, + } + } + + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + + hostClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects( + newHostPod("a1", agentName, clusterNamespace), // synced by this agent -> returned + newHostPod("b1", "node-b", clusterNamespace), // synced by another agent -> excluded + newHostPod("d1", agentName, "ns-other"), // another namespace -> excluded + ). + Build() + + p := Provider{ + Host: ClusterContext{Client: hostClient}, + Translator: translate.ToHostTranslator{ + ClusterName: clusterName, + ClusterNamespace: clusterNamespace, + }, + ClusterName: clusterName, + ClusterNamespace: clusterNamespace, + agentHostname: agentName, + logger: logr.Discard(), + } + + pods, err := p.GetPods(context.Background()) + require.NoError(t, err) + + names := map[string]bool{} + for _, pod := range pods { + names[pod.Name] = true + } + + // only a1 (synced by this agent, in this namespace) is returned. + assert.Equal(t, map[string]bool{"a1": true}, names) +} + +func TestUpdateMetadata(t *testing.T) { + hostPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "host-pod", + Namespace: "ns-test", + Labels: map[string]string{ + translate.ClusterNameLabel: "c-test", + translate.AgentNameLabel: "node-a", + "app": "nginx", + }, + Annotations: map[string]string{ + translate.ResourceNameAnnotation: "my-pod", + translate.ResourceNamespaceAnnotation: "default", + }, + }, + } + + virtualPod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-pod", + Namespace: "default", + Labels: map[string]string{"app": "nginx"}, + }, + } + + updateMetadata(hostPod, virtualPod) + + assert.Equal(t, "c-test", hostPod.Labels[translate.ClusterNameLabel]) + assert.Equal(t, "node-a", hostPod.Labels[translate.AgentNameLabel]) + assert.Equal(t, "my-pod", hostPod.Annotations[translate.ResourceNameAnnotation]) + assert.Equal(t, "default", hostPod.Annotations[translate.ResourceNamespaceAnnotation]) +} diff --git a/k3k-kubelet/translate/host.go b/k3k-kubelet/translate/host.go index d5ae4713..1b4f961e 100644 --- a/k3k-kubelet/translate/host.go +++ b/k3k-kubelet/translate/host.go @@ -14,15 +14,20 @@ import ( ) const ( + // MetadataPrefix is the common prefix for all k3k-managed labels and annotations. + MetadataPrefix = "k3k.io/" // ClusterNameLabel is the key for the label that contains the name of the virtual cluster // this resource was made in - ClusterNameLabel = "k3k.io/clusterName" + ClusterNameLabel = MetadataPrefix + "clusterName" + // AgentNameLabel is the key for the label that contains the name of the k3k-kubelet agent + // (the host node it runs on) that synced this resource + AgentNameLabel = MetadataPrefix + "agentName" // ResourceNameAnnotation is the key for the annotation that contains the original name of this // resource in the virtual cluster - ResourceNameAnnotation = "k3k.io/name" + ResourceNameAnnotation = MetadataPrefix + "name" // ResourceNamespaceAnnotation is the key for the annotation that contains the original namespace of this // resource in the virtual cluster - ResourceNamespaceAnnotation = "k3k.io/namespace" + ResourceNamespaceAnnotation = MetadataPrefix + "namespace" // MetadataNameField is the downwardapi field for object's name MetadataNameField = "metadata.name" // MetadataNamespaceField is the downward field for the object's namespace diff --git a/pkg/controller/cluster/cluster.go b/pkg/controller/cluster/cluster.go index 281eec4e..0cef7b6b 100644 --- a/pkg/controller/cluster/cluster.go +++ b/pkg/controller/cluster/cluster.go @@ -688,6 +688,7 @@ func (c *ClusterReconciler) ensureClusterService(ctx context.Context, cluster *v } else { maps.Copy(currentService.Annotations, expectedService.Annotations) } + currentService.Spec = expectedService.Spec return nil diff --git a/tests/cli/tests_suite_test.go b/tests/cli/tests_suite_test.go index 04bb1ac6..d1238cf9 100644 --- a/tests/cli/tests_suite_test.go +++ b/tests/cli/tests_suite_test.go @@ -15,8 +15,6 @@ import ( ) const ( - k3kNamespace = "k3k-system" - k3sVersion = "v1.36.2-k3s1" k3sOldVersion = "v1.36.0-k3s1" ) diff --git a/tests/e2e/cluster_kubelet_restart_test.go b/tests/e2e/cluster_kubelet_restart_test.go index c5164c2a..ca042d52 100644 --- a/tests/e2e/cluster_kubelet_restart_test.go +++ b/tests/e2e/cluster_kubelet_restart_test.go @@ -7,6 +7,7 @@ import ( "k8s.io/apimachinery/pkg/types" "k8s.io/kubernetes/pkg/api/v1/pod" + appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -188,4 +189,175 @@ var _ = Context("In a shared cluster", Label(e2eTestLabel), Label(slowTestsLabel Should(Succeed()) }) }) + + When("restarting the k3k-kubelet with Pods spread across nodes", func() { + const appLabel = "dangling-test" + + var ( + nodeCount int + virtualPodUIDs map[string]types.UID // virtual pod name -> UID + hostPodUIDs map[string]types.UID // host pod name -> UID + ) + + BeforeAll(func() { + ctx := context.Background() + + By("Listing the virtual cluster nodes") + + nodeList, err := virtualCluster.Client.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + Expect(err).NotTo(HaveOccurred()) + + nodeCount = len(nodeList.Items) + if nodeCount < 2 { + Skip("cross-node dangling-pod deletion can only be reproduced on a multi-node host cluster") + } + + By("Creating a Deployment with one Pod per node (required anti-affinity)") + + replicas := int32(nodeCount) + + deployment := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "dangling-test", + Namespace: "default", + }, + Spec: appsv1.DeploymentSpec{ + Replicas: &replicas, + Selector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": appLabel}, + }, + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{"app": appLabel}, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{ + Name: "nginx", + Image: "nginx", + }}, + // force one Pod per node, so some Pods necessarily live on a node + // other than any given k3k-kubelet instance's own node + Affinity: &corev1.Affinity{ + PodAntiAffinity: &corev1.PodAntiAffinity{ + RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{{ + LabelSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": appLabel}, + }, + TopologyKey: "kubernetes.io/hostname", + }}, + }, + }, + }, + }, + }, + } + + _, err = virtualCluster.Client.AppsV1().Deployments(deployment.Namespace).Create(ctx, deployment, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + + By("Waiting for all Pods to be Running on distinct nodes") + + Eventually(func(g Gomega) { + podList, err := virtualCluster.Client.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "app=" + appLabel}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(podList.Items).To(HaveLen(nodeCount)) + + nodes := map[string]struct{}{} + + for i := range podList.Items { + p := &podList.Items[i] + g.Expect(p.Status.Phase).To(Equal(corev1.PodRunning)) + g.Expect(p.Spec.NodeName).NotTo(BeEmpty()) + nodes[p.Spec.NodeName] = struct{}{} + } + + // the anti-affinity must have spread the Pods across every node + g.Expect(nodes).To(HaveLen(nodeCount)) + }). + WithPolling(time.Second). + WithTimeout(2 * time.Minute). + Should(Succeed()) + + By("Recording the virtual and host Pod UIDs before the restart") + + virtualPodUIDs = map[string]types.UID{} + hostPodUIDs = map[string]types.UID{} + + podList, err := virtualCluster.Client.CoreV1().Pods("default").List(ctx, metav1.ListOptions{LabelSelector: "app=" + appLabel}) + Expect(err).NotTo(HaveOccurred()) + + for i := range podList.Items { + vPod := &podList.Items[i] + virtualPodUIDs[vPod.Name] = vPod.UID + + hostPodName := translator.NamespacedName(vPod) + + hPod, err := k8s.CoreV1().Pods(hostPodName.Namespace).Get(ctx, hostPodName.Name, metav1.GetOptions{}) + Expect(err).NotTo(HaveOccurred()) + + hostPodUIDs[hostPodName.Name] = hPod.UID + } + }) + + It("should not delete Pods scheduled on other nodes after a restart", func() { + ctx := context.Background() + + By("Restarting every k3k-kubelet agent Pod (one per host node)") + + oldAgentPods := listAgentPods(ctx, virtualCluster) + Expect(oldAgentPods).NotTo(BeEmpty()) + + oldAgentUIDs := map[types.UID]bool{} + for _, agentPod := range oldAgentPods { + oldAgentUIDs[agentPod.UID] = true + } + + for _, agentPod := range oldAgentPods { + Expect(k8s.CoreV1().Pods(virtualCluster.Cluster.Namespace).Delete(ctx, agentPod.Name, metav1.DeleteOptions{})).To(Succeed()) + } + + By("Waiting for every k3k-kubelet agent Pod to be replaced and become Ready") + + Eventually(func(g Gomega) { + newAgentPods := listAgentPods(ctx, virtualCluster) + g.Expect(newAgentPods).To(HaveLen(len(oldAgentPods))) + + for _, agentPod := range newAgentPods { + g.Expect(oldAgentUIDs).NotTo(HaveKey(agentPod.UID), "Pod %q was not replaced", agentPod.Name) + + _, cond := pod.GetPodCondition(&agentPod.Status, corev1.PodReady) + g.Expect(cond).NotTo(BeNil()) + g.Expect(cond.Status).To(BeEquivalentTo(metav1.ConditionTrue)) + } + }). + WithPolling(time.Second). + WithTimeout(2 * time.Minute). + Should(Succeed()) + + // deleteDanglingPods runs asynchronously at each k3k-kubelet's startup, so poll over a + // window rather than checking once. If any workload Pod on another node is wrongly + // treated as dangling, its host Pod is deleted (Get returns NotFound / a new UID). + + By("Checking no workload Pod is deleted from the virtual or host cluster") + + Consistently(func(g Gomega) { + for name, uid := range virtualPodUIDs { + vPod, err := virtualCluster.Client.CoreV1().Pods("default").Get(ctx, name, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(vPod.UID).To(Equal(uid)) + g.Expect(vPod.Status.Phase).To(Equal(corev1.PodRunning)) + } + + for hostName, uid := range hostPodUIDs { + hPod, err := k8s.CoreV1().Pods(virtualCluster.Cluster.Namespace).Get(ctx, hostName, metav1.GetOptions{}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(hPod.UID).To(Equal(uid)) + g.Expect(hPod.Status.Phase).To(Equal(corev1.PodRunning)) + } + }). + WithPolling(5 * time.Second). + WithTimeout(time.Minute). + Should(Succeed()) + }) + }) })