UD 8 - Monitor Apache Kafka¶

1. Introducción¶

En un entorno de producción, la monitorización de Kafka (y cualquier otro de Big Data) es crucial para asegurar que el sistema esté funcionando de manera eficiente y sin interrupciones. Kafka es una plataforma de mensajería distribuida muy robusta, pero requiere una supervisión constante para garantizar que todos los componentes, como los brokers, controllers, productores y consumidores, estén operando correctamente. Sin una monitorización adecuada, es difícil detectar problemas de rendimiento, cuellos de botella, y errores en tiempo real que pueden afectar la calidad del servicio.

La monitorización de Kafka permite:

Rendimiento: Evaluar las métricas de throughput, latencia y el uso de recursos para garantizar un rendimiento óptimo.
Escalabilidad: Detectar problemas antes de que afecten a la escalabilidad del sistema.
Detección de fallos: Identificar problemas potenciales con los brokers, consumidores o productores que puedan estar interfiriendo con el flujo de datos.
Optimización: Tomar decisiones informadas sobre la configuración y la infraestructura para mejorar la eficiencia del sistema.

En esta guía, exploraremos cómo monitorizar un clúster de Kafka utilizando JMX (Java Management Extensions) para exponer métricas internas de Kafka, Prometheus para recolectar estas métricas, y Grafana para visualizar las métricas en paneles interactivos.

2. ¿Qué es JMX?¶

JMX (Java Management Extensions) es una tecnología de Java que permite gestionar y monitorizar aplicaciones, dispositivos y servicios basados en Java. Kafka, al ser una aplicación basada en Java, expone muchas de sus métricas internas a través de JMX.

JMX permite obtener métricas valiosas como el uso de CPU, la memoria, el rendimiento de los brokers, el rendimiento de los consumidores y los productores, la latencia, el número de particiones y más. Estas métricas son fundamentales para la administración eficiente de un clúster de Kafka.

3. ¿Qué es Prometheus?¶

Prometheus es una herramienta de monitoreo y recopilación de métricas de código abierto, diseñada para almacenar series temporales de datos. Funciona extrayendo métricas de las aplicaciones a través de un formato estándar que puede ser consultado mediante su lenguaje de consulta, PromQL.

Prometheus es ampliamente utilizado para la monitorización de sistemas distribuidos como Kafka debido a su capacidad para recolectar y almacenar grandes volúmenes de métricas de manera eficiente y en tiempo real.

Características principales de Prometheus:

Recolección de métricas basada en scrapes: Prometheus obtiene métricas mediante el mecanismo de scraping desde los endpoints HTTP de las aplicaciones.
Modelo de datos multidimensional: Las métricas son etiquetadas, lo que permite una rica semántica de consulta y filtrado.
Escalabilidad: Puede recolectar y almacenar grandes cantidades de datos sin comprometer el rendimiento.
Integración con Grafana (además de otros servicios): Prometheus se integra perfectamente con Grafana para crear dashboards visuales.

4. ¿Qué es Grafana?¶

Grafana es una plataforma de código abierto para la visualización y análisis de métricas. Es compatible con diversas fuentes de datos, y se usa principalmente para la visualización de series temporales de datos.

Cuando se combina con Prometheus, Grafana se convierte en una herramienta poderosa para representar visualmente las métricas de sistemas distribuidos como Kafka (y muchísimos otros). Con Grafana, puedes crear paneles interactivos y gráficos que muestran el rendimiento de Kafka en tiempo real, facilitando la detección de anomalías o problemas.

Características principales de Grafana:

Visualización de datos en tiempo real: Ofrece gráficos interactivos que se actualizan en tiempo real.
Paneles personalizables: Los usuarios pueden crear paneles personalizados según sus necesidades.
Soporte para múltiples fuentes de datos: Grafana puede conectarse a una amplia variedad de fuentes de datos, como Prometheus, Elasticsearch, InfluxDB, entre otras.

5. ¿Por qué usar JMX, Prometheus y Grafana para la monitorización de Kafka?¶

La combinación de JMX, Prometheus y Grafana es una solución altamente eficiente para la monitorización de Kafka por las siguientes razones:

JMX permite acceder directamente a las métricas internas de Kafka, proporcionando una visión profunda de su rendimiento y estado.
Prometheus es ideal para recolectar y almacenar estas métricas de manera eficiente, soportando grandes volúmenes de datos y permitiendo consultas de alto rendimiento con PromQL.
Grafana es una herramienta muy completa para visualizar las métricas recolectadas, permitiendo una monitorización en tiempo real con dashboards altamente personalizables y fáciles de usar.

