Merge pull request #26145 from Random-Liu/image-pulling-progress

Automatic merge from submit-queue

Kubelet: Periodically reporting image pulling progress in log

Addresses https://github.com/kubernetes/kubernetes/issues/26075#issuecomment-221129896 and https://github.com/kubernetes/kubernetes/pull/26122#issuecomment-221128397.

This PR changes kube_docker_client to log pulling progress every *10* seconds. We can't print all progress messages into the log, because there are too many. So I make it report newest progress every 10 seconds to reduce log spam.
If the image pulling is too slow or stuck, we'll see image pulling progress unchanged or changed little overtime.

The following is the result if I set the reporting interval to *1* second.
```
I0524 00:53:26.189086  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "17b6a9e179d7: Pulling fs layer "
I0524 00:53:27.189082  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "202e40f8bb3a: Download complete "
I0524 00:53:28.189160  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Downloading [=>                                                 ] 1.474 MB/48.35 MB"
I0524 00:53:29.189156  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Downloading [====>                                              ] 3.931 MB/48.35 MB"
I0524 00:53:30.189089  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Downloading [=========>                                         ] 8.847 MB/48.35 MB"
I0524 00:53:31.189089  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Downloading [==================>                                ] 18.19 MB/48.35 MB"
I0524 00:53:32.189076  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Downloading [=======================================>           ] 38.34 MB/48.35 MB"
I0524 00:53:33.189106  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Extracting [=============>                                     ] 12.78 MB/48.35 MB"
I0524 00:53:34.189067  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Extracting [==================================>                ] 33.42 MB/48.35 MB"
I0524 00:53:35.189083  145099 kube_docker_client.go:252] Pulling image "ubuntu:latest": "487bffc61de6: Extracting [==================================================>] 48.35 MB/48.35 MB"
I0524 00:53:35.376667  145099 kube_docker_client.go:254] Finish pulling image "ubuntu:latest": "Status: Downloaded newer image for ubuntu:latest"
```

Ref image pulling related issue #19077.

[![Analytics](https://kubernetes-site.appspot.com/UA-36037335-10/GitHub/.github/PULL_REQUEST_TEMPLATE.md?pixel)]()

@yujuhong @dchen1107 
/cc @kubernetes/sig-node

pkg/kubelet/dockertools/kube_docker_client.go

@@ -23,8 +23,11 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"sync"
 	"time"
 
+	"github.com/golang/glog"
+
 	dockermessage "github.com/docker/docker/pkg/jsonmessage"
 	dockerstdcopy "github.com/docker/docker/pkg/stdcopy"
 	dockerapi "github.com/docker/engine-api/client"
@@ -58,6 +61,9 @@ const (
 
 	// defaultShmSize is the default ShmSize to use (in bytes) if not specified.
 	defaultShmSize = int64(1024 * 1024 * 64)
+
+	// defaultImagePullingProgressReportInterval is the default interval of image pulling progress reporting.
+	defaultImagePullingProgressReportInterval = 10 * time.Second
 )
 
 // newKubeDockerClient creates an kubeDockerClient from an existing docker client.
@@ -192,6 +198,71 @@ func base64EncodeAuth(auth dockertypes.AuthConfig) (string, error) {
 	return base64.URLEncoding.EncodeToString(buf.Bytes()), nil
 }
 
+// progress is a wrapper of dockermessage.JSONMessage with a lock protecting it.
+type progress struct {
+	sync.RWMutex
+	// message stores the latest docker json message.
+	message *dockermessage.JSONMessage
+}
+
+func (p *progress) set(msg *dockermessage.JSONMessage) {
+	p.Lock()
+	defer p.Unlock()
+	p.message = msg
+}
+
+func (p *progress) get() string {
+	p.RLock()
+	defer p.RUnlock()
+	if p.message == nil {
+		return "No progress"
+	}
+	var prefix string
+	if p.message.ID != "" {
+		prefix = fmt.Sprintf("%s: ", p.message.ID)
+	}
+	if p.message.Progress == nil {
+		return fmt.Sprintf("%s%s", prefix, p.message.Status)
+	}
+	return fmt.Sprintf("%s%s %s", prefix, p.message.Status, p.message.Progress.String())
+}
+
+// progressReporter keeps the newest image pulling progress and periodically report the newest progress.
+type progressReporter struct {
+	progress
+	image    string
+	interval time.Duration
+	stopCh   chan struct{}
+}
+
+// newProgressReporter creates a new progressReporter for specific image with specified reporting interval
+func newProgressReporter(image string, interval time.Duration) *progressReporter {
+	return &progressReporter{image: image, interval: interval, stopCh: make(chan struct{})}
+}
+
+// start starts the progressReporter
+func (p *progressReporter) start() {
+	go func() {
+		ticker := time.NewTicker(p.interval)
+		defer ticker.Stop()
+		for {
+			// TODO(random-liu): Report as events.
+			select {
+			case <-ticker.C:
+				glog.V(2).Infof("Pulling image %q: %q", p.image, p.progress.get())
+			case <-p.stopCh:
+				glog.V(2).Infof("Stop pulling image %q: %q", p.image, p.progress.get())
+				return
+			}
+		}
+	}()
+}
+
+// stop stops the progressReporter
+func (p *progressReporter) stop() {
+	close(p.stopCh)
+}
+
 func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig, opts dockertypes.ImagePullOptions) error {
 	// RegistryAuth is the base64 encoded credentials for the registry
 	base64Auth, err := base64EncodeAuth(auth)
@@ -206,7 +277,9 @@ func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig,
 		return err
 	}
 	defer resp.Close()
-	// TODO(random-liu): Use the image pulling progress information.
+	reporter := newProgressReporter(image, defaultImagePullingProgressReportInterval)
+	reporter.start()
+	defer reporter.stop()
 	decoder := json.NewDecoder(resp)
 	for {
 		var msg dockermessage.JSONMessage
@@ -220,6 +293,7 @@ func (d *kubeDockerClient) PullImage(image string, auth dockertypes.AuthConfig,
 		if msg.Error != nil {
 			return msg.Error
 		}
+		reporter.set(&msg)
 	}
 	return nil
 }