Add update operations for VolumeSize, InstanceType, and BrokerCount (#55)

Issue [#2249](aws-controllers-k8s/community#2249)

Description of changes:
Also added reconcile before deletion attempt if cluster is 
not active. Before, if we tried deleting a cluster that was 
not active, the controller would set it to terminal, and does 
not reconcile.

Another change being added is a field in the Cluster status
called ClusterVersion. This field is returned by the sdkFind 
operation, and is crutial to make updates to VolumeSize,
InstanceType, and BrokerCount.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: michaelhtm

apis/v1alpha1/ack-generate-metadata.yaml

@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2025-02-20T18:10:11Z"
+  build_date: "2025-03-04T23:55:37Z"
   build_hash: a326346bd3a6973254d247c9ab2dc76790c36241
   go_version: go1.24.0
   version: v0.43.2
-api_directory_checksum: eda989f20dde9f1b4331ffa67dc3b9a5ef0d64e4
+api_directory_checksum: 36fbfad1e0bff98a14b120ba292a7f6b4e546fb4
 api_version: v1alpha1
 aws_sdk_go_version: v1.32.6
 generator_config_info:
-  file_checksum: 5ea49df43c7aef08a9ac8b7171e9f50c3ed82e13
+  file_checksum: c641b5dd9aa81f1f42655f2afe9fcfb9dc7de696
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

apis/v1alpha1/cluster.go

@@ -23,6 +23,9 @@ resources:
         CreateCluster:
           input_fields:
             ClusterName: Name
+        DescribeCluster:
+          input_fields:
+            ClusterName: Name
     hooks:
       sdk_read_one_post_set_output:
         template_path: hooks/cluster/sdk_read_one_post_set_output.go.tpl
@@ -95,6 +98,11 @@ resources:
       BootstrapBrokerStringVpcConnectivityTls:
         type: string
         is_read_only: true
+      CurrentVersion:
+        from:
+          operation: DescribeCluster
+          path: ClusterInfo.CurrentVersion
+        is_read_only: true
     tags:
       # TODO(jaypipes): Ignore tags for now... we will add support later
       ignore: true

apis/v1alpha1/generator.yaml

@@ -356,6 +356,9 @@ spec:
                   - type
                   type: object
                 type: array
+              currentVersion:
+                description: The current version of the MSK cluster.
+                type: string
               state:
                 description: |-
                   The state of the cluster. The possible states are ACTIVE, CREATING, DELETING,

apis/v1alpha1/zz_generated.deepcopy.go

@@ -23,6 +23,9 @@ resources:
         CreateCluster:
           input_fields:
             ClusterName: Name
+        DescribeCluster:
+          input_fields:
+            ClusterName: Name
     hooks:
       sdk_read_one_post_set_output:
         template_path: hooks/cluster/sdk_read_one_post_set_output.go.tpl
@@ -95,6 +98,11 @@ resources:
       BootstrapBrokerStringVpcConnectivityTls:
         type: string
         is_read_only: true
+      CurrentVersion:
+        from:
+          operation: DescribeCluster
+          path: ClusterInfo.CurrentVersion
+        is_read_only: true
     tags:
       # TODO(jaypipes): Ignore tags for now... we will add support later
       ignore: true

config/crd/bases/kafka.services.k8s.aws_clusters.yaml

@@ -356,6 +356,9 @@ spec:
                   - type
                   type: object
                 type: array
+              currentVersion:
+                description: The current version of the MSK cluster.
+                type: string
               state:
                 description: |-
                   The state of the cluster. The possible states are ACTIVE, CREATING, DELETING,

generator.yaml

@@ -18,10 +18,12 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	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"
@@ -38,6 +40,7 @@ var (
 		string(svcsdktypes.ClusterStateDeleting),
 		string(svcsdktypes.ClusterStateFailed),
 	}
+	RequeueAfterUpdateDuration = 15 * time.Second
 )
 
 var (
@@ -113,6 +116,18 @@ func clusterDeleting(r *resource) bool {
 	return cs == strings.ToLower(string(svcsdktypes.ClusterStateDeleting))
 }
 
+// requeueAfterAsyncUpdate returns a `ackrequeue.RequeueNeededAfter` struct
+// explaining the cluster cannot be modified until after the asynchronous update
+// has (first, started and then) completed and the cluster reaches an active
+// status.
+func requeueAfterAsyncUpdate() *ackrequeue.RequeueNeededAfter {
+	return ackrequeue.NeededAfter(
+		fmt.Errorf("cluster has started asynchronously updating, cannot be modified until '%s'",
+			"Active"),
+		RequeueAfterUpdateDuration,
+	)
+}
+
 func (rm *resourceManager) customUpdate(
 	ctx context.Context,
 	desired *resource,
@@ -133,12 +148,6 @@ func (rm *resourceManager) customUpdate(
 	// Copy status from latest since it has the current cluster state
 	updatedRes.ko.Status = latest.ko.Status
 
-	if clusterDeleting(latest) {
-		msg := "Cluster is currently being deleted"
-		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &msg, nil)
-		return updatedRes, requeueWaitWhileDeleting
-	}
-
 	if !clusterActive(latest) {
 		msg := "Cluster is in '" + *latest.ko.Status.State + "' state"
 		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &msg, nil)
@@ -149,16 +158,120 @@ func (rm *resourceManager) customUpdate(
 		return updatedRes, requeueWaitUntilCanModify(latest)
 	}
 
-	if delta.DifferentAt("Spec.AssociatedSCRAMSecrets") {
+	switch {
+	case delta.DifferentAt("Spec.ClientAuthentication"):
+		input := &svcsdk.UpdateSecurityInput{}
+		if desired.ko.Status.CurrentVersion != nil {
+			input.CurrentVersion = desired.ko.Status.CurrentVersion
+		}
+		if desired.ko.Status.ACKResourceMetadata.ARN != nil {
+			input.ClusterArn = (*string)(desired.ko.Status.ACKResourceMetadata.ARN)
+		}
+		if desired.ko.Spec.ClientAuthentication != nil {
+			f0 := &svcsdktypes.ClientAuthentication{}
+			if desired.ko.Spec.ClientAuthentication.SASL != nil {
+				f0f0 := &svcsdktypes.Sasl{}
+				if desired.ko.Spec.ClientAuthentication.SASL.IAM != nil &&
+					desired.ko.Spec.ClientAuthentication.SASL.IAM.Enabled != nil {
+					f0f0f0 := &svcsdktypes.Iam{
+						Enabled: desired.ko.Spec.ClientAuthentication.SASL.IAM.Enabled,
+					}
+					f0f0.Iam = f0f0f0
+				}
+				if desired.ko.Spec.ClientAuthentication.SASL.SCRAM != nil &&
+					desired.ko.Spec.ClientAuthentication.SASL.SCRAM.Enabled != nil {
+					f0f0f1 := &svcsdktypes.Scram{
+						Enabled: desired.ko.Spec.ClientAuthentication.SASL.SCRAM.Enabled,
+					}
+					f0f0.Scram = f0f0f1
+				}
+				f0.Sasl = f0f0
+			}
+			if desired.ko.Spec.ClientAuthentication.TLS != nil {
+				f0f1 := &svcsdktypes.Tls{}
+				if desired.ko.Spec.ClientAuthentication.TLS.CertificateAuthorityARNList != nil {
+					f0f1.CertificateAuthorityArnList = aws.ToStringSlice(desired.ko.Spec.ClientAuthentication.TLS.CertificateAuthorityARNList)
+				}
+				if desired.ko.Spec.ClientAuthentication.TLS.Enabled != nil {
+					f0f1.Enabled = desired.ko.Spec.ClientAuthentication.TLS.Enabled
+				}
+				f0.Tls = f0f1
+			}
+			if desired.ko.Spec.ClientAuthentication.Unauthenticated != nil &&
+				desired.ko.Spec.ClientAuthentication.Unauthenticated.Enabled != nil {
+				f0.Unauthenticated = &svcsdktypes.Unauthenticated{
+					Enabled: desired.ko.Spec.ClientAuthentication.Unauthenticated.Enabled,
+				}
+			}
+			input.ClientAuthentication = f0
+		}
+
+		_, err = rm.sdkapi.UpdateSecurity(ctx, input)
+		rm.metrics.RecordAPICall("UPDATE", "UpdateSecurity", err)
+		if err != nil {
+			return nil, err
+		}
+		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, nil, nil)
+		err = requeueAfterAsyncUpdate()
+
+	case delta.DifferentAt("Spec.AssociatedSCRAMSecrets"):
 		err = rm.syncAssociatedScramSecrets(ctx, updatedRes, latest)
 		if err != nil {
 			return nil, err
 		}
+		// Set synced condition to True after successful update
+		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, nil, nil)
+
+	case delta.DifferentAt("Spec.BrokerNodeGroupInfo.StorageInfo.EBSStorageInfo.VolumeSize"):
+		_, err := rm.sdkapi.UpdateBrokerStorage(ctx, &svcsdk.UpdateBrokerStorageInput{
+			ClusterArn:     (*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+			CurrentVersion: latest.ko.Status.CurrentVersion,
+			TargetBrokerEBSVolumeInfo: []svcsdktypes.BrokerEBSVolumeInfo{
+				{
+					KafkaBrokerNodeId: aws.String("ALL"),
+					VolumeSizeGB:      aws.Int32(int32(*desired.ko.Spec.BrokerNodeGroupInfo.StorageInfo.EBSStorageInfo.VolumeSize)),
+				},
+			},
+		})
+		rm.metrics.RecordAPICall("UPDATE", "UpdateBrokerStorage", err)
+		if err != nil {
+			return nil, err
+		}
+		message := fmt.Sprintf("kafka is updating broker storage")
+		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &message, nil)
+		err = requeueAfterAsyncUpdate()
+
+	case delta.DifferentAt("Spec.BrokerNodeGroupInfo.InstanceType"):
+		_, err := rm.sdkapi.UpdateBrokerType(ctx, &svcsdk.UpdateBrokerTypeInput{
+			ClusterArn:         (*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+			CurrentVersion:     latest.ko.Status.CurrentVersion,
+			TargetInstanceType: desired.ko.Spec.BrokerNodeGroupInfo.InstanceType,
+		})
+		rm.metrics.RecordAPICall("UPDATE", "UpdateBrokerType", err)
+		if err != nil {
+			return nil, err
+		}
+		message := fmt.Sprintf("kafka is updating broker instanceType")
+		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &message, nil)
+		err = requeueAfterAsyncUpdate()
+
+	case delta.DifferentAt("Spec.NumberOfBrokerNodes"):
+		_, err := rm.sdkapi.UpdateBrokerCount(ctx, &svcsdk.UpdateBrokerCountInput{
+			ClusterArn:                (*string)(latest.ko.Status.ACKResourceMetadata.ARN),
+			CurrentVersion:            latest.ko.Status.CurrentVersion,
+			TargetNumberOfBrokerNodes: aws.Int32(int32(*desired.ko.Spec.NumberOfBrokerNodes)),
+		})
+		rm.metrics.RecordAPICall("UPDATE", "UpdateBrokerCount", err)
+		if err != nil {
+			return nil, err
+		}
+		message := fmt.Sprintf("kafka is updating broker instanceType")
+		ackcondition.SetSynced(updatedRes, corev1.ConditionFalse, &message, nil)
+		err = requeueAfterAsyncUpdate()
+
 	}
 
-	// Set synced condition to True after successful update
-	ackcondition.SetSynced(updatedRes, corev1.ConditionTrue, nil, nil)
-	return updatedRes, nil
+	return updatedRes, err
 }
 
 // syncAssociatedScramSecrets examines the Secret ARNs in the supplied Cluster
@@ -242,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(
@@ -255,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
 }
 

helm/crds/kafka.services.k8s.aws_clusters.yaml

@@ -9,3 +9,10 @@
 	if err := rm.syncAssociatedScramSecrets(ctx, &resource{ko: groupCpy}, r); err != nil {
 		return nil, err
 	}
+	if !clusterActive(r) {
+		// doing this to avoid BadRequestException
+		return r, ackrequeue.NeededAfter(
+			fmt.Errorf("waiting for cluster to be active before deletion"),
+			ackrequeue.DefaultRequeueAfterDuration,
+		)
+	}