diff --git a/pkg/virtualkubelet/execute.go b/pkg/virtualkubelet/execute.go index 7b0b763d..71c29893 100644 --- a/pkg/virtualkubelet/execute.go +++ b/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 diff --git a/pkg/virtualkubelet/execute_test.go b/pkg/virtualkubelet/execute_test.go index 48b0bd77..7a6ab3c7 100644 --- a/pkg/virtualkubelet/execute_test.go +++ b/pkg/virtualkubelet/execute_test.go @@ -749,3 +749,97 @@ func TestResolveEnvFromRefsOptionalMissingSecret(t *testing.T) { 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") + } + }) +} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index a1f07406..8e6b9e53 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/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)