L'intégration Apache Kafka collecte des métriques d'agent, telles que les échecs et les requêtes de sujet. Il surveille également les partitions de l'agent. L'intégration collecte les journaux Kafka et les analyse dans une charge utile JSON. Le résultat inclut les champs de logger, de niveau et de message.
Pour en savoir plus sur Kafka, consultez la page kafka.apache.org/.
Prérequis
Pour collecter et ingérer des journaux et des métriques Kafka, vous devez installer l'agent Ops version 2.10.0 ou ultérieure.
Ce récepteur est compatible avec les versions 0.8 à 3.0.0 d'Apache Kafka.
Configurer votre instance Kafka
Pour exposer un point de terminaison JMX, vous devez définir la propriété système com.sun.management.jmxremote.port lors du démarrage de la JVM. Nous vous recommandons également de définir la propriété système com.sun.management.jmxremote.rmi.port sur le même port. Pour exposer un point de terminaison JMX à distance, vous devez également définir la propriété système java.rmi.server.hostname.
Par défaut, ces propriétés sont définies dans le fichier bin/kafka-run-class.sh d'un déploiement Kafka.
Pour définir les propriétés système à l'aide d'arguments de ligne de commande, ajoutez le préfixe -D au nom de la propriété lors du démarrage de la JVM. Par exemple, pour définir com.sun.management.jmxremote.port sur le port 9999, spécifiez les éléments suivants lors du démarrage de la JVM :
-Dcom.sun.management.jmxremote.port=9999
Configurer l'agent Ops pour Kafka
En suivant le guide de configuration de l'agent Ops, ajoutez les éléments requis pour collecter les journaux et les métriques de vos instances Kafka, puis redémarrez l'agent.
Exemple de configuration
La commande suivante crée le fichier de configuration permettant de collecter et d'ingérer des journaux et des métriques de Kafka, et de redémarrer l'agent Ops sous Linux.
sudo tee /etc/google-cloud-ops-agent/config.yaml > /dev/null << EOF
logging:
receivers:
kafka:
type: kafka
service:
pipelines:
kafka:
receivers:
- kafka
metrics:
receivers:
kafka:
type: kafka
service:
pipelines:
kafka:
receivers:
- kafka
EOF
sudo service google-cloud-ops-agent restart
Configurer la collecte de journaux
Pour ingérer des journaux à partir de Kafka, vous devez créer des récepteurs pour les journaux produits par Kafka, puis créer un pipeline pour les nouveaux récepteurs. Pour configurer un récepteur pour vos journaux kafka, spécifiez les champs suivants :
| Champ | Par défaut | Description |
|---|---|---|
type |
La valeur doit être égale à kafka. |
|
include_paths |
[/var/log/kafka/*.log] |
Liste des chemins d'accès du système de fichiers à lire en affichant chaque fichier. Un caractère générique (*) peut être utilisé dans les chemins d'accès. Exemple : /var/log/kafka*/*.log. |
exclude_paths |
Liste des formats de chemin d'accès au système de fichiers à exclure de l'ensemble correspondant à include_paths. |
|
wildcard_refresh_interval |
60s |
Intervalle d'actualisation pour les chemins d'accès de fichiers utilisant des caractères génériques dans include_paths. Renseigné sous la forme d'une durée, par exemple, 30s ou 2m. Cette propriété peut s'avérer utile lorsque le débit de journalisation est élevé et que les fichiers journaux sont alternés plus rapidement que l'intervalle par défaut. La valeur doit être un multiple de 1 s. |
Contenu consigné
Le logName des journaux kafka est dérivé des ID de récepteur spécifiés dans la configuration. Les champs détaillés dans l'entrée de journal (LogEntry) sont les suivants.
| Champ | Type | Description |
|---|---|---|
jsonPayload.source |
chaîne | Module et/ou thread d'où provient le journal. |
jsonPayload.logger |
chaîne | Nom du logger d'où provient le journal. |
jsonPayload.message |
chaîne | Message de journal, y compris la trace détaillée de la pile, le cas échéant. |
severity |
chaîne (LogSeverity) |
Niveau d'entrée de journal (traduit). |
timestamp |
chaîne (Timestamp) |
Heure de réception de la requête. |
Tous les champs vides ou manquants ne seront pas présents dans l'entrée de journal.
Configurer la collecte de métriques
Pour collecter des métriques à partir de Kafka, vous devez créer un récepteur pour les métriques Kafka, puis créer un pipeline pour le nouveau récepteur. Pour configurer un récepteur pour vos métriques Kafka, spécifiez les champs suivants :
| Champ | Par défaut | Description |
|---|---|---|
type |
La valeur doit être égale à kafka. |
|
stub_status_url |
localhost:9999 |
L'URL du service JMX, ou l'hôte et le port utilisés pour créer l'URL du service. Doit être au format service:jmx:<protocol>:<sap> ou host:port. Les valeurs du formulaire host:port seront utilisées pour créer une URL de service de service:jmx:rmi:///jndi/rmi://<host>:<port>/jmxrmi. |
collect_jvm_metrics |
true |
Cela configure le récepteur de manière à ce qu'il collecte également les métriques JVM compatibles. |
username |
Le nom d'utilisateur configuré si JMX est configuré pour exiger une authentification. | |
password |
Le mot de passe configuré si JMX est configuré pour exiger une authentification. | |
collection_interval |
60s |
Une valeur time.Duration, telle que 30s ou 5m. |
Métriques surveillées
Le tableau suivant fournit la liste des métriques que l'agent Ops collecte à partir de l'instance Kafka.
| Type de métrique | |
|---|---|
| Genre, type Ressources surveillées |
Libellés |
workload.googleapis.com/kafka.isr.operation.count
|
|
CUMULATIVE, INT64gce_instance |
operation
|
workload.googleapis.com/kafka.message.count
|
|
CUMULATIVE, INT64gce_instance |
|
workload.googleapis.com/kafka.network.io
|
|
CUMULATIVE, INT64gce_instance |
state
|
workload.googleapis.com/kafka.partition.count
|
|
GAUGE, INT64gce_instance |
|
workload.googleapis.com/kafka.partition.offline
|
|
GAUGE, INT64gce_instance |
|
workload.googleapis.com/kafka.partition.under_replicated
|
|
GAUGE, INT64gce_instance |
|
workload.googleapis.com/kafka.purgatory.size
|
|
GAUGE, INT64gce_instance |
type
|
workload.googleapis.com/kafka.request.count
|
|
CUMULATIVE, INT64gce_instance |
type
|
workload.googleapis.com/kafka.request.failed
|
|
CUMULATIVE, INT64gce_instance |
type
|
workload.googleapis.com/kafka.request.time.total
|
|
CUMULATIVE, INT64gce_instance |
type
|
Vérifier la configuration
Vous pouvez utiliser l'explorateur de journaux et l'explorateur de métriques pour vérifier que vous avez correctement configuré le récepteur Kafka. La collecte des journaux et des métriques par l'agent Ops peut prendre une ou deux minutes.
Pour vérifier que les journaux sont ingérés, accédez à l'explorateur de journaux et exécutez la requête suivante pour afficher les journaux Kafka :
resource.type="gce_instance"
logName=("projects/PROJECT_ID/logs/kafka")
Pour vérifier que les métriques sont ingérées, accédez à l'explorateur de métriques et exécutez la requête suivante dans l'onglet MQL.
fetch gce_instance
| metric 'workload.googleapis.com/kafka.request.count'
| align rate(1m)
| every 1m