Surveiller Apache Kafka avec Telegraf et Grafana dans Azure

La surveillance contribue à stimuler la productivité et améliore la disponibilité des services. Apache Kafka ne fait pas exception à cela. Alors qu’Apache Kafka gagne en popularité et devient la norme de facto pour la messagerie distribuée dans l’architecture Microservice, il est important pour les ingénieurs DevOps de comprendre certaines métriques clés telles que – Lag du consommateur ? Partitions sous-répliquées ? Ou votre cluster Kafka est-il en panne ? – pour disposer d’un cluster Kafka hautement fiable et disponible et rester au top de la santé des données et des applications.

Outils pour surveiller Apache Kafka

Il existe de nombreux outils disponibles pour surveiller Apache Kafka, notamment :

Terrier de LinkedIn
Gestionnaire Yahoo Kafka
KafDrop
Centre de contrôle Confluent – si vous utilisez Confluent Kafka et de nombreux autres outils

Nous pouvons également surveiller Apache Kafka à l’aide de Java Management Extensions (JMX) avec des clients JMX populaires tels que Prometheus, Telegraf, etc.

Dans cet article, je vais parcourir la surveillance d’Apache Kafka déployé sur Azure Cloud à l’aide de Telegraf et Grafana. Notez qu’Azure fournit également Kafka managé à l’aide de HDInsight Kafka. Mais si vous souhaitez créer un cluster Apache Kafka indépendant du cloud, il est important de disposer d’une solution de surveillance pour le cluster. Dans cet article, je vais vous guider sur l’installation, la configuration et l’exécution de solutions de surveillance à l’aide de l’agent Telegraf et Grafana, agent Jolokia, Azure Monitor. En conséquence, nous verrons les métriques du système, Kafka Broker, Kafka Consumer et Kafka Producer sur notre tableau de bord côté Grafana.

Examinons certains de ces composants :

Agent Jolokia

Jolokia est une approche basée sur un agent JMX-HTTP prenant en charge de nombreuses plates-formes. En plus des opérations JMX de base, il améliore la communication à distance JMX avec des fonctionnalités uniques telles que les requêtes en masse et les politiques de sécurité affinées.

Télégraphe

Telegraf est un processus serveur léger capable de collecter, traiter et agréger des métriques. Telegraf est un agent serveur piloté par plug-in pour la collecte et l’envoi de métriques et d’événements à partir de bases de données, de systèmes et de capteurs IoT.

Grafana

Grafana est une solution de tableau de bord front-end open source complète et populaire. Il s’agit d’un outil de visualisation conçu pour fonctionner avec une variété de sources de données telles que Graphite, InfluxDB, Elasticsearch, etc. Cette solution permet le développement rapide et facile de tableaux de bord pour les utilisateurs.

Moniteur Azure

Azure Monitor collecte la télémétrie de surveillance à partir de diverses sources locales et Azure. Azure Monitor permet de collecter des données de performances et d’utilisation granulaires, des journaux d’activité et de diagnostic et des notifications à partir des ressources Azure.

Surveillance de bâtiment Ssolution

Il s’agit d’un diagramme d’architecture de haut niveau pour créer une solution de surveillance pour Apache Kafka hébergée sur des machines virtuelles Azure.

Ici, j’ai créé 2 sous-réseaux – un pour le cluster Kafka et un autre pour les solutions de surveillance mises en place, afin de séparer la responsabilité d’accès. Lorsque vous configurez une solution de surveillance dans un sous-réseau distinct, assurez-vous d’avoir mis à jour les règles entrantes/sortantes dans le NSG respectif des sous-réseaux pour avoir des communications appropriées. Le sous-réseau de surveillance aura une VM pour héberger les agents Grafana et Telegraf.

Pour gérer la haute disponibilité, vous devez créer 2 machines virtuelles et utiliser Azure Application Gateway pour équilibrer la charge. Parcourons la configuration :

1. Installer Jolokia

Téléchargez la dernière version de Jolokia JVM-Agent à partir d’ici. (la version jolokia-jvm-1.7.1-agent.jar est la plus récente à ce jour)
Enregistrez jolokia-jvm-1.7.1-agent.jar sur le serveur Kafka dans/opt/kafka/libs ou n’importe où.
Configurez Kafka pour utiliser Jolokia :

Ajoutez l’extrait de code suivant à kafka-server-start.sh :

            export JMX_PORT=9999 export RMI_HOSTNAME=<KAFKA_SERVER_IP_ADDRESS> export KAFKA_JMX_OPTS="-javaagent:/opt/kafka/libs/jolokia-agent.jar=port=8778,host=$RMI_HOSTNAME -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=$RMI_HOSTNAME -Dcom.sun.management.jmxremote.rmi.port=$JMX_PORT"

Redémarrez le service Kafka.
Vérifiez que vous pouvez accéder à Jolokia sur le port 8778 en exécutant :

Boucle – http://KAFKA_SERVER_IP_ADDRESS:8778/jolokia/version

2. Installer l’agent Telegraf

Cette intégration utilise le plugin d’entrée Jolokia pour Telegraf pour obtenir les métriques Kafka via JMX.

Vous pouvez télécharger le dernier agent Telegraf à partir d’ici.

3. Configurer le plug-in d’entrée Jolokia

Créer un fichier appelé jolokia-kafka.conf dans /etc/telegraf/telegraf.d et entrez ce qui suit :

