Merge pull request #28116 from jsafrane/integration-startup

Automatic merge from submit-queue

Add integration test for volume controller startup.

Tests #28002 with real etcd (unit tests have a fake one with different behavior).

@kubernetes/sig-storage

test/integration/persistentvolumes/persistent_volumes_test.go

@@ -75,8 +75,8 @@ func getObjectCount() int {
 	return objectCount
 }
 
-func getSyncPeriod() time.Duration {
+func getSyncPeriod(syncPeriod time.Duration) time.Duration {
-	period := defaultSyncPeriod
+	period := syncPeriod
 	if s := os.Getenv("KUBE_INTEGRATION_PV_SYNC_PERIOD"); s != "" {
 		var err error
 		period, err = time.ParseDuration(s)
@@ -112,7 +112,7 @@ func TestPersistentVolumeRecycler(t *testing.T) {
 	ns := framework.CreateTestingNamespace("pv-recycler", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -165,7 +165,7 @@ func TestPersistentVolumeDeleter(t *testing.T) {
 	ns := framework.CreateTestingNamespace("pv-deleter", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -223,7 +223,7 @@ func TestPersistentVolumeBindRace(t *testing.T) {
 	ns := framework.CreateTestingNamespace("pv-bind-race", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, ctrl, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -293,7 +293,7 @@ func TestPersistentVolumeClaimLabelSelector(t *testing.T) {
 	ns := framework.CreateTestingNamespace("pvc-label-selector", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, controller, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, controller, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -372,7 +372,7 @@ func TestPersistentVolumeClaimLabelSelectorMatchExpressions(t *testing.T) {
 	ns := framework.CreateTestingNamespace("pvc-match-expresssions", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, controller, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, controller, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -470,7 +470,7 @@ func TestPersistentVolumeMultiPVs(t *testing.T) {
 	ns := framework.CreateTestingNamespace("multi-pvs", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, controller, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, controller, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -558,7 +558,7 @@ func TestPersistentVolumeMultiPVsPVCs(t *testing.T) {
 	ns := framework.CreateTestingNamespace("multi-pvs-pvcs", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, binder, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, binder, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -708,6 +708,133 @@ func TestPersistentVolumeMultiPVsPVCs(t *testing.T) {
 	testSleep()
 }
 
+// TestPersistentVolumeControllerStartup tests startup of the controller.
+// The controller should not unbind any volumes when it starts.
+func TestPersistentVolumeControllerStartup(t *testing.T) {
+	_, s := framework.RunAMaster(nil)
+	defer s.Close()
+
+	ns := framework.CreateTestingNamespace("controller-startup", s, t)
+	defer framework.DeleteTestingNamespace(ns, s, t)
+
+	objCount := getObjectCount()
+
+	const shortSyncPeriod = 2 * time.Second
+	syncPeriod := getSyncPeriod(shortSyncPeriod)
+
+	testClient, binder, watchPV, watchPVC := createClients(ns, t, s, shortSyncPeriod)
+	defer watchPV.Stop()
+	defer watchPVC.Stop()
+
+	// Create *bound* volumes and PVCs
+	pvs := make([]*api.PersistentVolume, objCount)
+	pvcs := make([]*api.PersistentVolumeClaim, objCount)
+	for i := 0; i < objCount; i++ {
+		pvName := "pv-startup-" + strconv.Itoa(i)
+		pvcName := "pvc-startup-" + strconv.Itoa(i)
+
+		pvc := createPVC(pvcName, ns.Name, "1G", []api.PersistentVolumeAccessMode{api.ReadWriteOnce})
+		pvc.Annotations = map[string]string{"annBindCompleted": ""}
+		pvc.Spec.VolumeName = pvName
+		newPVC, err := testClient.PersistentVolumeClaims(ns.Name).Create(pvc)
+		if err != nil {
+			t.Fatalf("Cannot create claim %q: %v", pvc.Name, err)
+		}
+		// Save Bound status as a separate transaction
+		newPVC.Status.Phase = api.ClaimBound
+		newPVC, err = testClient.PersistentVolumeClaims(ns.Name).UpdateStatus(newPVC)
+		if err != nil {
+			t.Fatalf("Cannot update claim status %q: %v", pvc.Name, err)
+		}
+		pvcs[i] = newPVC
+		// Drain watchPVC with all events generated by the PVC until it's bound
+		// We don't want to catch "PVC craated with Status.Phase == Pending"
+		// later in this test.
+		waitForAnyPersistentVolumeClaimPhase(watchPVC, api.ClaimBound)
+
+		pv := createPV(pvName, "/tmp/foo"+strconv.Itoa(i), "1G",
+			[]api.PersistentVolumeAccessMode{api.ReadWriteOnce}, api.PersistentVolumeReclaimRetain)
+		claimRef, err := api.GetReference(newPVC)
+		if err != nil {
+			glog.V(3).Infof("unexpected error getting claim reference: %v", err)
+			return
+		}
+		pv.Spec.ClaimRef = claimRef
+		newPV, err := testClient.PersistentVolumes().Create(pv)
+		if err != nil {
+			t.Fatalf("Cannot create volume %q: %v", pv.Name, err)
+		}
+		// Save Bound status as a separate transaction
+		newPV.Status.Phase = api.VolumeBound
+		newPV, err = testClient.PersistentVolumes().UpdateStatus(newPV)
+		if err != nil {
+			t.Fatalf("Cannot update volume status %q: %v", pv.Name, err)
+		}
+		pvs[i] = newPV
+		// Drain watchPV with all events generated by the PV until it's bound
+		// We don't want to catch "PV craated with Status.Phase == Pending"
+		// later in this test.
+		waitForAnyPersistentVolumePhase(watchPV, api.VolumeBound)
+	}
+
+	// Start the controller when all PVs and PVCs are already saved in etcd
+	binder.Run()
+	defer binder.Stop()
+
+	// wait for at least two sync periods for changes. No volume should be
+	// Released and no claim should be Lost during this time.
+	timer := time.NewTimer(2 * syncPeriod)
+	defer timer.Stop()
+	finished := false
+	for !finished {
+		select {
+		case volumeEvent := <-watchPV.ResultChan():
+			volume, ok := volumeEvent.Object.(*api.PersistentVolume)
+			if !ok {
+				continue
+			}
+			if volume.Status.Phase != api.VolumeBound {
+				t.Errorf("volume %s unexpectedly changed state to %s", volume.Name, volume.Status.Phase)
+			}
+
+		case claimEvent := <-watchPVC.ResultChan():
+			claim, ok := claimEvent.Object.(*api.PersistentVolumeClaim)
+			if !ok {
+				continue
+			}
+			if claim.Status.Phase != api.ClaimBound {
+				t.Errorf("claim %s unexpectedly changed state to %s", claim.Name, claim.Status.Phase)
+			}
+
+		case <-timer.C:
+			// Wait finished
+			glog.V(2).Infof("Wait finished")
+			finished = true
+		}
+	}
+
+	// check that everything is bound to something
+	for i := 0; i < objCount; i++ {
+		pv, err := testClient.PersistentVolumes().Get(pvs[i].Name)
+		if err != nil {
+			t.Fatalf("Unexpected error getting pv: %v", err)
+		}
+		if pv.Spec.ClaimRef == nil {
+			t.Fatalf("PV %q is not bound", pv.Name)
+		}
+		glog.V(2).Infof("PV %q is bound to PVC %q", pv.Name, pv.Spec.ClaimRef.Name)
+
+		pvc, err := testClient.PersistentVolumeClaims(ns.Name).Get(pvcs[i].Name)
+		if err != nil {
+			t.Fatalf("Unexpected error getting pvc: %v", err)
+		}
+		if pvc.Spec.VolumeName == "" {
+			t.Fatalf("PVC %q is not bound", pvc.Name)
+		}
+		glog.V(2).Infof("PVC %q is bound to PV %q", pvc.Name, pvc.Spec.VolumeName)
+	}
+}
+
 // TestPersistentVolumeProvisionMultiPVCs tests provisioning of many PVCs.
 // This test is configurable by KUBE_INTEGRATION_PV_* variables.
 func TestPersistentVolumeProvisionMultiPVCs(t *testing.T) {
@@ -717,7 +844,7 @@ func TestPersistentVolumeProvisionMultiPVCs(t *testing.T) {
 	ns := framework.CreateTestingNamespace("provision-multi-pvs", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, binder, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, binder, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -801,7 +928,7 @@ func TestPersistentVolumeMultiPVsDiffAccessModes(t *testing.T) {
 	ns := framework.CreateTestingNamespace("multi-pvs-diff-access", s, t)
 	defer framework.DeleteTestingNamespace(ns, s, t)
 
-	testClient, controller, watchPV, watchPVC := createClients(ns, t, s)
+	testClient, controller, watchPV, watchPVC := createClients(ns, t, s, defaultSyncPeriod)
 	defer watchPV.Stop()
 	defer watchPVC.Stop()
 
@@ -941,7 +1068,7 @@ func waitForAnyPersistentVolumeClaimPhase(w watch.Interface, phase api.Persisten
 	}
 }
 
-func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server) (*clientset.Clientset, *persistentvolumecontroller.PersistentVolumeController, watch.Interface, watch.Interface) {
+func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server, syncPeriod time.Duration) (*clientset.Clientset, *persistentvolumecontroller.PersistentVolumeController, watch.Interface, watch.Interface) {
 	// Use higher QPS and Burst, there is a test for race conditions which
 	// creates many objects and default values were too low.
 	binderClient := clientset.NewForConfigOrDie(&restclient.Config{
@@ -973,7 +1100,7 @@ func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server) (*client
 	plugins := []volume.VolumePlugin{plugin}
 	cloud := &fake_cloud.FakeCloud{}
 
-	syncPeriod := getSyncPeriod()
+	syncPeriod = getSyncPeriod(syncPeriod)
 	ctrl := persistentvolumecontroller.NewPersistentVolumeController(binderClient, syncPeriod, plugin, plugins, cloud, "", nil, nil, nil, true)
 
 	watchPV, err := testClient.PersistentVolumes().Watch(api.ListOptions{})