Merge pull request #19868 from jsafrane/devel/syncclaim-twice

pkg/controller/persistentvolume/persistentvolume_claim_binder_controller.go

@@ -335,7 +335,15 @@ func syncVolume(volumeIndex *persistentVolumeOrderedIndex, binderClient binderCl
 }
 
 func syncClaim(volumeIndex *persistentVolumeOrderedIndex, binderClient binderClient, claim *api.PersistentVolumeClaim) (err error) {
-	glog.V(5).Infof("Synchronizing PersistentVolumeClaim[%s]\n", claim.Name)
+	glog.V(5).Infof("Synchronizing PersistentVolumeClaim[%s] for binding", claim.Name)
+
+	// The claim may have been modified by parallel call to syncClaim, load
+	// the current version.
+	newClaim, err := binderClient.GetPersistentVolumeClaim(claim.Namespace, claim.Name)
+	if err != nil {
+		return fmt.Errorf("Cannot reload claim %s/%s: %v", claim.Namespace, claim.Name, err)
+	}
+	claim = newClaim
 
 	switch claim.Status.Phase {
 	case api.ClaimPending:

pkg/controller/persistentvolume/persistentvolume_claim_binder_controller_test.go

@@ -123,7 +123,6 @@ func TestClaimRace(t *testing.T) {
 
 	plugMgr := volume.VolumePluginMgr{}
 	plugMgr.InitPlugins(host_path.ProbeRecyclableVolumePlugins(newMockRecycler, volume.VolumeConfig{}), volume.NewFakeVolumeHost(tmpDir, nil, nil))
-
 	// adds the volume to the index, making the volume available
 	syncVolume(volumeIndex, mockClient, v)
 	if mockClient.volume.Status.Phase != api.VolumeAvailable {
@@ -133,6 +132,8 @@ func TestClaimRace(t *testing.T) {
 		t.Errorf("Expected to find volume in index but it did not exist")
 	}
 
+	// add the claim to fake API server
+	mockClient.UpdatePersistentVolumeClaim(c1)
 	// an initial sync for a claim matches the volume
 	err = syncClaim(volumeIndex, mockClient, c1)
 	if err != nil {
@@ -143,6 +144,8 @@ func TestClaimRace(t *testing.T) {
 	}
 
 	// before the volume gets updated w/ claimRef, a 2nd claim can attempt to bind and find the same volume
+	// add the 2nd claim to fake API server
+	mockClient.UpdatePersistentVolumeClaim(c2)
 	err = syncClaim(volumeIndex, mockClient, c2)
 	if err != nil {
 		t.Errorf("unexpected error for unmatched claim: %v", err)
@@ -470,6 +473,8 @@ func TestBindingWithExamples(t *testing.T) {
 		t.Errorf("Expected phase %s but got %s", api.VolumeAvailable, mockClient.volume.Status.Phase)
 	}
 
+	// add the claim to fake API server
+	mockClient.UpdatePersistentVolumeClaim(claim)
 	// an initial sync for a claim will bind it to an unbound volume
 	syncClaim(volumeIndex, mockClient, claim)
 
@@ -539,6 +544,14 @@ func TestCasting(t *testing.T) {
 		Status:     api.PersistentVolumeClaimStatus{Phase: api.ClaimBound},
 	}
 
+	// Inject mockClient into the binder. This prevents weird errors on stderr
+	// as the binder wants to load PV/PVC from API server.
+	mockClient := &mockBinderClient{
+		volume: pv,
+		claim:  pvc,
+	}
+	binder.client = mockClient
+
 	// none of these should fail casting.
 	// the real test is not failing when passed DeletedFinalStateUnknown in the deleteHandler
 	binder.addVolume(pv)

pkg/controller/persistentvolume/persistentvolume_provisioner_controller.go

@@ -153,6 +153,20 @@ func (controller *PersistentVolumeProvisionerController) handleUpdateClaim(oldOb
 }
 
 func (controller *PersistentVolumeProvisionerController) reconcileClaim(claim *api.PersistentVolumeClaim) error {
+	glog.V(5).Infof("Synchronizing PersistentVolumeClaim[%s] for dynamic provisioning", claim.Name)
+
+	// The claim may have been modified by parallel call to reconcileClaim, load
+	// the current version.
+	newClaim, err := controller.client.GetPersistentVolumeClaim(claim.Namespace, claim.Name)
+	if err != nil {
+		return fmt.Errorf("Cannot reload claim %s/%s: %v", claim.Namespace, claim.Name, err)
+	}
+	claim = newClaim
+	err = controller.claimStore.Update(claim)
+	if err != nil {
+		return fmt.Errorf("Cannot update claim %s/%s: %v", claim.Namespace, claim.Name, err)
+	}
+
 	if controller.provisioner == nil {
 		return fmt.Errorf("No provisioner configured for controller")
 	}

pkg/controller/persistentvolume/persistentvolume_provisioner_controller_test.go

@@ -105,6 +105,9 @@ func TestReconcileClaim(t *testing.T) {
 
 	// watch would have added the claim to the store
 	controller.claimStore.Add(pvc)
+	// store it in fake API server
+	mockClient.UpdatePersistentVolumeClaim(pvc)
+
 	err := controller.reconcileClaim(pvc)
 	if err != nil {
 		t.Errorf("Unexpected error: %v", err)
@@ -116,6 +119,8 @@ func TestReconcileClaim(t *testing.T) {
 	}
 
 	pvc.Annotations[qosProvisioningKey] = "foo"
+	// store it in fake API server
+	mockClient.UpdatePersistentVolumeClaim(pvc)
 
 	err = controller.reconcileClaim(pvc)
 	if err != nil {
@@ -130,6 +135,40 @@ func TestReconcileClaim(t *testing.T) {
 	if mockClient.volume.Spec.ClaimRef.Name != pvc.Name {
 		t.Errorf("Expected PV to be bound to %s but got %s", mockClient.volume.Spec.ClaimRef.Name, pvc.Name)
 	}
+
+	// the PVC should have correct annotation
+	if mockClient.claim.Annotations[pvProvisioningRequiredAnnotationKey] != pvProvisioningCompletedAnnotationValue {
+		t.Errorf("Annotation %q not set", pvProvisioningRequiredAnnotationKey)
+	}
+
+	// Run the syncClaim 2nd time to simulate periodic sweep running in parallel
+	// to the previous syncClaim. There is a lock in handleUpdateVolume(), so
+	// they will be called sequentially, but the second call will have old
+	// version of the claim.
+	oldPVName := mockClient.volume.Name
+
+	// Make the "old" claim
+	pvc2 := makeTestClaim()
+	pvc2.Annotations[qosProvisioningKey] = "foo"
+	// Add a dummy annotation so we recognize the claim was updated (i.e.
+	// stored in mockClient)
+	pvc2.Annotations["test"] = "test"
+
+	err = controller.reconcileClaim(pvc2)
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	// The 2nd PVC should be ignored, no new PV was created
+	if val, found := pvc2.Annotations[pvProvisioningRequiredAnnotationKey]; found {
+		t.Errorf("2nd PVC got unexpected annotation %q: %q", pvProvisioningRequiredAnnotationKey, val)
+	}
+	if mockClient.volume.Name != oldPVName {
+		t.Errorf("2nd PVC unexpectedly provisioned a new volume")
+	}
+	if _, found := mockClient.claim.Annotations["test"]; found {
+		t.Errorf("2nd PVC was unexpectedly updated")
+	}
 }
 
 func checkTagValue(t *testing.T, tags map[string]string, tag string, expectedValue string) {

test/integration/persistent_volumes_test.go

@@ -46,9 +46,11 @@ func TestPersistentVolumeRecycler(t *testing.T) {
 	defer s.Close()
 
 	deleteAllEtcdKeys()
-	binderClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}})
-	recyclerClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}})
-	testClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}})
+	// Use higher QPS and Burst, there is a test for race condition below, which
+	// creates many claims and default values were too low.
+	binderClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}, QPS: 1000, Burst: 100000})
+	recyclerClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}, QPS: 1000, Burst: 100000})
+	testClient := clientset.NewForConfigOrDie(&client.Config{Host: s.URL, ContentConfig: client.ContentConfig{GroupVersion: testapi.Default.GroupVersion()}, QPS: 1000, Burst: 100000})
 	host := volume.NewFakeVolumeHost("/tmp/fake", nil, nil)
 
 	plugins := []volume.VolumePlugin{&volume.FakeVolumePlugin{"plugin-name", host, volume.VolumeConfig{}, volume.VolumeOptions{}, 0, 0}}