resource "datadog_monitor" "msk_cpu_monitor" {
  count = var.msk_cpu_monitor_enabled ? 1 : 0
  name               = "${var.environment}-${var.cluster_name}-cpu-monitor"
  type               = "query alert"
  message            = "MSK CPU usage is high on Broker #{{broker_id}} in ${var.environment}-${var.cluster_name}. Notify: ${var.notification_endpoints}"
  escalation_message = "Escalation to ${var.escalation_notification_endpoints}"

  query = "avg(last_1h):avg:aws.kafka.cpu_system{${local.cluster_filter}} by {broker_id} + avg:aws.kafka.cpu_user{${local.cluster_filter}} by {broker_id} > ${var.msk_cpu_monitor_critical_threshold}"

  monitor_thresholds {
    warning  = var.msk_cpu_monitor_warning_threshold
    warning_recovery = 50
    critical = var.msk_cpu_monitor_critical_threshold
  }

  evaluation_delay    = 900 # Recommended for AWS metrics
  new_group_delay     = 900
  notify_by           = ["*"]
  include_tags        = false
  notify_no_data      = false
  notify_audit        = false
  renotify_interval   = 60
  timeout_h           = 0
  require_full_window = true
  tags                = local.dashboard_tags        
}

resource "datadog_monitor" "msk_memory_monitor" {
  count = var.msk_memory_monitor_enabled ? 1 : 0
  name               = "${var.environment}-${var.cluster_name}-memory-monitor"
  type               = "query alert"
  message            = "MSK Memory usage is high on Broker #{{broker_id}} in ${var.environment}-${var.cluster_name}. Notify: ${var.notification_endpoints}"
  escalation_message = "Escalation to ${var.escalation_notification_endpoints}"

  query = "avg(last_1h):avg:aws.kafka.heap_memory_after_gc{${local.cluster_filter}} by {broker_id} > ${var.msk_memory_monitor_critical_threshold}"

  monitor_thresholds {
    warning  = var.msk_memory_monitor_warning_threshold
    warning_recovery = 50
    critical = var.msk_memory_monitor_critical_threshold
  }

  evaluation_delay    = 900 # Recommended for AWS metrics
  new_group_delay     = 900
  notify_by           = ["*"]
  include_tags        = false
  notify_no_data      = false
  notify_audit        = false
  renotify_interval   = 60
  timeout_h           = 0
  require_full_window = true
  tags                = local.dashboard_tags
}

resource "datadog_monitor" "partition_limit_monitor" {
  count = var.msk_partition_limit_monitor_enabled ? 1 : 0
  name               = "${var.environment}-${var.cluster_name}-partition-limit-monitor"
  type               = "query alert"
  message            = "MSK Cluster has reached its Partition Per Broker Limit in ${var.environment}-${var.cluster_name}. Notify: ${var.notification_endpoints}"
  escalation_message = "Escalation to ${var.escalation_notification_endpoints}"

  query = "avg(last_1h):max:aws.kafka.partition_count{${local.cluster_filter}} by {broker_id} > ${local.broker_partition_limit}"

  monitor_thresholds {
    critical = local.broker_partition_limit
    critical_recovery = local.broker_partition_limit - 10
  }

  evaluation_delay    = 900 # Recommended for AWS metrics
  new_group_delay     = 900
  notify_by           = ["*"]
  include_tags        = false
  notify_no_data      = false
  notify_audit        = false
  renotify_interval   = 1440
  timeout_h           = 0
  require_full_window = true
  tags                = local.dashboard_tags
}