attachdetach controller: attach volumes immediately when Pod's PVCs are bound

- Add integration test for this feature

test/integration/volume/BUILD

@@ -20,6 +20,7 @@ go_test(
         "//pkg/controller/volume/attachdetach:go_default_library",
         "//pkg/controller/volume/attachdetach/cache:go_default_library",
         "//pkg/controller/volume/persistentvolume:go_default_library",
+        "//pkg/controller/volume/persistentvolume/options:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/testing:go_default_library",
         "//pkg/volume/util:go_default_library",

test/integration/volume/attach_detach_test.go

@@ -17,11 +17,13 @@ limitations under the License.
 package volume
 
 import (
+	"fmt"
 	"net/http/httptest"
 	"testing"
 	"time"
 
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -32,6 +34,8 @@ import (
 	fakecloud "k8s.io/kubernetes/pkg/cloudprovider/providers/fake"
 	"k8s.io/kubernetes/pkg/controller/volume/attachdetach"
 	volumecache "k8s.io/kubernetes/pkg/controller/volume/attachdetach/cache"
+	"k8s.io/kubernetes/pkg/controller/volume/persistentvolume"
+	persistentvolumeoptions "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/options"
 	"k8s.io/kubernetes/pkg/volume"
 	volumetest "k8s.io/kubernetes/pkg/volume/testing"
 	"k8s.io/kubernetes/pkg/volume/util"
@@ -73,8 +77,68 @@ func fakePodWithVol(namespace string) *v1.Pod {
 	return fakePod
 }
 
+func fakePodWithPVC(name, pvcName, namespace string) (*v1.Pod, *v1.PersistentVolumeClaim) {
+	fakePod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				{
+					Name:  "fake-container",
+					Image: "nginx",
+					VolumeMounts: []v1.VolumeMount{
+						{
+							Name:      "fake-mount",
+							MountPath: "/var/www/html",
+						},
+					},
+				},
+			},
+			Volumes: []v1.Volume{
+				{
+					Name: "fake-mount",
+					VolumeSource: v1.VolumeSource{
+						PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+							ClaimName: pvcName,
+						},
+					},
+				},
+			},
+			NodeName: "node-sandbox",
+		},
+	}
+	class := "fake-sc"
+	fakePVC := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      pvcName,
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{
+				v1.ReadWriteOnce,
+			},
+			Resources: v1.ResourceRequirements{
+				Requests: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): resource.MustParse("5Gi"),
+				},
+			},
+			StorageClassName: &class,
+		},
+	}
+	return fakePod, fakePVC
+}
+
 type podCountFunc func(int) bool
 
+var defaultTimerConfig = attachdetach.TimerConfig{
+	ReconcilerLoopPeriod:                              100 * time.Millisecond,
+	ReconcilerMaxWaitForUnmountDuration:               6 * time.Second,
+	DesiredStateOfWorldPopulatorLoopSleepPeriod:       1 * time.Second,
+	DesiredStateOfWorldPopulatorListPodsRetryDuration: 3 * time.Second,
+}
+
 // Via integration test we can verify that if pod delete
 // event is somehow missed by AttachDetach controller - it still
 // gets cleaned up by Desired State of World populator.
