KEP-5334: Image Pull Progress

Author: wzshiming

pkg/kubelet/container/runtime.go

@@ -151,6 +151,7 @@ type StreamingRuntime interface {
 	GetExec(ctx context.Context, id ContainerID, cmd []string, stdin, stdout, stderr, tty bool) (*url.URL, error)
 	GetAttach(ctx context.Context, id ContainerID, stdin, stdout, stderr, tty bool) (*url.URL, error)
 	GetPortForward(ctx context.Context, podName, podNamespace string, podUID types.UID, ports []int32) (*url.URL, error)
+	GetImagePullProgress(ctx context.Context, podName, podNamespace string, podUID types.UID) (*url.URL, error)
 }
 
 // ImageService interfaces allows to work with image service.

pkg/kubelet/container/testing/fake_runtime.go

@@ -511,6 +511,14 @@ func (f *FakeStreamingRuntime) GetPortForward(_ context.Context, podName, podNam
 	return &url.URL{Host: FakeHost}, f.Err
 }
 
+func (f *FakeStreamingRuntime) GetImagePullProgress(ctx context.Context, podName, podNamespace string, podUID types.UID) (*url.URL, error) {
+	f.Lock()
+	defer f.Unlock()
+
+	f.CalledFunctions = append(f.CalledFunctions, "GetImagePullProgress")
+	return &url.URL{Host: FakeHost}, f.Err
+}
+
 type FakeContainerCommandRunner struct {
 	// what to return
 	Stdout string

pkg/kubelet/kubelet_pods.go

@@ -44,6 +44,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1"
 	"k8s.io/klog/v2"
+	"k8s.io/kubelet/pkg/cri/streaming/imagepullprogress"
 	"k8s.io/kubelet/pkg/cri/streaming/portforward"
 	remotecommandserver "k8s.io/kubelet/pkg/cri/streaming/remotecommand"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
@@ -2450,6 +2451,24 @@ func (kl *Kubelet) GetPortForward(ctx context.Context, podName, podNamespace str
 	return kl.streamingRuntime.GetPortForward(ctx, podName, podNamespace, podUID, portForwardOpts.Ports)
 }
 
+// GetImagePullProgress gets the URL the image-pull-progress will be served from, or nil if the Kubelet will serve it.
+func (kl *Kubelet) GetImagePullProgress(ctx context.Context, podName, podNamespace string, podUID types.UID, portForwardOpts imagepullprogress.Options) (*url.URL, error) {
+	pods, err := kl.containerRuntime.GetPods(ctx, false)
+	if err != nil {
+		return nil, err
+	}
+	// Resolve and type convert back again.
+	// We need the static pod UID but the kubecontainer API works with types.UID.
+	podUID = types.UID(kl.podManager.TranslatePodUID(podUID))
+	podFullName := kubecontainer.BuildPodFullName(podName, podNamespace)
+	pod := kubecontainer.Pods(pods).FindPod(podFullName, podUID)
+	if pod.IsEmpty() {
+		return nil, fmt.Errorf("pod not found (%q)", podFullName)
+	}
+
+	return kl.streamingRuntime.GetImagePullProgress(ctx, podName, podNamespace, podUID)
+}
+
 // cleanupOrphanedPodCgroups removes cgroups that should no longer exist.
 // it reconciles the cached state of cgroupPods with the specified list of runningPods
 func (kl *Kubelet) cleanupOrphanedPodCgroups(pcm cm.PodContainerManager, cgroupPods map[types.UID]cm.CgroupName, possiblyRunningPods map[types.UID]sets.Empty) {

pkg/kubelet/kuberuntime/instrumented_services.go

@@ -272,6 +272,15 @@ func (in instrumentedRuntimeService) PortForward(ctx context.Context, req *runti
 	return resp, err
 }
 
+func (in instrumentedRuntimeService) ImagePullProgress(ctx context.Context, req *runtimeapi.ImagePullProgressRequest) (*runtimeapi.ImagePullProgressResponse, error) {
+	const operation = "image_pull_progress"
+	defer recordOperation(operation, time.Now())
+
+	resp, err := in.service.ImagePullProgress(ctx, req)
+	recordError(operation, err)
+	return resp, err
+}
+
 func (in instrumentedRuntimeService) UpdatePodSandboxResources(ctx context.Context, req *runtimeapi.UpdatePodSandboxResourcesRequest) (*runtimeapi.UpdatePodSandboxResourcesResponse, error) {
 	const operation = "update_podsandbox_resources"
 	defer recordOperation(operation, time.Now())

pkg/kubelet/kuberuntime/kuberuntime_sandbox.go

@@ -380,3 +380,22 @@ func (m *kubeGenericRuntimeManager) GetPortForward(ctx context.Context, podName,
 	}
 	return url.Parse(resp.Url)
 }
+
+// GetImagePullProgress gets the endpoint the runtime will serve the image-pull-progress request from.
+func (m *kubeGenericRuntimeManager) GetImagePullProgress(ctx context.Context, podName, podNamespace string, podUID kubetypes.UID) (*url.URL, error) {
+	sandboxIDs, err := m.getSandboxIDByPodUID(ctx, podUID, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to find sandboxID for pod %s: %v", format.PodDesc(podName, podNamespace, podUID), err)
+	}
+	if len(sandboxIDs) == 0 {
+		return nil, fmt.Errorf("failed to find sandboxID for pod %s", format.PodDesc(podName, podNamespace, podUID))
+	}
+	req := &runtimeapi.ImagePullProgressRequest{
+		PodSandboxId: sandboxIDs[0],
+	}
+	resp, err := m.runtimeService.ImagePullProgress(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	return url.Parse(resp.Url)
+}

pkg/kubelet/server/server_test.go

@@ -273,6 +273,21 @@ func (fk *fakeKubelet) GetPortForward(ctx context.Context, podName, podNamespace
 	return url.Parse(resp.GetUrl())
 }
 
+func (fk *fakeKubelet) GetPortForward(ctx context.Context, podName, podNamespace string, podUID types.UID, portForwardOpts portforward.V4Options) (*url.URL, error) {
+	if fk.getPortForwardCheck != nil {
+		fk.getPortForwardCheck(podName, podNamespace, podUID, portForwardOpts)
+	}
+	// Always use testPodSandboxID
+	resp, err := fk.streamingRuntime.GetPortForward(&runtimeapi.PortForwardRequest{
+		PodSandboxId: testPodSandboxID,
+		Port:         portForwardOpts.Ports,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return url.Parse(resp.GetUrl())
+}
+
 // Unused functions
 func (*fakeKubelet) GetNode() (*v1.Node, error)                       { return nil, nil }
 func (*fakeKubelet) GetNodeConfig() cm.NodeConfig                     { return cm.NodeConfig{} }

staging/src/k8s.io/client-go/tools/imagepullprogress/imagepullprogress.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package imagepullprogress
+
+type Layer struct {
+	Name     string `json:"name"`
+	Progress int64  `json:"progress"`
+	Total    int64  `json:"total"`
+	Error    string `json:"error,omitempty"`
+}