Fixing Delete acl functionality (#984)

* Fixing Delete acl functionality * Adding missing param * incorporating review comments
banzaicloud · Jun 9, 2023 · 115f380 · 115f380
1 parent 811b8f2
commit 115f380
Show file tree

Hide file tree

Showing 5 changed files with 35 additions and 16 deletions.
diff --git a/controllers/kafkauser_controller.go b/controllers/kafkauser_controller.go
@@ -332,8 +332,10 @@ func (r *KafkaUserReconciler) checkFinalizers(ctx context.Context, cluster *v1be
 	var err error
 	if util.StringSliceContains(instance.GetFinalizers(), userFinalizer) {
 		if len(instance.Spec.TopicGrants) > 0 {
-			if err = r.finalizeKafkaUserACLs(reqLogger, cluster, user); err != nil {
-				return requeueWithError(reqLogger, "failed to finalize kafkauser", err)
+			for _, topicGrant := range instance.Spec.TopicGrants {
+				if err = r.finalizeKafkaUserACLs(reqLogger, cluster, user, topicGrant.PatternType); err != nil {
+					return requeueWithError(reqLogger, "failed to finalize kafkauser", err)
+				}
 			}
 		}
 		// remove finalizer
@@ -350,7 +352,7 @@ func (r *KafkaUserReconciler) removeFinalizer(ctx context.Context, user *v1alpha
 	return err
 }
 
-func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, user string) error {
+func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, user string, patternType v1alpha1.KafkaPatternType) error {
 	if k8sutil.IsMarkedForDeletion(cluster.ObjectMeta) {
 		reqLogger.Info("Cluster is being deleted, skipping ACL deletion")
 		return nil
@@ -362,7 +364,7 @@ func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, clust
 		return err
 	}
 	defer close()
-	if err = broker.DeleteUserACLs(user); err != nil {
+	if err = broker.DeleteUserACLs(user, patternType); err != nil {
 		return err
 	}
 	return nil

diff --git a/controllers/tests/common_test.go b/controllers/tests/common_test.go
@@ -185,5 +185,5 @@ func resetMockKafkaClient(kafkaCluster *v1beta1.KafkaCluster) {
 	}
 
 	// delete all acls
-	_ = mockKafkaClient.DeleteUserACLs("")
+	_ = mockKafkaClient.DeleteUserACLs("", "")
 }
diff --git a/pkg/kafkaclient/client.go b/pkg/kafkaclient/client.go
@@ -42,7 +42,7 @@ type KafkaClient interface {
 	DescribeTopic(string) (*sarama.TopicMetadata, error)
 	CreateUserACLs(v1alpha1.KafkaAccessType, v1alpha1.KafkaPatternType, string, string) error
 	ListUserACLs() ([]sarama.ResourceAcls, error)
-	DeleteUserACLs(string) error
+	DeleteUserACLs(string, v1alpha1.KafkaPatternType) error
 
 	Brokers() map[int32]string
 	DescribeCluster() ([]*sarama.Broker, int32, error)

diff --git a/pkg/kafkaclient/users.go b/pkg/kafkaclient/users.go
@@ -68,19 +68,27 @@ func (k *kafkaClient) ListUserACLs() ([]sarama.ResourceAcls, error) {
 }
 
 // DeleteUserACLs removes all ACLs for a given user
-func (k *kafkaClient) DeleteUserACLs(dn string) (err error) {
+func (k *kafkaClient) DeleteUserACLs(dn string, patternType v1alpha1.KafkaPatternType) error {
+	if patternType == "" {
+		patternType = v1alpha1.KafkaPatternTypeDefault
+	}
+	aclPatternType := AclPatternTypeMapping(patternType)
 	matches, err := k.admin.DeleteACL(sarama.AclFilter{
-		Principal: &dn,
+		Principal:                 &dn,
+		ResourcePatternTypeFilter: aclPatternType,
+		Operation:                 sarama.AclOperationDelete,
+		ResourceType:              sarama.AclResourceTopic,
+		PermissionType:            sarama.AclPermissionAny,
 	}, false)
 	if err != nil {
-		return
+		return err
 	}
 	for _, x := range matches {
 		if x.Err != sarama.ErrNoError {
 			return x.Err
 		}
 	}
-	return
+	return nil
 }
 
 func (k *kafkaClient) createReadACLs(dn string, topic string, patternType sarama.AclResourcePatternType) (err error) {

diff --git a/pkg/kafkaclient/users_test.go b/pkg/kafkaclient/users_test.go
@@ -81,16 +81,25 @@ func TestCreateUserACLs(t *testing.T) {
 func TestDeleteUserACLs(t *testing.T) {
 	client := newOpenedMockClient()
 
-	if err := client.DeleteUserACLs("test-user"); err != nil {
-		t.Error("Expected no error, got:", err)
-	}
+	validPatternTypes := []v1alpha1.KafkaPatternType{
+		"any",
+		"literal",
+		"match",
+		"prefixed",
+		""}
 
-	if err := client.DeleteUserACLs("with-error"); err == nil {
-		t.Error("Expected error, got nil")
+	for _, patternType := range validPatternTypes {
+		if err := client.DeleteUserACLs("test-user", patternType); err != nil {
+			t.Error("Expected no error, got:", err)
+		}
+
+		if err := client.DeleteUserACLs("with-error", patternType); err == nil {
+			t.Error("Expected error, got nil")
+		}
 	}
 
 	client.admin, _ = newMockClusterAdminFailOps([]string{}, sarama.NewConfig())
-	if err := client.DeleteUserACLs("test-userr"); err == nil {
+	if err := client.DeleteUserACLs("test-userr", ""); err == nil {
 		t.Error("Expected error, got nil")
 	}
 }