Merge pull request kubernetes#98850 from yangjunmyfm192085/run-test14

Structured Logging migration: modify volume and container part logs o…

Author: k8s-ci-robot

pkg/kubelet/container/container_gc.go

@@ -82,6 +82,6 @@ func (cgc *realContainerGC) GarbageCollect() error {
 }
 
 func (cgc *realContainerGC) DeleteAllUnusedContainers() error {
-	klog.Infof("attempting to delete unused containers")
+	klog.InfoS("Attempting to delete unused containers")
 	return cgc.runtime.GarbageCollect(cgc.policy, cgc.sourcesReadyProvider.AllReady(), true)
 }

pkg/kubelet/container/helpers.go

@@ -31,7 +31,6 @@ import (
 	"k8s.io/client-go/tools/record"
 	runtimeapi "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
-	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	hashutil "k8s.io/kubernetes/pkg/util/hash"
 	"k8s.io/kubernetes/third_party/forked/golang/expansion"
 	utilsnet "k8s.io/utils/net"
@@ -82,13 +81,13 @@ func ShouldContainerBeRestarted(container *v1.Container, pod *v1.Pod, podStatus
 	}
 	// Check RestartPolicy for dead container
 	if pod.Spec.RestartPolicy == v1.RestartPolicyNever {
-		klog.V(4).Infof("Already ran container %q of pod %q, do nothing", container.Name, format.Pod(pod))
+		klog.V(4).InfoS("Already ran container, do nothing", "pod", klog.KObj(pod), "containerName", container.Name)
 		return false
 	}
 	if pod.Spec.RestartPolicy == v1.RestartPolicyOnFailure {
 		// Check the exit code.
 		if status.ExitCode == 0 {
-			klog.V(4).Infof("Already successfully ran container %q of pod %q, do nothing", container.Name, format.Pod(pod))
+			klog.V(4).InfoS("Already successfully ran container, do nothing", "pod", klog.KObj(pod), "containerName", container.Name)
 			return false
 		}
 	}
@@ -341,7 +340,7 @@ func MakePortMappings(container *v1.Container) (ports []PortMapping) {
 
 		// Protect against a port name being used more than once in a container.
 		if _, ok := names[name]; ok {
-			klog.Warningf("Port name conflicted, %q is defined more than once", name)
+			klog.InfoS("Port name conflicted, it is defined more than once", "portName", name)
 			continue
 		}
 		ports = append(ports, pm)

pkg/kubelet/container/runtime.go

@@ -204,7 +204,7 @@ func BuildContainerID(typ, ID string) ContainerID {
 func ParseContainerID(containerID string) ContainerID {
 	var id ContainerID
 	if err := id.ParseString(containerID); err != nil {
-		klog.Error(err)
+		klog.ErrorS(err, "Parsing containerID failed")
 	}
 	return id
 }

pkg/kubelet/kubelet_volumes.go

@@ -63,11 +63,11 @@ func (kl *Kubelet) podVolumesExist(podUID types.UID) bool {
 	// There are some volume plugins such as flexvolume might not have mounts. See issue #61229
 	volumePaths, err := kl.getMountedVolumePathListFromDisk(podUID)
 	if err != nil {
-		klog.Errorf("pod %q found, but error %v occurred during checking mounted volumes from disk", podUID, err)
+		klog.ErrorS(err, "Pod found, but error occurred during checking mounted volumes from disk", "podUID", podUID)
 		return true
 	}
 	if len(volumePaths) > 0 {
-		klog.V(4).Infof("pod %q found, but volumes are still mounted on disk %v", podUID, volumePaths)
+		klog.V(4).InfoS("Pod found, but volumes are still mounted on disk", "podUID", podUID, "volumePaths", volumePaths)
 		return true
 	}
 
@@ -86,7 +86,7 @@ func (kl *Kubelet) newVolumeMounterFromPlugins(spec *volume.Spec, pod *v1.Pod, o
 	if err != nil {
 		return nil, fmt.Errorf("failed to instantiate mounter for volume: %s using plugin: %s with a root cause: %v", spec.Name(), plugin.GetPluginName(), err)
 	}
-	klog.V(10).Infof("Using volume plugin %q to mount %s", plugin.GetPluginName(), spec.Name())
+	klog.V(10).InfoS("Using volume plugin for mount", "volumePluginName", plugin.GetPluginName(), "volumeName", spec.Name())
 	return physicalMounter, nil
 }
 
@@ -118,7 +118,7 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecon
 		// TODO: getMountedVolumePathListFromDisk() call may be redundant with
 		// kl.getPodVolumePathListFromDisk(). Can this be cleaned up?
 		if podVolumesExist := kl.podVolumesExist(uid); podVolumesExist {
-			klog.V(3).Infof("Orphaned pod %q found, but volumes are not cleaned up", uid)
+			klog.V(3).InfoS("Orphaned pod found, but volumes are not cleaned up", "podUID", uid)
 			continue
 		}
 
@@ -167,18 +167,18 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecon
 			continue
 		}
 
-		klog.V(3).Infof("Orphaned pod %q found, removing", uid)
+		klog.V(3).InfoS("Orphaned pod found, removing", "podUID", uid)
 		if err := removeall.RemoveAllOneFilesystem(kl.mounter, kl.getPodDir(uid)); err != nil {
-			klog.Errorf("Failed to remove orphaned pod %q dir; err: %v", uid, err)
+			klog.ErrorS(err, "Failed to remove orphaned pod dir", "podUID", uid)
 			orphanRemovalErrors = append(orphanRemovalErrors, err)
 		}
 	}
 
 	logSpew := func(errs []error) {
 		if len(errs) > 0 {
-			klog.Errorf("%v : There were a total of %v errors similar to this. Turn up verbosity to see them.", errs[0], len(errs))
+			klog.ErrorS(errs[0], "There were many similar errors. Turn up verbosity to see them.", "numErrs", len(errs))
 			for _, err := range errs {
-				klog.V(5).Infof("Orphan pod: %v", err)
+				klog.V(5).InfoS("Orphan pod", "err", err)
 			}
 		}
 	}

pkg/kubelet/pod_container_deletor.go

@@ -49,7 +49,7 @@ func newPodContainerDeletor(runtime kubecontainer.Runtime, containersToKeep int)
 		for {
 			id := <-buffer
 			if err := runtime.DeleteContainer(id); err != nil {
-				klog.Warningf("[pod_container_deletor] DeleteContainer returned error for (id=%v): %v", id, err)
+				klog.InfoS("DeleteContainer returned error", "containerID", id, "err", err)
 			}
 		}
 	}, 0, wait.NeverStop)