Esta combinación ofrece una solución completa y escalable para monitorizar un clúster de Kafka, ayudando a los administradores a detectar problemas rápidamente, optimizar el rendimiento y garantizar la disponibilidad continua del sistema.

6 Funcionamiento JMX + Prometheus + Grafana¶

6.1 Conectando JMX y Prometheus¶

Sin entrar en mucho detalle, el flujo de monitorización con JMX, Prometheus y Grafana es el siguiente:

Kafka expone una variedad de métricas a través de JMX, las cuales están relacionadas con los brokers, productores, consumidores, etc.

Por otro lado, Prometheus es un ecosistema con dos componentes principales:

el componente del lado del servidor
la configuración del lado del cliente.

El componente del lado del servidor se encarga de almacenar todas las métricas y de realizar el scraping de todos los clientes. Prometheus se diferencia de servicios como Elasticsearch y Splunk, que suelen utilizar un componente intermedio responsable de extraer los datos de los clientes y enviarlos a los servidores. Dado que no hay ningún componente intermedio que extraiga las métricas de Prometheus, todas las configuraciones relacionadas con las encuestas están presentes en el propio servidor.

El proceso tiene este aspecto:

Figura 8.1_Monitor_Kafka: Prometheus (Fuente: confluent)

Hay dos piezas centrales en este diagrama:

Servidor Prometheus: Este componente es responsable de sondear todos los procesos/clientes con sus métricas expuestas en un puerto específico. El servidor de Prometheus mantiene internamente un archivo de configuración que enumera todas las direcciones IP/nombres de host del servidor y los puertos en los que se exponen las métricas de Prometheus. La configuración de objetivos de scrape es el archivo que mantiene toda la asignación de objetivos dentro de Prometheus. Los objetivos de scrape son necesarios cuando estamos desplegando todo manualmente sin ningún tipo de automatización. Prometheus también admite módulos de descubrimiento de servicios, que puede aprovechar para descubrir cualquier servicio disponible que esté exponiendo métricas.
Procesos cliente: Todos los clientes que quieran aprovechar Prometheus necesitarán dos piezas de configuración. En primer lugar, deben utilizar la biblioteca de clientes de Prometheus para exponer métricas en un formato compatible con Prometheus (OpenMetrics). En segundo lugar, deben utilizar un archivo de configuración YAML para extraer métricas JMX. Este archivo de configuración se utiliza para convertir, renombrar y filtrar algunos de los atributos para su consumo.

6.2 Connecting Grafana a Prometheus¶

Ahora que tenemos nuestros datos de métricas transmitiéndose al servidor Prometheus, podemos empezar a crear paneles de control de nuestras métricas. La herramienta elegida en nuestra pila es Grafana. Conceptualmente, este es el aspecto que tendrá el proceso una vez que hayamos conectado Grafana a Prometheus:

Figura 8.2_Monitor_Kafka: Connecting Prometheus y Grafana (Fuente: confluent)

Hay dos maneras de conectar Grafana con Prometheus: Podemos configurar la conexión desde la GUI de Grafana, o podemos añadir los detalles de la conexión a las configuraciones de Grafana antes del arranque. Hay artículos muy detallados disponibles en la documentación de Grafana. Seguiremos esta documentación en el siguiente apartado donde explicamos la instalación y configuración.

7. Configuración de JMX + Prometheus¶

7.1 Habilitar JMX en Kafka¶

En primer lugar, necesitamos añadir la librería JMX Exporter. Descarga la última versión.

wget https://github.com/prometheus/jmx_exporter/releases/download/1.2.0/jmx_prometheus_javaagent-1.2.0.jar

Guardamos la librería en el directorio de librerías de Kafka

mv jmx_prometheus_javaagent-1.2.0.jar /opt/kafka_2.13-4.0.0/libs

Necesitamos configurar nuestro JMX Exporter para que sepa lo que va a extraer de Kafka. Para explicarlo brevemente, la configuración es una colección de regexps que nombra y filtra las métricas para Prometheus. Gracias a Prometheus, tenemos configuraciones de ejemplo en su repositorio de Github. Usaremos la configuración de ejemplo mas reciente (kafka-kraft-3_0_0.yml) en esta configuración.

wget https://raw.githubusercontent.com/prometheus/jmx_exporter/refs/heads/main/examples/kafka-kraft-3_0_0.yml

