Merge pull request #116706 from pacoxu/deflake-kubemark-data-race

kubelet: fix data races

cmd/kubelet/app/options/options.go

@@ -20,6 +20,7 @@ package options
 import (
 	"fmt"
 	_ "net/http/pprof" // Enable pprof HTTP handlers.
+	"path/filepath"
 	"strings"
 
 	"github.com/spf13/pflag"
@@ -138,7 +139,7 @@ func NewKubeletFlags() *KubeletFlags {
 	return &KubeletFlags{
 		ContainerRuntimeOptions: *NewContainerRuntimeOptions(),
 		CertDirectory:           "/var/lib/kubelet/pki",
-		RootDirectory:           defaultRootDir,
+		RootDirectory:           filepath.Clean(defaultRootDir),
 		MaxContainerCount:       -1,
 		MaxPerPodContainerCount: 1,
 		MinimumGCAge:            metav1.Duration{Duration: 0},

pkg/kubelet/kubelet.go

@@ -523,7 +523,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		kubeClient:                              kubeDeps.KubeClient,
 		heartbeatClient:                         kubeDeps.HeartbeatClient,
 		onRepeatedHeartbeatFailure:              kubeDeps.OnHeartbeatFailure,
-		rootDirectory:                           rootDirectory,
+		rootDirectory:                           filepath.Clean(rootDirectory),
 		resyncInterval:                          kubeCfg.SyncFrequency.Duration,
 		sourcesReady:                            config.NewSourcesReady(kubeDeps.PodConfig.SeenAllSources),
 		registerNode:                            registerNode,
@@ -1321,7 +1321,9 @@ func (kl *Kubelet) RlimitStats() (*statsapi.RlimitStats, error) {
 // 4.  the pod-resources directory
 // 5.  the checkpoint directory
 func (kl *Kubelet) setupDataDirs() error {
-	kl.rootDirectory = filepath.Clean(kl.rootDirectory)
+	if cleanedRoot := filepath.Clean(kl.rootDirectory); cleanedRoot != kl.rootDirectory {
+		return fmt.Errorf("rootDirectory not in canonical form: expected %s, was %s", cleanedRoot, kl.rootDirectory)
+	}
 	pluginRegistrationDir := kl.getPluginsRegistrationDir()
 	pluginsDir := kl.getPluginsDir()
 	if err := os.MkdirAll(kl.getRootDir(), 0750); err != nil {

pkg/kubelet/runonce_test.go

@@ -19,6 +19,7 @@ package kubelet
 import (
 	"context"
 	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -81,7 +82,7 @@ func TestRunOnce(t *testing.T) {
 	}
 	defer os.RemoveAll(basePath)
 	kb := &Kubelet{
-		rootDirectory:    basePath,
+		rootDirectory:    filepath.Clean(basePath),
 		recorder:         &record.FakeRecorder{},
 		cadvisor:         cadvisor,
 		nodeLister:       testNodeLister{},

pkg/kubelet/volumemanager/reconciler/reconciler_common.go

@@ -18,6 +18,7 @@ package reconciler
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -139,6 +140,8 @@ type reconciler struct {
 	volumePluginMgr               *volumepkg.VolumePluginMgr
 	skippedDuringReconstruction   map[v1.UniqueVolumeName]*globalVolumeInfo
 	kubeletPodsDir                string
+	// lock protects timeOfLastSync for updating and checking
+	timeOfLastSyncLock          sync.Mutex
 	timeOfLastSync              time.Time
 	volumesFailedReconstruction []podVolume
 	volumesNeedDevicePath       []v1.UniqueVolumeName

pkg/kubelet/volumemanager/reconciler/reconstruct_common.go

@@ -72,10 +72,14 @@ type globalVolumeInfo struct {
 }
 
 func (rc *reconciler) updateLastSyncTime() {
+	rc.timeOfLastSyncLock.Lock()
+	defer rc.timeOfLastSyncLock.Unlock()
 	rc.timeOfLastSync = time.Now()
 }
 
 func (rc *reconciler) StatesHasBeenSynced() bool {
+	rc.timeOfLastSyncLock.Lock()
+	defer rc.timeOfLastSyncLock.Unlock()
 	return !rc.timeOfLastSync.IsZero()
 }