diff --git a/README.md b/README.md
index fb5e81c..b57dde1 100644
--- a/README.md
+++ b/README.md
@@ -1 +1,27 @@
-# terraform-datadog-kafka
\ No newline at end of file
+
+[//]: # (This file is generated. Do not edit)
+
+# kafka
+
+TOC:
+<!--ts-->
+   * [kafka](#kafka)
+      * [Module Variables](#module-variables)
+
+<!-- Added by: sjuuljanssen, at: za 13 mrt 2021 15:51:21 CET -->
+
+<!--te-->
+
+## Module Variables
+
+| variable             | default  | required | description  |
+|----------------------|----------|----------|--------------|
+| env                  |          | Yes      |              |
+| alert_env            |          | Yes      |              |
+| service              | Kafka    | No       |              |
+| notification_channel |          | Yes      |              |
+| additional_tags      | []       | No       |              |
+| filter_str           |          | Yes      |              |
+| is_hosted_service    | False    | No       |              |
+
+
diff --git a/bytesin-high-variables.tf b/bytesin-high-variables.tf
new file mode 100644
index 0000000..e1b07a3
--- /dev/null
+++ b/bytesin-high-variables.tf
@@ -0,0 +1,57 @@
+variable "bytesin_high_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "bytesin_high_warning" {
+  type    = number
+  default = 2500000
+}
+
+variable "bytesin_high_critical" {
+  type    = number
+  default = 5000000
+}
+
+variable "bytesin_high_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "bytesin_high_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "bytesin_high_note" {
+  type    = string
+  default = ""
+}
+
+variable "bytesin_high_docs" {
+  type    = string
+  default = <<EOFF
+  NOTE: This is based on a baseline and might need adjusting further down the road. 
+                                        
+  Generally, disk throughput tends to be the main bottleneck in Kafka performance. However, that’s not to say that the network is never a bottleneck. Network throughput can affect Kafka’s performance if you are sending messages across data centers, if your topics have a large number of consumers, or if your replicas are catching up to their leaders. Tracking network throughput on your brokers gives you more information as to where potential bottlenecks may lie, and can inform decisions like whether or not you should enable end-to-end compression of your messages.
+
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "bytesin_high_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "bytesin_high_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "bytesin_high_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/bytesin-high.tf b/bytesin-high.tf
new file mode 100644
index 0000000..4895700
--- /dev/null
+++ b/bytesin-high.tf
@@ -0,0 +1,29 @@
+locals {
+  bytesin_high_filter               = var.bytesin_high_filter_override != "" ? var.bytesin_high_filter_override : var.filter_str
+  bytesin_high_notification_channel = var.bytesin_high_alerting_enabled ? var.notification_channel : ""
+}
+
+module "bytesin_high" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "BytesIn unusually high"
+  query            = "avg(${var.bytesin_high_evaluation_period}):avg:kafka.net.bytes_in.rate{${local.bytesin_high_filter}} by {host} > ${var.bytesin_high_critical}"
+  alert_message    = "The Plan specified {{threshold}} APM Analyzed Spans. The current estimate ({{value}}) exceeds the plan"
+  recovery_message = "Analyzed APM Spans have recovered"
+
+  # monitor level vars
+  enabled            = var.bytesin_high_enabled
+  alerting_enabled   = var.bytesin_high_alerting_enabled
+  critical_threshold = var.bytesin_high_critical
+  warning_threshold  = var.bytesin_high_warning
+  priority           = var.bytesin_high_priority
+  severity           = var.bytesin_high_severity
+  docs               = var.bytesin_high_docs
+  note               = var.bytesin_high_note
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/bytesout_high-variables.tf b/bytesout_high-variables.tf
new file mode 100644
index 0000000..5733cf7
--- /dev/null
+++ b/bytesout_high-variables.tf
@@ -0,0 +1,57 @@
+variable "bytesout_high_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "bytesout_high_warning" {
+  type    = number
+  default = 2500000
+}
+
+variable "bytesout_high_critical" {
+  type    = number
+  default = 5000000
+}
+
+variable "bytesout_high_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "bytesout_high_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "bytesout_high_note" {
+  type    = string
+  default = ""
+}
+
+variable "bytesout_high_docs" {
+  type    = string
+  default = <<EOFF
+  NOTE: This is based on a baseline and might need adjusting further down the road.
+
+  Generally, disk throughput tends to be the main bottleneck in Kafka performance. However, that’s not to say that the network is never a bottleneck. Network throughput can affect Kafka’s performance if you are sending messages across data centers, if your topics have a large number of consumers, or if your replicas are catching up to their leaders. Tracking network throughput on your brokers gives you more information as to where potential bottlenecks may lie, and can inform decisions like whether or not you should enable end-to-end compression of your messages.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "bytesout_high_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "bytesout_high_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "bytesout_high_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/bytesout_high.tf b/bytesout_high.tf
new file mode 100644
index 0000000..1eb88cf
--- /dev/null
+++ b/bytesout_high.tf
@@ -0,0 +1,29 @@
+locals {
+  bytesout_high_filter               = var.bytesout_high_filter_override != "" ? var.bytesout_high_filter_override : var.filter_str
+  bytesout_high_notification_channel = var.bytesout_high_alerting_enabled ? var.notification_channel : ""
+}
+
+module "bytesout_high" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "BytesOut unusually high"
+  query            = "avg(${var.bytesout_high_evaluation_period}):avg:kafka.net.bytes_out.rate{${local.bytesout_high_filter}} by {host} > ${var.bytesout_high_critical}"
+  alert_message    = "Kafka is sending an unusual high network traffic"
+  recovery_message = "Analyzed APM Spans have recovered"
+
+  # monitor level vars
+  enabled            = var.bytesout_high_enabled
+  alerting_enabled   = var.bytesout_high_alerting_enabled
+  critical_threshold = var.bytesout_high_critical
+  warning_threshold  = var.bytesout_high_warning
+  priority           = var.bytesout_high_priority
+  severity           = var.bytesout_high_severity
+  docs               = var.bytesout_high_docs
+  note               = var.bytesout_high_note
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/fetch_purgatory_size-variables.tf b/fetch_purgatory_size-variables.tf
new file mode 100644
index 0000000..c799fdb
--- /dev/null
+++ b/fetch_purgatory_size-variables.tf
@@ -0,0 +1,58 @@
+variable "fetch_purgatory_size_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "fetch_purgatory_size_warning" {
+  type    = number
+  default = 70000
+}
+
+variable "fetch_purgatory_size_critical" {
+  type    = number
+  default = 100000
+}
+
+variable "fetch_purgatory_size_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "fetch_purgatory_size_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "fetch_purgatory_size_note" {
+  type    = string
+  default = ""
+}
+
+variable "fetch_purgatory_size_docs" {
+  type    = string
+  default = <<EOFF
+  The request purgatory serves as a temporary holding pen for produce and fetch requests waiting to be satisfied.
+  Fetch requests are added to purgatory if there is not enough data to fulfill the request (fetch.min.bytes on consumers) until the time specified by fetch.wait.max.ms is reached or enough data becomes available
+
+  Keeping an eye on the size of purgatory is useful to determine the underlying causes of latency. Increases in consumer fetch times, for example, can be easily explained if there is a corresponding increase in the number of fetch requests in purgatory.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "fetch_purgatory_size_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "fetch_purgatory_size_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "fetch_purgatory_size_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/fetch_purgatory_size.tf b/fetch_purgatory_size.tf
new file mode 100644
index 0000000..cf91f1d
--- /dev/null
+++ b/fetch_purgatory_size.tf
@@ -0,0 +1,29 @@
+locals {
+  fetch_purgatory_size_filter               = var.fetch_purgatory_size_filter_override != "" ? var.fetch_purgatory_size_filter_override : var.filter_str
+  fetch_purgatory_size_notification_channel = var.fetch_purgatory_size_alerting_enabled ? var.notification_channel : ""
+}
+
+module "fetch_purgatory_size" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Fetch Purgatory Size"
+  query            = "avg(${var.fetch_purgatory_size_evaluation_period}):max:kafka.request.fetch_request_purgatory.size{${local.fetch_purgatory_size_filter}} by {host} > ${var.fetch_purgatory_size_critical}"
+  alert_message    = "Purgatory size ({{value}}) larger than usual."
+  recovery_message = "Purgatory size ({{values}} recovered"
+
+  # monitor level vars
+  enabled            = var.fetch_purgatory_size_enabled
+  alerting_enabled   = var.fetch_purgatory_size_alerting_enabled
+  critical_threshold = var.fetch_purgatory_size_critical
+  warning_threshold  = var.fetch_purgatory_size_warning
+  priority           = var.fetch_purgatory_size_priority
+  severity           = var.fetch_purgatory_size_severity
+  docs               = var.fetch_purgatory_size_docs
+  note               = var.fetch_purgatory_size_note
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/in_sync_nodes_dropped-variables.tf b/in_sync_nodes_dropped-variables.tf
new file mode 100644
index 0000000..33096d7
--- /dev/null
+++ b/in_sync_nodes_dropped-variables.tf
@@ -0,0 +1,55 @@
+variable "in_sync_nodes_dropped_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "in_sync_nodes_dropped_warning" {
+  type    = number
+  default = 0
+}
+
+variable "in_sync_nodes_dropped_critical" {
+  type    = number
+  default = 0
+}
+
+variable "in_sync_nodes_dropped_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "in_sync_nodes_dropped_severity" {
+  type    = string
+  default = "major"
+}
+
+variable "in_sync_nodes_dropped_note" {
+  type    = string
+  default = ""
+}
+
+variable "in_sync_nodes_dropped_docs" {
+  type    = string
+  default = <<EOFF
+  The number of in-sync replicas (ISRs) for a particular partition should remain fairly static, except when you are expanding your broker cluster or removing partitions. In order to maintain high availability, a healthy Kafka cluster requires a minimum number of ISRs for failover. A replica could be removed from the ISR pool if it has not contacted the leader for some time (configurable with the replica.socket.timeout.ms parameter). You should investigate any flapping in the values of these metrics, and any increase in IsrShrinksPerSec without a corresponding increase in IsrExpandsPerSec shortly thereafter.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "in_sync_nodes_dropped_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "in_sync_nodes_dropped_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "in_sync_nodes_dropped_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/in_sync_nodes_dropped.tf b/in_sync_nodes_dropped.tf
new file mode 100644
index 0000000..01e323b
--- /dev/null
+++ b/in_sync_nodes_dropped.tf
@@ -0,0 +1,29 @@
+locals {
+  in_sync_nodes_dropped_filter               = var.in_sync_nodes_dropped_filter_override != "" ? var.in_sync_nodes_dropped_filter_override : var.filter_str
+  in_sync_nodes_dropped_notification_channel = var.in_sync_nodes_dropped_alerting_enabled ? var.notification_channel : ""
+}
+
+module "in_sync_nodes_dropped" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "In Sync Nodes dropped"
+  query            = "avg(${var.in_sync_nodes_dropped_evaluation_period}):max:kafka.replication.isr_shrinks.rate{${local.in_sync_nodes_dropped_filter}} by {aiven-service} - max:kafka.replication.isr_expands.rate{${local.in_sync_nodes_dropped_filter}} by {aiven-service} > ${var.in_sync_nodes_dropped_critical}"
+  alert_message    = "The number of in Sync Nodes dropped compared to the number of In Sync Nodes that were added"
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.in_sync_nodes_dropped_enabled
+  alerting_enabled   = var.in_sync_nodes_dropped_alerting_enabled
+  critical_threshold = var.in_sync_nodes_dropped_critical
+  # warning_threshold  = var.in_sync_nodes_dropped_warning
+  priority           = var.in_sync_nodes_dropped_priority
+  severity           = var.in_sync_nodes_dropped_severity
+  docs               = var.in_sync_nodes_dropped_docs
+  note               = var.in_sync_nodes_dropped_note
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/leader_election_occurring-variables.tf b/leader_election_occurring-variables.tf
new file mode 100644
index 0000000..54ac59a
--- /dev/null
+++ b/leader_election_occurring-variables.tf
@@ -0,0 +1,60 @@
+variable "leader_election_occurring_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "leader_election_occurring_warning" {
+  type    = number
+  default = 0
+}
+
+variable "leader_election_occurring_critical" {
+  type    = number
+  default = 0
+}
+
+variable "leader_election_occurring_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "leader_election_occurring_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "leader_election_occurring_note" {
+  type    = string
+  default = ""
+}
+
+variable "leader_election_occurring_docs" {
+  type    = string
+  default = <<EOFF
+  When a partition leader dies, an election for a new leader is triggered. A partition leader is considered “dead” if it fails to maintain its session with ZooKeeper. Unlike ZooKeeper’s Zab, Kafka does not employ a majority-consensus algorithm for leadership election. Instead, Kafka’s quorum is composed of the set of all in-sync replicas (ISRs) for a particular partition. Replicas are considered in-sync if they are caught-up to the leader, which means that any replica in the ISR can be promoted to the leader. 
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "leader_election_occurring_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "leader_election_occurring_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "leader_election_occurring_require_full_window" {
+  type = bool
+  default = false
+}
+
+variable "leader_election_occurring_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/leader_election_occurring.tf b/leader_election_occurring.tf
new file mode 100644
index 0000000..fe4a8b6
--- /dev/null
+++ b/leader_election_occurring.tf
@@ -0,0 +1,30 @@
+locals {
+  leader_election_occurring_filter               = var.leader_election_occurring_filter_override != "" ? var.leader_election_occurring_filter_override : var.filter_str
+  leader_election_occurring_notification_channel = var.leader_election_occurring_alerting_enabled ? var.notification_channel : ""
+}
+
+module "leader_election_occurring" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Leader Election occurring"
+  query            = "max(${var.leader_election_occurring_evaluation_period}):avg:kafka.replication.leader_elections.rate{${local.leader_election_occurring_filter}} by {aiven-service} > ${var.leader_election_occurring_critical}"
+  alert_message    = "Kafka is electing leaders"
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.leader_election_occurring_enabled
+  alerting_enabled   = var.leader_election_occurring_alerting_enabled
+  critical_threshold = var.leader_election_occurring_critical
+  # warning_threshold  = var.leader_election_occurring_warning
+  priority           = var.leader_election_occurring_priority
+  severity           = var.leader_election_occurring_severity
+  docs               = var.leader_election_occurring_docs
+  note               = var.leader_election_occurring_note
+  require_full_window = var.leader_election_occurring_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/main.tf b/main.tf
new file mode 100644
index 0000000..e382714
--- /dev/null
+++ b/main.tf
@@ -0,0 +1,32 @@
+module "system" {
+  source = "git@github.com:kabisa/terraform-datadog-system.git?ref=0.7"
+
+  locked               = var.locked
+  additional_tags      = var.additional_tags
+  alert_env            = var.alert_env
+  env                  = var.env
+  filter_str           = var.filter_str
+  notification_channel = var.notification_channel
+  service              = var.service
+  bytes_sent_docs      = "If you are monitoring Kafka’s bytes in/out metric, you are getting Kafka’s side of the story. To get a full picture of network usage on your host, you need to monitor host-level network throughput, especially if your Kafka brokers are hosting other network services. High network usage could be a symptom of degraded performance—if you are seeing high network use, correlating with TCP retransmissions and dropped packet errors could help in determining if the performance issue is network-related.\nhttps://www.datadoghq.com/blog/monitoring-kafka-performance-metrics"
+  bytes_received_docs  = "If you are monitoring Kafka’s bytes in/out metric, you are getting Kafka’s side of the story. To get a full picture of network usage on your host, you need to monitor host-level network throughput, especially if your Kafka brokers are hosting other network services. High network usage could be a symptom of degraded performance—if you are seeing high network use, correlating with TCP retransmissions and dropped packet errors could help in determining if the performance issue is network-related.\nhttps://www.datadoghq.com/blog/monitoring-kafka-performance-metrics"
+
+  # kafka uses nearly all memory
+  memory_free_percent_enabled = false
+  disk_free_docs              = "Because Kafka persists all data to disk, it is necessary to monitor the amount of free disk space available to Kafka. Kafka will fail should its disk become full, so it’s very important that you keep track of disk growth over time, and set alerts to inform administrators at an appropriate amount of time before disk space is all but used up.\nhttps://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/"
+  memory_free_bytes_enabled   = true
+  memory_free_bytes_critical  = 30000000
+  memory_free_bytes_warning   = 50000000
+  # memory is a saw tooth chart (garbage collection cycles)
+  memory_free_bytes_evaluation_period = "last_15m"
+
+  # hosted service
+  # we don't find everything as important as it's the job of the service
+  # we do want to know what's going on so we downplay the severity
+  dd_agent_enabled           = !var.is_hosted_service
+  dd_agent_data_enabled      = !var.is_hosted_service
+  reboot_enabled             = !var.is_hosted_service
+  disk_free_severity         = var.is_hosted_service ? "minor" : "major"
+  disk_io_wait_severity      = var.is_hosted_service ? "minor" : "major"
+  memory_free_bytes_severity = var.is_hosted_service ? "minor" : "major"
+}
\ No newline at end of file
diff --git a/multiple_active_controllers-variables.tf b/multiple_active_controllers-variables.tf
new file mode 100644
index 0000000..6d4ebff
--- /dev/null
+++ b/multiple_active_controllers-variables.tf
@@ -0,0 +1,60 @@
+variable "multiple_active_controllers_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "multiple_active_controllers_warning" {
+  type    = number
+  default = 0
+}
+
+variable "multiple_active_controllers_critical" {
+  type    = number
+  default = 1
+}
+
+variable "multiple_active_controllers_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "multiple_active_controllers_severity" {
+  type    = string
+  default = "major"
+}
+
+variable "multiple_active_controllers_note" {
+  type    = string
+  default = ""
+}
+
+variable "multiple_active_controllers_docs" {
+  type    = string
+  default = <<EOFF
+  The first node to boot in a Kafka cluster automatically becomes the controller, and there can be only one. The controller in a Kafka cluster is responsible for maintaining the list of partition leaders, and coordinating leadership transitions (in the event a partition leader becomes unavailable). If it becomes necessary to replace the controller, ZooKeeper chooses a new controller randomly from the pool of brokers. The sum of ActiveControllerCount across all of your brokers should always equal one, and you should alert on any other value that lasts for longer than one second.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "multiple_active_controllers_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "multiple_active_controllers_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "multiple_active_controllers_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "multiple_active_controllers_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/multiple_active_controllers.tf b/multiple_active_controllers.tf
new file mode 100644
index 0000000..a349f95
--- /dev/null
+++ b/multiple_active_controllers.tf
@@ -0,0 +1,30 @@
+locals {
+  multiple_active_controllers_filter               = var.multiple_active_controllers_filter_override != "" ? var.multiple_active_controllers_filter_override : var.filter_str
+  multiple_active_controllers_notification_channel = var.multiple_active_controllers_alerting_enabled ? var.notification_channel : ""
+}
+
+module "multiple_active_controllers" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Multiple Active controllers"
+  query            = "avg(${var.multiple_active_controllers_evaluation_period}):max:kafka.replication.active_controller_count{${local.multiple_active_controllers_filter}} by {aiven-project} > ${var.multiple_active_controllers_critical}"
+  alert_message    = "Multiple active Kafka controllers. There should be one and only one."
+  recovery_message = "Kafka controller count recovered"
+
+  # monitor level vars
+  enabled            = var.multiple_active_controllers_enabled
+  alerting_enabled   = var.multiple_active_controllers_alerting_enabled
+  critical_threshold = var.multiple_active_controllers_critical
+  # warning_threshold  = var.multiple_active_controllers_warning
+  priority           = var.multiple_active_controllers_priority
+  severity           = var.multiple_active_controllers_severity
+  docs               = var.multiple_active_controllers_docs
+  note               = var.multiple_active_controllers_note
+  require_full_window = var.multiple_active_controllers_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/no_active_controllers-variables.tf b/no_active_controllers-variables.tf
new file mode 100644
index 0000000..d8e3cf1
--- /dev/null
+++ b/no_active_controllers-variables.tf
@@ -0,0 +1,60 @@
+variable "no_active_controllers_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "no_active_controllers_warning" {
+  type    = number
+  default = 0
+}
+
+variable "no_active_controllers_critical" {
+  type    = number
+  default = 1
+}
+
+variable "no_active_controllers_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "no_active_controllers_severity" {
+  type    = string
+  default = "major"
+}
+
+variable "no_active_controllers_note" {
+  type    = string
+  default = ""
+}
+
+variable "no_active_controllers_docs" {
+  type    = string
+  default = <<EOFF
+  The first node to boot in a Kafka cluster automatically becomes the controller, and there can be only one. The controller in a Kafka cluster is responsible for maintaining the list of partition leaders, and coordinating leadership transitions (in the event a partition leader becomes unavailable). If it becomes necessary to replace the controller, ZooKeeper chooses a new controller randomly from the pool of brokers. The sum of ActiveControllerCount across all of your brokers should always equal one, and you should alert on any other value that lasts for longer than one second.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "no_active_controllers_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "no_active_controllers_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "no_active_controllers_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "no_active_controllers_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/no_active_controllers.tf b/no_active_controllers.tf
new file mode 100644
index 0000000..facb42f
--- /dev/null
+++ b/no_active_controllers.tf
@@ -0,0 +1,30 @@
+locals {
+  no_active_controllers_filter               = var.no_active_controllers_filter_override != "" ? var.no_active_controllers_filter_override : var.filter_str
+  no_active_controllers_notification_channel = var.no_active_controllers_alerting_enabled ? var.notification_channel : ""
+}
+
+module "no_active_controllers" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "No Active controllers"
+  query            = "avg(${var.no_active_controllers_evaluation_period}):max:kafka.replication.active_controller_count{${local.no_active_controllers_filter}} by {aiven-project} < ${var.no_active_controllers_critical}"
+  alert_message    = "No active Kafka controllers. There should be one and only one."
+  recovery_message = "Kafka controller count recovered"
+
+  # monitor level vars
+  enabled            = var.no_active_controllers_enabled
+  alerting_enabled   = var.no_active_controllers_alerting_enabled
+  critical_threshold = var.no_active_controllers_critical
+  # warning_threshold  = var.no_active_controllers_warning
+  priority           = var.no_active_controllers_priority
+  severity           = var.no_active_controllers_severity
+  docs               = var.no_active_controllers_docs
+  note               = var.no_active_controllers_note
+  require_full_window = var.no_active_controllers_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/offline_partitions-variables.tf b/offline_partitions-variables.tf
new file mode 100644
index 0000000..8ce6c08
--- /dev/null
+++ b/offline_partitions-variables.tf
@@ -0,0 +1,60 @@
+variable "offline_partitions_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "offline_partitions_warning" {
+  type    = number
+  default = 0
+}
+
+variable "offline_partitions_critical" {
+  type    = number
+  default = 0
+}
+
+variable "offline_partitions_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "offline_partitions_severity" {
+  type    = string
+  default = "major"
+}
+
+variable "offline_partitions_note" {
+  type    = string
+  default = ""
+}
+
+variable "offline_partitions_docs" {
+  type    = string
+  default = <<EOFF
+  This metric reports the number of partitions without an active leader. Because all read and write operations are only performed on partition leaders, you should alert on a non-zero value for this metric to prevent service interruptions. Any partition without an active leader will be completely inaccessible, and both consumers and producers of that partition will be blocked until a leader becomes available.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "offline_partitions_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "offline_partitions_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "offline_partitions_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "offline_partitions_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/offline_partitions.tf b/offline_partitions.tf
new file mode 100644
index 0000000..327188d
--- /dev/null
+++ b/offline_partitions.tf
@@ -0,0 +1,30 @@
+locals {
+  offline_partitions_filter               = var.offline_partitions_filter_override != "" ? var.offline_partitions_filter_override : var.filter_str
+  offline_partitions_notification_channel = var.offline_partitions_alerting_enabled ? var.notification_channel : ""
+}
+
+module "offline_partitions" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Offline Partitions"
+  query            = "avg(${var.offline_partitions_evaluation_period}):max:kafka.replication.offline_partitions_count{${local.offline_partitions_filter}} > ${var.offline_partitions_critical}"
+  alert_message    = "Offline partitions detected {{value}}"
+  recovery_message = "Kafka controller count recovered"
+
+  # monitor level vars
+  enabled            = var.offline_partitions_enabled
+  alerting_enabled   = var.offline_partitions_alerting_enabled
+  critical_threshold = var.offline_partitions_critical
+  # warning_threshold  = var.offline_partitions_warning
+  priority           = var.offline_partitions_priority
+  severity           = var.offline_partitions_severity
+  docs               = var.offline_partitions_docs
+  note               = var.offline_partitions_note
+  require_full_window = var.offline_partitions_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/produce_purgatory_size-variables.tf b/produce_purgatory_size-variables.tf
new file mode 100644
index 0000000..0a927f5
--- /dev/null
+++ b/produce_purgatory_size-variables.tf
@@ -0,0 +1,63 @@
+variable "produce_purgatory_size_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "produce_purgatory_size_warning" {
+  type    = number
+  default = 100
+}
+
+variable "produce_purgatory_size_critical" {
+  type    = number
+  default = 200
+}
+
+variable "produce_purgatory_size_evaluation_period" {
+  type    = string
+  default = "last_15m"
+}
+
+variable "produce_purgatory_size_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "produce_purgatory_size_note" {
+  type    = string
+  default = ""
+}
+
+variable "produce_purgatory_size_docs" {
+  type    = string
+  default = <<EOFF
+  The request purgatory serves as a temporary holding pen for produce and fetch requests waiting to be satisfied.
+  If request.required.acks=-1, all produce requests will end up in purgatory until the partition leader receives an acknowledgment from all followers.
+
+  Keeping an eye on the size of purgatory is useful to determine the underlying causes of latency. Increases in consumer fetch times, for example, can be easily explained if there is a corresponding increase in the number of fetch requests in purgatory.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "produce_purgatory_size_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "produce_purgatory_size_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "produce_purgatory_size_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "produce_purgatory_size_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/produce_purgatory_size.tf b/produce_purgatory_size.tf
new file mode 100644
index 0000000..1763ee8
--- /dev/null
+++ b/produce_purgatory_size.tf
@@ -0,0 +1,30 @@
+locals {
+  produce_purgatory_size_filter               = var.produce_purgatory_size_filter_override != "" ? var.produce_purgatory_size_filter_override : var.filter_str
+  produce_purgatory_size_notification_channel = var.produce_purgatory_size_alerting_enabled ? var.notification_channel : ""
+}
+
+module "produce_purgatory_size" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Produce Purgatory Size"
+  query            = "avg(${var.produce_purgatory_size_evaluation_period}):max:kafka.request.producer_request_purgatory.size{${local.produce_purgatory_size_filter}} > ${var.produce_purgatory_size_critical}"
+  alert_message    = "Purgatory size ({{value}}) larger than usual."
+  recovery_message = "Purgatory size ({{values}} recovered"
+
+  # monitor level vars
+  enabled            = var.produce_purgatory_size_enabled
+  alerting_enabled   = var.produce_purgatory_size_alerting_enabled
+  critical_threshold = var.produce_purgatory_size_critical
+  warning_threshold  = var.produce_purgatory_size_warning
+  priority           = var.produce_purgatory_size_priority
+  severity           = var.produce_purgatory_size_severity
+  docs               = var.produce_purgatory_size_docs
+  note               = var.produce_purgatory_size_note
+  require_full_window = var.produce_purgatory_size_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/provider.tf b/provider.tf
new file mode 100644
index 0000000..faaf377
--- /dev/null
+++ b/provider.tf
@@ -0,0 +1,8 @@
+terraform {
+  required_providers {
+    datadog = {
+      source  = "DataDog/datadog"
+      version = ">= 2.21"
+    }
+  }
+}
\ No newline at end of file
diff --git a/unclean_leader_election-variables.tf b/unclean_leader_election-variables.tf
new file mode 100644
index 0000000..ca278df
--- /dev/null
+++ b/unclean_leader_election-variables.tf
@@ -0,0 +1,60 @@
+variable "unclean_leader_election_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unclean_leader_election_warning" {
+  type    = number
+  default = 100
+}
+
+variable "unclean_leader_election_critical" {
+  type    = number
+  default = 0
+}
+
+variable "unclean_leader_election_evaluation_period" {
+  type    = string
+  default = "last_5m"
+}
+
+variable "unclean_leader_election_severity" {
+  type    = string
+  default = "major"
+}
+
+variable "unclean_leader_election_note" {
+  type    = string
+  default = ""
+}
+
+variable "unclean_leader_election_docs" {
+  type    = string
+  default = <<EOFF
+  Unclean leader elections occur when there is no qualified partition leader among Kafka brokers. Normally, when a broker that is the leader for a partition goes offline, a new leader is elected from the set of ISRs for the partition. Unclean leader election is disabled by default in Kafka version 0.11 and newer, meaning that a partition is taken offline if it does not have any ISRs to elect as the new leader. If Kafka is configured to allow an unclean leader election, a leader is chosen from the out-of-sync replicas, and any messages that were not synced prior to the loss of the former leader are lost forever. Essentially, unclean leader elections sacrifice consistency for availability. You should alert on this metric, as it signals data loss.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "unclean_leader_election_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "unclean_leader_election_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unclean_leader_election_require_full_window" {
+  type = bool
+  default = false
+}
+
+variable "unclean_leader_election_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unclean_leader_election.tf b/unclean_leader_election.tf
new file mode 100644
index 0000000..633fcb9
--- /dev/null
+++ b/unclean_leader_election.tf
@@ -0,0 +1,30 @@
+locals {
+  unclean_leader_election_filter               = var.unclean_leader_election_filter_override != "" ? var.unclean_leader_election_filter_override : var.filter_str
+  unclean_leader_election_notification_channel = var.unclean_leader_election_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unclean_leader_election" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unclean Leader Election"
+  query            = "avg(${var.unclean_leader_election_evaluation_period}):max:kafka.replication.unclean_leader_elections.rate{${local.unclean_leader_election_filter}} by {aiven-project} > ${var.unclean_leader_election_critical}"
+  alert_message    = "Unclean Leader Election Detected"
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unclean_leader_election_enabled
+  alerting_enabled   = var.unclean_leader_election_alerting_enabled
+  critical_threshold = var.unclean_leader_election_critical
+  # warning_threshold  = var.unclean_leader_election_warning
+  priority           = var.unclean_leader_election_priority
+  severity           = var.unclean_leader_election_severity
+  docs               = var.unclean_leader_election_docs
+  note               = var.unclean_leader_election_note
+  require_full_window = var.unclean_leader_election_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/under_replicated_partitions-variables.tf b/under_replicated_partitions-variables.tf
new file mode 100644
index 0000000..45a63fa
--- /dev/null
+++ b/under_replicated_partitions-variables.tf
@@ -0,0 +1,60 @@
+variable "under_replicated_partitions_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "under_replicated_partitions_warning" {
+  type    = number
+  default = 100
+}
+
+variable "under_replicated_partitions_critical" {
+  type    = number
+  default = 0
+}
+
+variable "under_replicated_partitions_evaluation_period" {
+  type    = string
+  default = "last_15m"
+}
+
+variable "under_replicated_partitions_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "under_replicated_partitions_note" {
+  type    = string
+  default = ""
+}
+
+variable "under_replicated_partitions_docs" {
+  type    = string
+  default = <<EOFF
+  If a broker becomes unavailable, the value of UnderReplicatedPartitions will increase sharply. Since Kafka’s high-availability guarantees cannot be met without replication, investigation is certainly warranted should this metric value exceed zero for extended time periods.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/
+  EOFF
+}
+
+variable "under_replicated_partitions_filter_override" {
+  type    = string
+  default = ""
+}
+
+variable "under_replicated_partitions_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "under_replicated_partitions_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "under_replicated_partitions_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/under_replicated_partitions.tf b/under_replicated_partitions.tf
new file mode 100644
index 0000000..e04d9df
--- /dev/null
+++ b/under_replicated_partitions.tf
@@ -0,0 +1,30 @@
+locals {
+  under_replicated_partitions_filter               = var.under_replicated_partitions_filter_override != "" ? var.under_replicated_partitions_filter_override : var.filter_str
+  under_replicated_partitions_notification_channel = var.under_replicated_partitions_alerting_enabled ? var.notification_channel : ""
+}
+
+module "under_replicated_partitions" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "UnderReplicated Partitions"
+  query            = "avg(${var.under_replicated_partitions_evaluation_period}):avg:kafka.replication.under_replicated_partitions{${local.under_replicated_partitions_filter}} by {aiven-service} > ${var.under_replicated_partitions_critical}"
+  alert_message    = "Kafka cluster is not fully replicated."
+  recovery_message = "Kafka cluster replication recovered"
+
+  # monitor level vars
+  enabled            = var.under_replicated_partitions_enabled
+  alerting_enabled   = var.under_replicated_partitions_alerting_enabled
+  critical_threshold = var.under_replicated_partitions_critical
+  # warning_threshold  = var.under_replicated_partitions_warning
+  priority           = var.under_replicated_partitions_priority
+  severity           = var.under_replicated_partitions_severity
+  docs               = var.under_replicated_partitions_docs
+  note               = var.under_replicated_partitions_note
+  require_full_window = var.under_replicated_partitions_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/unusual_consumer_fetch_time-variables.tf b/unusual_consumer_fetch_time-variables.tf
new file mode 100644
index 0000000..8628b3c
--- /dev/null
+++ b/unusual_consumer_fetch_time-variables.tf
@@ -0,0 +1,63 @@
+variable "unusual_consumer_fetch_time_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_consumer_fetch_time_warning" {
+  type    = number
+  default = 800
+}
+
+variable "unusual_consumer_fetch_time_critical" {
+  type    = number
+  default = 1000
+}
+
+variable "unusual_consumer_fetch_time_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "unusual_consumer_fetch_time_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "unusual_consumer_fetch_time_note" {
+  type    = string
+  default = ""
+}
+
+variable "unusual_consumer_fetch_time_docs" {
+  type    = string
+  default = <<EOFF
+  produce: requests from producers to send data
+  fetch-consumer: requests from consumers to get new data
+  fetch-follower: requests from brokers that are the followers of a partition to get new data
+
+  Under normal conditions, this value should be fairly static, with minimal fluctuations. If you are seeing anomalous behavior, you may want to check the individual queue, local, remote and response values to pinpoint the exact request segment that is causing the slowdown.
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/#metric-to-watch-totaltimems
+  EOFF
+}
+
+variable "unusual_consumer_fetch_time_filter_override" {
+  type    = string
+  default = "aiven-service:kafka-prod"
+}
+
+variable "unusual_consumer_fetch_time_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_consumer_fetch_time_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "unusual_consumer_fetch_time_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unusual_consumer_fetch_time.tf b/unusual_consumer_fetch_time.tf
new file mode 100644
index 0000000..d87ef3a
--- /dev/null
+++ b/unusual_consumer_fetch_time.tf
@@ -0,0 +1,30 @@
+locals {
+  unusual_consumer_fetch_time_filter               = var.unusual_consumer_fetch_time_filter_override != "" ? var.unusual_consumer_fetch_time_filter_override : var.filter_str
+  unusual_consumer_fetch_time_notification_channel = var.unusual_consumer_fetch_time_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unusual_consumer_fetch_time" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unusual Consumer Fetch Time"
+  query            = "avg(${var.unusual_consumer_fetch_time_evaluation_period}):avg:kafka.request.fetch_consumer.time.avg{${local.unusual_consumer_fetch_time_filter}} > ${var.unusual_consumer_fetch_time_critical}"
+  alert_message    = ""
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unusual_consumer_fetch_time_enabled
+  alerting_enabled   = var.unusual_consumer_fetch_time_alerting_enabled
+  critical_threshold = var.unusual_consumer_fetch_time_critical
+  warning_threshold  = var.unusual_consumer_fetch_time_warning
+  priority           = var.unusual_consumer_fetch_time_priority
+  severity           = var.unusual_consumer_fetch_time_severity
+  docs               = var.unusual_consumer_fetch_time_docs
+  note               = var.unusual_consumer_fetch_time_note
+  require_full_window = var.unusual_consumer_fetch_time_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/unusual_fetch_failures-variables.tf b/unusual_fetch_failures-variables.tf
new file mode 100644
index 0000000..0434e52
--- /dev/null
+++ b/unusual_fetch_failures-variables.tf
@@ -0,0 +1,66 @@
+variable "unusual_fetch_failures_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_fetch_failures_warning" {
+  type    = number
+  default = 40
+}
+
+variable "unusual_fetch_failures_critical" {
+  type    = number
+  default = 50
+}
+
+variable "unusual_fetch_failures_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "unusual_fetch_failures_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "unusual_fetch_failures_note" {
+  type    = string
+  default = ""
+}
+
+variable "unusual_fetch_failures_docs" {
+  type    = string
+  default = <<EOFF
+  produce: requests from producers to send data
+  fetch-consumer: requests from consumers to get new data
+  fetch-follower: requests from brokers that are the followers of a partition to get new data
+
+  Under normal conditions, this value should be fairly static, with minimal fluctuations. If you are seeing anomalous behavior, you may want to check the individual queue, local, remote and response values to pinpoint the exact request segment that is causing the slowdown.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/#metric-to-watch-totaltimems
+  
+  This monitor checks if there's an unusual high amount of failures. Which might be indicative of the application not being able to perform its task
+  EOFF
+}
+
+variable "unusual_fetch_failures_filter_override" {
+  type    = string
+  default = "aiven-service:kafka-prod"
+}
+
+variable "unusual_fetch_failures_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_fetch_failures_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "unusual_fetch_failures_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unusual_fetch_failures.tf b/unusual_fetch_failures.tf
new file mode 100644
index 0000000..d87d89d
--- /dev/null
+++ b/unusual_fetch_failures.tf
@@ -0,0 +1,30 @@
+locals {
+  unusual_fetch_failures_filter               = var.unusual_fetch_failures_filter_override != "" ? var.unusual_fetch_failures_filter_override : var.filter_str
+  unusual_fetch_failures_notification_channel = var.unusual_fetch_failures_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unusual_fetch_failures" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unusual Fetch Failures"
+  query            = "avg(${var.unusual_fetch_failures_evaluation_period}):avg:kafka.request.fetch.failed.rate{${local.unusual_fetch_failures_filter}} > ${var.unusual_fetch_failures_critical}"
+  alert_message    = ""
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unusual_fetch_failures_enabled
+  alerting_enabled   = var.unusual_fetch_failures_alerting_enabled
+  critical_threshold = var.unusual_fetch_failures_critical
+  warning_threshold  = var.unusual_fetch_failures_warning
+  priority           = var.unusual_fetch_failures_priority
+  severity           = var.unusual_fetch_failures_severity
+  docs               = var.unusual_fetch_failures_docs
+  note               = var.unusual_fetch_failures_note
+  require_full_window = var.unusual_fetch_failures_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/unusual_follower_fetch_time-variables.tf b/unusual_follower_fetch_time-variables.tf
new file mode 100644
index 0000000..62e9f06
--- /dev/null
+++ b/unusual_follower_fetch_time-variables.tf
@@ -0,0 +1,64 @@
+variable "unusual_follower_fetch_time_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_follower_fetch_time_warning" {
+  type    = number
+  default = 800
+}
+
+variable "unusual_follower_fetch_time_critical" {
+  type    = number
+  default = 1000
+}
+
+variable "unusual_follower_fetch_time_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "unusual_follower_fetch_time_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "unusual_follower_fetch_time_note" {
+  type    = string
+  default = ""
+}
+
+variable "unusual_follower_fetch_time_docs" {
+  type    = string
+  default = <<EOFF
+  produce: requests from producers to send data
+  fetch-consumer: requests from consumers to get new data
+  fetch-follower: requests from brokers that are the followers of a partition to get new data
+  
+  Under normal conditions, this value should be fairly static, with minimal fluctuations. If you are seeing anomalous behavior, you may want to check the individual queue, local, remote and response values to pinpoint the exact request segment that is causing the slowdown.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/#metric-to-watch-totaltimems
+  EOFF
+}
+
+variable "unusual_follower_fetch_time_filter_override" {
+  type    = string
+  default = "aiven-service:kafka-prod"
+}
+
+variable "unusual_follower_fetch_time_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_follower_fetch_time_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "unusual_follower_fetch_time_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unusual_follower_fetch_time.tf b/unusual_follower_fetch_time.tf
new file mode 100644
index 0000000..c4cc42d
--- /dev/null
+++ b/unusual_follower_fetch_time.tf
@@ -0,0 +1,30 @@
+locals {
+  unusual_follower_fetch_time_filter               = var.unusual_follower_fetch_time_filter_override != "" ? var.unusual_follower_fetch_time_filter_override : var.filter_str
+  unusual_follower_fetch_time_notification_channel = var.unusual_follower_fetch_time_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unusual_follower_fetch_time" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unusual Follower Fetch Time"
+  query            = "avg(${var.unusual_follower_fetch_time_evaluation_period}):avg:kafka.request.fetch_follower.time.avg{${local.unusual_follower_fetch_time_filter}} > ${var.unusual_follower_fetch_time_critical}"
+  alert_message    = "The TotalTimeMs metric family measures the total time taken to service a request (be it a produce, fetch-consumer, or fetch-follower request):"
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unusual_follower_fetch_time_enabled
+  alerting_enabled   = var.unusual_follower_fetch_time_alerting_enabled
+  critical_threshold = var.unusual_follower_fetch_time_critical
+  warning_threshold  = var.unusual_follower_fetch_time_warning
+  priority           = var.unusual_follower_fetch_time_priority
+  severity           = var.unusual_follower_fetch_time_severity
+  docs               = var.unusual_follower_fetch_time_docs
+  note               = var.unusual_follower_fetch_time_note
+  require_full_window = var.unusual_follower_fetch_time_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/unusual_produce_failures-variables.tf b/unusual_produce_failures-variables.tf
new file mode 100644
index 0000000..df8de6b
--- /dev/null
+++ b/unusual_produce_failures-variables.tf
@@ -0,0 +1,68 @@
+variable "unusual_produce_failures_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_produce_failures_warning" {
+  type    = number
+  default = 40
+}
+
+variable "unusual_produce_failures_critical" {
+  type    = number
+  default = 50
+}
+
+variable "unusual_produce_failures_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "unusual_produce_failures_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "unusual_produce_failures_note" {
+  type    = string
+  default = ""
+}
+
+variable "unusual_produce_failures_docs" {
+  type    = string
+  default = <<EOFF
+  The TotalTimeMs metric family measures the total time taken to service a request (be it a produce, fetch-consumer, or fetch-follower request):
+  
+  produce: requests from producers to send data
+  fetch-consumer: requests from consumers to get new data
+  fetch-follower: requests from brokers that are the followers of a partition to get new data
+
+  Under normal conditions, this value should be fairly static, with minimal fluctuations. If you are seeing anomalous behavior, you may want to check the individual queue, local, remote and response values to pinpoint the exact request segment that is causing the slowdown.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/#metric-to-watch-totaltimems
+  
+  This monitor checks if there's an unusual high amount of failures. Which might be indicative of the application not being able to perform its task
+  EOFF
+}
+
+variable "unusual_produce_failures_filter_override" {
+  type    = string
+  default = "aiven-service:kafka-prod"
+}
+
+variable "unusual_produce_failures_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_produce_failures_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "unusual_produce_failures_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unusual_produce_failures.tf b/unusual_produce_failures.tf
new file mode 100644
index 0000000..f49a6d2
--- /dev/null
+++ b/unusual_produce_failures.tf
@@ -0,0 +1,30 @@
+locals {
+  unusual_produce_failures_filter               = var.unusual_produce_failures_filter_override != "" ? var.unusual_produce_failures_filter_override : var.filter_str
+  unusual_produce_failures_notification_channel = var.unusual_produce_failures_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unusual_produce_failures" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unusual Produce Failures"
+  query            = "avg(${var.unusual_produce_failures_evaluation_period}):avg:kafka.request.produce.failed.rate{${local.unusual_produce_failures_filter}} > ${var.unusual_produce_failures_critical}"
+  alert_message    = ""
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unusual_produce_failures_enabled
+  alerting_enabled   = var.unusual_produce_failures_alerting_enabled
+  critical_threshold = var.unusual_produce_failures_critical
+  warning_threshold  = var.unusual_produce_failures_warning
+  priority           = var.unusual_produce_failures_priority
+  severity           = var.unusual_produce_failures_severity
+  docs               = var.unusual_produce_failures_docs
+  note               = var.unusual_produce_failures_note
+  require_full_window = var.unusual_produce_failures_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/unusual_produce_time-variables.tf b/unusual_produce_time-variables.tf
new file mode 100644
index 0000000..d176539
--- /dev/null
+++ b/unusual_produce_time-variables.tf
@@ -0,0 +1,68 @@
+variable "unusual_produce_time_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_produce_time_warning" {
+  type    = number
+  default = 10
+}
+
+variable "unusual_produce_time_critical" {
+  type    = number
+  default = 20
+}
+
+variable "unusual_produce_time_evaluation_period" {
+  type    = string
+  default = "last_30m"
+}
+
+variable "unusual_produce_time_severity" {
+  type    = string
+  default = "minor"
+}
+
+variable "unusual_produce_time_note" {
+  type    = string
+  default = ""
+}
+
+variable "unusual_produce_time_docs" {
+  type    = string
+  default = <<EOFF
+  The TotalTimeMs metric family measures the total time taken to service a request (be it a produce, fetch-consumer, or fetch-follower request):
+  
+  produce: requests from producers to send data
+  fetch-consumer: requests from consumers to get new data
+  fetch-follower: requests from brokers that are the followers of a partition to get new data
+
+  Under normal conditions, this value should be fairly static, with minimal fluctuations. If you are seeing anomalous behavior, you may want to check the individual queue, local, remote and response values to pinpoint the exact request segment that is causing the slowdown.
+  
+  https://www.datadoghq.com/blog/monitoring-kafka-performance-metrics/#metric-to-watch-totaltimems
+  
+  This monitor checks if there's an unusual high amount of failures. Which might be indicative of the application not being able to perform its task
+  EOFF
+}
+
+variable "unusual_produce_time_filter_override" {
+  type    = string
+  default = "aiven-service:kafka-prod"
+}
+
+variable "unusual_produce_time_alerting_enabled" {
+  type    = bool
+  default = true
+}
+
+variable "unusual_produce_time_require_full_window" {
+  type = bool
+  default = true
+}
+
+variable "unusual_produce_time_priority" {
+  description = "Number from 1 (high) to 5 (low)."
+
+  type    = number
+  default = null
+}
\ No newline at end of file
diff --git a/unusual_produce_time.tf b/unusual_produce_time.tf
new file mode 100644
index 0000000..51106cf
--- /dev/null
+++ b/unusual_produce_time.tf
@@ -0,0 +1,30 @@
+locals {
+  unusual_produce_time_filter               = var.unusual_produce_time_filter_override != "" ? var.unusual_produce_time_filter_override : var.filter_str
+  unusual_produce_time_notification_channel = var.unusual_produce_time_alerting_enabled ? var.notification_channel : ""
+}
+
+module "unusual_produce_time" {
+  source = "git@github.com:kabisa/terraform-datadog-generic-monitor.git?ref=0.5.2"
+
+  name             = "Unusual Produce Time"
+  query            = "avg(${var.unusual_produce_time_evaluation_period}):avg:kafka.request.produce.time.avg{${local.unusual_produce_time_filter}} > ${var.unusual_produce_time_critical}"
+  alert_message    = ""
+  recovery_message = ""
+
+  # monitor level vars
+  enabled            = var.unusual_produce_time_enabled
+  alerting_enabled   = var.unusual_produce_time_alerting_enabled
+  critical_threshold = var.unusual_produce_time_critical
+  warning_threshold  = var.unusual_produce_time_warning
+  priority           = var.unusual_produce_time_priority
+  severity           = var.unusual_produce_time_severity
+  docs               = var.unusual_produce_time_docs
+  note               = var.unusual_produce_time_note
+  require_full_window = var.unusual_produce_time_require_full_window
+
+  # module level vars
+  env                  = var.alert_env
+  service              = var.service
+  notification_channel = var.notification_channel
+  additional_tags      = var.additional_tags
+}
diff --git a/variables.tf b/variables.tf
new file mode 100644
index 0000000..bd9ca73
--- /dev/null
+++ b/variables.tf
@@ -0,0 +1,35 @@
+variable "env" {
+  type = string
+}
+
+variable "alert_env" {
+  type = string
+}
+
+variable "service" {
+  type    = string
+  default = "Kafka"
+}
+
+variable "notification_channel" {
+  type = string
+}
+
+variable "additional_tags" {
+  type    = list(string)
+  default = []
+}
+
+variable "filter_str" {
+  type = string
+}
+
+variable "is_hosted_service" {
+  type    = bool
+  default = false
+}
+
+variable "locked" {
+  type    = bool
+  default = true
+}