Merge pull request kubernetes#61019 from ianchakeres/e2e-lv-prov-bind-disc

Automatic merge from submit-queue (batch tested with PRs 61644, 61624, 61743, 61019, 61287). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Added e2e test for local-volume provisioner that does not create PV for discovered non-bind-mounted filesystem.

**What this PR does / why we need it**:

For v2+ of the [local volume provisioner](https://github.com/kubernetes-incubator/external-storage/tree/master/local-volume) non-bind mounted filesystems in a discovery directory will no longer result in local PVs. This change was put in place to handle the non-atomic nature of other methods of adding directories (e.g. kubernetes-retired/external-storage#482).

This PR tests this change in behavior, and it validates that non-bind mounted directories within a discovery directory do NOT result in a local PV.

**Which issue(s) this PR fixes**:
Fixes kubernetes#61020

**Special notes for your reviewer**:

This test can be executed using the following commands:
```
KUBE_FEATURE_GATES="BlockVolume=true" NUM_NODES=1 go run hack/e2e.go -- --up

go run hack/e2e.go -- --test --test_args='--ginkgo.focus=PersistentVolumes-local.*Local.*volume.*provisioner'
```

If you get the logs from a local volume provisioner pod, you will see the following log messages:
```
$ kubectl logs local-volume-provisioner-94ddb -n e2e-tests-persistent-local-volumes-test-6ls4z
<snip>
I0311 19:01:30.350504       1 controller.go:73] Controller started
E0311 19:01:30.350849       1 discovery.go:172] Path "/mnt/local-storage/notbindmount" is not an actual mountpoint
```


**Release note**:
```release-note
NONE
```

Author: Kubernetes Submit Queue

test/e2e/storage/persistent_volumes-local.go

@@ -159,6 +159,8 @@ var _ = utils.SIGDescribe("PersistentVolumes-local ", func() {
 	)
 
 	BeforeEach(func() {
+		framework.SkipUnlessProviderIs(framework.ProvidersWithSSH...)
+
 		// Get all the schedulable nodes
 		nodes := framework.GetReadySchedulableNodesOrDie(f.ClientSet)
 		Expect(len(nodes.Items)).NotTo(BeZero(), "No available nodes for scheduling")
@@ -403,6 +405,32 @@ var _ = utils.SIGDescribe("PersistentVolumes-local ", func() {
 			By("Deleting provisioner daemonset")
 			deleteProvisionerDaemonset(config)
 		})
+		It("should not create local persistent volume for filesystem volume that was not bind mounted", func() {
+
+			directoryPath := filepath.Join(config.discoveryDir, "notbindmount")
+			By("Creating a directory, not bind mounted, in discovery directory")
+			mkdirCmd := fmt.Sprintf("mkdir -p %v -m 777", directoryPath)
+			err := framework.IssueSSHCommand(mkdirCmd, framework.TestContext.Provider, config.node0)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Starting a provisioner daemonset")
+			createProvisionerDaemonset(config)
+
+			By("Allowing provisioner to run for 30s and discover potential local PVs")
+			time.Sleep(30 * time.Second)
+
+			By("Examining provisioner logs for not an actual mountpoint message")
+			provisionerPodName := findProvisionerDaemonsetPodName(config)
+			logs, err := framework.GetPodLogs(config.client, config.ns, provisionerPodName, "" /*containerName*/)
+			Expect(err).NotTo(HaveOccurred(),
+				"Error getting logs from pod %s in namespace %s", provisionerPodName, config.ns)
+
+			expectedLogMessage := "Path \"/mnt/local-storage/notbindmount\" is not an actual mountpoint"
+			Expect(strings.Contains(logs, expectedLogMessage)).To(BeTrue())
+
+			By("Deleting provisioner daemonset")
+			deleteProvisionerDaemonset(config)
+		})
 		It("should discover dynamicly created local persistent volume mountpoint in discovery directory", func() {
 			By("Starting a provisioner daemonset")
 			createProvisionerDaemonset(config)
@@ -1397,6 +1425,22 @@ func createProvisionerDaemonset(config *localTestConfig) {
 	framework.WaitForControlledPodsRunning(config.client, config.ns, daemonSetName, kind)
 }
 
+func findProvisionerDaemonsetPodName(config *localTestConfig) string {
+	podList, err := config.client.CoreV1().Pods(config.ns).List(metav1.ListOptions{})
+	if err != nil {
+		framework.Failf("could not get the pod list: %v", err)
+		return ""
+	}
+	pods := podList.Items
+	for _, pod := range pods {
+		if strings.HasPrefix(pod.Name, daemonSetName) && pod.Spec.NodeName == config.node0.Name {
+			return pod.Name
+		}
+	}
+	framework.Failf("Unable to find provisioner daemonset pod on node0")
+	return ""
+}
+
 func deleteProvisionerDaemonset(config *localTestConfig) {
 	ds, err := config.client.ExtensionsV1beta1().DaemonSets(config.ns).Get(daemonSetName, metav1.GetOptions{})
 	if ds == nil {