Fix fake runtime's image pull

Fixed fake_runtime.PullImage to be blocked at the fake image pulling behavior,
not just before before returning

pkg/kubelet/container/testing/fake_runtime.go

@@ -322,6 +322,30 @@ func (f *FakeRuntime) GetContainerLogs(_ context.Context, pod *v1.Pod, container
 func (f *FakeRuntime) PullImage(ctx context.Context, image kubecontainer.ImageSpec, creds []credentialprovider.TrackedAuthConfig, podSandboxConfig *runtimeapi.PodSandboxConfig) (string, *credentialprovider.TrackedAuthConfig, error) {
 	f.Lock()
 	f.CalledFunctions = append(f.CalledFunctions, "PullImage")
+
+	if f.imagePullTokenBucket == nil {
+		f.imagePullTokenBucket = make(chan bool, 1)
+	}
+	if f.imagePullErrBucket == nil {
+		f.imagePullErrBucket = make(chan error, 1)
+	}
+
+	blockImagePulls := f.BlockImagePulls
+	f.Unlock()
+
+	if blockImagePulls {
+		// Block the function before adding the image to f.ImageList
+		select {
+		case <-ctx.Done():
+		case <-f.imagePullTokenBucket:
+		case pullImageErr := <-f.imagePullErrBucket:
+			return "", nil, pullImageErr
+		}
+	}
+
+	f.Lock()
+	defer f.Unlock()
+
 	if f.Err == nil {
 		i := kubecontainer.Image{
 			ID:   image.Image,
@@ -336,28 +360,7 @@ func (f *FakeRuntime) PullImage(ctx context.Context, image kubecontainer.ImageSp
 		retCreds = &creds[0]
 	}
 
-	if !f.BlockImagePulls {
-		f.Unlock()
-		return image.Image, retCreds, f.Err
-	}
-
-	retErr := f.Err
-	if f.imagePullTokenBucket == nil {
-		f.imagePullTokenBucket = make(chan bool, 1)
-	}
-	if f.imagePullErrBucket == nil {
-		f.imagePullErrBucket = make(chan error, 1)
-	}
-	// Unlock before waiting for UnblockImagePulls calls, to avoid deadlock.
-	f.Unlock()
-	select {
-	case <-ctx.Done():
-	case <-f.imagePullTokenBucket:
-	case pullImageErr := <-f.imagePullErrBucket:
-		return image.Image, retCreds, pullImageErr
-	}
-
-	return image.Image, retCreds, retErr
+	return image.Image, retCreds, f.Err
 }
 
 // UnblockImagePulls unblocks a certain number of image pulls, if BlockImagePulls is true.

pkg/kubelet/images/image_manager.go

@@ -181,6 +181,11 @@ func (m *imageManager) EnsureImageExists(ctx context.Context, objRef *v1.ObjectR
 	if imageRef != "" && !utilfeature.DefaultFeatureGate.Enabled(features.KubeletEnsureSecretPulledImages) {
 		msg := fmt.Sprintf("Container image %q already present on machine", requestedImage)
 		m.logIt(objRef, v1.EventTypeNormal, events.PulledImage, logPrefix, msg, klog.Info)
+
+		// we need to stop recording if it is still in progress, as the image
+		// has already been pulled by another pod.
+		m.podPullingTimeRecorder.RecordImageFinishedPulling(pod.UID)
+
 		return imageRef, msg, nil
 	}
 

pkg/kubelet/images/image_manager_test.go

@@ -1161,6 +1161,13 @@ func TestParallelPodPullingTimeRecorderWithErr(t *testing.T) {
 			UID:             "bar1",
 			ResourceVersion: "42",
 		}}
+	pod1SandboxConfig := &runtimeapi.PodSandboxConfig{
+		Metadata: &runtimeapi.PodSandboxMetadata{
+			Name:      pod1.Name,
+			Namespace: pod1.Namespace,
+			Uid:       string(pod1.UID),
+		},
+	}
 
 	pod2 := &v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
@@ -1169,8 +1176,16 @@ func TestParallelPodPullingTimeRecorderWithErr(t *testing.T) {
 			UID:             "bar2",
 			ResourceVersion: "42",
 		}}
+	pod2SandboxConfig := &runtimeapi.PodSandboxConfig{
+		Metadata: &runtimeapi.PodSandboxMetadata{
+			Name:      pod2.Name,
+			Namespace: pod2.Namespace,
+			Uid:       string(pod2.UID),
+		},
+	}
 
 	pods := [2]*v1.Pod{pod1, pod2}
+	podSandboxes := [2]*runtimeapi.PodSandboxConfig{pod1SandboxConfig, pod2SandboxConfig}
 
 	testCase := &pullerTestCase{
 		containerImage: "missing_image",
@@ -1196,7 +1211,7 @@ func TestParallelPodPullingTimeRecorderWithErr(t *testing.T) {
 	for i := 0; i < 2; i++ {
 		wg.Add(1)
 		go func(i int) {
-			_, _, _ = puller.EnsureImageExists(ctx, nil, pods[i], container.Image, testCase.pullSecrets, nil, "", container.ImagePullPolicy)
+			_, _, _ = puller.EnsureImageExists(ctx, nil, pods[i], container.Image, testCase.pullSecrets, podSandboxes[i], "", container.ImagePullPolicy)
 			wg.Done()
 		}(i)
 	}
@@ -1437,7 +1452,10 @@ func TestEnsureImageExistsWithServiceAccountCoordinates(t *testing.T) {
 			fakeClock := testingclock.NewFakeClock(time.Now())
 			fakeRuntime := &ctest.FakeRuntime{T: t}
 			fakeRecorder := testutil.NewFakeRecorder()
-			fakePodPullingTimeRecorder := &mockPodPullingTimeRecorder{}
+			fakePodPullingTimeRecorder := &mockPodPullingTimeRecorder{
+				startedPullingRecorded:  make(map[types.UID]bool),
+				finishedPullingRecorded: make(map[types.UID]bool),
+			}
 
 			fakeRuntime.ImageList = []Image{{ID: "present_image:latest"}}
 
@@ -1525,7 +1543,10 @@ func TestEnsureImageExistsWithNodeCredentialsOnly(t *testing.T) {
 	fakeClock := testingclock.NewFakeClock(time.Now())
 	fakeRuntime := &ctest.FakeRuntime{T: t}
 	fakeRecorder := testutil.NewFakeRecorder()
-	fakePodPullingTimeRecorder := &mockPodPullingTimeRecorder{}
+	fakePodPullingTimeRecorder := &mockPodPullingTimeRecorder{
+		startedPullingRecorded:  make(map[types.UID]bool),
+		finishedPullingRecorded: make(map[types.UID]bool),
+	}
 
 	fakeRuntime.ImageList = []Image{{ID: "present_image:latest"}}