Trazas y Métricas de Kafka con OpenTelemetry

12 de abril de 2023

por Leonardo Torres Altez

• Kafka
• Confluent
• Observabilidad
• OpenTelemetry

En este artículo voy a explicar cómo exponer las trazas y métricas de una aplicacion consumidora o productora Kafka.

Introducción

La Observabilidad es una característica del software cada vez más importante por la creciente cantidad de datos de las aplicaciones desplegadas en cloud como microservicios.

La observabilidad se compone de tres características de software: que se pueda generar y buscar logs, que se pueda medir el rendimiento (métricas) y que se pueda ver logs de una petición y su contexto (trazas). En esta ocasión voy a configurar métricas y trazas.

Las métricas son los indicadores de performance. Usualmente el cliente Kafka ya las implementa out-of-the-box, como es el caso de Spring Kafka. En este artículo explico cómo implementar métricas custom.

Las trazas, a diferencia de las métricas y logs, nos indican por qué componente/microservicio/clase ha pasado un mensaje específico junto con su contexto asociado. Por ejemplo: fecha, hora, instancia de microsevicio, entorno,etc. Las trazas son muy útiles para troubleshoting.

OpenTelemetry es una aplicación impulsada por CNCF y es un estándar para transportar métricas y trazas usando el protocolo abierto OLTP. Esto hace que cualquier sistema de tipo dashboad/indexador/alertas pueda explotar las métricas y trazas, como por ejemplo Jaeger, Elastic, Prometheus, Kibana. La página oficial es: https://opentelemetry.io/

La aplicación de ejemplo

En la aplicación de ejemplo tenemos los siguientes microservicios:

• Kafkaproducer: kafka client encargado de producir el mensaje "Books"
• Custommetricsdemo: kafka streams encargado de consumir el mensaje "Books".

Estos microservicios usan las siguientes piezas para implementar la observabilidad de trazas y métricas:

1. OpenTelemetry agent: agente java que instrumenta las llamadas hacia el api java kafka. En esta oportunidad muestro el como configurar el instrumentador automático usando el agente java. También es posible hacerlo manualmente, es decir creando las trazas/spans manualmente. Para lo cual seguir la siguiente página.

2. OpenTelemetry collector: servicio encargado de recibir las llamadas del agente y enviarlas hacia el sistema de dashboard/indexado/alertado. OpenTelemetry tiene soporte limitado para Kafka Streams, dada la complejidad que puede tener una topología kstreams. Por ejemplo los state stores no soportan el traspaso de headers kafka que es donde Otel guarda los metadatos de trazabilidad. Para ver el detalle del soporte OpenTelemtry para Kafka Streams recomiendo escuchar este podcast de Confluent.

3. Prometheus: servicio con dashboards ofrecidos por grafana y con alertado.

4. Jaeger: servicio con dashboard estilo árbol para las trazas.

Configuración

A continuación mencionamos las partes importantes del proyecto. Siempre tenemos disponible todo el código fuente en el link github.

https://github.com/leonardotorresaltez/kafka-opentelemetry

El fichero docker-compose con todos los servicios es el siguiente:

version: "2"
services:

  #primer microservicio productor kafka  
  kafkaproducer:
    build:
      dockerfile: Dockerfile
      context: ./kafkaproducer
    restart: always
    environment:
      - OTEL_EXPORTER_OTLP_ENDPOINT=otel-collector:4317
    ports:
      - "8080:8080"
    depends_on:
      - broker
      - schema-registry

  #segundo microservicio consumidor kafka    
  custom-metrics-demo:
    build:
      dockerfile: Dockerfile
      context: ./custom-metrics-demo
    restart: always
    environment:
      - OTEL_EXPORTER_OTLP_ENDPOINT=otel-collector:4317
    ports:
      - "8090:8090"
    depends_on:
      - broker
      - schema-registry

  #componente kafka confluent
  zookeeper:
    image: confluentinc/cp-zookeeper:6.2.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
  
  #componente kafka confluent
  broker:
    image: confluentinc/cp-server:6.2.1
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101" 
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  #componente kafka confluent
  schema-registry:
    image: confluentinc/cp-schema-registry:6.2.1
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081" 
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  #componente kafka confluent
  control-center:
    image: confluentinc/cp-enterprise-control-center:6.2.1
    hostname: control-center
    container_name: control-center
    depends_on:
      - broker
      - schema-registry
    ports:
      - "9021:9021"
    environment:
      CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
      CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONTROL_CENTER_REPLICATION_FACTOR: 1
      CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
      CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
      CONFLUENT_METRICS_TOPIC_REPLICATION: 1
      PORT: 9021


  # Jaeger, visualizador de trazas
  jaeger-all-in-one:
    image: jaegertracing/all-in-one:latest
    restart: always
    ports:
      - "16686:16686"
      - "14268:14268"
      - "14250"



  # Otel Collector
  otel-collector:
    image: ${OTELCOL_IMG}
    restart: always
    command: ["--config=/etc/otel-collector-config.yaml", "${OTELCOL_ARGS}"]
    volumes:
      - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml
    ports:
      - "1888:1888"   # pprof extension
      - "8888:8888"   # Prometheus metrics exposed by the collector
      - "8889:8889"   # Prometheus exporter metrics
      - "13133:13133" # health_check extension
      - "4317:4317"   # OTLP gRPC receiver
      - "55679:55679" # zpages extension
    depends_on:
      - jaeger-all-in-one
      - zipkin-all-in-one

  #visualizador de metricas 
  prometheus:
    container_name: prometheus
    image: prom/prometheus:latest
    restart: always
    volumes:
      - ./prometheus.yaml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"