@@ -76,7 +76,7 @@ func getContainersToDeleteInPod(filterContainerID string, podStatus *kubecontain
 	}(filterContainerID, podStatus)
 
 	if filterContainerID != "" && matchedContainer == nil {
-		klog.Warningf("Container %q not found in pod's containers", filterContainerID)
+		klog.InfoS("Container not found in pod's containers", "containerID", filterContainerID)
 		return containerStatusbyCreatedList{}
 	}
 
@@ -110,7 +110,7 @@ func (p *podContainerDeletor) deleteContainersInPod(filterContainerID string, po
 		select {
 		case p.worker <- candidate.ID:
 		default:
-			klog.Warningf("Failed to issue the request to remove container %v", candidate.ID)
+			klog.InfoS("Failed to issue the request to remove container", "containerID", candidate.ID)
 		}
 	}
 }

pkg/kubelet/pod_workers.go

@@ -33,7 +33,6 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/eviction"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
-	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	"k8s.io/kubernetes/pkg/kubelet/util/queue"
 )
 
@@ -188,7 +187,7 @@ func (p *podWorkers) managePodLoop(podUpdates <-chan UpdatePodOptions) {
 		}
 		if err != nil {
 			// IMPORTANT: we do not log errors here, the syncPodFn is responsible for logging errors
-			klog.Errorf("Error syncing pod %s (%q), skipping: %v", update.Pod.UID, format.Pod(update.Pod), err)
+			klog.ErrorS(err, "Error syncing pod, skipping", "pod", klog.KObj(update.Pod), "podUID", update.Pod.UID)
 		}
 		p.wrapUp(update.Pod.UID, err)
 	}