Remarque : remplacez KAFKA_SERVER_IP_ADDRESS par l’adresse IP du serveur Kafka :

## Read JMX metrics through Jolokia
 [[inputs.jolokia2_agent]]
   ## An array of Kafka servers URI to gather stats.
   urls = ["http://KAFKA_SERVER_IP_ADDRESS:8778/jolokia"]
   name_prefix = "kafka."

   ## List of metrics collected on above servers
   [[inputs.jolokia2_agent.metric]]
     name = "heap_memory_usage"
     mbean  = "java.lang:type=Memory"
     paths = ["HeapMemoryUsage"]

   ## This collects thread counts metrics.
   [[inputs.jolokia2_agent.metric]]
     name = "thread_count"
     mbean  = "java.lang:type=Threading"
     paths = ["TotalStartedThreadCount","ThreadCount","DaemonThreadCount","PeakThreadCount"]

   ## This collects garbage collection metrics.
   [[inputs.jolokia2_agent.metric]]
     name = "garbage_collector"
     mbean  = "java.lang:type=GarbageCollector,name=*"
     paths = ["CollectionCount","CollectionTime"]
     tag_keys = ["name"]

   # Kafka Server Broker Topic Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "server_brokertopics_messagesinpersec"
     mbean  = "kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec"
   [[inputs.jolokia2_agent.metric]]
     name = "server_brokertopics_bytesinpersec"
     mbean  = "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec"
   [[inputs.jolokia2_agent.metric]]
     name = "server_brokertopics_bytesoutpersec"
     mbean  = "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec"

   # Kafka Server Request Handler Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "server_requesthandler_avgidlepct"
     mbean  = "kafka.server:name=RequestHandlerAvgIdlePercent,type=KafkaRequestHandlerPool"

   # Kafka Server Delayed Operation Purgatory Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "server_delayedoperationpugatory_fetch"
     mbean  = "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Fetch"
   [[inputs.jolokia2_agent.metric]]
     name = "server_delayedoperationpugatory_produce"
     mbean  = "kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=Produce"

   # Kafka Server Replica Fetcher Manager Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicafetchmanager.maxlag"
     mbean  = "kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica"

   # Kafka Server Replica Manager Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicamanager_underreplicated"
     mbean  = "kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions"
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicamanager_partitioncount"
     mbean  = "kafka.server:type=ReplicaManager,name=PartitionCount"
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicamanager_leadercount"
     mbean  = "kafka.server:type=ReplicaManager,name=LeaderCount"
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicamanager_isrshrinkspersec"
     mbean  = "kafka.server:type=ReplicaManager,name=IsrShrinksPerSec"
   [[inputs.jolokia2_agent.metric]]
     name = "server_replicamanager_isrexpandspersec"
     mbean  = "kafka.server:type=ReplicaManager,name=IsrExpandsPerSec"

   # Kafka Network Request Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_requests_fetch_consumer"
     mbean  = "kafka.network:type=RequestMetrics,name=RequestsPerSec,request=FetchConsumer"
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_requests_fetch_follower"
     mbean  = "kafka.network:type=RequestMetrics,name=RequestsPerSec,request=FetchFollower"
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_requests_produce"
     mbean  = "kafka.network:type=RequestMetrics,name=RequestsPerSec,request=Produce"
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_totaltime_fetch_consumer"
     mbean  = "kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchConsumer"
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_totaltime_fetch_follower"
     mbean  = "kafka.network:type=RequestMetrics,name=TotalTimeMs,request=FetchFollower"
   [[inputs.jolokia2_agent.metric]]
     name = "network_requestmetrics_totaltime_produce"
     mbean  = "kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce"

   # Kafka Network Processor Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "network_processor_avgidlepct"
     mbean  = "kafka.network:name=NetworkProcessorAvgIdlePercent,type=SocketServer"

   # Kafka Controller Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "controller_activecontrollers"
     mbean  = "kafka.controller:type=KafkaController,name=ActiveControllerCount"
   [[inputs.jolokia2_agent.metric]]
     name = "controller_offlinepartitions"
     mbean  = "kafka.controller:type=KafkaController,name=OfflinePartitionsCount"
   [[inputs.jolokia2_agent.metric]]
     name = "controller_stats_leaderelectionrateandtime"
     mbean  = "kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs"
   [[inputs.jolokia2_agent.metric]]
     name = "controller_stats_uncleanleaderelections"
     mbean  = "kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec"

   # Zookeeper Metrics
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_disconnects"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperDisconnectsPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_sync_connects"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperSyncConnectsPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_auth_failures"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_readonly_connects"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperReadOnlyConnectsPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_authentications"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperSaslAuthenticationsPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]
   [[inputs.jolokia2_agent.metric]]
     name = "zookeeper_expires"
     mbean  = "kafka.server:type=SessionExpireListener,name=ZooKeeperExpiresPerSec"
     paths = ["Count","OneMinuteRate","FiveMinuteRate","FifteenMinuteRate","MeanRate"]

Étendez les métriques ci-dessus pour toutes les métriques JMX afin d’avoir une vue complète d’Apache Kafka. Vous pouvez vous référer à la documentation Apache Kafka pour les points de terminaison et les métriques JMX.

4. Configurer le plug-in de sortie Azure Monitor

[[outputs.azure_monitor]]
  ## Timeout for HTTP writes.
  # timeout = "20s"

  ## Set the...