data "aws_caller_identity" "current" {
}

data "datadog_role" "datadog_admin_role" {
  filter = "Datadog Admin Role"
}

# Datadog MSK timeboard
resource "datadog_dashboard" "datadog_msk" {
  title            = "${var.environment}-${var.cluster_name}-MSK-timeboard"
  description      = "Default terraformed timeboard for MSK"
  restricted_roles = [data.datadog_role.datadog_admin_role.id]
  layout_type      = "ordered"

  # Mean latency in milliseconds for ZooKeeper requests from broker.
  widget {
    timeseries_definition {
      title = "Mean latency in milliseconds for ZooKeeper requests from broker"


      request {
        q = "avg:aws.kafka.zookeeper_request_latency_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The size in bytes of memory that is in use for the broker. Shown as byte.
  widget {
    timeseries_definition {
      title = "Broker used memory in bytes"

      request {
        q = "max:aws.kafka.memory_used{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The size in bytes of memory that is free and available for the broker.
  widget {
    timeseries_definition {
      title = "Broker free memory in bytes"

      request {
        q = "max:aws.kafka.memory_free{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }

  }

  # The percentage of CPU in kernel space.
  widget {
    timeseries_definition {
      title = "Percentage of CPU in kernel space"

      request {
        q = "max:aws.kafka.cpu_system{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The percentage of the root disk used by the broker.
  widget {
    timeseries_definition {
      title = "Percentage of root disk used by the broker"

      request {
        q = "max:aws.kafka.root_disk_used{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The percentage of disk space used for application logs.
  widget {
    timeseries_definition {
      title = "Percentage of disk space used for application logs"

      request {
        q = "max:aws.kafka.kafka_app_logs_disk_used{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The percentage of disk space used for data logs.
  widget {
    timeseries_definition {
      title = "Percentage of disk space used for data logs"

      request {
        q = "max:aws.kafka.kafka_data_logs_disk_used{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of network receive errors for the broker.
  widget {
    timeseries_definition {
      title = "Network receive errors for the broker"

      request {
        q = "sum:aws.kafka.network_rx_errors{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of network transmit errors for the broker.
  widget {
    timeseries_definition {
      title = "Network transmit errors for the broker"

      request {
        q = "sum:aws.kafka.network_tx_errors{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of dropped receive packets.
  widget {
    timeseries_definition {
      title = "Dropped receive packets"

      request {
        q = "sum:aws.kafka.network_rx_dropped{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of dropped transmit packets.
  widget {
    timeseries_definition {
      title = "Dropped transmit packets"

      request {
        q = "sum:aws.kafka.network_tx_dropped{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of packets recieved by the broker.
  widget {
    timeseries_definition {
      title = "Packets recieved by the broker."

      request {
        q = "sum:aws.kafka.network_rx_packets{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of packets transmitted by the broker.
  widget {
    timeseries_definition {
      title = "Packets transmitted by the broker"

      request {
        q = "sum:aws.kafka.network_tx_packets{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of incoming messages per second for the broker.
  widget {
    timeseries_definition {
      title = "Incoming messages per second for the broker"

      request {
        q = "sum:aws.kafka.messages_in_per_sec{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of leader replicas.
  widget {
    timeseries_definition {
      title = "Number of leader replicas"

      request {
        q = "sum:aws.kafka.leader_count{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of partitions for the broker.
  widget {
    timeseries_definition {
      title = "Partitions for the broker"

      request {
        q = "sum:aws.kafka.partition_count{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that request messages spend in the queue.
  widget {
    timeseries_definition {
      title = "Average time in milliseconds request messages spend in the queue"

      request {
        q = "avg:aws.kafka.produce_request_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that response messages spend in the queue.
  widget {
    timeseries_definition {
      title = "Average time in milliseconds response messages spend in the queue"

      request {
        q = "avg:aws.kafka.produce_response_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds spent on sending response messages.
  widget {
    timeseries_definition {
      title = "Average time in milliseconds spent on sending response messages"

      request {
        q = "avg:aws.kafka.produce_response_send_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean produce time in milliseconds.
  widget {
    timeseries_definition {
      title = "Mean produce time in milliseconds"

      request {
        q = "avg:aws.kafka.produce_total_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of messages received from clients per second.
  widget {
    timeseries_definition {
      title = "Messages received from clients per second"

      request {
        q = "sum:aws.kafka.messages_in_per_sec{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of produce message conversions per second for the broker.
  widget {
    timeseries_definition {
      title = "Produce message conversions per second for the broker"

      request {
        q = "sum:aws.kafka.produce_message_conversions_per_sec{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean total time in milliseconds that consumers spend on fetching data from the broker.
  widget {
    timeseries_definition {
      title = "Mean total time consumers spend on fetching data from the broker"

      request {
        q = "avg:aws.kafka.fetch_consumer_total_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean total time in milliseconds that followers spend on fetching data from the broker.
  widget {
    timeseries_definition {
      title = "Mean total time followers spend on fetching data from the broker"

      request {
        q = "avg:aws.kafka.fetch_follower_total_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that the consumer request waits in the request queue.
  widget {
    timeseries_definition {
      title = "Average time in milliseconds consumer request waits in the request queue"

      request {
        q = "avg:aws.kafka.fetch_consumer_request_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that the follower request waits in the request queue.
  widget {
    timeseries_definition {
      title = "Mean time in millisecond follower request waits in the request queue"

      request {
        q = "avg:aws.kafka.fetch_follower_request_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that the consumer request waits in the response queue.
  widget {
    timeseries_definition {
      title = "Mean time in milliseconds consumer request waits in the response queue"

      request {
        q = "avg:aws.kafka.fetch_consumer_response_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds that the follower request waits in the response queue.
  widget {
    timeseries_definition {
      title = "Mean time in milliseconds follower request waits in the response queue"

      request {
        q = "avg:aws.kafka.fetch_follower_response_queue_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds for the consumer to send a response.
  widget {
    timeseries_definition {
      title = "Mean time in milliseconds for the consumer to send a response"

      request {
        q = "avg:aws.kafka.consumer_response_send_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The mean time in milliseconds for the follower to send a response.
  widget {
    timeseries_definition {
      title = "Mean time in milliseconds for the follower to send a response"

      request {
        q = "avg:aws.kafka.fetch_follower_response_send_time_ms_mean{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The average produce throttle time in milliseconds.
  widget {
    timeseries_definition {
      title = "Average produce throttle time in milliseconds"

      request {
        q = "avg:aws.kafka.produce_throttle_time{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of messages in the throttle queue.
  widget {
    timeseries_definition {
      title = "Number of messages in the throttle queue"

      request {
        q = "sum:aws.kafka.produce_throttle_queue_size{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The average fetch throttle time in milliseconds.
  widget {
    timeseries_definition {
      title = " Average fetch throttle time in milliseconds"

      request {
        q = "avg:aws.kafka.fetch_throttle_time{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of messages in the throttle queue.
  widget {
    timeseries_definition {
      title = "Messages in the throttle queue"

      request {
        q = "sum:aws.kafka.fetch_throttle_queue_size{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The average request throttle time in milliseconds.
  widget {
    timeseries_definition {
      title = "Average request throttle time in milliseconds"

      request {
        q = "sum:aws.kafka.request_throttle_time{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The average time spent in broker network and I/O threads to process requests that are exempt throttled.
  widget {
    timeseries_definition {
      title = "Average time spent in broker network and I/O threads to process requests that are exempt throttled"

      request {
        q = "avg:aws.kafka.request_time{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  # The number of messages in the throttle queue.
  widget {
    timeseries_definition {
      title = "Number of messages in the throttle queue"

      request {
        q = "sum:aws.kafka.request_throttle_queue_size{cluster_name:${var.environment}-${var.cluster_name}}"
      }
    }
  }

  tags = local.dashboard_tags
}

resource "datadog_dashboard" "datadog_msk_throughput" {
  count            = var.environment != "dev" ? 1 : 0
  title            = "${upper(var.environment)} MSK ${var.cluster_name} clients and throughput metrics"
  description      = "MSK client count, bytes IO and lag per topic."
  restricted_roles = [data.datadog_role.datadog_admin_role.id]
  layout_type      = "ordered"

  dynamic "widget" {
    for_each = local.timeseries_definitions

    content {
      timeseries_definition {
        title          = widget.value["title"]
        show_legend    = widget.value["show_legend"]
        legend_size    = widget.value["legend_size"]
        legend_columns = widget.value["legend_columns"]
        legend_layout  = "vertical"

        yaxis {
          include_zero = true
          scale        = widget.value["yaxis_scale"]
        }

        request {
          q = widget.value["query"]
        }
      }
      widget_layout {
        width  = widget.value["width"]
        height = widget.value["height"]
        x      = widget.value["pos_x"]
        y      = widget.value["pos_y"]
      }
    }
  }

  tags = local.dashboard_tags
}