pkg/kubelet/runonce.go

@@ -51,15 +51,15 @@ func (kl *Kubelet) RunOnce(updates <-chan kubetypes.PodUpdate) ([]RunPodResult,
 	// If the container logs directory does not exist, create it.
 	if _, err := os.Stat(ContainerLogsDir); err != nil {
 		if err := kl.os.MkdirAll(ContainerLogsDir, 0755); err != nil {
-			klog.Errorf("Failed to create directory %q: %v", ContainerLogsDir, err)
+			klog.ErrorS(err, "Failed to create directory", "path", ContainerLogsDir)
 		}
 	}
 
 	select {
 	case u := <-updates:
-		klog.Infof("processing manifest with %d pods", len(u.Pods))
+		klog.InfoS("Processing manifest with pods", "numPods", len(u.Pods))
 		result, err := kl.runOnce(u.Pods, runOnceRetryDelay)
-		klog.Infof("finished processing %d pods", len(u.Pods))
+		klog.InfoS("Finished processing pods", "numPods", len(u.Pods))
 		return result, err
 	case <-time.After(runOnceManifestDelay):
 		return nil, fmt.Errorf("no pod manifest update after %v", runOnceManifestDelay)
@@ -85,27 +85,27 @@ func (kl *Kubelet) runOnce(pods []*v1.Pod, retryDelay time.Duration) (results []
 		}(pod)
 	}
 
-	klog.Infof("Waiting for %d pods", len(admitted))
+	klog.InfoS("Waiting for pods", "numPods", len(admitted))
 	failedPods := []string{}
 	for i := 0; i < len(admitted); i++ {
 		res := <-ch
 		results = append(results, res)
 		if res.Err != nil {
-			faliedContainerName, err := kl.getFailedContainers(res.Pod)
+			failedContainerName, err := kl.getFailedContainers(res.Pod)
 			if err != nil {
-				klog.Infof("unable to get failed containers' names for pod %q, error:%v", format.Pod(res.Pod), err)
+				klog.InfoS("Unable to get failed containers' names for pod", "pod", klog.KObj(res.Pod), "err", err)
 			} else {
-				klog.Infof("unable to start pod %q because container:%v failed", format.Pod(res.Pod), faliedContainerName)
+				klog.InfoS("Unable to start pod because container failed", "pod", klog.KObj(res.Pod), "containerName", failedContainerName)
 			}
 			failedPods = append(failedPods, format.Pod(res.Pod))
 		} else {
-			klog.Infof("started pod %q", format.Pod(res.Pod))
+			klog.InfoS("Started pod", "pod", klog.KObj(res.Pod))
 		}
 	}
 	if len(failedPods) > 0 {
 		return results, fmt.Errorf("error running pods: %v", failedPods)
 	}
-	klog.Infof("%d pods started", len(pods))
+	klog.InfoS("Pods started", "numPods", len(pods))
 	return results, err
 }
 
@@ -120,14 +120,14 @@ func (kl *Kubelet) runPod(pod *v1.Pod, retryDelay time.Duration) error {
 		}
 
 		if kl.isPodRunning(pod, status) {
-			klog.Infof("pod %q containers running", format.Pod(pod))
+			klog.InfoS("Pod's containers running", "pod", klog.KObj(pod))
 			return nil
 		}