@@ -94,7 +158,7 @@ func TestPodDeletionWithDswp(t *testing.T) {
 	ns := framework.CreateTestingNamespace(namespaceName, server, t)
 	defer framework.DeleteTestingNamespace(ns, server, t)
 
-	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+	testClient, ctrl, _, informers := createAdClients(ns, t, server, defaultSyncPeriod, defaultTimerConfig)
 	pod := fakePodWithVol(namespaceName)
 	podStopCh := make(chan struct{})
 
@@ -160,7 +224,7 @@ func TestPodUpdateWithWithADC(t *testing.T) {
 	ns := framework.CreateTestingNamespace(namespaceName, server, t)
 	defer framework.DeleteTestingNamespace(ns, server, t)
 
-	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+	testClient, ctrl, _, informers := createAdClients(ns, t, server, defaultSyncPeriod, defaultTimerConfig)
 
 	pod := fakePodWithVol(namespaceName)
 	podStopCh := make(chan struct{})
@@ -228,7 +292,7 @@ func TestPodUpdateWithKeepTerminatedPodVolumes(t *testing.T) {
 	ns := framework.CreateTestingNamespace(namespaceName, server, t)
 	defer framework.DeleteTestingNamespace(ns, server, t)
 
-	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+	testClient, ctrl, _, informers := createAdClients(ns, t, server, defaultSyncPeriod, defaultTimerConfig)
 
 	pod := fakePodWithVol(namespaceName)
 	podStopCh := make(chan struct{})
@@ -320,7 +384,7 @@ func waitForPodFuncInDSWP(t *testing.T, dswp volumecache.DesiredStateOfWorld, ch
 	}
 }
 
-func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, syncPeriod time.Duration) (*clientset.Clientset, attachdetach.AttachDetachController, informers.SharedInformerFactory) {
+func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, syncPeriod time.Duration, timers attachdetach.TimerConfig) (*clientset.Clientset, attachdetach.AttachDetachController, *persistentvolume.PersistentVolumeController, informers.SharedInformerFactory) {
 	config := restclient.Config{
 		Host:          server.URL,
 		ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}},
@@ -346,12 +410,6 @@ func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, sy
 	plugins := []volume.VolumePlugin{plugin}
 	cloud := &fakecloud.FakeCloud{}
 	informers := informers.NewSharedInformerFactory(testClient, resyncPeriod)
-	timers := attachdetach.TimerConfig{
-		ReconcilerLoopPeriod:                              100 * time.Millisecond,
-		ReconcilerMaxWaitForUnmountDuration:               6 * time.Second,
-		DesiredStateOfWorldPopulatorLoopSleepPeriod:       1 * time.Second,
-		DesiredStateOfWorldPopulatorListPodsRetryDuration: 3 * time.Second,
-	}
 	ctrl, err := attachdetach.NewAttachDetachController(
 		testClient,
 		informers.Core().V1().Pods(),
@@ -368,7 +426,27 @@ func createAdClients(ns *v1.Namespace, t *testing.T, server *httptest.Server, sy
 	if err != nil {
 		t.Fatalf("Error creating AttachDetach : %v", err)
 	}
-	return testClient, ctrl, informers
+
+	// create pv controller
+	controllerOptions := persistentvolumeoptions.NewPersistentVolumeControllerOptions()
+	params := persistentvolume.ControllerParameters{
+		KubeClient:                testClient,
+		SyncPeriod:                controllerOptions.PVClaimBinderSyncPeriod,
+		VolumePlugins:             plugins,
+		Cloud:                     nil,
+		ClusterName:               "volume-test-cluster",
+		VolumeInformer:            informers.Core().V1().PersistentVolumes(),
+		ClaimInformer:             informers.Core().V1().PersistentVolumeClaims(),
+		ClassInformer:             informers.Storage().V1().StorageClasses(),
+		PodInformer:               informers.Core().V1().Pods(),
+		NodeInformer:              informers.Core().V1().Nodes(),
+		EnableDynamicProvisioning: false,
+	}
+	pvCtrl, err := persistentvolume.NewController(params)
+	if err != nil {
+		t.Fatalf("Failed to create PV controller: %v", err)
+	}
+	return testClient, ctrl, pvCtrl, informers
 }
 
 // Via integration test we can verify that if pod add
@@ -391,7 +469,7 @@ func TestPodAddedByDswp(t *testing.T) {
 	ns := framework.CreateTestingNamespace(namespaceName, server, t)
 	defer framework.DeleteTestingNamespace(ns, server, t)
 
-	testClient, ctrl, informers := createAdClients(ns, t, server, defaultSyncPeriod)
+	testClient, ctrl, _, informers := createAdClients(ns, t, server, defaultSyncPeriod, defaultTimerConfig)
 
 	pod := fakePodWithVol(namespaceName)
 	podStopCh := make(chan struct{})
@@ -446,3 +524,91 @@ func TestPodAddedByDswp(t *testing.T) {
 
 	close(stopCh)
 }
+
+func TestPVCBoundWithADC(t *testing.T) {
+	_, server, closeFn := framework.RunAMaster(framework.NewIntegrationTestMasterConfig())
+	defer closeFn()
+	namespaceName := "test-pod-deletion"
+
+	ns := framework.CreateTestingNamespace(namespaceName, server, t)
+	defer framework.DeleteTestingNamespace(ns, server, t)
+
+	testClient, ctrl, pvCtrl, informers := createAdClients(ns, t, server, defaultSyncPeriod, attachdetach.TimerConfig{
+		ReconcilerLoopPeriod:                        100 * time.Millisecond,
+		ReconcilerMaxWaitForUnmountDuration:         6 * time.Second,
+		DesiredStateOfWorldPopulatorLoopSleepPeriod: 24 * time.Hour,
+		// Use high duration to disable DesiredStateOfWorldPopulator.findAndAddActivePods loop in test.
+		DesiredStateOfWorldPopulatorListPodsRetryDuration: 24 * time.Hour,
+	})
+
+	node := &v1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "node-sandbox",
+			Annotations: map[string]string{
+				util.ControllerManagedAttachAnnotation: "true",
+			},
+		},
+	}
+	if _, err := testClient.Core().Nodes().Create(node); err != nil {
+		t.Fatalf("Failed to created node : %v", err)
+	}
+
+	// pods with pvc not bound
+	pvcs := []*v1.PersistentVolumeClaim{}
+	for i := 0; i < 3; i++ {
+		pod, pvc := fakePodWithPVC(fmt.Sprintf("fakepod-pvcnotbound-%d", i), fmt.Sprintf("fakepvc-%d", i), namespaceName)
+		if _, err := testClient.Core().Pods(pod.Namespace).Create(pod); err != nil {
+			t.Errorf("Failed to create pod : %v", err)
+		}
+		if _, err := testClient.Core().PersistentVolumeClaims(pvc.Namespace).Create(pvc); err != nil {
+			t.Errorf("Failed to create pvc : %v", err)
+		}
+		pvcs = append(pvcs, pvc)
+	}
+	// pod with no pvc
+	podNew := fakePodWithVol(namespaceName)
+	podNew.SetName("fakepod")
+	if _, err := testClient.Core().Pods(podNew.Namespace).Create(podNew); err != nil {
+		t.Errorf("Failed to create pod : %v", err)
+	}
+
+	// start controller loop
+	stopCh := make(chan struct{})
+	informers.Start(stopCh)
+	informers.WaitForCacheSync(stopCh)
+	go ctrl.Run(stopCh)
+	go pvCtrl.Run(stopCh)
+
+	waitToObservePods(t, informers.Core().V1().Pods().Informer(), 4)
+	// Give attachdetach controller enough time to populate pods into DSWP.
+	time.Sleep(10 * time.Second)
+	waitForPodFuncInDSWP(t, ctrl.GetDesiredStateOfWorld(), 60*time.Second, "expected 1 pod in dsw", 1)
+	for _, pvc := range pvcs {
+		createPVForPVC(t, testClient, pvc)
+	}
+	waitForPodFuncInDSWP(t, ctrl.GetDesiredStateOfWorld(), 60*time.Second, "expected 4 pods in dsw after PVCs are bound", 4)
+	close(stopCh)
+}
+
+// Create PV for PVC, pv controller will bind them together.
+func createPVForPVC(t *testing.T, testClient *clientset.Clientset, pvc *v1.PersistentVolumeClaim) {
+	pv := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("fakepv-%s", pvc.Name),
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity:    pvc.Spec.Resources.Requests,
+			AccessModes: pvc.Spec.AccessModes,
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				HostPath: &v1.HostPathVolumeSource{
+					Path: "/var/www/html",
+				},
+			},
+			ClaimRef:         &v1.ObjectReference{Name: pvc.Name, Namespace: pvc.Namespace},
+			StorageClassName: *pvc.Spec.StorageClassName,
+		},
+	}
+	if _, err := testClient.Core().PersistentVolumes().Create(pv); err != nil {
+		t.Errorf("Failed to create pv : %v", err)
+	}
+}