Check for volume-subpaths directory in orpahaned pod cleanup

pkg/kubelet/BUILD

@@ -165,6 +165,7 @@ go_test(
         "kubelet_pods_windows_test.go",
         "kubelet_resources_test.go",
         "kubelet_test.go",
+        "kubelet_volumes_linux_test.go",
         "kubelet_volumes_test.go",
         "main_test.go",
         "oom_watcher_test.go",

pkg/kubelet/config/defaults.go

@@ -19,6 +19,7 @@ package config
 const (
 	DefaultKubeletPodsDirName                = "pods"
 	DefaultKubeletVolumesDirName             = "volumes"
+	DefaultKubeletVolumeSubpathsDirName      = "volume-subpaths"
 	DefaultKubeletVolumeDevicesDirName       = "volumeDevices"
 	DefaultKubeletPluginsDirName             = "plugins"
 	DefaultKubeletPluginsRegistrationDirName = "plugins_registry"

pkg/kubelet/kubelet_getters.go

@@ -25,7 +25,7 @@ import (
 	cadvisorapiv1 "github.com/google/cadvisor/info/v1"
 	"k8s.io/klog"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	"k8s.io/kubernetes/pkg/kubelet/config"
@@ -99,6 +99,13 @@ func (kl *Kubelet) getPodDir(podUID types.UID) string {
 	return filepath.Join(kl.getPodsDir(), string(podUID))
 }
 
+// getPodVolumesSubpathsDir returns the full path to the per-pod subpaths directory under
+// which subpath volumes are created for the specified pod.  This directory may not
+// exist if the pod does not exist or subpaths are not specified.
+func (kl *Kubelet) getPodVolumeSubpathsDir(podUID types.UID) string {
+	return filepath.Join(kl.getPodDir(podUID), config.DefaultKubeletVolumeSubpathsDirName)
+}
+
 // getPodVolumesDir returns the full path to the per-pod data directory under
 // which volumes are created for the specified pod.  This directory may not
 // exist if the pod does not exist.
@@ -315,6 +322,19 @@ func (kl *Kubelet) getMountedVolumePathListFromDisk(podUID types.UID) ([]string,
 	return mountedVolumes, nil
 }
 
+// podVolumesSubpathsDirExists returns true if the pod volume-subpaths directory for
+// a given pod exists
+func (kl *Kubelet) podVolumeSubpathsDirExists(podUID types.UID) (bool, error) {
+	podVolDir := kl.getPodVolumeSubpathsDir(podUID)
+
+	if pathExists, pathErr := volumeutil.PathExists(podVolDir); pathErr != nil {
+		return true, fmt.Errorf("Error checking if path %q exists: %v", podVolDir, pathErr)
+	} else if !pathExists {
+		return false, nil
+	}
+	return true, nil
+}
+
 // GetVersionInfo returns information about the version of cAdvisor in use.
 func (kl *Kubelet) GetVersionInfo() (*cadvisorapiv1.VersionInfo, error) {
 	return kl.cadvisor.VersionInfo()

pkg/kubelet/kubelet_volumes.go

@@ -19,7 +19,7 @@ package kubelet
 import (
 	"fmt"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -114,6 +114,8 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecon
 		}
 		// If volumes have not been unmounted/detached, do not delete directory.
 		// Doing so may result in corruption of data.
+		// 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)
 			continue
@@ -128,6 +130,18 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*v1.Pod, runningPods []*kubecon
 			orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("Orphaned pod %q found, but volume paths are still present on disk", uid))
 			continue
 		}
+
+		// If there are any volume-subpaths, do not cleanup directories
+		volumeSubpathExists, err := kl.podVolumeSubpathsDirExists(uid)
+		if err != nil {
+			orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("Orphaned pod %q found, but error %v occurred during reading of volume-subpaths dir from disk", uid, err))
+			continue
+		}
+		if volumeSubpathExists {
+			orphanVolumeErrors = append(orphanVolumeErrors, fmt.Errorf("Orphaned pod %q found, but volume subpaths are still present on disk", uid))
+			continue
+		}
+
 		klog.V(3).Infof("Orphaned pod %q found, removing", 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)

pkg/kubelet/kubelet_volumes_linux_test.go

+// +build linux
+
+/*
+Copyright 2018 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 kubelet
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	_ "k8s.io/kubernetes/pkg/apis/core/install"
+)
+
+func validateDirExists(dir string) error {
+	_, err := ioutil.ReadDir(dir)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func validateDirNotExists(dir string) error {
+	_, err := ioutil.ReadDir(dir)
+	if os.IsNotExist(err) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	return fmt.Errorf("dir %q still exists", dir)
+}
+
+func TestCleanupOrphanedPodDirs(t *testing.T) {
+	testCases := map[string]struct {
+		pods         []*v1.Pod
+		prepareFunc  func(kubelet *Kubelet) error
+		validateFunc func(kubelet *Kubelet) error
+		expectErr    bool
+	}{
+		"nothing-to-do": {},
+		"pods-dir-not-found": {
+			prepareFunc: func(kubelet *Kubelet) error {
+				return os.Remove(kubelet.getPodsDir())
+			},
+			expectErr: true,
+		},
+		"pod-doesnot-exist-novolume": {
+			prepareFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return os.MkdirAll(filepath.Join(podDir, "not/a/volume"), 0750)
+			},
+			validateFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return validateDirNotExists(filepath.Join(podDir, "not"))
+			},
+		},
+		"pod-exists-with-volume": {
+			pods: []*v1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "pod1",
+						UID:  "pod1uid",
+					},
+				},
+			},
+			prepareFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return os.MkdirAll(filepath.Join(podDir, "volumes/plugin/name"), 0750)
+			},
+			validateFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return validateDirExists(filepath.Join(podDir, "volumes/plugin/name"))
+			},
+		},
+		"pod-doesnot-exist-with-volume": {
+			prepareFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return os.MkdirAll(filepath.Join(podDir, "volumes/plugin/name"), 0750)
+			},
+			validateFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return validateDirExists(filepath.Join(podDir, "volumes/plugin/name"))
+			},
+		},
+		"pod-doesnot-exist-with-subpath": {
+			prepareFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return os.MkdirAll(filepath.Join(podDir, "volume-subpaths/volume/container/index"), 0750)
+			},
+			validateFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return validateDirExists(filepath.Join(podDir, "volume-subpaths/volume/container/index"))
+			},
+		},
+		"pod-doesnot-exist-with-subpath-top": {
+			prepareFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return os.MkdirAll(filepath.Join(podDir, "volume-subpaths"), 0750)
+			},
+			validateFunc: func(kubelet *Kubelet) error {
+				podDir := kubelet.getPodDir("pod1uid")
+				return validateDirExists(filepath.Join(podDir, "volume-subpaths"))
+			},
+		},
+		// TODO: test volume in volume-manager
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			testKubelet := newTestKubelet(t, false /* controllerAttachDetachEnabled */)
+			defer testKubelet.Cleanup()
+			kubelet := testKubelet.kubelet
+
+			if tc.prepareFunc != nil {
+				if err := tc.prepareFunc(kubelet); err != nil {
+					t.Fatalf("%s failed preparation: %v", name, err)
+				}
+			}
+
+			err := kubelet.cleanupOrphanedPodDirs(tc.pods, nil)
+			if tc.expectErr && err == nil {
+				t.Errorf("%s failed: expected error, got success", name)
+			}
+			if !tc.expectErr && err != nil {
+				t.Errorf("%s failed: got error %v", name, err)
+			}
+
+			if tc.validateFunc != nil {
+				if err := tc.validateFunc(kubelet); err != nil {
+					t.Errorf("%s failed validation: %v", name, err)
+				}
+			}
+
+		})
+	}
+}