La movemos a la carpeta de configuración de kafka

mv kafka-kraft-3_0_0.yml /opt/kafka_2.13-4.0.0/config/jmx-exporter-kafka.yml

Añadimos JMX a Kafka. Para ello, tenemos que indicarlo cada vez que usamos kafka. Por tanto, debemos abrir el script de inicio de los servidores de kafka (kafka-server-start.sh) y añadir JMX Exporter con su configuración correspondiente.
Primero hacemos una copia de respaldo y abrimos el script

cp /opt/kafka_2.13-4.0.0/bin/kafka-server-start.sh /opt/kafka_2.13-4.0.0/bin/kafka-server-start.sh.bak
nano /opt/kafka_2.13-4.0.0/bin/kafka-server-start.sh

Añadimos al final la configuración de JMX Exporter. Debemos colocar esta línea antes de la linea de la ejecución de exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@". Por tanto, la colocamos, por ejemplo, al principio. Hemos elegido el puerto 9091 para exponer los datos a Prometheus. Puedes usar el que quieras

export KAFKA_OPTS="-javaagent:/opt/kafka_2.13-4.0.0/libs/jmx_prometheus_javaagent-1.2.0.jar=9091:/opt/kafka_2.13-4.0.0/config/jmx-exporter-kafka.yml"

En el caso de que tuviéramos kafka en ejecución, lo reiniciamos.

sudo systemctl daemon-reload
sudo systemctl restart kafka

7.2 Prometheus¶

Descarga la última versión LTS de Prometheus

wget https://github.com/prometheus/prometheus/releases/download/v2.53.4/prometheus-2.53.4.linux-amd64.tar.gz
tar -xzf prometheus-2.53.4.linux-amd64.tar.gz

Movemos el directorio a nuestro directorio /opt para una correcta organización

sudo mv prometheus-2.53.4.linux-amd64 /opt/prometheus-2.53.4

Accedemos al directorio de Prometheus

ls /opt/prometheus-2.53.4

Vemos que existe un fichero de configuración prometheus.yml. Hacemos una copia de respaldo

cp /opt/prometheus-2.53.4/prometheus.yml /opt/prometheus-2.53.4/prometheus.yml.bak

Abrimos el archivo de configuración

nano /opt/prometheus-2.53.4/prometheus.yml

Como vemos, ya existe una configuración de scraping del propio prometheus en job_name. Como puedes observar, para añadir diferentes scrapings(en este caso para nuestro JMX Exporter de Kafka), debemos añadir un configurar en un nuevo job_name. Añadimos la configuración indicando también el correspondiente endpoint en el puerto 9091.
Por tanto, añadimos debajo (recuerda respetar tabulaciones y formatos)

  - job_name: "kafka"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: ["localhost:9091"]

El resultado sería.

# my global config
global:
  scrape_interval: 15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
  # scrape_timeout is set to the global default (10s).

# Alertmanager configuration
alerting:
  alertmanagers:
    - static_configs:
        - targets:
          # - alertmanager:9093

# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
  # - "first_rules.yml"
  # - "second_rules.yml"

# A scrape configuration containing exactly one endpoint to scrape:
# Here it's Prometheus itself.
scrape_configs:
  # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
  - job_name: "prometheus"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: ["localhost:9090"]

  - job_name: "kafka"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: ["localhost:9091"]

En el caso de que tuviéramos prometheus en ejecución, lo reiniciamos.

sudo systemctl daemon-reload
sudo systemctl restart prometheus

8. Ejemplo 1. JMX + Prometheus¶

Antes de seguir con la configuración con Grafana, vamos a comprobar que nuestra configuración es correcta. Lanzamos un cluster de kafka en modo server y vemos si JMX Exporter da métricas correctamente y éstas son recogidas por prometheus.
Iniciamos Kafka

#Genera un cluster UUID 
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

#Si no te permite formatear (con el comando siguiente) es debido a que se mantienen los logs del ejemplo anterior. debes borrarlos
#sudo rm -r /tmp/kraft-combined-logs/

#Formatea el directorio de log
bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

#Ejecuta el servidor Kafka
bin/kafka-server-start.sh config/server.properties

Comprobamos que se ha expuesto correctamente el endpoint en 9091:

sudo ss -tunelp | grep 9091

tcp   LISTEN 0      3                  *:9091             *:*    users:(("java",pid=9692,fd=115)) uid:1000 ino:34770 sk:f cgroup:/user.slice/user-1000.slice/session-1.scope v6only:0 <->

