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

@@ -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

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])
+}

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

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

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"
 )

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())
+		})
+	})
 })