Add update operations for VolumeSize, InstanceType, and BrokerCount

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.

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-02-27T21:18:28Z"
   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

@@ -6,4 +6,4 @@ kind: Kustomization
 images:
 - name: controller
   newName: public.ecr.aws/aws-controllers-k8s/kafka-controller
-  newTag: 1.0.5
+  newTag: 1.0.2

apis/v1alpha1/zz_generated.deepcopy.go

@@ -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,

config/controller/kustomization.yaml

@@ -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

@@ -1,8 +1,8 @@
 apiVersion: v1
 name: kafka-chart
 description: A Helm chart for the ACK service controller for Amazon Managed Streaming for Apache Kafka (MSK)
-version: 1.0.5
-appVersion: 1.0.5
+version: 1.0.2
+appVersion: 1.0.2
 home: https://github.com/aws-controllers-k8s/kafka-controller
 icon: https://raw.githubusercontent.com/aws/eks-charts/master/docs/logo/aws.png
 sources:

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,

helm/Chart.yaml

@@ -1,5 +1,5 @@
 {{ .Chart.Name }} has been installed.
-This chart deploys "public.ecr.aws/aws-controllers-k8s/kafka-controller:1.0.5".
+This chart deploys "public.ecr.aws/aws-controllers-k8s/kafka-controller:1.0.2".
 
 Check its status by running:
   kubectl --namespace {{ .Release.Namespace }} get pods -l "app.kubernetes.io/instance={{ .Release.Name }}"

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

@@ -4,7 +4,7 @@
 
 image:
   repository: public.ecr.aws/aws-controllers-k8s/kafka-controller
-  tag: 1.0.5
+  tag: 1.0.2
   pullPolicy: IfNotPresent
   pullSecrets: []
 

helm/templates/NOTES.txt

@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	svcapitypes "github.com/aws-controllers-k8s/kafka-controller/apis/v1alpha1"
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
@@ -38,6 +39,7 @@ var (
 		string(svcsdktypes.ClusterStateDeleting),
 		string(svcsdktypes.ClusterStateFailed),
 	}
+	RequeueAfterUpdateDuration = 15 * time.Second
 )
 
 var (
@@ -113,6 +115,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 +147,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 +157,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

helm/values.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,
+		)
+	}

pkg/resource/cluster/hooks.go

@@ -1,3 +1,8 @@
+	if resp.ClusterInfo.CurrentVersion != nil {
+		ko.Status.CurrentVersion = resp.ClusterInfo.CurrentVersion
+	} else {
+		ko.Status.CurrentVersion = nil
+	}
 	if !clusterActive(&resource{ko}) {
 		// Setting resource synced condition to false will trigger a requeue of
 		// the resource. No need to return a requeue error here.

pkg/resource/cluster/sdk.go

@@ -16,6 +16,9 @@ spec:
     clientSubnets:
       - $SUBNET_ID_1
       - $SUBNET_ID_2
+    storageInfo:
+      ebsStorageInfo:
+        volumeSize: 10
   kafkaVersion: "3.3.1"
   # NOTE(jaypipes): Number of broker nodes need to be a multiple of the number
   # of subnets