Y comprobamos que JMX Exporter está funcionando correctamente. Podemos hacerlo con curl o en el navegador en el puerto :9091/metrics

curl http://localhost:9091/metrics

# TYPE jmx_config_reload_failure_total counter
jmx_config_reload_failure_total 0.0
# HELP jmx_config_reload_success_total Number of times configuration have successfully been reloaded.
# TYPE jmx_config_reload_success_total counter
jmx_config_reload_success_total 0.0
# HELP jmx_exporter_build_info JMX Exporter build information
# TYPE jmx_exporter_build_info gauge
jmx_exporter_build_info{name="jmx_prometheus_javaagent",version="1.2.0"} 1
# HELP jmx_scrape_cached_beans Number of beans with their matching rule cached
# TYPE jmx_scrape_cached_beans gauge
jmx_scrape_cached_beans 0.0
# HELP jmx_scrape_duration_seconds Time this JMX scrape took, in seconds.
# TYPE jmx_scrape_duration_seconds gauge
jmx_scrape_duration_seconds 0.667689572
# HELP jmx_scrape_error Non-zero if this scrape failed.
# TYPE jmx_scrape_error gauge
jmx_scrape_error 0.0
# HELP jvm_buffer_pool_capacity_bytes Bytes capacity of a given JVM buffer pool.
# TYPE jvm_buffer_pool_capacity_bytes gauge
jvm_buffer_pool_capacity_bytes{pool="direct"} 1574526.0
jvm_buffer_pool_capacity_bytes{pool="mapped"} 2.0971516E7
jvm_buffer_pool_capacity_bytes{pool="mapped - 'non-volatile memory'"} 0.0
# HELP jvm_buffer_pool_used_buffers Used buffers of a given JVM buffer pool.
# TYPE jvm_buffer_pool_used_buffers gauge
jvm_buffer_pool_used_buffers{pool="direct"} 11.0
jvm_buffer_pool_used_buffers{pool="mapped"} 2.0
jvm_buffer_pool_used_buffers{pool="mapped - 'non-volatile memory'"} 0.0
# HELP jvm_buffer_pool_used_bytes Used bytes of a given JVM buffer pool.
# TYPE jvm_buffer_pool_used_bytes gauge
jvm_buffer_pool_used_bytes{pool="direct"} 1574526.0
jvm_buffer_pool_used_bytes{pool="mapped"} 2.0971516E7
jvm_buffer_pool_used_bytes{pool="mapped - 'non-volatile memory'"} 0.0
# HELP jvm_classes_currently_loaded The number of classes that are currently loaded in the JVM
# TYPE jvm_classes_currently_loaded gauge
jvm_classes_currently_loaded 9132.0

Iniciamos Prometheus. Vamos a su directorio /opt/prometheus-2.53.4. Levantamos prometheus

./prometheus --config.file=prometheus.yml

Comprobamos que prometheus captura correctamente el endpoint http://192.168.56.10:9090/targets (recuerda que accedemos desde nuestro host con esta configuración de red a nuestra máquina)

Figura 8.3_Monitor_Kafka: Ejemplo1: WebUI Prometheus

También podemos comprobar que los datos de kafka se capturan correctamente desde Prometheus buscando kafka.

Figura 8.4_Monitor_Kafka: Ejemplo1: WebUI Prometheus 2

Puedes elegir cualquiera de las métricas

Figura 8.5_Monitor_Kafka: Ejemplo1: WebUI Prometheus 3

9. Configuración de Grafana¶

Toda la documentación de Grafana OSS la tenemos en su página oficial
Descarga e instala la última versión de Grafana.

sudo apt-get install -y adduser libfontconfig1 musl
wget https://dl.grafana.com/enterprise/release/grafana-enterprise_11.6.1_amd64.deb
sudo dpkg -i grafana-enterprise_11.6.1_amd64.deb

Inicia Grafana

systemctl start grafana-server

Grafana como servicio

Si quieres habilitar grafana para que arranque automáticamente usando systemd, ejecuta:

sudo systemctl daemon-reload
sudo systemctl enable grafana-server

Accedemos a su WebUI en el puerto 3000 http://localhost:3000, en nuestro caso http://192.168.56.10:3000/ desde el host
Entramos con las credenciales usuario admin y pass admin. Cámbiala a tu criterio.

Figura 8.6_Monitor_Kafka: Grafana y Prometheus 1

