terraform.aws-baseline-infra/modules/ManagementGovernance/Monitoring.Kafka/main.tf

resource "aws_cloudwatch_metric_alarm" "Kafka-ZooKeeperRequestLatencyMsMean" {
  alarm_name                = "${var.cw-alarm-prefix}:Kafka:ZooKeeperRequestLatencyMsMean:${var.cluster-name}"
  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
  }
  tags = var.default-tags
  lifecycle {
    ignore_changes = [tags]
  }
}

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
}

/*
output debug {
  value = data.aws_msk_broker_nodes.msk-broker.node_info_list
}
*/
/*
module "msk-brokers" {
  source        = "../../util/resource-list"
  resource-type = "kafka-brokers"
  query-input   = 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.cw-alarm-prefix}:Kafka:CpuUsage:${var.cluster-name}-${each.value}"
  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"
  }

  tags = var.default-tags
  lifecycle {
    ignore_changes = [tags]
  }
}

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.cw-alarm-prefix}:Kafka:KafkaDataLogsDiskUsed:${var.cluster-name}-${each.value}"
  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
  }
  tags = var.default-tags
  lifecycle {
    ignore_changes = [tags]
  }
}

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.cw-alarm-prefix}:Kafka:HeapMemoryAfterGC:${var.cluster-name}-${each.value}"
  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
  }
  tags = var.default-tags
  lifecycle {
    ignore_changes = [tags]
  }
}