DENA-671: Validate cleanup policy (#18)

* Validate cleanup policy

* Added cleanup policy to docs

* Do not export func

Author: sbuliarca

rules/msk_topic_config.go

@@ -2,6 +2,8 @@ package rules
 
 import (
 	"fmt"
+	"slices"
+	"strings"
 
 	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/hcl/v2/gohcl"
@@ -93,6 +95,10 @@ func (r *MSKTopicConfigRule) validateTopicConfig(runner tflint.Runner, topic *hc
 	if err := r.validateCompressionType(runner, configAttr, configKeyToPairMap); err != nil {
 		return err
 	}
+
+	if err := r.validateCleanupPolicy(runner, configAttr, configKeyToPairMap); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -202,7 +208,7 @@ func (r *MSKTopicConfigRule) validateCompressionType(
 			},
 		)
 		if err != nil {
-			return fmt.Errorf("emitting issue with fix: no replication factor: %w", err)
+			return fmt.Errorf("emitting issue with fix: no compression type: %w", err)
 		}
 		return nil
 	}
@@ -229,3 +235,58 @@ func (r *MSKTopicConfigRule) validateCompressionType(
 	}
 	return nil
 }
+
+const (
+	cleanupPolicyKey     = "cleanup.policy"
+	cleanupPolicyDefault = "delete"
+)
+
+var (
+	cleanupPolicyDefaultFix  = fmt.Sprintf(`"%s" = "%s"`, cleanupPolicyKey, cleanupPolicyDefault)
+	cleanupPolicyValidValues = []string{"delete", "compact"}
+)
+
+func (r *MSKTopicConfigRule) validateCleanupPolicy(
+	runner tflint.Runner,
+	config *hclext.Attribute,
+	configKeyToPairMap map[string]hcl.KeyValuePair,
+) error {
+	cpPair, hasCp := configKeyToPairMap[cleanupPolicyKey]
+	if !hasCp {
+		err := runner.EmitIssueWithFix(
+			r,
+			fmt.Sprintf("missing %s: using default '%s'", cleanupPolicyKey, cleanupPolicyDefault),
+			config.Range,
+			func(f tflint.Fixer) error {
+				return f.InsertTextAfter(config.Expr.StartRange(), "\n"+cleanupPolicyDefaultFix)
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("emitting issue with fix: no cleanup policy: %w", err)
+		}
+		return nil
+	}
+
+	var cpVal string
+	diags := gohcl.DecodeExpression(cpPair.Value, nil, &cpVal)
+	if diags.HasErrors() {
+		return diags
+	}
+	if !slices.Contains(cleanupPolicyValidValues, cpVal) {
+		err := runner.EmitIssue(
+			r,
+			fmt.Sprintf(
+				"invalid %s: it must be one of [%s], but currently is '%s'",
+				cleanupPolicyKey,
+				strings.Join(cleanupPolicyValidValues, ", "),
+				cpVal,
+			),
+			cpPair.Value.Range(),
+		)
+		if err != nil {
+			return fmt.Errorf("emitting issue: invalid cleanup policy: %w", err)
+		}
+		return nil
+	}
+	return nil
+}

rules/msk_topic_config.md

@@ -5,6 +5,7 @@
 An MSK topic configuration must comply with the following rules:
 - the replication factor must be equal to 3, because we are deploying across 3 availability zones and this is the minimum we can run, since min-in-sync replicas is set to 2. 
 - the 'compression.type' must always be set to `zstd`. This is a very good compression algorithm, and it is set by default for the producer in our [kafka lib](https://github.com/utilitywarehouse/uwos-go/tree/main/pubsub/kafka)
+- the 'cleanup.policy' must be specified and must be one of 'delete' or 'compact'. If not specified, it is set automatically on 'delete'. See [kafka spec](https://kafka.apache.org/30/generated/topic_config.html#topicconfigs_cleanup.policy)
 
 ## Example
 
@@ -16,6 +17,7 @@ resource "kafka_topic" "good_topic" {
   replication_factor = 3
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }
 
@@ -36,6 +38,14 @@ resource "kafka_topic" "topic_with_wrong_compression_type" {
     "compression.type" = "gzip"
   }
 }
+
+# topic with cleanup policy
+resource "kafka_topic" "topic_with_wrong_cleanup_policy" {
+  name = "wrong-topic"
+  config = {
+    "cleanup.policy" = "invalid-value"
+  }
+}
 ```
 
 ## How To Fix

rules/msk_topic_config_test.go

@@ -25,6 +25,7 @@ func Test_MSKTopicConfigRule(t *testing.T) {
 resource "kafka_topic" "topic_without_repl_factor_and_name" {
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			expected: []*helper.Issue{
@@ -47,6 +48,7 @@ resource "kafka_topic" "topic_without_repl_factor" {
   name = "topic_without_repl_factor"
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			fixed: `
@@ -55,6 +57,7 @@ resource "kafka_topic" "topic_without_repl_factor" {
   replication_factor = 3
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			expected: []*helper.Issue{
@@ -77,6 +80,7 @@ resource "kafka_topic" "topic_with_incorrect_repl_factor" {
   replication_factor = 10
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			fixed: `
@@ -85,6 +89,7 @@ resource "kafka_topic" "topic_with_incorrect_repl_factor" {
   replication_factor = 3
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			expected: []*helper.Issue{
@@ -125,6 +130,7 @@ resource "kafka_topic" "topic_without_compression_type" {
   name               = "topic_without_compression_type"
   replication_factor = 3
   config = {
+    "cleanup.policy"   = "delete"
   }
 }`,
 			fixed: `
@@ -133,6 +139,7 @@ resource "kafka_topic" "topic_without_compression_type" {
   replication_factor = 3
   config = {
     "compression.type" = "zstd"
+    "cleanup.policy"   = "delete"
   }
 }`,
 			expected: []*helper.Issue{
@@ -142,7 +149,7 @@ resource "kafka_topic" "topic_without_compression_type" {
 					Range: hcl.Range{
 						Filename: fileName,
 						Start:    hcl.Pos{Line: 5, Column: 3},
-						End:      hcl.Pos{Line: 6, Column: 4},
+						End:      hcl.Pos{Line: 7, Column: 4},
 					},
 				},
 			},
@@ -154,6 +161,7 @@ resource "kafka_topic" "topic_with_wrong_compression_type" {
   name               = "topic_with_wrong_compression_type"
   replication_factor = 3
   config = {
+    "cleanup.policy"   = "delete"
     "compression.type" = "gzip"
   }
 }`,
@@ -162,6 +170,7 @@ resource "kafka_topic" "topic_with_wrong_compression_type" {
   name               = "topic_with_wrong_compression_type"
   replication_factor = 3
   config = {
+    "cleanup.policy"   = "delete"
     "compression.type" = "zstd"
   }
 }`,
@@ -171,8 +180,62 @@ resource "kafka_topic" "topic_with_wrong_compression_type" {
 					Message: "the compression.type value must be equal to 'zstd'",
 					Range: hcl.Range{
 						Filename: fileName,
-						Start:    hcl.Pos{Line: 6, Column: 5},
-						End:      hcl.Pos{Line: 6, Column: 23},
+						Start:    hcl.Pos{Line: 7, Column: 5},
+						End:      hcl.Pos{Line: 7, Column: 23},
+					},
+				},
+			},
+		},
+		{
+			name: "missing cleanup policy",
+			input: `
+resource "kafka_topic" "topic_without_cleanup_policy" {
+  name               = "topic_without_cleanup_policy"
+  replication_factor = 3
+  config = {
+    "compression.type" = "zstd"
+  }
+}`,
+			fixed: `
+resource "kafka_topic" "topic_without_cleanup_policy" {
+  name               = "topic_without_cleanup_policy"
+  replication_factor = 3
+  config = {
+    "cleanup.policy"   = "delete"
+    "compression.type" = "zstd"
+  }
+}`,
+			expected: []*helper.Issue{
+				{
+					Rule:    rule,
+					Message: "missing cleanup.policy: using default 'delete'",
+					Range: hcl.Range{
+						Filename: fileName,
+						Start:    hcl.Pos{Line: 5, Column: 3},
+						End:      hcl.Pos{Line: 7, Column: 4},
+					},
+				},
+			},
+		},
+		{
+			name: "invalid cleanup policy value",
+			input: `
+resource "kafka_topic" "topic_with_invalid_cleanup_policy" {
+  name               = "topic_with_invalid_cleanup_policy"
+  replication_factor = 3
+  config = {
+    "cleanup.policy"   = "invalid-value"
+    "compression.type" = "zstd"
+  }
+}`,
+			expected: []*helper.Issue{
+				{
+					Rule:    rule,
+					Message: "invalid cleanup.policy: it must be one of [delete, compact], but currently is 'invalid-value'",
+					Range: hcl.Range{
+						Filename: fileName,
+						Start:    hcl.Pos{Line: 6, Column: 26},
+						End:      hcl.Pos{Line: 6, Column: 41},
 					},
 				},
 			},
@@ -184,6 +247,7 @@ resource "kafka_topic" "good topic" {
   name               = "good_topic"
   replication_factor = 3
   config = {
+    "cleanup.policy"   = "delete"
     "compression.type" = "zstd"
   }
 }`,