Vamos a añadir los datos de monitorización de Kafka (Recuerda tener levantado kafka y prometheus)
Vamos a Connections -> Data sources -> Add data source

Figura 8.7_Monitor_Kafka: Grafana y Prometheus 2

Elegimos Prometheus y añadimos la configuración http://localhost:9090

Figura 8.8_Monitor_Kafka: Grafana y Prometheus 3

Marcamos "Save & test"

Figura 8.9_Monitor_Kafka: Grafana y Prometheus 4

Ahora necesitamos crear un dashboard para monitorizar visualmente los datos en tiempo real. Podemos crearlos nosotros manualmente, o usar algunas ya creados para importarlos. Puedes consultar dashboard en la página oficial de dashboards de grafana. Elegimos el que más nos interese de kafka
Si queremos elegir uno sólo habría que añadir el ćodigo identificador. Para que veamos como sería cargar uno, usaremos como ejemplo este o este
Para ello nos vamos a Dashboards -> New -> Import

Figura 8.10_Monitor_Kafka: Grafana y Prometheus 5

Cargamos el dashboard indicando el id del dashboard. En nuestro caso el 11962 o 18276
Marcamos el identificador y Load

Figura 8.11_Monitor_Kafka: Grafana y Prometheus 5

Configuramos el dashboard. Hay que indicarle nuestro data source prometheus.

Figura 8.12_Monitor_Kafka: Grafana y Prometheus 7

Versiones

Hay que tener en cuenta que cada dashboards está basado en unas versiones determinadas, tanto de la versión de kafka, como cual ha sido la configuración de la plantilla del scraping, si hemos usado JVM Exporter o Kafka Exporter,... (kafka-kraft-3_0_0.yml en kafka 4.0 y JMX Exporter en nuestro caso). Por tanto, ninguno de estos dashboards nos funcionaría. Pero si es interesante saber como hacerlo para cuando haya alguno o queramos usarlo para otro tipo de servicio

Haz lo mismo con el otro dashboard o el que quieras usar según tus necesidades.
Nosotros vamos a desarrollar nuestro propio dashboard. Para ver las métricas que podemos añadir, podemos verlas directamente desde prometheus. Pare ello nos vamos a Graph http://192.168.56.10:9090/graph y le damos al botón de open metrics explorer que está justo antes del botón "execute"

Figura 8.13_Monitor_Kafka: Grafana y Prometheus 8

He generado un pequeño dashboard para que puedas importarlo y usarlo. Haz las modificaciones que creas oportunas. Recuerda, este dashboard está configurado y debe ser usado para nuestra configuración de JMX Exporter + Kafka 4.0 + kafka-kraft-3_0_0.yml. Puedes descargar el json e importarlo como nuevo dashboard
Observa la captura de como se vería para este pequeño ejemplo. Al no añadir ningún topic, consumidor,... no están completos los datos del dashboards. Este ejemplo nos sirve para ver como funcionan los dashboards y preparar una plantilla. En el siguiente ejemplo ejecutaremos un ejemplo más completo para verlo en funcionamiento.

Figura 8.14_Monitor_Kafka: Grafana y Prometheus 9

10. Ejemplo 2. JMX + Prometheus + Grafana¶

Para ver grafana en funcionamiento, vamos a ejecutar el Ejemplo 3 de Kafka de la unidad anterior, donde configuramos 3 controller con 3 brokers, 1 topic junto a un consumidor y un productor.

Severs y puertos de kafka para el ejemplo

Para este caso, para poder realizar el ejemplo, tenemos 3 opciones

Levantar un server en cada una de las máquinas que tenemos y usar el mismo puerto para JMX
Levantar todos los servers de Kafka en una máquina y usar un puerto para cada uno de ellos
Levantar todo el stack con docker (Se desarrollará en la siguiente Unidad)

Ambos casos tienen sus dificultades. En el primero necesitamos configurar el cluster de Kafka cada uno en un nodo, por ejemplo, en master 1 controller, y un broker en cada nodo (nodo1-> broker1, nodo2 -> broker2,..). En el segundo, el script de inicio de los servers de kafka debe tener mayor configuración, ya que debemos indicar en el script que arranque cada server en un puerto diferente.

Nosotros, elegiremos la segunda opción, así podemos establecer un ejemplo (NO EN PRODUCCIÓN) del funcionamiento de grafana con un posible cluster con quorum dinámico de varios controllers y brokers.