-		klog.Infof("pod %q containers not running: syncing", format.Pod(pod))
+		klog.InfoS("Pod's containers not running: syncing", "pod", klog.KObj(pod))
 
-		klog.Infof("Creating a mirror pod for static pod %q", format.Pod(pod))
+		klog.InfoS("Creating a mirror pod for static pod", "pod", klog.KObj(pod))
 		if err := kl.podManager.CreateMirrorPod(pod); err != nil {
-			klog.Errorf("Failed creating a mirror pod %q: %v", format.Pod(pod), err)
+			klog.ErrorS(err, "Failed creating a mirror pod", "pod", klog.KObj(pod))
 		}
 		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
 		if err = kl.syncPod(syncPodOptions{
@@ -142,7 +142,7 @@ func (kl *Kubelet) runPod(pod *v1.Pod, retryDelay time.Duration) error {
 			return fmt.Errorf("timeout error: pod %q containers not running after %d retries", format.Pod(pod), runOnceMaxRetries)
 		}
 		// TODO(proppy): health checking would be better than waiting + checking the state at the next iteration.
-		klog.Infof("pod %q containers synced, waiting for %v", format.Pod(pod), delay)
+		klog.InfoS("Pod's containers synced, waiting", "pod", klog.KObj(pod), "duration", delay)
 		time.Sleep(delay)
 		retry++
 		delay *= runOnceRetryDelayBackoff
@@ -154,7 +154,7 @@ func (kl *Kubelet) isPodRunning(pod *v1.Pod, status *kubecontainer.PodStatus) bo
 	for _, c := range pod.Spec.Containers {
 		cs := status.FindContainerStatusByName(c.Name)
 		if cs == nil || cs.State != kubecontainer.ContainerStateRunning {
-			klog.Infof("Container %q for pod %q not running", c.Name, format.Pod(pod))
+			klog.InfoS("Container not running", "pod", klog.KObj(pod), "containerName", c.Name)
 			return false
 		}
 	}

pkg/kubelet/volume_host.go

@@ -72,7 +72,7 @@ func NewInitializedVolumePluginMgr(
 		csiDriversSynced = csiDriverInformer.Informer().HasSynced
 
 	} else {
-		klog.Warning("kubeClient is nil. Skip initialization of CSIDriverLister")
+		klog.InfoS("KubeClient is nil. Skip initialization of CSIDriverLister")
 	}
 
 	kvh := &kubeletVolumeHost{
@@ -176,13 +176,13 @@ func (kvh *kubeletVolumeHost) CSIDriversSynced() cache.InformerSynced {
 // WaitForCacheSync is a helper function that waits for cache sync for CSIDriverLister
 func (kvh *kubeletVolumeHost) WaitForCacheSync() error {
 	if kvh.csiDriversSynced == nil {
-		klog.Error("csiDriversSynced not found on KubeletVolumeHost")
+		klog.ErrorS(nil, "CsiDriversSynced not found on KubeletVolumeHost")
 		return fmt.Errorf("csiDriversSynced not found on KubeletVolumeHost")
 	}
 
 	synced := []cache.InformerSynced{kvh.csiDriversSynced}
 	if !cache.WaitForCacheSync(wait.NeverStop, synced...) {
-		klog.Warning("failed to wait for cache sync for CSIDriverLister")
+		klog.InfoS("Failed to wait for cache sync for CSIDriverLister")
 		return fmt.Errorf("failed to wait for cache sync for CSIDriverLister")
 	}
 

pkg/kubelet/volumemanager/cache/actual_state_of_world.go

@@ -461,9 +461,7 @@ func (asw *actualStateOfWorld) addVolume(
 	} else {
 		// If volume object already exists, update the fields such as device path
 		volumeObj.devicePath = devicePath
-		klog.V(2).Infof("Volume %q is already added to attachedVolume list, update device path %q",
-			volumeName,
-			devicePath)
+		klog.V(2).InfoS("Volume is already added to attachedVolume list, update device path", "volumeName", volumeName, "path", devicePath)
 	}
 	asw.attachedVolumes[volumeName] = volumeObj
 
@@ -530,9 +528,7 @@ func (asw *actualStateOfWorld) MarkVolumeAsResized(
 			podName,
 			volumeName)
 	}
-
-	klog.V(5).Infof("Volume %s(OuterVolumeSpecName %s) of pod %s has been resized",
-		volumeName, podObj.outerVolumeSpecName, podName)
+	klog.V(5).InfoS("Pod volume has been resized", "uniquePodName", podName, "volumeName", volumeName, "outerVolumeSpecName", podObj.outerVolumeSpecName)
 	podObj.fsResizeRequired = false
 	asw.attachedVolumes[volumeName].mountedPods[podName] = podObj
 	return nil
@@ -548,12 +544,7 @@ func (asw *actualStateOfWorld) MarkRemountRequired(
 				asw.volumePluginMgr.FindPluginBySpec(podObj.volumeSpec)
 			if err != nil || volumePlugin == nil {
 				// Log and continue processing
-				klog.Errorf(
-					"MarkRemountRequired failed to FindPluginBySpec for pod %q (podUid %q) volume: %q (volSpecName: %q)",
-					podObj.podName,
-					podObj.podUID,
-					volumeObj.volumeName,
-					podObj.volumeSpec.Name())
+				klog.ErrorS(nil, "MarkRemountRequired failed to FindPluginBySpec for volume", "uniquePodName", podObj.podName, "podUID", podObj.podUID, "volumeName", volumeName, "volumeSpecName", podObj.volumeSpec.Name())
 				continue
 			}
 
@@ -572,33 +563,27 @@ func (asw *actualStateOfWorld) MarkFSResizeRequired(
 	defer asw.Unlock()
 	volumeObj, volumeExists := asw.attachedVolumes[volumeName]
 	if !volumeExists {
-		klog.Warningf("MarkFSResizeRequired for volume %s failed as volume not exist", volumeName)
+		klog.InfoS("MarkFSResizeRequired for volume failed as volume does not exist", "volumeName", volumeName)
 		return
 	}
 
 	podObj, podExists := volumeObj.mountedPods[podName]
 	if !podExists {
-		klog.Warningf("MarkFSResizeRequired for volume %s failed "+
-			"as pod(%s) not exist", volumeName, podName)
+		klog.InfoS("MarkFSResizeRequired for volume failed because the pod does not exist", "uniquePodName", podName, "volumeName", volumeName)
 		return
 	}
 
 	volumePlugin, err :=
 		asw.volumePluginMgr.FindNodeExpandablePluginBySpec(podObj.volumeSpec)
 	if err != nil || volumePlugin == nil {
 		// Log and continue processing
-		klog.Errorf(
-			"MarkFSResizeRequired failed to find expandable plugin for pod %q volume: %q (volSpecName: %q)",
-			podObj.podName,
-			volumeObj.volumeName,
-			podObj.volumeSpec.Name())
+		klog.ErrorS(nil, "MarkFSResizeRequired failed to find expandable plugin for volume", "uniquePodName", podObj.podName, "volumeName", volumeObj.volumeName, "volumeSpecName", podObj.volumeSpec.Name())
 		return
 	}
 
 	if volumePlugin.RequiresFSResize() {
 		if !podObj.fsResizeRequired {
-			klog.V(3).Infof("PVC volume %s(OuterVolumeSpecName %s) of pod %s requires file system resize",
-				volumeName, podObj.outerVolumeSpecName, podName)
+			klog.V(3).InfoS("PVC volume of the pod requires file system resize", "uniquePodName", podName, "volumeName", volumeName, "outerVolumeSpecName", podObj.outerVolumeSpecName)
 			podObj.fsResizeRequired = true
 		}
 		asw.attachedVolumes[volumeName].mountedPods[podName] = podObj