# General variables
variable "environment" {
  description = "Name of the environment, e.g. dev, qa, prod"
  default     = "dev"
}

variable "environment_type" {
  description = "Type of environment: valid options are fargate"
  default     = "fargate"
}

variable "service_name" {
  description = "Service name, without environment prefix"
  default     = "ows-service"
}

variable "application_family" {
  type        = string
  description = "Application family to which this service belongs"
}

variable "teams" {
  type        = set(string)
  description = "A set of Datadog Teams to associate the resources with"
  default     = []
}

variable "additional_tags" {
  type        = map(string)
  description = "Optional map of additional tags to set on resources. These will be combined with programmatically set required tags."
  default     = {}
}

variable "notification_endpoints" {
  description = "Endpoints for all alerts. Can be email, slack, pagerduty, etc"
}

variable "escalation_notification_endpoints" {
  description = "Endpoints for escalation alerts. Can be email, slack, pagerdut, etc"
}

variable "alert_notification_endpoints" {
  type        = string
  description = "Additional endpoints to send only alert notifications to. Can be email, slack, pagerduty, etc."
  default     = ""
}

variable "no_data_notification_endpoints" {
  type        = string
  description = "Additional endpoints to send only no data notifications to. Can be email, slack, pagerduty, etc."
  default     = ""
}

# AWS variables
variable "aws_region" {
  description = "AWS region from which to pull metrics"
  default     = "us-east-1"
}

variable "healthy_tasks_monitor_enabled" {
  description = "Whether or not healthy tasks monitor is enabled"
  default     = true
}

variable "healthy_tasks_monitor_silenced" {
  description = "Whether or not healthy tasks monitor is silenced (creates a 'datadog_downtime_schedule' resource)"
  default     = false
}

variable "healthy_tasks_critical_number" {
  description = "Number of healthy_tasks for critical alerts"
  default     = 1
}

variable "healthy_tasks_ok_number" {
  description = "Number of healthy_tasks that are ok"
  default     = 0
}

variable "healthy_tasks_evaluation_window" {
  description = "How long to use for evaluation window"
  default     = "last_10m"
}

variable "healthy_tasks_notification_overrides" {
  description = "Override the notification settings for the healthy tasks monitor. Note if this is specified then the notification settings for the monitor will be determined solely by this variable - no settings will be inherited from the defaults."
  type = object({
    endpoints            = string
    escalation_endpoints = string
    alert_endpoints      = optional(string, "")
    no_data_endpoints    = optional(string, "")
  })
  default = null
}

variable "service_cpu_monitor_enabled" {
  description = "Whether or not service cpu monitor is enabled"
  default     = true
}

variable "service_cpu_monitor_silenced" {
  description = "Whether or not service cpu monitor is silenced (creates a 'datadog_downtime_schedule' resource)"
  default     = false
}

variable "service_cpu_critical_number" {
  description = "CPU utilization percentage (per 10m) for critical alerts"
  default     = 90
}

variable "service_cpu_critical_recovery_number" {
  description = "CPU utilization percentage (per 10m) for critical recovery"
  default     = 85
}

variable "service_cpu_time_window" {
  description = "CPU utilization time window"
  default     = "last_10m"
}

variable "service_cpu_monitor_evaluation_function" {
  description = "CPU utilization evaluation function"
  default     = "avg"

  validation {
    condition     = contains(["avg", "max", "min", "sum"], var.service_cpu_monitor_evaluation_function)
    error_message = "Allowed values for service_cpu_monitor_evaluation_function are \"avg\", \"max\", \"min\", or \"sum\"."
  }
}

variable "service_cpu_warning_number" {
  description = "CPU utilization percentage (per 10m) for warnings"
  default     = 80
}

variable "service_cpu_warning_recovery_number" {
  description = "CPU utilization percentage (per 10m) for warning recovery"
  default     = 75
}