Necesitamos cambiar el script de arranque de los servidores de Kafka. Para ello vamos a hacer una copia del ya existente y vamos modificarlo. Este nuevo script será el que usemos para este ejemplo.

cp /opt/kafka_2.13-4.0.0/bin/kafka-server-start.sh /opt/kafka_2.13-4.0.0/bin/kafka-server-start_ejemplo2_mon_kafka.sh

A abrimos el script

nano /opt/kafka_2.13-4.0.0/bin/kafka-server-start_ejemplo2_mon_kafka.sh

Sustituimos la línea de configuración de nuestro ejemplo anterior por el siguiente código.Hemos elegido los puertos 11000 para exponer los datos a Prometheus. Puedes usar los que quieras

# --- INICIO DE MODIFICACIÓN PARA JMX EXPORTER ---

# Establecemos KAFKA_OPTS por defecto si no está definido
KAFKA_OPTS=${KAFKA_OPTS:-}

# Encontrar la ruta al archivo server.properties (o cualquier archivo .properties)
PROPERTIES_FILE=""
for arg in "$@"; do
  if [[ "$arg" == *".properties" ]]; then
    PROPERTIES_FILE="$arg"
    break
  fi
done

# Variables para JMX Exporter
JMX_AGENT_PATH="/opt/kafka_2.13-4.0.0/libs/jmx_prometheus_javaagent-1.2.0.jar"
# Ruta al archivo de configuración del JMX Exporter. Si no usas uno, déjala vacía pero mantén el ':' en el argumento final.
JMX_CONFIG_FILE="/opt/kafka_2.13-4.0.0/config/jmx-exporter-kafka.yml"

# Lógica para determinar el puerto y configurar KAFKA_OPTS
NODE_ID=""
JMX_PORT=""
JMX_JAVAAGENT_ARG=""

# 1. Intentar extraer node.id del archivo properties
if [ -n "$PROPERTIES_FILE" ] && [ -f "$PROPERTIES_FILE" ]; then
    NODE_ID=$(grep "^[[:space:]]*node.id[[:space:]]*=" "$PROPERTIES_FILE" | head -1 | sed "s/^[[:space:]]*node.id[[:space:]]*=//" | cut -d'#' -f1 | tr -d '[:space:]')
fi

# 2. Validar NODE_ID y calcular JMX_PORT
if ! [[ "$NODE_ID" =~ ^[0-9]+$ ]]; then
    echo "WARNING: JMX Exporter: No se pudo determinar el node.id del archivo $PROPERTIES_FILE o no es un número válido ('$NODE_ID')." >&2
    echo "WARNING: JMX Exporter NO será cargado para esta instancia de Kafka." >&2
# 3. Validar si el JAR del JMX Exporter existe y es legible
elif [ ! -r "$JMX_AGENT_PATH" ]; then
    echo "ERROR: JMX Exporter: El archivo JAR NO se encuentra o no es legible: $JMX_AGENT_PATH" >&2
    echo "WARNING: JMX Exporter NO será cargado para esta instancia de Kafka." >&2
# 4. Validar si el archivo de configuración existe y es legible (si se especificó uno)
elif [ -n "$JMX_CONFIG_FILE" ] && [ ! -r "$JMX_CONFIG_FILE" ]; then
    echo "ERROR: JMX Exporter: El archivo de configuración NO se encuentra o no es legible: $JMX_CONFIG_FILE" >&2
    echo "WARNING: JMX Exporter NO será cargado para esta instancia de Kafka." >&2
# 5. Si todas las validaciones pasan, construir el argumento y configurar KAFKA_OPTS
else
    JMX_BASE_PORT=11000 # Puerto base, JMX_PORT = JMX_BASE_PORT + NODE_ID
    JMX_PORT=$((JMX_BASE_PORT + NODE_ID))

    # Construir el argumento del javaagent: =puerto:/ruta/config (o =puerto:)
    JMX_JAVAAGENT_ARG="-javaagent:$JMX_AGENT_PATH=$JMX_PORT"
    if [ -n "$JMX_CONFIG_FILE" ]; then
        JMX_JAVAAGENT_ARG="${JMX_JAVAAGENT_ARG}:${JMX_CONFIG_FILE}" # Añadir : y la ruta del archivo
    else
        JMX_JAVAAGENT_ARG="${JMX_JAVAAGENT_ARG}:" # Si no hay archivo, añadir solo el :
    fi


    # Añadir el argumento del javaagent a KAFKA_OPTS
    if [ -z "$KAFKA_OPTS" ]; then
      export KAFKA_OPTS="$JMX_JAVAAGENT_ARG"
    else
      export KAFKA_OPTS="$KAFKA_OPTS $JMX_JAVAAGENT_ARG" # Añadir un espacio antes por si ya hay opciones
    fi

    echo "DEBUG: JMX Exporter: Configurado en puerto $JMX_PORT para node.id=$NODE_ID." >&2
    echo "DEBUG: JMX Exporter: KAFKA_OPTS resultante: '$KAFKA_OPTS'" >&2
