fix: do not submit a pod deleted while its volumes are resolved

pkg/virtualkubelet/execute.go

@@ -28,6 +28,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	trace "go.opentelemetry.io/otel/trace"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -62,6 +63,14 @@ const (
 	PodPhaseCompleted  = "Completed"
 )
 
+// volumeRetryInterval is how often a missing ConfigMap or Secret is looked up again
+// before the pod is submitted.
+var volumeRetryInterval = 2 * time.Second
+
+// errPodDeletedBeforeCreation tells the caller the pod was deleted before it was
+// submitted to the plugin, so there is nothing left to create.
+var errPodDeletedBeforeCreation = errors.New("pod deleted before its remote creation")
+
 func parseDisableOffloadContainers(pod *v1.Pod) map[string]bool {
 	disabledContainers := make(map[string]bool)
 
@@ -914,9 +923,11 @@ func remoteExecutionHandleVolumes(ctx context.Context, p *Provider, pod *v1.Pod,
 	endTime := startTime.Add(5 * time.Minute)
 
 	_, err := p.clientSet.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
+	if apierrors.IsNotFound(err) {
+		return errPodDeletedBeforeCreation
+	}
 	if err != nil {
-		log.G(ctx).Warning("Deleted Pod before actual creation")
-		return nil
+		log.G(ctx).Warningf("unable to check that pod %s still exists before resolving its volumes: %v", pod.Name, err)
 	}
 	// Sometime the get secret or configmap can fail because it didn't have time to initialize, thus this
 	// is not a true failure. We use this flag to wait.
@@ -1024,8 +1035,17 @@ func remoteExecutionHandleVolumes(ctx context.Context, p *Provider, pod *v1.Pod,
 				}
 
 				if failedAndWait {
-					log.G(ctx).Warningf("volume %s not ready, sleeping 2s, attempt %f / 5 minutes max", volume.Name, time.Since(startTime).Minutes())
-					time.Sleep(2 * time.Second)
+					log.G(ctx).Warningf("volume %s not ready, sleeping %s, attempt %f / 5 minutes max", volume.Name, volumeRetryInterval, time.Since(startTime).Minutes())
+					time.Sleep(volumeRetryInterval)
+					// A pod deleted while it waits for its volumes is dropped from p.pods. Stop
+					// waiting then: submitting it once the volume shows up would leave a remote
+					// job that nothing deletes.
+					p.podsMu.RLock()
+					_, tracked := p.pods[string(pod.UID)]
+					p.podsMu.RUnlock()
+					if !tracked {
+						return errPodDeletedBeforeCreation
+					}
 					continue
 				}
 				pod.Status.Phase = v1.PodPending

pkg/virtualkubelet/execute_test.go

@@ -749,3 +749,97 @@
 	assert.Empty(t, container.Env)
 	assert.Empty(t, container.EnvFrom)
 }
+
+func TestRemoteExecutionHandleVolumesPodDeletedBeforeCreation(t *testing.T) {
+	pod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      testPodName,
+			Namespace: testNamespace,
+		},
+	}
+	p := &Provider{
+		clientSet: fake.NewSimpleClientset(),
+		notifier:  func(*v1.Pod) {},
+	}
+
+	err := remoteExecutionHandleVolumes(t.Context(), p, pod, &types.PodCreateRequests{})
+	assert.ErrorIs(t, err, errPodDeletedBeforeCreation)
+}
+
+func TestRemoteExecutionHandleVolumesWaitingForConfigMap(t *testing.T) {
+	volumeRetryInterval = 10 * time.Millisecond
+	t.Cleanup(func() { volumeRetryInterval = 2 * time.Second })
+
+	newPod := func() *v1.Pod {
+		return &v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      testPodName,
+				Namespace: testNamespace,
+				UID:       "uid-waiting-for-volumes",
+			},
+			Spec: v1.PodSpec{
+				Volumes: []v1.Volume{
+					{
+						Name: "config",
+						VolumeSource: v1.VolumeSource{
+							ConfigMap: &v1.ConfigMapVolumeSource{
+								LocalObjectReference: v1.LocalObjectReference{Name: testConfigMapName},
+							},
+						},
+					},
+				},
+			},
+		}
+	}
+	newProvider := func(pod *v1.Pod) *Provider {
+		return &Provider{
+			clientSet: fake.NewSimpleClientset(pod.DeepCopy()),
+			notifier:  func(*v1.Pod) {},
+			pods:      map[string]*v1.Pod{string(pod.UID): pod},
+		}
+	}
+
+	t.Run("submits the volume once it shows up", func(t *testing.T) {
+		pod := newPod()
+		p := newProvider(pod)
+		req := &types.PodCreateRequests{}
+
+		done := make(chan error, 1)
+		go func() { done <- remoteExecutionHandleVolumes(t.Context(), p, pod.DeepCopy(), req) }()
+		time.Sleep(50 * time.Millisecond)
+		_, err := p.clientSet.CoreV1().ConfigMaps(testNamespace).Create(t.Context(), &v1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{Name: testConfigMapName, Namespace: testNamespace},
+		}, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+			require.Len(t, req.ConfigMaps, 1)
+			assert.Equal(t, testConfigMapName, req.ConfigMaps[0].Name)
+		case <-time.After(2 * time.Second):
+			t.Fatal("still waiting for a ConfigMap that exists")
+		}
+	})
+
+	t.Run("stops when the pod is deleted while waiting", func(t *testing.T) {
+		pod := newPod()
+		p := newProvider(pod)
+		req := &types.PodCreateRequests{}
+
+		done := make(chan error, 1)
+		go func() { done <- remoteExecutionHandleVolumes(t.Context(), p, pod.DeepCopy(), req) }()
+		time.Sleep(50 * time.Millisecond)
+		p.podsMu.Lock()
+		delete(p.pods, string(pod.UID))
+		p.podsMu.Unlock()
+
+		select {
+		case err := <-done:
+			assert.ErrorIs(t, err, errPodDeletedBeforeCreation)
+			assert.Empty(t, req.ConfigMaps)
+		case <-time.After(2 * time.Second):
+			t.Fatal("still waiting for the volumes of a deleted pod")
+		}
+	})
+}

pkg/virtualkubelet/virtualkubelet.go

@@ -8,6 +8,7 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	mathrand "math/rand"
@@ -2149,8 +2150,8 @@ func (p *Provider) CreatePod(ctx context.Context, pod *v1.Pod) error {
 		}
 		err := RemoteExecution(ctx, p.config, p, pod, CREATE)
 		if err != nil {
-			if err.Error() == "Deleted pod before actual creation" {
-				log.G(ctx).Warn(err)
+			if errors.Is(err, errPodDeletedBeforeCreation) {
+				log.G(ctx).Warningf("Pod %s/%s not submitted: %v", pod.Namespace, pod.Name, err)
 			} else {
 				log.G(ctx).Error(err)
 				p.handleRemoteExecutionFailure(ctx, pod, podIP, err)