diff --git a/pkg/kubelet/container/testing/fake_runtime.go b/pkg/kubelet/container/testing/fake_runtime.go index 1ad90651a6d..554a9e07919 100644 --- a/pkg/kubelet/container/testing/fake_runtime.go +++ b/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. diff --git a/pkg/kubelet/images/image_manager.go b/pkg/kubelet/images/image_manager.go index d4bba346be7..b318b347984 100644 --- a/pkg/kubelet/images/image_manager.go +++ b/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 } diff --git a/pkg/kubelet/images/image_manager_test.go b/pkg/kubelet/images/image_manager_test.go index fd6c4449855..ad460e097f9 100644 --- a/pkg/kubelet/images/image_manager_test.go +++ b/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"}}