fi

# --- FIN DE MODIFICACIÓN PARA JMX EXPORTER ---

Hacemos una copia del fichero de configuración prometheus.yml para adaptarlo a esta configuración.

cp /opt/prometheus-2.53.4/prometheus.yml /opt/prometheus-2.53.4/prometheus_ejemplo2_mon_kafka.yml

Abrimos el archivo de configuración

nano /opt/prometheus-2.53.4/prometheus_ejemplo2_mon_kafka.yml

Añadimos los diferentes scrapings. La configuración de los endpoint a partir del puerto 11000.

  - job_name: "kafka"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: [
        "localhost:11001", # Controller 1 (node.id=1)
        "localhost:11002", # Controller 2 (node.id=2)
        "localhost:11003", # Controller 3 (node.id=3)
        "localhost:11005", # Broker 1 (node.id=5)
        "localhost:11006", # Broker 2 (node.id=6)
        "localhost:11007"  # Broker 3 (node.id=7)
      ]

El resultado sería.

# my global config
global:
  scrape_interval: 15s # Set the scrape interval to every 15 seconds. Default is every 1 minute.
  evaluation_interval: 15s # Evaluate rules every 15 seconds. The default is every 1 minute.
  # scrape_timeout is set to the global default (10s).

# Alertmanager configuration
alerting:
  alertmanagers:
    - static_configs:
        - targets:
          # - alertmanager:9093

# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
  # - "first_rules.yml"
  # - "second_rules.yml"

# A scrape configuration containing exactly one endpoint to scrape:
# Here it's Prometheus itself.
scrape_configs:
  # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
  - job_name: "prometheus"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: ["localhost:9090"]

  - job_name: "kafka"

    # metrics_path defaults to '/metrics'
    # scheme defaults to 'http'.

    static_configs:
      - targets: [
        "localhost:11001", # Controller 1 (node.id=1)
        "localhost:11002", # Controller 2 (node.id=2)
        "localhost:11003", # Controller 3 (node.id=3)
        "localhost:11005", # Broker 1 (node.id=5)
        "localhost:11006", # Broker 2 (node.id=6)
        "localhost:11007"  # Broker 3 (node.id=7)
      ]

Arrancamos el ejemplo
Generamos los uuid

#Generamos un cluster UUID y los IDs de los controllers
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_1_UUID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_2_UUID="$(bin/kafka-storage.sh random-uuid)"
CONTROLLER_3_UUID="$(bin/kafka-storage.sh random-uuid)"

10. Damos formato a los controllers de Kafka indicando los controllers iniciales (Recuerda la documentación)

#Formateamos los directorios de log indicando los controllers iniciales
#sudo rm -r /opt/kafka/ejemplo3/logs/*
#controller1
bin/kafka-storage.sh format --cluster-id ${KAFKA_CLUSTER_ID} \
                     --initial-controllers "1@localhost:9096:${CONTROLLER_1_UUID},2@localhost:9097:${CONTROLLER_2_UUID},3@localhost:9098:${CONTROLLER_3_UUID}" \
                     --config /opt/kafka/ejemplo3/config/controller1.properties

#controller2
bin/kafka-storage.sh format --cluster-id ${KAFKA_CLUSTER_ID} \
                     --initial-controllers "1@localhost:9096:${CONTROLLER_1_UUID},2@localhost:9097:${CONTROLLER_2_UUID},3@localhost:9098:${CONTROLLER_3_UUID}" \
                     --config /opt/kafka/ejemplo3/config/controller2.properties

#controller3
bin/kafka-storage.sh format --cluster-id ${KAFKA_CLUSTER_ID} \
                     --initial-controllers "1@localhost:9096:${CONTROLLER_1_UUID},2@localhost:9097:${CONTROLLER_2_UUID},3@localhost:9098:${CONTROLLER_3_UUID}" \
                     --config /opt/kafka/ejemplo3/config/controller3.properties

Damos formato a los logs de los brokers

