kafka connect integration with datadog

[amitca71]

Hi,
I use MSK and deployed strimzi KafkaConnect (my kafka cluster in on MSK). I'm able to get the metrics, but was not able to see them in datadog (i followed the datadog tutorial, and it worked: https://docs.datadoghq.com/agent/kubernetes/prometheus/), but couldnt make it happen with the strimzi prometheous... i added the annotation part to the prometheus operator, but this didnt work. i tryed to add it to prometheus.yaml, but then i got : error: error validating "prometheus.yaml": error validating data: ValidationError(Prometheus.spec): unknown field "template" in com.coreos.monitoring.v1.Prometheus.spec; if you choose to ignore these errors, turn validation off with --validate=false
Thanks,
Amit

[iMajna]

Well, I believe you need to approach it differently.
I haven't tried it but I have done it for Kafka/Zookeeper/CruiseControl and many other and for each worked out.

First of all you need to enable metrics exporting if it is not by default:

In your KafkaConnect CR under spec.kafka:

    jmxOptions: {}
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          key: kafka-metrics-config.yml
          name: kafka-zk-metrics

and then again under spec.kafka you would have something like this:

    template:
      pod:
        metadata:
          annotations:
            ad.datadoghq.com/kafka-connect.check_names: '["jmx"]'
            ad.datadoghq.com/kafka.init_configs: '[{"is_jmx": true, "collect_default_metrics": true}]'
            ad.datadoghq.com/kafka.instances: '[{"host": "%%host%%","port":"<port>"}]'

Now, this will only extract metrics about resources such as cpu/memory and etc.

If you want more metrics related to kafkaconnect producer and consumer (connectors) you have 2 options:

1. Openmetrics (this is generating custom metrics/tags and it can be quite $$$)
2. Potentially Kafka plugin of Datadog?

OpenMetrics

This is how I did it for Zookeeper:

    template:
      pod:
        metadata:
          annotations:
            ad.datadoghq.com/zookeeper.check_names: '["openmetrics"]'
            ad.datadoghq.com/zookeeper.init_configs: '[{}]'
            ad.datadoghq.com/zookeeper.instances: |
              [
                {
                  "prometheus_url": "http://%%host%%:9404/metrics",
                  "namespace": "zookeeper",
                  "metrics": [
                    "zookeeper_leader",
                    "process_open_fds",
                    "process_max_fds",
                    "zookeeper_tick",
                    "zookeeper_quorumsize",
                    "zookeeper_avgrequestlatency"
                  ]
                }
              ]
            ad.datadoghq.com/zookeeper.logs: '[{"source":"zookeeper","service":"zookeeper"}]'

