Merge pull request #33616 from jingxu97/statusupdater-9-22

pkg/kubelet/kubelet_getters.go

@@ -239,9 +239,9 @@ func (kl *Kubelet) GetExtraSupplementalGroupsForPod(pod *api.Pod) []int64 {
 	return kl.volumeManager.GetExtraSupplementalGroupsForPod(pod)
 }
 
-// getPodVolumeNameListFromDisk returns a list of the volume names by reading the
+// getPodVolumePathListFromDisk returns a list of the volume paths by reading the
 // volume directories for the given pod from the disk.
-func (kl *Kubelet) getPodVolumeNameListFromDisk(podUID types.UID) ([]string, error) {
+func (kl *Kubelet) getPodVolumePathListFromDisk(podUID types.UID) ([]string, error) {
 	volumes := []string{}
 	podVolDir := kl.getPodVolumesDir(podUID)
 	volumePluginDirs, err := ioutil.ReadDir(podVolDir)
@@ -254,9 +254,11 @@ func (kl *Kubelet) getPodVolumeNameListFromDisk(podUID types.UID) ([]string, err
 		volumePluginPath := path.Join(podVolDir, volumePluginName)
 		volumeDirs, err := util.ReadDirNoStat(volumePluginPath)
 		if err != nil {
-			return volumes, err
+			return volumes, fmt.Errorf("Could not read directory %s: %v", volumePluginPath, err)
+		}
+		for _, volumeDir := range volumeDirs {
+			volumes = append(volumes, path.Join(volumePluginPath, volumeDir))
 		}
-		volumes = append(volumes, volumeDirs...)
 	}
 	return volumes, nil
 }

pkg/kubelet/kubelet_node_status.go

@@ -342,6 +342,8 @@ func (kl *Kubelet) tryUpdateNodeStatus() error {
 	}
 	// Update the current status on the API server
 	updatedNode, err := kl.kubeClient.Core().Nodes().UpdateStatus(node)
+	// If update finishes sucessfully, mark the volumeInUse as reportedInUse to indicate
+	// those volumes are already updated in the node's status
 	if err == nil {
 		kl.volumeManager.MarkVolumesAsReportedInUse(
 			updatedNode.Status.VolumesInUse)
@@ -882,9 +884,13 @@ func (kl *Kubelet) recordNodeSchedulableEvent(node *api.Node) {
 	}
 }
 
-// Update VolumesInUse field in Node Status
+// Update VolumesInUse field in Node Status only after states are synced up at least once
+// in volume reconciler.
 func (kl *Kubelet) setNodeVolumesInUseStatus(node *api.Node) {
-	node.Status.VolumesInUse = kl.volumeManager.GetVolumesInUse()
+	// Make sure to only update node status after reconciler starts syncing up states
+	if kl.volumeManager.ReconcilerStatesHasBeenSynced() {
+		node.Status.VolumesInUse = kl.volumeManager.GetVolumesInUse()
+	}
 }
 
 // setNodeStatus fills in the Status fields of the given Node, overwriting

pkg/kubelet/kubelet_volumes.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/securitycontext"
 	"k8s.io/kubernetes/pkg/types"
 	utilerrors "k8s.io/kubernetes/pkg/util/errors"
+	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/util/selinux"
 	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/volume"
@@ -153,8 +154,20 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(
 			continue
 		}
 		// Check whether volume is still mounted on disk. If so, do not delete directory
-		if volumeNames, err := kl.getPodVolumeNameListFromDisk(uid); err != nil || len(volumeNames) != 0 {
-			glog.V(3).Infof("Orphaned pod %q found, but volumes are still mounted; err: %v, volumes: %v ", uid, err, volumeNames)
+		volumePaths, err := kl.getPodVolumePathListFromDisk(uid)
+		if err != nil {
+			glog.Errorf("Orphaned pod %q found, but error %v occured during reading volume dir from disk", uid, err)
+			continue
+		} else if len(volumePaths) > 0 {
+			for _, path := range volumePaths {
+				notMount, err := mount.IsNotMountPoint(path)
+				if err == nil && notMount {
+					glog.V(2).Infof("Volume path %q is no longer mounted, remove it", path)
+					os.Remove(path)
+				} else {
+					glog.Errorf("Orphaned pod %q found, but it might still mounted with error %v", uid, err)
+				}
+			}
 			continue
 		}
 

pkg/kubelet/volumemanager/cache/actual_state_of_world.go

@@ -186,6 +186,7 @@ func IsRemountRequiredError(err error) bool {
 type actualStateOfWorld struct {
 	// nodeName is the name of this node. This value is passed to Attach/Detach
 	nodeName types.NodeName
+
 	// attachedVolumes is a map containing the set of volumes the kubelet volume
 	// manager believes to be successfully attached to this node. Volume types
 	// that do not implement an attacher interface are assumed to be in this
@@ -193,6 +194,7 @@ type actualStateOfWorld struct {
 	// The key in this map is the name of the volume and the value is an object
 	// containing more information about the attached volume.
 	attachedVolumes map[api.UniqueVolumeName]attachedVolume
+
 	// volumePluginMgr is the volume plugin manager used to create volume
 	// plugin objects.
 	volumePluginMgr *volume.VolumePluginMgr

pkg/kubelet/volumemanager/cache/desired_state_of_world.go

@@ -58,7 +58,8 @@ type DesiredStateOfWorld interface {
 	// ReportedInUse value is reset to false. The default ReportedInUse value
 	// for a newly created volume is false.
 	// When set to true this value indicates that the volume was successfully
-	// added to the VolumesInUse field in the node's status.
+	// added to the VolumesInUse field in the node's status. Mount operation needs
+	// to check this value before issuing the operation.
 	// If a volume in the reportedVolumes list does not exist in the list of
 	// volumes that should be attached to this node, it is skipped without error.
 	MarkVolumesReportedInUse(reportedVolumes []api.UniqueVolumeName)

pkg/kubelet/volumemanager/reconciler/reconciler_test.go

@@ -42,8 +42,8 @@ import (
 const (
 	// reconcilerLoopSleepDuration is the amount of time the reconciler loop
 	// waits between successive executions
-	reconcilerLoopSleepDuration      time.Duration = 0 * time.Millisecond
-	reconcilerReconstructSleepPeriod time.Duration = 10 * time.Minute
+	reconcilerLoopSleepDuration     time.Duration = 0 * time.Millisecond
+	reconcilerSyncStatesSleepPeriod time.Duration = 10 * time.Minute
 	// waitForAttachTimeout is the maximum amount of time a
 	// operationexecutor.Mount call will wait for a volume to be attached.
 	waitForAttachTimeout time.Duration     = 1 * time.Second
@@ -65,7 +65,7 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		dsw,
@@ -102,7 +102,7 @@ func Test_Run_Positive_VolumeAttachAndMount(t *testing.T) {
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		dsw,
@@ -173,7 +173,7 @@ func Test_Run_Positive_VolumeMountControllerAttachEnabled(t *testing.T) {
 		kubeClient,
 		true, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		dsw,
@@ -245,7 +245,7 @@ func Test_Run_Positive_VolumeAttachMountUnmountDetach(t *testing.T) {
 		kubeClient,
 		false, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		dsw,
@@ -328,7 +328,7 @@ func Test_Run_Positive_VolumeUnmountControllerAttachEnabled(t *testing.T) {
 		kubeClient,
 		true, /* controllerAttachDetachEnabled */
 		reconcilerLoopSleepDuration,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		dsw,

pkg/kubelet/volumemanager/volume_manager.go

@@ -49,9 +49,9 @@ const (
 	// between successive executions
 	reconcilerLoopSleepPeriod time.Duration = 100 * time.Millisecond
 
-	// reconcilerReconstructSleepPeriod is the amount of time the reconciler reconstruct process
+	// reconcilerSyncStatesSleepPeriod is the amount of time the reconciler reconstruct process
 	// waits between successive executions
-	reconcilerReconstructSleepPeriod time.Duration = 3 * time.Minute
+	reconcilerSyncStatesSleepPeriod time.Duration = 3 * time.Minute
 
 	// desiredStateOfWorldPopulatorLoopSleepPeriod is the amount of time the
 	// DesiredStateOfWorldPopulator loop waits between successive executions
@@ -115,7 +115,7 @@ type VolumeManager interface {
 	// from annotations on persistent volumes that the pod depends on.
 	GetExtraSupplementalGroupsForPod(pod *api.Pod) []int64
 
-	// Returns a list of all volumes that implement the volume.Attacher
+	// GetVolumesInUse returns a list of all volumes that implement the volume.Attacher
 	// interface and are currently in use according to the actual and desired
 	// state of the world caches. A volume is considered "in use" as soon as it
 	// is added to the desired state of world, indicating it *should* be
@@ -126,6 +126,11 @@ type VolumeManager interface {
 	// restarts.
 	GetVolumesInUse() []api.UniqueVolumeName
 
+	// ReconcilerStatesHasBeenSynced returns true only after the actual states in reconciler
+	// has been synced at least once after kubelet starts so that it is safe to update mounted
+	// volume list retrieved from actual state.
+	ReconcilerStatesHasBeenSynced() bool
+
 	// VolumeIsAttached returns true if the given volume is attached to this
 	// node.
 	VolumeIsAttached(volumeName api.UniqueVolumeName) bool
@@ -168,7 +173,7 @@ func NewVolumeManager(
 		kubeClient,
 		controllerAttachDetachEnabled,
 		reconcilerLoopSleepPeriod,
-		reconcilerReconstructSleepPeriod,
+		reconcilerSyncStatesSleepPeriod,
 		waitForAttachTimeout,
 		nodeName,
 		vm.desiredStateOfWorld,
@@ -305,6 +310,10 @@ func (vm *volumeManager) GetVolumesInUse() []api.UniqueVolumeName {
 	return volumesToReportInUse
 }
 
+func (vm *volumeManager) ReconcilerStatesHasBeenSynced() bool {
+	return vm.reconciler.StatesHasBeenSynced()
+}
+
 func (vm *volumeManager) VolumeIsAttached(
 	volumeName api.UniqueVolumeName) bool {
 	return vm.actualStateOfWorld.VolumeExists(volumeName)

pkg/util/mount/mount_linux.go

@@ -101,9 +101,10 @@ func isBind(options []string) (bool, []string) {
 
 // doMount runs the mount command.
 func doMount(mountCmd string, source string, target string, fstype string, options []string) error {
-	glog.V(5).Infof("Mounting %s %s %s %v with command: %q", source, target, fstype, options, mountCmd)
+	glog.V(4).Infof("Mounting %s %s %s %v with command: %q", source, target, fstype, options, mountCmd)
 	mountArgs := makeMountArgs(source, target, fstype, options)
 
+	glog.V(4).Infof("Mounting cmd (%s) with arguments (%s)", mountCmd, mountArgs)
 	command := exec.Command(mountCmd, mountArgs...)
 	output, err := command.CombinedOutput()
 	if err != nil {
@@ -135,7 +136,7 @@ func makeMountArgs(source, target, fstype string, options []string) []string {
 
 // Unmount unmounts the target.
 func (mounter *Mounter) Unmount(target string) error {
-	glog.V(5).Infof("Unmounting %s", target)
+	glog.V(4).Infof("Unmounting %s", target)
 	command := exec.Command("umount", target)
 	output, err := command.CombinedOutput()
 	if err != nil {
@@ -156,6 +157,10 @@ func (*Mounter) List() ([]MountPoint, error) {
 // will return true. When in fact /tmp/b is a mount point. If this situation
 // if of interest to you, don't use this function...
 func (mounter *Mounter) IsLikelyNotMountPoint(file string) (bool, error) {
+	return IsNotMountPoint(file)
+}
+
+func IsNotMountPoint(file string) (bool, error) {
 	stat, err := os.Stat(file)
 	if err != nil {
 		return true, err
@@ -173,9 +178,10 @@ func (mounter *Mounter) IsLikelyNotMountPoint(file string) (bool, error) {
 }
 
 // DeviceOpened checks if block device in use by calling Open with O_EXCL flag.
-// Returns true if open returns errno EBUSY, and false if errno is nil.
-// Returns an error if errno is any error other than EBUSY.
-// Returns with error if pathname is not a device.
+// If pathname is not a device, log and return false with nil error.
+// If open returns errno EBUSY, return true with nil error.
+// If open returns nil, return false with nil error.
+// Otherwise, return false with error
 func (mounter *Mounter) DeviceOpened(pathname string) (bool, error) {
 	return exclusiveOpenFailsOnDevice(pathname)
 }
@@ -187,12 +193,17 @@ func (mounter *Mounter) PathIsDevice(pathname string) (bool, error) {
 }
 
 func exclusiveOpenFailsOnDevice(pathname string) (bool, error) {
-	if isDevice, err := pathIsDevice(pathname); !isDevice {
+	isDevice, err := pathIsDevice(pathname)
+	if err != nil {
 		return false, fmt.Errorf(
 			"PathIsDevice failed for path %q: %v",
 			pathname,
 			err)
 	}
+	if !isDevice {
+		glog.Errorf("Path %q is not refering to a device.", pathname)
+		return false, nil
+	}
 	fd, errno := syscall.Open(pathname, syscall.O_RDONLY|syscall.O_EXCL, 0)
 	// If the device is in use, open will return an invalid fd.
 	// When this happens, it is expected that Close will fail and throw an error.

pkg/volume/gce_pd/gce_pd.go

@@ -247,7 +247,7 @@ func (b *gcePersistentDiskMounter) SetUp(fsGroup *int64) error {
 func (b *gcePersistentDiskMounter) SetUpAt(dir string, fsGroup *int64) error {
 	// TODO: handle failed mounts here.
 	notMnt, err := b.mounter.IsLikelyNotMountPoint(dir)
-	glog.V(4).Infof("PersistentDisk set up: %s %v %v, pd name %v readOnly %v", dir, !notMnt, err, b.pdName, b.readOnly)
+	glog.V(4).Infof("GCE PersistentDisk set up: Dir (%s) PD name (%q) Mounted (%t) Error (%v), ReadOnly (%t)", dir, b.pdName, !notMnt, err, b.readOnly)
 	if err != nil && !os.IsNotExist(err) {
 		glog.Errorf("cannot validate mount point: %s %v", dir, err)
 		return err

pkg/volume/util/operationexecutor/operation_executor.go

@@ -758,11 +758,12 @@ func (oe *operationExecutor) generateMountVolumeFunc(
 			}
 
 			glog.Infof(
-				"MountVolume.MountDevice succeeded for volume %q (spec.Name: %q) pod %q (UID: %q).",
+				"MountVolume.MountDevice succeeded for volume %q (spec.Name: %q) pod %q (UID: %q) device mount path %q",
 				volumeToMount.VolumeName,
 				volumeToMount.VolumeSpec.Name(),
 				volumeToMount.PodName,
-				volumeToMount.Pod.UID)
+				volumeToMount.Pod.UID,
+				deviceMountPath)
 
 			// Update actual state of world to reflect volume is globally mounted
 			markDeviceMountedErr := actualStateOfWorld.MarkDeviceAsMounted(