A continuación, todas las partes implicadas:

Microservicios clientes kafka

Ambos microservicios: kafkaproducer y custometricsdemo, están iniciados con el agente OpenTelemetry que captura todas las llamadas del cliente kafka.

Ambos tienen que ser iniciados usando el agente y con dos variables de entorno:

• OTEL_EXPORTER_OTLP_ENDPOINT : indica el endpoint de Otel collector
• OTEL_SERVICE_NAME : indica el nombre del microservicio

Los demás parámetros se encuetran en esta página.

OpenTelemetry Collector

Es el colector que se encarga de agregar todas las trazas en formato único OLTP y enviarlas a los distintos sistemas de métricas/trazas.

El colector tiene un fichero de configuracion donde se indican:

• receivers: protocolos de mensajes que entran
• exporters: protocolos de mensajes que salen
• procesos: transformaciones de los mensajes

El fichero de configuración otel-collector-config.yaml es el siguiente:

receivers:
  otlp:
    protocols:
      grpc:

exporters:
  prometheus:
    endpoint: "0.0.0.0:8889"
    const_labels:
      label1: value1

  logging:


  jaeger:
    endpoint: jaeger-all-in-one:14250
    tls:
      insecure: true

processors:
  batch:

extensions:
  health_check:
  pprof:
    endpoint: :1888
  zpages:
    endpoint: :55679

service:
  extensions: [pprof, health_check]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [logging, jaeger]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [logging, prometheus]

Como se puede ver, se configuran los receivers y exporters.

Jaeger

Es la aplicación que muestra las trazas en una vista de grafo para ver por donde fue cada llamada y su contexto asociado. Está configurado como una de las salidas de Otel collector.

Prometheus

Es la aplicación que muestra las métricas. Está configurado como una de las salidas de Otel collector.

El fichero de configuración prometheus.yml es el siguiente:

scrape_configs:
  - job_name: 'otel-collector'
    scrape_interval: 10s
    static_configs:
      - targets: ['otel-collector:8889']
      - targets: ['otel-collector:8888']
  - job_name: 'kafkaproducer'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['kafkaproducer:8080']
  - job_name: 'custom-metrics-demo'
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['custom-metrics-demo:8090'] 

Como se puede ver, se configuran los endpoints donde se recogen las métricas que corresponden a cada microservicio.

Cómo ejecutar la aplicación de ejemplo

Para ejecutar el ejemplo seguir los siguientes pasos:

--paso 1 descargar codigo fuente 
git clone https://github.com/leonardotorresaltez/kafka-opentelemetry
--paso 2 compilar los proyectos de microservicios
> mvn package
--paso 3 inicar los servicios
>docker-compose up -d
--paso 4 provocar la produccion de un mensaje
> curl http://localhost:8080/produce-fake-data
--paso 5 ver la consola de jaeger y prometheus las trazas y métricas

La vista en Jaeguer http://localhost:16686/ : Como podemos ver la traza de producción de mensaje rest se propaga hacia en micro productor y luego consumidor kafka. Podemos cuánto ha tardado en cada caso y su contexto asociado.

La vista en Prometheus http://localhost:9090/ : Como podemos ver las métricas kafka están disponibles.

Conclusión

La observabilidad es una característica del software cada vez mas importante y hay muchas soluciones distintas en el mercado, lo cual hace difícil su integración con nuestras aplicaciones. Aquí es donde OpenTelemetry tiene su lugar haciendo de 'puente' y agregador de métricas y trazas para enviarlas a distintas soluciones.