Kafka plugin from Datadog

    template:
      pod:
        metadata:
          annotations:
            ad.datadoghq.com/kafka.check_names: '["kafka"]'
            ad.datadoghq.com/kafka.init_configs: |
              [
                {
                  "is_jmx": true,
                  "collect_default_metrics": true,
                  "conf": [
                    {
                      "include": {
                        "domain": "kafka.log",
                        "bean_regex": "kafka\\.log:type=Log,name=Size,topic=(.*?),partition=(.*?)(?:,|$)",
                        "tags": { "topic": "$1", "partition": "$2" },
                        "attribute": {
                          "Value": {
                            "alias": "kafka.log.partition.size",
                            "metric_type": "gauge"
                          }
                        }
                      }
                    },
....
....
            ad.datadoghq.com/kafka.instances: '[{"host": "%%host%%","port":"9999"}]'

I would only change domains for kafkaconnect related ones, maybe it would work (haven't tried but in process in couple of days from now)

Hope some of this helps you. Hit me up if you have some questions.

[hanlin1412sp]

Hi Ivan Majnarić,

I have a kafka connect and snowflake-sink connector working by using strimzi in my local machine. I am trying to integrate datadog to get jmx metrics from kafak-cluster and sink connector. Any articles you would like me to recommend to follow how to install datadog-agent with strimzi into k8s and how to integrate datadog to get the metrics?

Thank you.

[romankydybets]

this is how we did
we have this comment
https://github.com/orgs/strimzi/discussions/4287#discussioncomment-931377
and datadog docs about openmetrics
https://docs.datadoghq.com/containers/kubernetes/prometheus/?tab=kubernetesadv2

also keep we allow our datadog agent take this metrics from strimzi namespace

here is the snipped or kafka-exporter from

kafkaExporter:
    template:
      pod:
        # taking this from this thread
        # https://github.com/orgs/strimzi/discussions/4287#discussioncomment-931377
        metadata:
            annotations: 
              ad.datadoghq.com/<kafka-exporter-container-name>.checks: |
                {
                  "openmetrics": {
                    "init_config": {},
                    "instances": [
                      {
                        "openmetrics_endpoint": "http://%%host%%:9404/metrics",
                        "namespace": "<your namespace>",
                        "metrics": [
                          {"go_gc_duration_seconds":"go_gc_duration_seconds"},
                          {"go_goroutines":"go_goroutines"},
                          {"go_memstats_alloc_bytes":"go_memstats_alloc_bytes"},
                          {"go_memstats_alloc_bytes_total":"go_memstats_alloc_bytes_total"},
                          {"go_memstats_buck_hash_sys_bytes":"go_memstats_buck_hash_sys_bytes"},
                          {"go_memstats_frees_total":"go_memstats_frees_total"},
                          {"go_memstats_gc_sys_bytes":"go_memstats_gc_sys_bytes"},
                          {"go_memstats_heap_alloc_bytes":"go_memstats_heap_alloc_bytes"},
                          {"go_memstats_heap_idle_bytes":"go_memstats_heap_idle_bytes"},
                          {"go_memstats_heap_inuse_bytes":"go_memstats_heap_inuse_bytes"},
                          {"go_memstats_heap_objects":"go_memstats_heap_objects"},
                          {"go_memstats_heap_released_bytes_total":"go_memstats_heap_released_bytes_total"},
                          {"go_memstats_heap_sys_bytes":"go_memstats_heap_sys_bytes"},
                          {"go_memstats_last_gc_time_seconds":"go_memstats_last_gc_time_seconds"},
                          {"go_memstats_lookups_total":"go_memstats_lookups_total"},
                          {"go_memstats_mallocs_total":"go_memstats_mallocs_total"},
                          {"go_memstats_mcache_inuse_bytes":"go_memstats_mcache_inuse_bytes"},
                          {"go_memstats_mcache_sys_bytes":"go_memstats_mcache_sys_bytes"},
                          {"go_memstats_mspan_inuse_bytes":"go_memstats_mspan_inuse_bytes"},
                          {"go_memstats_mspan_sys_bytes":"go_memstats_mspan_sys_bytes"},
                          {"go_memstats_next_gc_bytes":"go_memstats_next_gc_bytes"},
                          {"go_memstats_other_sys_bytes":"go_memstats_other_sys_bytes"},
                          {"go_memstats_stack_inuse_bytes":"go_memstats_stack_inuse_bytes"},
                          {"go_memstats_stack_sys_bytes":"go_memstats_stack_sys_bytes"},
                          {"go_memstats_sys_bytes":"go_memstats_sys_bytes"},
                          {"kafka_brokers":"kafka_brokers"},
                          {"kafka_consumergroup_current_offset":"kafka_consumergroup_current_offset"},
                          {"kafka_consumergroup_lag":"kafka_consumergroup_lag"},
                          {"kafka_exporter_build_info":"kafka_exporter_build_info"},
                          {"kafka_topic_partition_in_sync_replica":"kafka_topic_partition_in_sync_replica"},
                          {"kafka_topic_partition_current_offset":"kafka_topic_partition_current_offset"},
                          {"kafka_topic_partition_leader":"kafka_topic_partition_leader"},
                          {"kafka_topic_partition_leader_is_preferred":"kafka_topic_partition_leader_is_preferred"},
                          {"kafka_topic_partition_oldest_offset":"kafka_topic_partition_oldest_offset"},
                          {"kafka_topic_partition_replicas":"kafka_topic_partition_replicas"},
                          {"kafka_topic_partition_under_replicated_partition":"kafka_topic_partition_under_replicated_partition"},
                          {"kafka_topic_partitions":"kafka_topic_partitions"},
                          {"process_cpu_seconds_total":"process_cpu_seconds_total"},
                          {"process_max_fds":"process_max_fds"},
                          {"process_open_fds":"process_open_fds"},
                          {"process_resident_memory_bytes":"process_resident_memory_bytes"},
                          {"process_start_time_seconds":"process_start_time_seconds"},
                          {"process_virtual_memory_bytes":"process_virtual_memory_bytes"}
                        ]
                      }
                    ]
                  }
                }