variable "service_cpu_ok_number" {
  description = "CPU utilization percentage (per 10m) that are ok"
  default     = 70
}

variable "service_cpu_notification_overrides" {
  description = "Override the notification settings for the service cpu monitor. Note if this is specified then the notification settings for the monitor will be determined solely by this variable - no settings will be inherited from the defaults."
  type = object({
    endpoints            = string
    escalation_endpoints = string
    alert_endpoints      = optional(string, "")
    no_data_endpoints    = optional(string, "")
  })
  default = null
}

# Error monitor configuration
variable "error_monitor_enabled" {
  type        = bool
  description = "Whether or not to enable a monitor for error in connector logs."
  default     = false
}

variable "error_monitor_critical_number_errors" {
  description = "The number of errors in the query interval below which to trigger a critical alert"
  default     = 2
}

variable "error_monitor_query_interval" {
  type        = string
  description = "The query interval to use for error in logs."
  default     = "5m"
}

variable "error_monitor_extra_query" {
  type        = string
  description = "Extra conditions to append to error query."
  default     = ""
}

# DLQ Topic monitor configuration
variable "dlq_topic_name" {
  type        = string
  description = "Name of dlq topic on var.monitor_kafka_topics_cluster_name cluster"
  default     = ""
}

variable "dlq_number_errors" {
  description = "The number of messages in dlq topic in the query interval which to trigger a critical alert"
  default     = 2
}

variable "dlq_query_interval" {
  type        = string
  description = "The query interval to use for messages in dlq topic."
  default     = "last_5m"
}


# debezium seconds_since_last_event monitor configuration
variable "seconds_since_last_event_critical" {
  description = "Delay in seconds considered critical."
  default     = 300
}
variable "seconds_since_last_event_warning" {
  description = "Delay in seconds considered warning."
  default     = 150
}

variable "seconds_since_last_event_no_data_timeframe" {
  description = "No data time frame for seconds since last event monitor, in minutes."
  default     = 15
}

# debezium seconds_behind_source monitor configuration
variable "seconds_behind_source_critical" {
  description = "Delay in seconds considered critical."
  default     = 300
}
variable "seconds_behind_source_warning" {
  description = "Delay in seconds considered warning."
  default     = 150
}


# Monitor reated Kafka Topic
variable "monitor_kafka_topics" {
  type        = list(string)
  description = "List of kafka topics"
  default     = []
}

variable "monitor_kafka_topics_cluster_name" {
  type        = string
  description = "Name of kafka cluster that hosts these topics."
  default     = "qa-managed-kafka-cdc-destination"
}

# To show appropriate JMX Metrics @todo use it
variable "connector_type" {
  type        = string
  description = "Type of connector."
  default     = ""
}


