resource "aws_cloudwatch_metric_alarm" "Kafka-ZooKeeperRequestLatencyMsMean" {
  alarm_name                = "${var.settings.ZooKeeperRequestLatencyMsMean.ecccode}-Kafka_${var.cluster-name}-ZooKeeperRequestLatencyMsMean"
  comparison_operator       = var.settings.ZooKeeperRequestLatencyMsMean.comparison_operator
  evaluation_periods        = var.settings.ZooKeeperRequestLatencyMsMean.evaluation_periods
  metric_name               = "ZooKeeperRequestLatencyMsMean"
  period                    = var.settings.ZooKeeperRequestLatencyMsMean.period
  statistic                 = var.settings.ZooKeeperRequestLatencyMsMean.statistic
  threshold                 = var.settings.ZooKeeperRequestLatencyMsMean.threshold
  alarm_description         = "Kafka:ZooKeeperRequestLatencyMsMean"
  namespace                 = "AWS/Kafka"
  insufficient_data_actions = []
  actions_enabled           = var.actions-enabled
  alarm_actions             = [var.settings.ZooKeeperRequestLatencyMsMean.action]
  ok_actions                = [var.settings.ZooKeeperRequestLatencyMsMean.action]
  dimensions = {
    "Cluster Name" = var.cluster-name
  }
}

data "aws_msk_cluster" "msk-cluster" {
  cluster_name = var.cluster-name
}

data "aws_msk_broker_nodes" "msk-broker" {
  cluster_arn = data.aws_msk_cluster.msk-cluster.arn
}

resource "aws_cloudwatch_metric_alarm" "Kafka-CpuUserSystem" {
  for_each                  = toset([for i in data.aws_msk_broker_nodes.msk-broker.node_info_list[*].broker_id : tostring(i)])
  alarm_name                = "${var.settings.CpuUserSystem.ecccode}-Kafka_${var.cluster-name}-${each.value}-CpuUsage"
  comparison_operator       = var.settings.CpuUserSystem.comparison_operator
  evaluation_periods        = var.settings.CpuUserSystem.evaluation_periods
  threshold                 = var.settings.CpuUserSystem.threshold
  alarm_description         = "Kafka:ZooKeeperRequestLatencyMsMean"
  insufficient_data_actions = []
  actions_enabled           = var.actions-enabled
  alarm_actions             = [var.settings.CpuUserSystem.action]
  ok_actions                = [var.settings.CpuUserSystem.action]
  metric_query {
    id = "m1"
    metric {
      metric_name = "CpuUser"
      namespace   = "AWS/Kafka"
      period      = var.settings.CpuUserSystem.period
      stat        = var.settings.CpuUserSystem.statistic
      dimensions = {
        "Cluster Name" = var.cluster-name
        "Broker ID"    = each.value
      }
    }
  }

  metric_query {
    id = "m2"
    metric {
      metric_name = "CpuSystem"
      namespace   = "AWS/Kafka"
      period      = var.settings.CpuUserSystem.period
      stat        = var.settings.CpuUserSystem.statistic
      dimensions = {
        "Cluster Name" = var.cluster-name
        "Broker ID"    = each.value
      }
    }
  }

  metric_query {
    id          = "e1"
    expression  = "m1 + m2"
    label       = "CpuUserSystem"
    return_data = "true"
  }
}

resource "aws_cloudwatch_metric_alarm" "Kafka-KafkaDataLogsDiskUsed" {
  for_each                  = toset([for i in data.aws_msk_broker_nodes.msk-broker.node_info_list[*].broker_id : tostring(i)])
  alarm_name                = "${var.settings.KafkaDataLogsDiskUsed.ecccode}-Kafka_${var.cluster-name}-${each.value}-KafkaDataLogsDiskUsed"
  comparison_operator       = var.settings.KafkaDataLogsDiskUsed.comparison_operator
  evaluation_periods        = var.settings.KafkaDataLogsDiskUsed.evaluation_periods
  metric_name               = "KafkaDataLogsDiskUsed"
  period                    = var.settings.KafkaDataLogsDiskUsed.period
  statistic                 = var.settings.KafkaDataLogsDiskUsed.statistic
  threshold                 = var.settings.KafkaDataLogsDiskUsed.threshold
  alarm_description         = "Kafka:KafkaDataLogsDiskUsed"
  namespace                 = "AWS/Kafka"
  insufficient_data_actions = []
  actions_enabled           = var.actions-enabled
  alarm_actions             = [var.settings.KafkaDataLogsDiskUsed.action]
  ok_actions                = [var.settings.KafkaDataLogsDiskUsed.action]
  dimensions = {
    "Cluster Name" = var.cluster-name
    "Broker ID"    = each.value
  }
}

resource "aws_cloudwatch_metric_alarm" "Kafka-HeapMemoryAfterGC" {
  for_each                  = toset([for i in data.aws_msk_broker_nodes.msk-broker.node_info_list[*].broker_id : tostring(i)])
  alarm_name                = "${var.settings.HeapMemoryAfterGC.ecccode}-Kafka_${var.cluster-name}-${each.value}-HeapMemoryAfterGC"
  comparison_operator       = var.settings.HeapMemoryAfterGC.comparison_operator
  evaluation_periods        = var.settings.HeapMemoryAfterGC.evaluation_periods
  metric_name               = "HeapMemoryAfterGC"
  period                    = var.settings.HeapMemoryAfterGC.period
  statistic                 = var.settings.HeapMemoryAfterGC.statistic
  threshold                 = var.settings.HeapMemoryAfterGC.threshold
  alarm_description         = "Kafka:HeapMemoryAfterGC"
  namespace                 = "AWS/Kafka"
  insufficient_data_actions = []
  actions_enabled           = var.actions-enabled
  alarm_actions             = [var.settings.HeapMemoryAfterGC.action]
  ok_actions                = [var.settings.HeapMemoryAfterGC.action]
  dimensions = {
    "Cluster Name" = var.cluster-name
    "Broker ID"    = each.value
  }
}