Skip processing of Attach/DetachError changes

Do not process "VolumeAttachment changed" event when only
Status.AttachError/DetachError is changed. Such VolumeAttachment was
written by the controller and retry with exponential backoff is already
enqueued.

pkg/controller/controller.go

@@ -23,6 +23,7 @@ import (
 
 	"k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1beta1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
 	coreinformers "k8s.io/client-go/informers/core/v1"
@@ -130,8 +131,13 @@ func (ctrl *CSIAttachController) vaAdded(obj interface{}) {
 
 // vaUpdated reacts to a VolumeAttachment update
 func (ctrl *CSIAttachController) vaUpdated(old, new interface{}) {
-	va := new.(*storage.VolumeAttachment)
-	ctrl.vaQueue.Add(va.Name)
+	oldVA := old.(*storage.VolumeAttachment)
+	newVA := new.(*storage.VolumeAttachment)
+	if shouldEnqueueVAChange(oldVA, newVA) {
+		ctrl.vaQueue.Add(newVA.Name)
+	} else {
+		glog.V(3).Infof("Ignoring VolumeAttachment %q change", newVA.Name)
+	}
 }
 
 // vaDeleted reacts to a VolumeAttachment deleted
@@ -210,3 +216,30 @@ func (ctrl *CSIAttachController) syncPV() {
 	}
 	ctrl.handler.SyncNewOrUpdatedPersistentVolume(pv)
 }
+
+// shouldEnqueueVAChange checks if a changed VolumeAttachment should be enqueued.
+// It filters out changes in Status.Attach/DetachError - these were posted by the controller
+// just few moments ago. If they were enqueued, Attach()/Detach() would be called again,
+// breaking exponential backoff.
+func shouldEnqueueVAChange(old, new *storage.VolumeAttachment) bool {
+	if old.ResourceVersion == new.ResourceVersion {
+		// This is most probably periodic sync, enqueue it
+		return true
+	}
+	if new.Status.AttachError == nil && new.Status.DetachError == nil && old.Status.AttachError == nil && old.Status.DetachError == nil {
+		// The difference between old and new must be elsewhere than Status.Attach/DetachError
+		return true
+	}
+
+	sanitized := new.DeepCopy()
+	sanitized.ResourceVersion = old.ResourceVersion
+	sanitized.Status.AttachError = old.Status.AttachError
+	sanitized.Status.DetachError = old.Status.DetachError
+
+	if equality.Semantic.DeepEqual(old, sanitized) {
+		// The objects are the same except Status.Attach/DetachError.
+		// Don't enqueue them.
+		return false
+	}
+	return true
+}

pkg/controller/controller_test.go

@@ -0,0 +1,131 @@
+/*
+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 controller
+
+import (
+	"testing"
+
+	storage "k8s.io/api/storage/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestShouldEnqueueVAChange(t *testing.T) {
+	va1 := &storage.VolumeAttachment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "foo",
+			ResourceVersion: "1",
+		},
+		Spec: storage.VolumeAttachmentSpec{
+			Attacher: "1",
+		},
+		Status: storage.VolumeAttachmentStatus{
+			Attached: false,
+		},
+	}
+
+	va1WithAttachError := va1.DeepCopy()
+	va1.Status.AttachError = &storage.VolumeError{
+		Message: "mock error1",
+		Time:    metav1.Time{},
+	}
+
+	va1WithDetachError := va1.DeepCopy()
+	va1.Status.DetachError = &storage.VolumeError{
+		Message: "mock error1",
+		Time:    metav1.Time{},
+	}
+
+	va2ChangedSpec := va1.DeepCopy()
+	va2ChangedSpec.ResourceVersion = "2"
+	va2ChangedSpec.Spec.Attacher = "2"
+
+	va2ChangedMetadata := va1.DeepCopy()
+	va2ChangedMetadata.ResourceVersion = "2"
+	va2ChangedMetadata.Annotations = map[string]string{"foo": "bar"}
+
+	va2ChangedAttachError := va1.DeepCopy()
+	va2ChangedAttachError.ResourceVersion = "2"
+	va2ChangedAttachError.Status.AttachError = &storage.VolumeError{
+		Message: "mock error2",
+		Time:    metav1.Time{},
+	}
+
+	va2ChangedDetachError := va1.DeepCopy()
+	va2ChangedDetachError.ResourceVersion = "2"
+	va2ChangedDetachError.Status.DetachError = &storage.VolumeError{
+		Message: "mock error2",
+		Time:    metav1.Time{},
+	}
+
+	tests := []struct {
+		name           string
+		oldVA, newVA   *storage.VolumeAttachment
+		expectedResult bool
+	}{
+		{
+			name:           "periodic sync",
+			oldVA:          va1,
+			newVA:          va1,
+			expectedResult: true,
+		},
+		{
+			name:           "changed spec",
+			oldVA:          va1,
+			newVA:          va2ChangedSpec,
+			expectedResult: true,
+		},
+		{
+			name:           "changed metadata",
+			oldVA:          va1,
+			newVA:          va2ChangedMetadata,
+			expectedResult: true,
+		},
+		{
+			name:           "added attachError",
+			oldVA:          va1,
+			newVA:          va2ChangedAttachError,
+			expectedResult: false,
+		},
+		{
+			name:           "added detachError",
+			oldVA:          va1,
+			newVA:          va2ChangedDetachError,
+			expectedResult: false,
+		},
+		{
+			name:           "changed attachError",
+			oldVA:          va1WithAttachError,
+			newVA:          va2ChangedAttachError,
+			expectedResult: false,
+		},
+		{
+			name:           "changed detachError",
+			oldVA:          va1WithDetachError,
+			newVA:          va2ChangedDetachError,
+			expectedResult: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := shouldEnqueueVAChange(test.oldVA, test.newVA)
+			if result != test.expectedResult {
+				t.Errorf("Error: expected result %t, got %t", test.expectedResult, result)
+			}
+		})
+	}
+}