# Local variables
locals {
  healthy_tasks_filter = {
    fargate = "servicename:${var.environment}-${var.service_name}"
  }

  service_cpu_monitor_filter = {
    fargate = "servicename:${var.environment}-${var.service_name}"
  }

  service_cpu_monitor_metric = {
    fargate = "aws.ecs.cpuutilization.maximum"
  }

  # Only team tags are supported for dashboards
  dashboard_tags = [for team in var.teams : "team:${team}"]

  # Concatenate required and user-supplied tags
  combined_resource_tags = flatten(concat(
    [
      "env:${var.environment}",
      "environment:${var.environment}",
      "service:${var.service_name}",
      "service_name:${var.service_name}",
      "application_family:${var.application_family}"
    ],
    [for tag_name, tag_value in var.additional_tags : "${tag_name}:${tag_value}"],
    [for team in var.teams : "team:${team}"],
  ))

  default_notification_settings = {
    endpoints            = var.notification_endpoints
    escalation_endpoints = var.escalation_notification_endpoints
    alert_endpoints      = var.alert_notification_endpoints
    no_data_endpoints    = var.no_data_notification_endpoints
  }

  combined_monitor_kafka_topics = flatten([
    for topic in var.monitor_kafka_topics : {
      topic = lower(topic),
      cluster = var.monitor_kafka_topics_cluster_name,
      alias = element(split(".", topic), length(split(".", topic))-1)
    }
  ])

  default_sink_widgets = [
      # widget connect task_count
      {
        show_legend = true,
        title = "Connector Task Count",
        display_type = "line",
        formulas = [
          {
            formula_expression = "query1"
            palette = "green"
            palette_index = 5
            alias = "running tasks"
          },
          {
            formula_expression = "query2"
            palette = "warm"
            palette_index = 4
            alias = "failed tasks"
          },
          {
            formula_expression = "query3"
            palette = "green"
            palette_index = 1
            alias = "restarting tasks"
          },
          {
            formula_expression = "query4"
            palette = "gray"
            palette_index = 1
            alias = "destroyed tasks"
          }
        ],
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.connect.connector_running_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.kafka.connect.connector_failed_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query3"
            query = "sum:jmx.kafka.connect.connector_restarting_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query4"
            query = "sum:jmx.kafka.connect.connector_destroyed_task_count{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget Total number of records read vs output
      {
        show_legend = true,
        display_type = "line",
        title = "Total number of records read vs output",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "records output"
          },
          {
            formula_expression = "query2"
            palette = "classic"
            palette_index = 1
            alias = "polled by the task"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "max:jmx.kafka.connect.sink_record_send_total{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "max:jmx.kafka.connect.sink_record_read_total{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = []
      },
      # widget Records read but not committed
      {
        show_legend = true,
        display_type = "line",
        title = "Records read from Kafka but not yet committed",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "polled by the task"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "max:jmx.kafka.connect.sink_record_active_count{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget number of active connections
      {
        show_legend = true,
        display_type = "area",
        title = "Number of active connections",
        formulas = [
          {
            formula_expression = "query2"
            palette = "classic"
            palette_index = 1
            alias = "connection closed"
          },
          {
            formula_expression = "query3"
            palette = "classic"
            palette_index = 1
            alias = "connection created"
          },
        ]
        metric_queries = [
          {
            name  = "query2"
            query = "sum:jmx.kafka.consumer.connection_close_total{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query3"
            query = "sum:jmx.kafka.consumer.connection_creation_total{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget Error and Retry count
      {
        show_legend = true,
        display_type = "line",
        title = "Error and Retry count",
        formulas = [
          {
            formula_expression = "query1"
            palette = "warm"
            palette_index = 1
            alias = "error"
          },
          {
            formula_expression = "query2"
            palette = "warm"
            palette_index = 4
            alias = "failures"
          },
          {
            formula_expression = "query3"
            palette = "gray"
            palette_index = 1
            alias = "skipped"
          },
          {
            formula_expression = "query4"
            palette = "green"
            palette_index = 1
            alias = "retries"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.connect.total_record_errors{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.kafka.connect.total_record_failures{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query3"
            query = "sum:jmx.kafka.connect.total_records_skipped{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query4"
            query = "sum:jmx.kafka.connect.total_retries{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget put batch time
      {
        show_legend = true,
        display_type = "line",
        title = "seconds taken to put a batch of sinks records",
        formulas = [
          {
            formula_expression = "query1 / 100"
            palette = "classic"
            palette_index = 1
            alias = "put time"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "max:jmx.kafka.connect.put_batch_avg_time_ms{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = []
      },
      # widget Current offset number
      {
        show_legend = true,
        display_type = "line",
        title = "Current offset number",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "offset"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.connect.offset_commit_seq_no{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget requests sent per second
      {
        show_legend = true,
        display_type = "line",
        title = "Requests sent per second",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "request-rate"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.consumer.request_rate{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
  ]

  default_source_widgets = [
    # widget connect task_count
    {
      show_legend = true,
      title = "Connector Task Count",
      display_type = "line",
      formulas = [
        {
          formula_expression = "query1"
          palette = "green"
          palette_index = 5
          alias = "running tasks"
        },
        {
          formula_expression = "query2"
          palette = "warm"
          palette_index = 4
          alias = "failed tasks"
        },
        {
          formula_expression = "query3"
          palette = "green"
          palette_index = 1
          alias = "restarting tasks"
        },
        {
          formula_expression = "query4"
          palette = "gray"
          palette_index = 1
          alias = "destroyed tasks"
        }
      ],
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.connect.connector_running_task_count{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query2"
          query = "sum:jmx.kafka.connect.connector_failed_task_count{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query3"
          query = "sum:jmx.kafka.connect.connector_restarting_task_count{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query4"
          query = "sum:jmx.kafka.connect.connector_destroyed_task_count{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
    },
    # widget average number of bytes sent
    {
      show_legend = true,
      display_type = "line",
      title = "average number of bytes sent",
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "batch_size_avg"
        }
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.producer.batch_size_avg{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Producer Connected and connection_count
    {
      show_legend = true,
      display_type = "line",
      title = "Producer Connected and connection_count",
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "active connections by producer"
        },
        {
          formula_expression = "query2"
          palette = "green"
          palette_index = 4
          alias = "Connected to DB?"
        }
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.producer.connection_count{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query2"
          query = "avg:jmx.debezium.mysql.connected{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Records polled
    {
      show_legend = true,
      display_type = "line",
      title = "Records polled Vs written to Kafka",
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "written to kafka"
        },
        {
          formula_expression = "query2"
          palette = "classic"
          palette_index = 1
          alias = "polled but not written"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.connect.source_record_write_total{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query2"
          query = "sum:jmx.kafka.connect.source_record_active_count{task_name:${var.environment}-${var.service_name}}"
        },
      ],
      event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
    },
    # widget records retried, error and sent
    {
      show_legend = true,
      display_type = "line",
      title = "Records retried, error and sent",
      formulas = [
        {
          formula_expression = "query1"
          palette = "warm"
          palette_index = 4
          alias = "error records"
        },
        {
          formula_expression = "query2"
          palette = "green"
          palette_index = 2
          alias = "retried records"
        },
        {
          formula_expression = "query3"
          palette = "green"
          palette_index = 5
          alias = "total records sent"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.producer.record_error_total{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query2"
          query = "sum:jmx.kafka.producer.record_retry_total{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query3"
          query = "sum:jmx.kafka.producer.record_send_total{task_name:${var.environment}-${var.service_name}}"
        },
      ],
      event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
    },
    # widget waiting-threads
    {
      show_legend = true,
      display_type = "line",
      title = "Waiting-threads",
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "threads blocked"
        }
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "sum:jmx.kafka.producer.waiting_threads{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
  ]

  debezium_widgets = [
      # widget connect task_count
      {
        show_legend = true,
        title = "Connector Task Count",
        display_type = "line",
        formulas = [
          {
            formula_expression = "query1"
            palette = "green"
            palette_index = 5
            alias = "running tasks"
          },
          {
            formula_expression = "query2"
            palette = "warm"
            palette_index = 4
            alias = "failed tasks"
          },
          {
            formula_expression = "query3"
            palette = "green"
            palette_index = 1
            alias = "restarting tasks"
          },
          {
            formula_expression = "query4"
            palette = "gray"
            palette_index = 1
            alias = "destroyed tasks"
          }
        ],
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.connect.connector_running_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.kafka.connect.connector_failed_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query3"
            query = "sum:jmx.kafka.connect.connector_restarting_task_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query4"
            query = "sum:jmx.kafka.connect.connector_destroyed_task_count{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget SecondsSinceLastEvent
      {
        show_legend = true,
        title = "SecondsSinceLastEvent",
        display_type = "line",
        formulas = [
          {
            formula_expression = "query1 / 100"
            palette = "classic"
            palette_index = 1
            alias = "seconds_since_last_event"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "avg:jmx.debezium.mysql.milli_seconds_since_last_event{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = []
      },
      # widget SecondsBehindSource
      {
        show_legend = true,
        title = "SecondsBehindSource",
        display_type = "line",
        formulas = [
          {
            formula_expression = "query1 / 100"
            palette = "classic"
            palette_index = 1
            alias = "seconds_behind_source"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "avg:jmx.debezium.mysql.milli_seconds_behind_source{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = []
      },
      # widget SecondsBehindSource
      {
        show_legend = true,
        display_type = "bars",
        title = "Number Of Events Seen",
        formulas = [
          {
            formula_expression = "query3"
            palette = "cool"
            palette_index = 1
            alias = "update_events"
          },
          {
            formula_expression = "query4"
            palette = "cool"
            palette_index = 4
            alias = "delete_events"
          },
          {
            formula_expression = "query2"
            palette = "cool"
            palette_index = 6
            alias = "create_events"
          },
        ]
        metric_queries = [
          {
            name  = "query3"
            query = "avg:jmx.debezium.mysql.total_number_of_update_events_seen{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query4"
            query = "avg:jmx.debezium.mysql.total_number_of_delete_events_seen{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "avg:jmx.debezium.mysql.total_number_of_create_events_seen{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget average number of bytes sent
      {
        show_legend = true,
        display_type = "line",
        title = "average number of bytes sent",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "batch_size_avg"
          }
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.producer.batch_size_avg{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = []
      },
      # widget Producer Connected and connection_count
      {
        show_legend = true,
        display_type = "line",
        title = "Producer Connected and connection_count",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "active connections by producer"
          },
          {
            formula_expression = "query2"
            palette = "green"
            palette_index = 4
            alias = "Connected to DB?"
          }
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.producer.connection_count{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "avg:jmx.debezium.mysql.connected{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = []
      },
      # widget BinlogPosition
      {
        show_legend = true,
        display_type = "line",
        title = "Binlog Position",
        formulas = [
          {
            formula_expression = "query1 / 100"
            palette = "classic"
            palette_index = 1
            alias = "binlog position"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.debezium.mysql.binlog_position{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget QueueRemainingCapacity
      {
        show_legend = true,
        display_type = "line",
        title = "Queue Remaining Capacity",
        formulas = [
          {
            formula_expression = "round(query1 / query2) * 100"
            palette = "classic"
            palette_index = 1
            alias = "QueueCapacity"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.debezium.mysql.queue_remaining_capacity{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.debezium.mysql.queue_total_capacity{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = []
      },
      # widget Records polled
      {
        show_legend = true,
        display_type = "line",
        title = "Records polled Vs written to Kafka",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "written to kafka"
          },
          {
            formula_expression = "query2"
            palette = "classic"
            palette_index = 1
            alias = "polled but not written"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.connect.source_record_write_total{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.kafka.connect.source_record_active_count{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget records retried, error and sent
      {
        show_legend = true,
        display_type = "line",
        title = "Records retried, error and sent",
        formulas = [
          {
            formula_expression = "query1"
            palette = "warm"
            palette_index = 4
            alias = "error records"
          },
          {
            formula_expression = "query2"
            palette = "green"
            palette_index = 2
            alias = "retried records"
          },
          {
            formula_expression = "query3"
            palette = "green"
            palette_index = 5
            alias = "total records sent"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.producer.record_error_total{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.kafka.producer.record_retry_total{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query3"
            query = "sum:jmx.kafka.producer.record_send_total{task_name:${var.environment}-${var.service_name}}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget Snapshot events
      {
        show_legend = true,
        display_type = "line",
        title = "Snapshot events",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "total_events_seen"
          },
          {
            formula_expression = "query2"
            palette = "classic"
            palette_index = 1
            alias = "filtered_events"
          },
          {
            formula_expression = "query3"
            palette = "classic"
            palette_index = 1
            alias = "snapshot_running"
          },
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "max:jmx.debezium.mysql.total_number_of_events_seen{task_name:${var.environment}-${var.service_name}, context:snapshot}"
          },
          {
            name  = "query2"
            query = "max:jmx.debezium.mysql.total_number_of_events_seen{task_name:${var.environment}-${var.service_name}, context:snapshot}"
          },
          {
            name  = "query3"
            query = "avg:jmx.debezium.mysql.snapshot_running{task_name:${var.environment}-${var.service_name}, context:snapshot}"
          },
        ],
        event_query = ["@title:\"ECS Deployment on service ${var.environment}-${var.service_name}\""]
      },
      # widget Schema history
      {
        show_legend = true,
        display_type = "line",
        title = "Schema history",
        formulas = [
          {
            formula_expression = "query1 / 100"
            palette = "classic"
            palette_index = 1
            alias = "seconds since last applied changes"
          },
          {
            formula_expression = "query2 / 100"
            palette = "classic"
            palette_index = 1
            alias = "seconds since last recovered changes"
          }
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.debezium.mysql.milli_seconds_since_last_applied_change{task_name:${var.environment}-${var.service_name}}"
          },
          {
            name  = "query2"
            query = "sum:jmx.debezium.mysql.milli_seconds_since_last_recovered_change{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = []
      },
      # widget waiting-threads
      {
        show_legend = true,
        display_type = "line",
        title = "Waiting-threads",
        formulas = [
          {
            formula_expression = "query1"
            palette = "classic"
            palette_index = 1
            alias = "threads blocked"
          }
        ]
        metric_queries = [
          {
            name  = "query1"
            query = "sum:jmx.kafka.producer.waiting_threads{task_name:${var.environment}-${var.service_name}}"
          }
        ],
        event_query = []
      },
    ]

  ksqldb_widgets = [
    # widget Active and Running Queries
    {
      show_legend = true,
      display_type = "line",
      title = "Active and Running queries",
      yaxis_include_zero = false,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "active queries"
        },
        {
          formula_expression = "query2"
          palette = "green"
          palette_index = 5
          alias = "running queries"
        }
      ],
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.num_active_queries{task_name:${var.environment}-${var.service_name}}"
        },
        {
          name  = "query2"
          query = "avg:jmx.io.confluent.ksql.metrics.running_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Total bytes consumed
    {
      show_legend = true,
      display_type = "bars",
      title = "Total bytes consumed across all queries",
      yaxis_include_zero = false,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "total bytes consumed"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.bytes_consumed_total{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Newly created queries
    {
      show_legend = true,
      display_type = "line",
      title = "Newly created queries",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "newly created queries"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.created_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Queries with errors
    {
      show_legend = true,
      display_type = "line",
      title = "Queries with errors",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "errored queries"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "max:jmx.io.confluent.ksql.metrics.error_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Query error rate
    {
      show_legend = true,
      display_type = "line",
      title = "Queries error rate",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1 * 100"
          palette = "classic"
          palette_index = 1
          alias = "queries error %"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.error_rate{task_name:${var.environment}-${var.service_name}}.as_rate()"
        }
      ],
      event_query = []
    },
    # widget Total consumed messages
    {
      show_legend = true,
      display_type = "bars",
      title = "Total Consumed messages across all queries",
      yaxis_include_zero = false,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "total consumed messages"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "max:jmx.io.confluent.ksql.metrics.messages_consumed_total{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Messages produced per sec
    {
      show_legend = true,
      display_type = "bars",
      title = "Messages produced per sec",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "messages produced/sec"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.messages_produced_per_sec{task_name:${var.environment}-${var.service_name}}.as_rate()"
        }
      ],
      event_query = []
    },
    # widget Messages consumed per sec
    {
      show_legend = true,
      display_type = "bars",
      title = "Messages consumed per sec",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "messages consumed/sec"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.messages_consumed_per_sec{task_name:${var.environment}-${var.service_name}}.as_rate()"
        }
      ],
      event_query = []
    },
    # widget Not running queries
    {
      show_legend = true,
      display_type = "line",
      title = "Not running queries",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "not running queries"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "max:jmx.io.confluent.ksql.metrics.not_running_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Queries pending shutdown
    {
      show_legend = true,
      display_type = "line",
      title = "Queries pending shutdown",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "queries pending shutdown"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.pending_shutdown_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },

    # widget Queries undergoing rebalance
    {
      show_legend = true,
      display_type = "line",
      title = "Queries undergoing rebalance",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "queries rebalancing"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.rebalancing_queries{task_name:${var.environment}-${var.service_name}}"
        }
      ],
      event_query = []
    },
    # widget Total messages produced per query
    {
      show_legend = true,
      display_type = "line",
      title = "Total messages produced per query",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "msgs produced per query"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "max:jmx.io.confluent.ksql.metrics.total_messages{task_name:${var.environment}-${var.service_name}, type:producer-metrics} by {key}"
        }
      ],
      event_query = []
    },
    # widget Total messages produced per second per query
    {
      show_legend = true,
      display_type = "bars",
      title = "Total messages produced per second per query",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "msgs produced per second per query"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.messages_per_sec{task_name:${var.environment}-${var.service_name}, type:producer-metrics} by {key}.as_rate()"
        }
      ],
      event_query = []
    },
    # widget Total consumed messages per query
    {
      show_legend = true,
      display_type = "line",
      title = "Total consumed messages per query",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "msgs consumed per query"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "max:jmx.io.confluent.ksql.metrics.consumer_total_messages{task_name:${var.environment}-${var.service_name}, type:consumer-metrics} by {key}"
        }
      ],
      event_query = []
    },
     # widget Total consumed messages per second per query
    {
      show_legend = true,
      display_type = "bars",
      title = "Total consumed messages per second query",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "msgs consumed per query per second"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.consumer_messages_per_sec{task_name:${var.environment}-${var.service_name}, type:consumer-metrics} by {key}.as_rate()"
        }
      ],
      event_query = []
    },
     # widget Total PULL query requests
    {
      show_legend = true,
      display_type = "bars",
      title = "Total PULL query requests",
      yaxis_include_zero = false,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "pull queries"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.pull_query_requests_total{task_name:${var.environment}-${var.service_name}, type:_confluent-ksql-pull-query}"
        }
      ],
      event_query = []
    },
     # widget Total Errors in PULL query
    {
      show_legend = true,
      display_type = "line",
      title = "Total Errors in PULL queries",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "error pull queries"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.pull_query_requests_error_total{task_name:${var.environment}-${var.service_name}, type:_confluent-ksql-pull-query}"
        }
      ],
      event_query = []
    },
     # widget PULL queries requests average latency
    {
      show_legend = true,
      display_type = "line",
      title = "PULL queries requests average latency",
      yaxis_include_zero = true,
      formulas = [
        {
          formula_expression = "query1"
          palette = "classic"
          palette_index = 1
          alias = "pull queries avg latency"
        },
      ]
      metric_queries = [
        {
          name  = "query1"
          query = "avg:jmx.io.confluent.ksql.metrics.pull_query_requests_latency_avg{task_name:${var.environment}-${var.service_name}, type:_confluent-ksql-pull-query}"
        }
      ],
      event_query = []
    }
  ]

  jmx_widgets_map = {
    "debezium_source": local.debezium_widgets,
    "default_sink": local.default_sink_widgets,
    "default_source": local.default_source_widgets,
    "ksqldb_server": local.ksqldb_widgets,
  }
  # select the right JMX widget
  jmx_widgets = lookup(local.jmx_widgets_map, var.connector_type, local.default_source_widgets)

  debezium_seconds_since_last_event_monitor_enabled = var.connector_type == "debezium_source" ? true : false
  debezium_seconds_behind_source_monitor_enabled    = var.connector_type == "debezium_source" ? true : false
}