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

@@ -57,10 +57,12 @@ import (
 	// Do some initialization to decode the query parameters correctly.
 	"k8s.io/apiserver/pkg/server/healthz"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	imagepullprogress "k8s.io/client-go/tools/imagepullprogress"
 	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	zpagesfeatures "k8s.io/component-base/zpages/features"
 	"k8s.io/component-base/zpages/flagz"
 	"k8s.io/kubelet/pkg/cri/streaming"
+	imagepullprogressserver "k8s.io/kubelet/pkg/cri/streaming/imagepullprogress"
 	"k8s.io/kubelet/pkg/cri/streaming/portforward"
 	remotecommandserver "k8s.io/kubelet/pkg/cri/streaming/remotecommand"
 	_ "k8s.io/kubernetes/pkg/apis/core/install"
@@ -78,15 +80,16 @@ const (
 )
 
 type fakeKubelet struct {
-	podByNameFunc       func(namespace, name string) (*v1.Pod, bool)
-	machineInfoFunc     func() (*cadvisorapi.MachineInfo, error)
-	podsFunc            func() []*v1.Pod
-	runningPodsFunc     func(ctx context.Context) ([]*v1.Pod, error)
-	logFunc             func(w http.ResponseWriter, req *http.Request)
-	runFunc             func(podFullName string, uid types.UID, containerName string, cmd []string) ([]byte, error)
-	getExecCheck        func(string, types.UID, string, []string, remotecommandserver.Options)
-	getAttachCheck      func(string, types.UID, string, remotecommandserver.Options)
-	getPortForwardCheck func(string, string, types.UID, portforward.V4Options)
+	podByNameFunc             func(namespace, name string) (*v1.Pod, bool)
+	machineInfoFunc           func() (*cadvisorapi.MachineInfo, error)
+	podsFunc                  func() []*v1.Pod
+	runningPodsFunc           func(ctx context.Context) ([]*v1.Pod, error)
+	logFunc                   func(w http.ResponseWriter, req *http.Request)
+	runFunc                   func(podFullName string, uid types.UID, containerName string, cmd []string) ([]byte, error)
+	getExecCheck              func(string, types.UID, string, []string, remotecommandserver.Options)
+	getAttachCheck            func(string, types.UID, string, remotecommandserver.Options)
+	getPortForwardCheck       func(string, string, types.UID, portforward.V4Options)
+	getImagePullProgressCheck func(podName, podNamespace string, podUID types.UID, imagePullProgressOpts imagepullprogressserver.Options)
 
 	containerLogsFunc func(ctx context.Context, podFullName, containerName string, logOptions *v1.PodLogOptions, stdout, stderr io.Writer) error
 	hostnameFunc      func() string
@@ -169,9 +172,10 @@ func (fk *fakeKubelet) SyncLoopHealthCheck(req *http.Request) error {
 }
 
 type fakeRuntime struct {
-	execFunc        func(string, []string, io.Reader, io.WriteCloser, io.WriteCloser, bool, <-chan remotecommand.TerminalSize) error
-	attachFunc      func(string, io.Reader, io.WriteCloser, io.WriteCloser, bool, <-chan remotecommand.TerminalSize) error
-	portForwardFunc func(string, int32, io.ReadWriteCloser) error
+	execFunc              func(string, []string, io.Reader, io.WriteCloser, io.WriteCloser, bool, <-chan remotecommand.TerminalSize) error
+	attachFunc            func(string, io.Reader, io.WriteCloser, io.WriteCloser, bool, <-chan remotecommand.TerminalSize) error
+	portForwardFunc       func(string, int32, io.ReadWriteCloser) error
+	imagePullProgressFunc func(string, chan<- imagepullprogress.Progress) error
 }
 
 func (f *fakeRuntime) Exec(_ context.Context, containerID string, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error {
@@ -186,6 +190,10 @@ func (f *fakeRuntime) PortForward(_ context.Context, podSandboxID string, port i
 	return f.portForwardFunc(podSandboxID, port, stream)
 }
 
+func (f *fakeRuntime) ImagePullProgress(_ context.Context, podSandboxID string, progress chan<- imagepullprogress.Progress) error {
+	return f.imagePullProgressFunc(podSandboxID, progress)
+}
+
 type testStreamingServer struct {
 	streaming.Server
 	fakeRuntime    *fakeRuntime
@@ -273,6 +281,20 @@ func (fk *fakeKubelet) GetPortForward(ctx context.Context, podName, podNamespace
 	return url.Parse(resp.GetUrl())
 }
 
+func (fk *fakeKubelet) GetImagePullProgress(ctx context.Context, podName, podNamespace string, podUID types.UID, imagePullProgressOpts imagepullprogressserver.Options) (*url.URL, error) {
+	if fk.getImagePullProgressCheck != nil {
+		fk.getImagePullProgressCheck(podName, podNamespace, podUID, imagePullProgressOpts)
+	}
+	// Always use testPodSandboxID
+	resp, err := fk.streamingRuntime.GetImagePullProgress(&runtimeapi.ImagePullProgressRequest{
+		PodSandboxId: testPodSandboxID,
+	})
+	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 Progress struct {
+	Name     string `json:"name"`
+	Progress int64  `json:"progress"`
+	Total    int64  `json:"total"`
+	Error    string `json:"error,omitempty"`
+}