#Formateamos los directorios de log de los brokers
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /opt/kafka/ejemplo3/config/broker1.properties
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /opt/kafka/ejemplo3/config/broker2.properties
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /opt/kafka/ejemplo3/config/broker3.properties

Iniciamos los servers(3 controllers y 3 brokers) cada uno en una terminal

#Ejecutamos los servidores Kafka (uno en cada terminal)
#Observa en el log al arrancar que indica que Levanta JMX y el puerto correspondiente al node.id
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/controller1.properties
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/controller2.properties
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/controller3.properties
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/broker1.properties
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/broker2.properties
bin/kafka-server-start_ejemplo2_mon_kafka.sh /opt/kafka/ejemplo3/config/broker3.properties

Comprobamos que se ha expuesto correctamente el endpoint en 1100*:

sudo ss -tunelp | grep 1100*

tcp   LISTEN 0      3                       *:11006            *:*    users:(("java",pid=11101,fd=115)) uid:1000 ino:39171 sk:4 cgroup:/user.slice/user-1000.slice/session-1.scope v6only:0 <-> 
tcp   LISTEN 0      3                       *:11007            *:*    users:(("java",pid=11562,fd=115)) uid:1000 ino:40394 sk:5 cgroup:/user.slice/user-1000.slice/session-7.scope v6only:0 <-> 
tcp   LISTEN 0      3                       *:11005            *:*    users:(("java",pid=10685,fd=115)) uid:1000 ino:38804 sk:6 cgroup:/user.slice/user-1000.slice/session-3.scope v6only:0 <-> 
tcp   LISTEN 0      3                       *:11002            *:*    users:(("java",pid=9819,fd=115)) uid:1000 ino:37271 sk:7 cgroup:/user.slice/user-1000.slice/session-5.scope v6only:0 <->  
tcp   LISTEN 0      3                       *:11003            *:*    users:(("java",pid=10266,fd=115)) uid:1000 ino:38499 sk:8 cgroup:/user.slice/user-1000.slice/session-4.scope v6only:0 <-> 
tcp   LISTEN 0      3                       *:11001            *:*    users:(("java",pid=9428,fd=115)) uid:1000 ino:38008 sk:9 cgroup:/user.slice/user-1000.slice/session-2.scope v6only:0 <->  
tcp   LISTEN 0      50     [::ffff:127.0.0.1]:9093             *:*    users:(("java",pid=11101,fd=156)) uid:1000 ino:39338 sk:c cgroup:/user.slice/user-1000.slice/session-1.scope v6only:0 <-> 
tcp   LISTEN 0      50                      *:36061            *:*    users:(("java",pid=11101,fd=120)) uid:1000 ino:39174 sk:e cgroup:/user.slice/user-1000.slice/session-1.scope v6only:0 <->

Iniciamos Prometheus. Vamos a su directorio /opt/prometheus-2.53.4. Levantamos prometheus

./prometheus --config.file=prometheus_ejemplo2_mon_kafka.yml

Figura 8.15_Monitor_Kafka: Ejemplo2: WebUI Prometheus

Iniciamos Grafana

systemctl start grafana-server

Creamos el topic con factor de replica 3 y 3 particiones. El topic debe conectarse a un broker.

bin/kafka-topics.sh --create --topic financial-transactions --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3

Podemos ver la descripción del topic creado

bin/kafka-topics.sh --describe --topic financial-transactions --bootstrap-server localhost:9092

bin/kafka-topics.sh --describe --topic financial-transactions --bootstrap-server localhost:9092
Topic: financial-transactions   TopicId: MjtMKW06RNyR0aZWDOiTdg PartitionCount: 3   ReplicationFactor: 3Configs: segment.bytes=1073741824
    Topic: financial-transactionsPartition: 0   Leader: 5   Replicas: 5,6,7 Isr: 5,6,7  Elr: LastKnownElr: 
    Topic: financial-transactionsPartition: 1   Leader: 6   Replicas: 6,7,5 Isr: 6,7,5  Elr: LastKnownElr: 
    Topic: financial-transactionsPartition: 2   Leader: 7   Replicas: 7,5,6 Isr: 7,5,6  Elr: LastKnownElr:

Comprobamos el dashboard de Grafana

Figura 8.16_Monitor_Kafka: Ejemplo2: Dashboard Grafana

Lanzamos el productor

python3 /opt/kafka/ejemplo3/producer.py

Lanzamos el consumidor

python3 /opt/kafka/ejemplo3/consumer.py