Apache Kafka

Die Apache Kafka-Integration erfasst Broker-Messwerte wie Themenanfragen und Fehler. Außerdem werden die Partitionen auf dem Broker überwacht. Die Integration erfasst Kafka-Logs und parst sie in eine JSON-Nutzlast. Das Ergebnis enthält Felder für Logger, Ebene und Nachricht.

Weitere Informationen zu Kafka finden Sie unter kafka.apache.org/.

Vorbereitung

Sie müssen Ops-Agent Version 2.10.0 oder höher installieren, um Kafka-Logs und -Messwerte zu erfassen und aufzunehmen.

Dieser Empfänger unterstützt die Apache Kafka-Versionen 0.8 bis 3.0.0.

Kafka-Instanz konfigurieren

Wenn Sie einen JMX-Endpunkt verfügbar machen möchten, müssen Sie beim Starten der JVM das Systemattribut com.sun.management.jmxremote.port festlegen. Außerdem empfehlen wir, das Systemattribut com.sun.management.jmxremote.rmi.port auf denselben Port festzulegen. Wenn Sie einen JMX-Endpunkt remote verfügbar machen möchten, müssen Sie auch das Systemattribut java.rmi.server.hostname festlegen.

Standardmäßig werden diese Attribute in der Datei bin/kafka-run-class.sh einer Kafka-Bereitstellung festgelegt.

Wenn Sie Systemattribute mithilfe von Befehlszeilenargumenten festlegen möchten, stellen Sie beim Starten der JVM dem Attributnamen -D voran. Wenn Sie beispielsweise com.sun.management.jmxremote.port auf den Port 9999 festlegen möchten, geben Sie beim Starten der JVM Folgendes an:

-Dcom.sun.management.jmxremote.port=9999

Ops-Agent für Kafka konfigurieren

Fügen Sie die erforderlichen Elemente zum Erfassen von Logs und Messwerten aus den Kafka-Instanzen gemäß der Anleitung unter Ops-Agent konfigurieren hinzu und starten Sie den Agent neu.

Konfigurationsbeispiel

Der folgende Befehl erstellt die Konfigurationsdatei, um Logs und Messwerte für Kafka zu erfassen und aufzunehmen, und startet den Ops-Agent unter Linux neu.

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

Logerfassung konfigurieren

Um Logs von Kafka aufzunehmen, müssen Sie Empfänger für die von Kafka erzeugten Logs erstellen und dann eine Pipeline für die neuen Empfänger erstellen. Geben Sie die folgenden Felder an, um einen Empfänger für Ihre kafka-Logs zu konfigurieren:

Feld	Standard	Beschreibung
type		Der Wert muss kafka betragen.
include_paths	[/var/log/kafka/*.log]	Eine Liste mit Dateisystempfaden, die durch Tailing jeder Datei gelesen werden sollen. In den Pfaden kann ein Platzhalter (*) verwendet werden. Beispiel: /var/log/kafka*/*.log.
exclude_paths		Eine Liste von Dateisystempfadmustern, die aus dem mit include_paths übereinstimmenden Satz ausgeschlossen werden sollen.
wildcard_refresh_interval	60s	Das Intervall, in dem Platzhalterdateipfade in include_paths aktualisiert werden. Wird als Zeitdauer angegeben, z. B. 30s oder 2m. Dieses Attribut kann bei hohen Logging-Durchsätzen nützlich sein, wenn Logdateien schneller als das Standardintervall rotiert werden. Muss ein Vielfaches von 1s sein.

Was wird protokolliert?

Die logName-Logs der Logs kafka werden von den Empfänger-IDs abgeleitet, die in der Konfiguration angegeben sind. Detaillierte Felder in LogEntry sind:

Feld	Typ	Beschreibung
jsonPayload.source	String	Modul und/oder Thread, von dem das Log stammt.
jsonPayload.logger	String	Name des Loggers, von dem das Log stammt.
jsonPayload.message	String	Lognachricht, einschließlich eines detaillierten Stacktrace, sofern angegeben.
severity	String (LogSeverity)	Ebene des Logeintrags (übersetzt).
timestamp	String (Timestamp)	Zeitpunkt des Eingangs der Anfrage.

Leere oder fehlende Felder werden nicht im Logeintrag angezeigt.

Messwerterfassung konfigurieren

Um Messwerte aus Kafka zu erfassen, müssen Sie einen Empfänger für Kafka-Messwerte erstellen und dann eine Pipeline für den neuen Empfänger erstellen. Um einen Empfänger für Ihre Kafka-Messwerte zu konfigurieren, geben Sie die folgenden Felder an:

Feld	Standard	Beschreibung
type		Der Wert muss kafka betragen.
stub_status_url	localhost:9999	Die JMX-Dienst-URL oder der Host und Port, die zum Erstellen der Dienst-URL verwendet werden. Muss das Format service:jmx:<protocol>:<sap> oder host:port haben. Werte im Format host:port werden verwendet, um die Dienst-URL service:jmx:rmi:///jndi/rmi://<host>:<port>/jmxrmi zu erstellen.
collect_jvm_metrics	true	Konfiguriert den Empfänger so, dass auch die unterstützten JVM-Messwerte erfasst werden.
username		Der konfigurierte Nutzername, wenn JMX für eine erforderliche Authentifizierung konfiguriert ist.
password		Das konfigurierte Passwort, wenn JMX für eine erforderliche Authentifizierung konfiguriert ist.
collection_interval	60s	Ein Wert für time.Duration, wie z. B. 30s oder 5m.

Was wird überwacht?

Die folgende Tabelle enthält die Liste der Messwerte, die der Ops-Agent aus der Kafka-Instanz erfasst.

Messwerttyp
Art, Typ Überwachte Ressourcen	Label
workload.googleapis.com/kafka.isr.operation.count
CUMULATIVE, INT64 gce_instance	operation
workload.googleapis.com/kafka.message.count
CUMULATIVE, INT64 gce_instance
workload.googleapis.com/kafka.network.io
CUMULATIVE, INT64 gce_instance	state
workload.googleapis.com/kafka.partition.count
GAUGE, INT64 gce_instance
workload.googleapis.com/kafka.partition.offline
GAUGE, INT64 gce_instance
workload.googleapis.com/kafka.partition.under_replicated
GAUGE, INT64 gce_instance
workload.googleapis.com/kafka.purgatory.size
GAUGE, INT64 gce_instance	type
workload.googleapis.com/kafka.request.count
CUMULATIVE, INT64 gce_instance	type
workload.googleapis.com/kafka.request.failed
CUMULATIVE, INT64 gce_instance	type
workload.googleapis.com/kafka.request.time.total
CUMULATIVE, INT64 gce_instance	type

Konfiguration prüfen

Mit dem Log-Explorer und dem Metrics Explorer können Sie prüfen, ob Sie den Kafka-Empfänger richtig konfiguriert haben. Es kann ein bis zwei Minuten dauern, bis der Ops-Agent beginnt, Logs und Messwerte zu erfassen.

Um zu prüfen, ob die Logs aufgenommen wurden, rufen Sie den Log-Explorer auf. Führen Sie dann die folgende Abfrage aus, um die Kafka-Logs aufzurufen:

resource.type="gce_instance"
logName=("projects/PROJECT_ID/logs/kafka")

Um zu prüfen, ob die Messwerte aufgenommen wurden, rufen Sie den Metrics Explorer auf und führen Sie die folgende Abfrage auf dem Tab MQL aus.

fetch gce_instance
| metric 'workload.googleapis.com/kafka.request.count'
| align rate(1m)
| every 1m

Nächste Schritte

Eine Anleitung zur Installation von Ops-Agent mit Ansible zum Konfigurieren einer Drittanbieteranwendung und zum Installieren eines Beispieldashboards finden Sie im Video Ops-Agent installieren, um Fehler in Drittanbieteranwendungen zu beheben.