From 690e8e17664a899d3d4ecff131b2b3a864942e6a Mon Sep 17 00:00:00 2001
From: cfln123 <c.neto@base2services.com>
Date: Tue, 9 Jul 2024 14:46:05 -0300
Subject: [PATCH] Add Kafka Cluster and Topic monitoring

---
 lib/cfnguardian/models/alarm.rb            | 28 +++++++++
 lib/cfnguardian/resources/kafka_cluster.rb | 68 ++++++++++++++++++++++
 lib/cfnguardian/resources/kafka_topic.rb   | 13 +++++
 3 files changed, 109 insertions(+)
 create mode 100644 lib/cfnguardian/resources/kafka_cluster.rb
 create mode 100644 lib/cfnguardian/resources/kafka_topic.rb

diff --git a/lib/cfnguardian/models/alarm.rb b/lib/cfnguardian/models/alarm.rb
index 0e22d4b..6bed0e0 100644
--- a/lib/cfnguardian/models/alarm.rb
+++ b/lib/cfnguardian/models/alarm.rb
@@ -394,6 +394,34 @@ def initialize(resource,environment)
       end
     end
 
+    class KafkaClusterAlarm < BaseAlarm
+      def initialize(resource)
+        super(resource)
+        @group = 'KafkaCluster'
+        @namespace = 'AWS/Kafka'
+        @dimensions = { ClusterName: resource['Id'] }
+        @statistic = 'Average'
+        @evaluation_periods = 1
+        @datapoints_to_alarm = 1
+        @period = 300
+        @treat_missing_data = 'breaching'
+      end
+    end
+
+    class KafkaTopicAlarm < BaseAlarm
+      def initialize(resource)
+        super(resource)
+        @group = 'KafkaTopic'
+        @namespace = 'AWS/Kafka'
+        @dimensions = { ClusterName: resource['ClusterName'], Topic: resource['Id'] }
+        @statistic = 'Average'
+        @evaluation_periods = 1
+        @datapoints_to_alarm = 1
+        @period = 300
+        @treat_missing_data = 'breaching'
+      end
+    end
+
     class LambdaAlarm < BaseAlarm
       def initialize(resource)
         super(resource)
diff --git a/lib/cfnguardian/resources/kafka_cluster.rb b/lib/cfnguardian/resources/kafka_cluster.rb
new file mode 100644
index 0000000..0d0df6e
--- /dev/null
+++ b/lib/cfnguardian/resources/kafka_cluster.rb
@@ -0,0 +1,68 @@
+module CfnGuardian::Resource
+  class KafkaCluster < Base
+
+    def default_alarms    
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'CPUUserCritical'
+      alarm.metric_name = 'CPUUser'
+      alarm.threshold = 80
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'CPUUserWarning'
+      alarm.metric_name = 'CPUUser'
+      alarm.threshold = 50
+      alarm.alarm_action = 'Warning'
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'KafkaDataLogsDiskUsedCritical'
+      alarm.metric_name = 'KafkaDataLogsDiskUsed'
+      alarm.threshold = 85
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'KafkaDataLogsDiskUsedWarning'
+      alarm.metric_name = 'KafkaDataLogsDiskUsed'
+      alarm.threshold = 70
+      alarm.alarm_action = 'Warning'
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'BurstBalance'
+      alarm.metric_name = 'BurstBalance'
+      alarm.threshold = 1
+      alarm.comparison_operator = 'LessThanThreshold'
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'MemoryFreeCritical'
+      alarm.metric_name = 'MemoryFree'
+      alarm.threshold = 10
+      alarm.comparison_operator = 'LessThanThreshold'
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'MemoryFreeWarning'
+      alarm.metric_name = 'MemoryFree'
+      alarm.threshold = 50
+      alarm.alarm_action = 'Warning'
+      alarm.comparison_operator = 'LessThanThreshold'
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'NetworkRxErrorsCritical'
+      alarm.metric_name = 'NetworkRxErrors'
+      alarm.threshold = 10
+      @alarms.push(alarm)
+
+      alarm = CfnGuardian::Models::KafkaClusterAlarm.new(@resource)
+      alarm.name = 'NetworkRxErrorsWarning'
+      alarm.metric_name = 'NetworkRxErrors'
+      alarm.threshold = 5
+      alarm.alarm_action = 'Warning'
+      @alarms.push(alarm)
+    end
+    
+  end
+end
diff --git a/lib/cfnguardian/resources/kafka_topic.rb b/lib/cfnguardian/resources/kafka_topic.rb
new file mode 100644
index 0000000..1911331
--- /dev/null
+++ b/lib/cfnguardian/resources/kafka_topic.rb
@@ -0,0 +1,13 @@
+module CfnGuardian::Resource
+  class KafkaTopic < Base
+
+    def default_alarms    
+      alarm = CfnGuardian::Models::KafkaTopicAlarm.new(@resource)
+      alarm.name = 'MessagesInPerSec'
+      alarm.metric_name = 'MessagesInPerSec'
+      alarm.threshold = 5
+      alarm.comparison_operator = 'LessThanThreshold'
+      @alarms.push(alarm)
+    end
+  end
+end