resource "datadog_monitor" "kafka_connector_os_sink_gp_v2_offset_lag_monitor" {
  count   = 1
  name    = "${var.environment}-${local.gp_service_name}-consumer-offset-lag"
  type    = "metric alert"
  message = "Consumer offset lag is to high for the last 24h. Notify: ${var.notification_endpoints} ${var.escalation_notification_endpoints}"

  query = "avg(last_3h):avg:aws.kafka.sum_offset_lag{environment:${var.environment},consumer_group:${local.gp_consumer_group_name},topic:stream.globalparticipant.elasticsearch.cdc.partitioned.v5} by {topic} > 50000"

  monitor_thresholds {
    critical = 50000
  }

  notify_no_data      = true
  no_data_timeframe   = 60
  notify_audit        = false
  renotify_interval   = 60
  timeout_h           = 0
  require_full_window = false

  tags = flatten(concat([
    "environment:${var.environment}",
    "service_name:${local.gp_service_name}",
    ],
    [for tag_name, tag_value in var.additional_tags : "${tag_name}:${tag_value}"]
  ))
}

resource "datadog_monitor" "kafka_connector_os_sink_gsr_v2_offset_lag_monitor" {
  count   = 1
  name    = "${var.environment}-${local.gsr_service_name}-consumer-offset-lag"
  type    = "metric alert"
  message = "Consumer offset lag is to high for the last 24h. Notify: ${var.notification_endpoints} ${var.escalation_notification_endpoints}"

  query = "avg(last_3h):avg:aws.kafka.sum_offset_lag{environment:${var.environment} AND consumer_group:${local.gsr_consumer_group_name} AND (topic:stream.globalSoundRecording.elasticSearch.cdc.partitioned.V5 )} by {topic} > 50000"

  monitor_thresholds {
    critical = 50000
  }

  notify_no_data      = true
  no_data_timeframe   = 60
  notify_audit        = false
  renotify_interval   = 60
  timeout_h           = 0
  require_full_window = false

  tags = flatten(concat([
    "environment:${var.environment}",
    "service_name:${local.gsr_service_name}",
    ],
    [for tag_name, tag_value in var.additional_tags : "${tag_name}:${tag_value}"]
  ))
}


resource "datadog_monitor" "kafka_connector_os_sink_lp_offset_lag_monitor" {
  count   = 1
  name    = "${var.environment}-${local.lp_service_name}-consumer-offset-lag"
  type    = "metric alert"
  message = "Consumer offset lag is to high for the last 24h. Notify: ${var.notification_endpoints} ${var.escalation_notification_endpoints}"

  query = "avg(last_3h):avg:aws.kafka.sum_offset_lag{environment:${var.environment} AND consumer_group:${local.lp_consumer_group_name} and ( topic:topic.musicGraphV5.labelParticipant.elasticSearch.V5 )} by {topic} > 50000"

  monitor_thresholds {
    critical = 50000
  }

  notify_no_data      = true
  no_data_timeframe   = 60
  notify_audit        = false
  renotify_interval   = 60
  timeout_h           = 0
  require_full_window = false

  tags = flatten(concat([
    "environment:${var.environment}",
    "service_name:${local.lp_service_name}",
    ],
    [for tag_name, tag_value in var.additional_tags : "${tag_name}:${tag_value}"]
  ))
}