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

@@ -133,17 +133,39 @@ def test_crud(self, simple_cluster):
         updates = {
             "spec": {
                 "associatedSCRAMSecrets": [secret_1, secret_2],
+            },
+        }
+        k8s.patch_custom_resource(ref, updates)
+        time.sleep(CHECK_STATUS_WAIT_SECONDS)
+        assert k8s.wait_on_condition(
+            ref,
+            "ACK.ResourceSynced",
+            "True",
+            wait_periods=MODIFY_WAIT_AFTER_SECONDS,
+        )
+
+        cluster.wait_until(
+            cluster_arn,
+            cluster.state_matches("ACTIVE"),
+        )
+
+        latest_secrets = cluster.get_associated_scram_secrets(cluster_arn)
+        assert len(latest_secrets) == 2
+        assert secret_1 in latest_secrets and secret_2 in latest_secrets
+
+        updates = {
+            "spec": {
                 'brokerNodeGroupInfo': {
                     "storageInfo": {
                         "ebsStorageInfo": {
                             "volumeSize": updated_volume_size
                         }
                     }
                 }
-            },
+            }
         }
         k8s.patch_custom_resource(ref, updates)
-
+        time.sleep(MODIFY_WAIT_AFTER_SECONDS)
         assert k8s.wait_on_condition(
             ref,
             "ACK.ResourceSynced",
@@ -155,19 +177,15 @@ def test_crud(self, simple_cluster):
             cluster_arn,
             cluster.state_matches("ACTIVE"),
         )
-
-        latest_secrets = cluster.get_associated_scram_secrets(cluster_arn)
-        assert len(latest_secrets) == 2
-        assert secret_1 in latest_secrets and secret_2 in latest_secrets
-
+        time.sleep(MODIFY_WAIT_AFTER_SECONDS)
 
         latest_cluster = cluster.get_by_arn(cluster_arn)
         assert latest_cluster is not None
 
         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 = {