# ── agent configuration ─────────────────────────────────────────────────
[agent]
  ## How often to gather metrics
  interval       = "10s"
  ## How often to write metrics to outputs
  flush_interval = "10s"
  ## Enable detailed logs of each scrape and write
  debug          = true
  ## Show info and debug logs
  quiet          = false

# ── outputs: write to InfluxDB v2 ────────────────────────────────────────
[[outputs.influxdb_v2]]
  urls         = ["http://influxdb:8086"]
  token        = "$INFLUX_TOKEN"
  organization = "local"
  bucket       = "kafka_metrics"

# ── inputs: Jolokia (Kafka JMX) ──────────────────────────────────────────
[[inputs.jolokia2_agent]]
  urls        = ["http://kafka:8778/jolokia"]
  name_prefix = "kafka_"

  # ── host CPU usage ───────────────────────────────────────────────────────
  [[inputs.cpu]]
    ## report total combined CPU usage rather than per‐core
    percpu        = false
    totalcpu      = true
    ## don’t emit raw time fields
    collect_cpu_time = false
    report_active = false

  ## ── core broker ingress
  [[inputs.jolokia2_agent.metric]]
    name  = "messages_in_total"
    mbean = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec"
    paths = ["Count"]

  [[inputs.jolokia2_agent.metric]]
    name  = "bytes_in_total"
    mbean = "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec"
    paths = ["Count"]

  ## ── under-replicated partitions
  [[inputs.jolokia2_agent.metric]]
    name  = "under_replicated_partitions"
    mbean = "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
    paths = ["Value"]

  ## ── core broker egress
  [[inputs.jolokia2_agent.metric]]
    name  = "bytes_out_total"
    mbean = "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec"
    paths = ["Count"]

  [[inputs.jolokia2_agent.metric]]
    name  = "messages_out_total"
    mbean = "kafka.server:type=BrokerTopicMetrics,name=MessagesOutPerSec"
    paths = ["Count"]

  ## ── per-topic ingress (tagged by topic)
  [[inputs.jolokia2_agent.metric]]
    name     = "bytes_in_per_topic"
    mbean    = "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=*"
    paths    = ["Count"]
    tag_keys = ["topic"]

  [[inputs.jolokia2_agent.metric]]
    name     = "messages_in_per_topic"
    mbean    = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=*"
    paths    = ["Count"]
    tag_keys = ["topic"]

  ## ── per-topic log size (tagged by topic & partition)
  [[inputs.jolokia2_agent.metric]]
    name     = "topic_size"
    mbean    = "kafka.log:type=Log,topic=*,partition=*,name=Size"
    paths    = ["Value"]
    tag_keys = ["topic","partition"]

  ## ── partition count (global)
  [[inputs.jolokia2_agent.metric]]
    name  = "partition_count"
    mbean = "kafka.server:type=ReplicaManager,name=PartitionCount"
    paths = ["Value"]

  ## ── request latency & failure rate (tagged by request)
  [[inputs.jolokia2_agent.metric]]
    name     = "request_latency_ms"
    mbean    = "kafka.network:type=RequestMetrics,name=TotalTimeMs,request=*"
    paths    = ["Mean"]
    tag_keys = ["request"]

  [[inputs.jolokia2_agent.metric]]
    name     = "failed_request_percent"
    mbean    = "kafka.network:type=RequestMetrics,name=ErrorPercent,request=*"
    paths    = ["Count"]
    tag_keys = ["request"]

  ## ── delayed (queued) fetch & produce requests
  [[inputs.jolokia2_agent.metric]]
    name  = "delayed_fetch_requests"
    mbean = "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedFetchRequests"
    paths = ["Value"]

  [[inputs.jolokia2_agent.metric]]
    name  = "delayed_produce_requests"
    mbean = "kafka.server:type=DelayedOperationPurgatory,name=NumDelayedProduceRequests"
    paths = ["Value"]

  ## ── topic offsets for backlog calculation (tagged by topic & partition)
  [[inputs.jolokia2_agent.metric]]
    name     = "log_end_offset"
    mbean    = "kafka.log:type=Log,topic=*,partition=*,name=LogEndOffset"
    paths    = ["Value"]
    tag_keys = ["topic","partition"]

  [[inputs.jolokia2_agent.metric]]
    name     = "log_start_offset"
    mbean    = "kafka.log:type=Log,topic=*,partition=*,name=LogStartOffset"
    paths    = ["Value"]
    tag_keys = ["topic","partition"]

# ── inputs: basic host stats (optional but nice) ─────────────────────────
[[inputs.mem]]
[[inputs.net]]