Handle BatchAssociateScramSecret errors

Author: michaelhtm

pkg/resource/cluster/hooks.go

@@ -23,6 +23,7 @@ import (
 	svcapitypes "github.com/aws-controllers-k8s/kafka-controller/apis/v1alpha1"
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
 	ackcondition "github.com/aws-controllers-k8s/runtime/pkg/condition"
+	ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
 	ackrequeue "github.com/aws-controllers-k8s/runtime/pkg/requeue"
 	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
 	ackutil "github.com/aws-controllers-k8s/runtime/pkg/util"
@@ -354,6 +355,16 @@ func (rm *resourceManager) getAssociatedScramSecrets(
 	return res, err
 }
 
+type unprocessedSecret struct {
+	errorCode    string
+	errorMessage string
+	secretArn    string
+}
+
+func (us unprocessedSecret) String() string {
+	return fmt.Sprintf("ErrorCode: %s, ErrorMessage %s, SecretArn: %s", us.errorCode, us.errorMessage, us.secretArn)
+}
+
 // batchAssociateScramSecret associates the supplied scram secrets to the supplied Cluster
 // resource
 func (rm *resourceManager) batchAssociateScramSecret(
@@ -367,14 +378,27 @@ func (rm *resourceManager) batchAssociateScramSecret(
 
 	input := &svcsdk.BatchAssociateScramSecretInput{}
 	input.ClusterArn = (*string)(r.ko.Status.ACKResourceMetadata.ARN)
-	// Convert []*string to []string
-	unrefSecrets := make([]string, len(secretARNs))
-	for i, s := range secretARNs {
-		unrefSecrets[i] = *s
-	}
-	input.SecretArnList = unrefSecrets
-	_, err = rm.sdkapi.BatchAssociateScramSecret(ctx, input)
+	input.SecretArnList = aws.ToStringSlice(secretARNs)
+	resp, err := rm.sdkapi.BatchAssociateScramSecret(ctx, input)
 	rm.metrics.RecordAPICall("UPDATE", "BatchAssociateScramSecret", err)
+	if err != nil {
+		return err
+	}
+
+	if len(resp.UnprocessedScramSecrets) > 0 {
+		unprocessedSecrets := []unprocessedSecret{}
+		for _, uss := range resp.UnprocessedScramSecrets {
+			us := unprocessedSecret{
+				errorCode:    aws.ToString(uss.ErrorCode),
+				errorMessage: aws.ToString(uss.ErrorMessage),
+				secretArn:    aws.ToString(uss.SecretArn),
+			}
+			unprocessedSecrets = append(unprocessedSecrets, us)
+		}
+
+		return ackerr.NewTerminalError(fmt.Errorf("Cant attach secret arns: %v", unprocessedSecrets))
+	}
+
 	return err
 }
 

test/e2e/tests/test_cluster.py

@@ -33,7 +33,8 @@
 # even trying to fetch a cluster's state.
 CREATE_WAIT_AFTER_SECONDS = 180
 DELETE_WAIT_SECONDS = 300
-MODIFY_WAIT_AFTER_SECONDS = 1800
+MODIFY_WAIT_AFTER_SECONDS = 10
+LONG_UPDATE_WAIT = 600
 
 # Time to wait after the cluster has changed status, for the CR to update
 CHECK_STATUS_WAIT_SECONDS = 60
@@ -127,23 +128,14 @@ def test_crud(self, simple_cluster):
         assert len(latest_secrets) == 1
         assert secret_1 in latest_secrets
 
-        updated_volume_size = cr['spec']['brokerNodeGroupInfo']['storageInfo']['ebsStorageInfo']['volumeSize'] + 10
-
         # associate one more secret to the cluster
         updates = {
             "spec": {
                 "associatedSCRAMSecrets": [secret_1, secret_2],
-                'brokerNodeGroupInfo': {
-                    "storageInfo": {
-                        "ebsStorageInfo": {
-                            "volumeSize": updated_volume_size
-                        }
-                    }
-                }
             },
         }
         k8s.patch_custom_resource(ref, updates)
-
+        time.sleep(CHECK_STATUS_WAIT_SECONDS)
         assert k8s.wait_on_condition(
             ref,
             "ACK.ResourceSynced",
@@ -160,14 +152,41 @@ def test_crud(self, simple_cluster):
         assert len(latest_secrets) == 2
         assert secret_1 in latest_secrets and secret_2 in latest_secrets
 
+        updated_volume_size = cr['spec']['brokerNodeGroupInfo']['storageInfo']['ebsStorageInfo']['volumeSize'] + 10
+        updates = {
+            "spec": {
+                'brokerNodeGroupInfo': {
+                    "storageInfo": {
+                        "ebsStorageInfo": {
+                            "volumeSize": updated_volume_size
+                        }
+                    }
+                }
+            }
+        }
+        k8s.patch_custom_resource(ref, updates)
+        time.sleep(CHECK_STATUS_WAIT_SECONDS)
+        assert k8s.wait_on_condition(
+            ref,
+            "ACK.ResourceSynced",
+            "True",
+            wait_periods=LONG_UPDATE_WAIT,
+        )
+
+        cluster.wait_until(
+            cluster_arn,
+            cluster.state_matches("ACTIVE"),
+        )
 
         latest_cluster = cluster.get_by_arn(cluster_arn)
         assert latest_cluster is not None
+        
+        cr = k8s.get_resource(ref)
 
         latest_volume = latest_cluster['BrokerNodeGroupInfo']["StorageInfo"]["EbsStorageInfo"]["VolumeSize"]
         desired_volume = cr['spec']['brokerNodeGroupInfo']['storageInfo']['ebsStorageInfo']['volumeSize']
 
-        assert latest_volume == desired_volume == updated_volume_size
+        assert latest_volume == desired_volume
 
         # remove all associated secrets
         updates = {