🥇Bestimmen Sie die geeignete Clustergröße für Kafka in Kubernetes

Anmerkung des Übersetzers.: In diesem Artikel teilt die Firma Banzai Cloud ein Beispiel für die Nutzung ihrer speziellen Tools zur Erleichterung des Betriebs von Kafka innerhalb von Kubernetes. Die angeführten Anleitungen veranschaulichen, wie man die optimale Größe der Infrastruktur bestimmen und Kafka selbst konfigurieren kann, um die erforderliche Durchsatzkapazität zu erreichen.

Apache Kafka ist eine verteilte Streaming-Plattform zur Erstellung zuverlässiger, skalierbarer und hochleistungsfähiger Echtzeitsysteme. Ihre beeindruckenden Möglichkeiten lassen sich durch Kubernetes erweitern. Zu diesem Zweck haben wir entwickelt den Open Source-Kafka-Operator und ein Tool namens Supertubes. Diese ermöglichen es, Kafka in Kubernetes zu betreiben und ihre verschiedenen Funktionen zu nutzen, wie z.B. die feine Abstimmung der Broker-Konfiguration, das Scaling basierend auf Metriken mit Rebalancing, Rack-Awareness, „sanftes“ (graceful) Rollout von Updates usw.

Testen Sie Supertubes in Ihrem Cluster:
curl https://getsupertubes.sh | sh und supertubes install -a --no-democluster --kubeconfig
Oder wenden Sie sich an Dokumentation. Sie können auch mehr über einige Funktionen von Kafka lesen, deren Nutzung durch Supertubes und den Kafka-Operator automatisiert ist. Darüber haben wir bereits in unserem Blog geschrieben:

Oh nein! Schon wieder ein Kafka-Operator für Kubernetes;

Überwachen und Betreiben von Kafka basierend auf Prometheus-Metriken;

Kafka-Rack-Awareness in Kubernetes;

Apache Kafka über Istio - Benchmark;

Benutzerauthentifizierte und zugriffskontrollierte Cluster mit dem Kafka-Operator;

Rolling Upgrade von Kafka und dynamische Konfiguration in Kubernetes;

Envoy-Protokollfilter für Kafka, gemeshte.

Wenn Sie sich entscheiden, einen Kafka-Cluster in Kubernetes bereitzustellen, werden Sie wahrscheinlich auf das Problem stoßen, die optimale Größe der zugrunde liegenden Infrastruktur zu bestimmen und die Kafka-Konfiguration fein abzustimmen, um die Anforderungen an die Durchsatzkapazität zu erfüllen. Die maximale Leistung jedes Brokers wird durch die Leistungsfähigkeit der Komponenten der zugrunde liegenden Infrastruktur bestimmt, wie z.B. RAM, CPU, Datentransferrate, Netzwerkbandbreite usw.

Idealerweise sollte die Broker-Konfiguration so beschaffen sein, dass alle Infrastrukturkomponenten ihre maximalen Möglichkeiten voll ausschöpfen. In der Realität gestaltet sich eine solche Einrichtung jedoch als ziemlich komplex. Wahrscheinlicher ist es, dass die Benutzer die Broker-Konfiguration so anpassen, dass die Nutzung eines oder zweier Komponenten (Festplatte, Speicher oder Prozessor) maximiert wird. Im Allgemeinen zeigt ein Broker die maximale Leistung, wenn seine Konfiguration es erlaubt, die langsamste Komponente "voll auszureizen". So können wir einen groben Überblick über die Last bekommen, die ein Broker bewältigen kann.

Theoretisch können wir auch die Anzahl der Broker abschätzen, die erforderlich sind, um mit einer bestimmten Last zu arbeiten. In der Praxis gibt es jedoch so viele Konfigurationsmöglichkeiten auf den verschiedenen Ebenen, dass es ziemlich schwierig (wenn nicht sogar unmöglich) ist, die potenzielle Leistung einer bestimmten Konfiguration zu bewerten. Mit anderen Worten, es ist sehr schwierig, eine Konfiguration basierend auf einer bestimmten Leistung zu planen.

Für Benutzer von Supertubes verwenden wir normalerweise folgenden Ansatz: Wir beginnen mit einer bestimmten Konfiguration (Infrastruktur + Einstellungen), messen deren Leistung, passen die Broker-Einstellungen an und wiederholen den Prozess erneut. Dies geschieht, bis das Potenzial der langsamsten Infrastrukturkomponente vollständig ausgeschöpft ist.

Auf diese Weise erhalten wir eine klarere Vorstellung davon, wie viele Broker der Cluster benötigt, um mit einer bestimmten Last umzugehen (die Anzahl der Broker hängt auch von anderen Faktoren ab, wie der minimalen Anzahl an Nachrichten-Replikaten zur Gewährleistung der Ausfallsicherheit, der Anzahl der Partition-Leiter usw.). Darüber hinaus erhalten wir einen Überblick über den Infrastrukturkomponenten, für den eine vertikale Skalierung wünschenswert ist.

In diesem Artikel werden die Schritte behandelt, die wir unternehmen, um die langsamsten Komponenten in den Anfangskonfigurationen "voll auszureizen" und die Durchsatzrate des Kafka-Clusters zu messen. Eine hochverfügbare Konfiguration erfordert mindestens drei laufende Broker (min.insync.replicas=3), verteilt auf drei verschiedene Verfügbarkeitszonen. Für die Konfiguration, Skalierung und Überwachung der Kubernetes-Infrastruktur nutzen wir unsere eigene Container-Management-Plattform für hybride Clouds — Pipeline. Sie unterstützt On-Premise (Bare Metal, VMware) und fünf Arten von Clouds (Alibaba, AWS, Azure, Google, Oracle) sowie ihre beliebigen Kombinationen.

Gedanken zur Infrastruktur und Konfiguration des Kafka-Clusters

Für die nachstehenden Beispiele haben wir AWS als Cloud-Anbieter und EKS als Kubernetes-Distribution gewählt. Eine ähnliche Konfiguration kann mit PKE — der Kubernetes-Distribution von Banzai Cloud, die CNCF-zertifiziert ist.

Festplatte

Amazon bietet verschiedene EBS-Volumentypen. Grundlage gp2 und io1 sind SSDs, aber um eine hohe Durchsatzrate zu gewährleisten, gp2 verbraucht es angesammelte Kredite (I/O-Credits), weshalb wir den Typ bevorzugten, io1, der eine stabile hohe Durchsatzrate bietet.

Instanztypen

Die Leistung von Kafka hängt stark vom Seiten-Cache des Betriebssystems ab, weshalb wir Instanzen mit ausreichend RAM für Broker (JVM) und Seiten-Cache benötigen. Die Instanz c5.2xlarge ist ein guter Start, da sie über 16 GB RAM verfügt und für die Arbeit mit EBS optimiert ist.Ein Nachteil besteht darin, dass sie ihre maximale Leistung nicht länger als 30 Minuten in einem Zeitraum von 24 Stunden aufrechterhalten kann. Wenn die Arbeitslast eine maximale Leistung über einen längeren Zeitraum erfordert, sollten andere Instanztypen in Betracht gezogen werden. Genau das haben wir getan und uns für c5.4xlargeentschieden. Sie bietet eine maximale Durchsatzrate von 593,75 MB/s. Die maximale Durchsatzrate des EBS-Volumes io1 liegt über der der Instanz c5.4xlarge, weshalb das langsamste Element der Infrastruktur offenbar die I/O-Durchsatzrate dieses Instanztyps ist (was auch durch die Ergebnisse unserer Lasttests bestätigt werden sollte).

Netzwerk

Die Netzwerkbandbreite sollte im Vergleich zur Leistung der VM-Instanz und der Festplatte ausreichend groß sein, andernfalls wird das Netzwerk zum Engpass. In unserem Fall unterstützt die Netzwerkschnittstelle c5.4xlarge Geschwindigkeiten von bis zu 10 Gbit/s, was deutlich über der I/O-Durchsatzrate der VM-Instanz liegt.

Bereitstellung von Brokern

Broker sollten auf dedizierte Knoten in Kubernetes bereitgestellt werden, um Konkurrenz um CPU-, Speicher-, Netzwerk- und Festplattressourcen mit anderen Prozessen zu vermeiden.

Java-Version

Eine logische Wahl ist Java 11, da es mit Docker kompatibel ist, sodass die JVM die verfügbaren Prozessoren und den Speicher im Container, in dem der Broker läuft, korrekt erkennt. Da die CPU-Grenzen wichtig sind, legt die JVM intern und transparent die Anzahl der GC-Threads und JIT-Compiler-Threads fest. Wir haben das Kafka-Image verwendet banzaicloud/kafka:2.13-2.4.0, das die Version Kafka 2.4.0 (Scala 2.13) auf Java 11 umfasst.

Wenn Sie mehr über Java/JVM auf Kubernetes erfahren möchten, werfen Sie einen Blick auf unsere folgenden Veröffentlichungen:

Warum meine Java-Anwendung OOMKilled ist;

Wie man Container für Java 10-Anwendungen richtig dimensioniert.

Broker-Speichereinstellungen

Es gibt zwei Schlüsselaspekte bei den Speichereinstellungen des Brokers: die Einstellungen für die JVM und für das Kubernetes-Pod. Das für das Pod festgelegte Speicherlimit sollte höher sein als die maximale Heap-Größe, damit die JVM Platz für den Java-Metaspeicher hat, der im eigenen Speicher liegt, und für den Seiten-Cache des Betriebssystems, den Kafka aktiv nutzt. In unseren Tests haben wir Kafka-Broker mit den Parametern -Xmx4G -Xms2G, und das Speicherlimit für das Pod betrug 10 Gi. Beachten Sie, dass die Speichereinstellungen für die JVM automatisch mit Hilfe von -XX:MaxRAMPercentage und -X:MinRAMPercentagebasierend auf dem Speicherlimit für das Pod ermittelt werden können.

CPU-Einstellungen des Brokers

Im Allgemeinen kann die Leistung gesteigert werden, indem der Parallelismus durch die Erhöhung der Anzahl der von Kafka verwendeten Threads maximiert wird. Je mehr Prozessoren für Kafka verfügbar sind, desto besser. In unserem Test begannen wir mit einem Limit von 6 Prozessoren und erhöhten die Anzahl schrittweise (iterativ) auf 15. Außerdem haben wir num.network.threads=12 in den Broker-Einstellungen festgelegt, um die Anzahl der Threads zu erhöhen, die Daten aus dem Netzwerk empfangen und versenden. Als wir feststellten, dass die Follower-Broker die Replikate nicht schnell genug erhalten konnten, erhöhten wir num.replica.fetchers auf 4, um die Geschwindigkeit zu erhöhen, mit der die Follower-Broker Nachrichten von den Leadern replizierten.

Lastgenerator

Es muss sichergestellt werden, dass das Potenzial des gewählten Lastgenerators nicht erschöpft ist, bevor das Kafka-Cluster (dessen Benchmark durchgeführt wird) seine maximale Last erreicht. Mit anderen Worten, es ist notwendig, eine Vorabanalyse der Möglichkeiten des Lastgenerierungstools durchzuführen und die geeigneten Instanztypen mit ausreichender Anzahl an Prozessoren und Speicher auszuwählen. In diesem Fall wird unser Tool mehr Last erzeugen, als das Kafka-Cluster verarbeiten kann. Nach zahlreichen Tests haben wir uns für drei Instanzen entschieden, c5.4xlargein denen jeweils ein Generator gestartet wurde.

Benchmarking

Die Leistungsmessung ist ein iterativer Prozess, der folgende Phasen umfasst:

Einrichtung der Infrastruktur (EKS-Cluster, Kafka-Cluster, Lastgenerierungstool sowie Prometheus und Grafana);
Lastgenerierung über einen bestimmten Zeitraum, um zufällige Abweichungen in den gesammelten Leistungskennzahlen herauszufiltern;
Anpassung der Infrastruktur und der Broker-Konfiguration basierend auf den beobachteten Leistungskennzahlen;
Wiederholung des Prozesses, bis das erforderliche Durchsatzlevel des Kafka-Clusters erreicht ist. Dabei sollte es stabil reproduzierbar sein und minimale Variationen im Durchsatz zeigen.

Im folgenden Abschnitt werden die Schritte beschrieben, die im Verlauf des Benchmarks des Testclusters ausgeführt wurden.

Werkzeuge

Für das schnelle Deployment der Grundkonfiguration, die Lastgenerierung und die Leistungsmessung wurden folgende Tools verwendet:

Banzai Cloud Pipeline zur Organisation des EKS-Clusters von Amazon mit Prometheus (zur Erfassung der Kafka- und Infrastrukturmetriken) und Grafana (zur Visualisierung dieser Metriken). Wir haben integrierte in Pipeline Dienste genutzt, die föderales Monitoring, zentrale Protokollsammlung, Schwachstellenscans, Wiederherstellung nach Ausfällen, Unternehmenssicherheit und vieles mehr bieten.
Sangrenel ist ein Tool für Lasttests des Kafka-Clusters.
Grafana-Dashboards zur Visualisierung der Kafka- und Infrastrukturmetriken: Kubernetes Kafka, Node Exporter.
Supertubes CLI für die einfachste Konfiguration eines Kafka-Clusters in Kubernetes. Zookeeper, Kafka-Operator, Envoy und viele andere Komponenten sind installiert und ordnungsgemäß konfiguriert, um ein produktionsbereites Kafka-Cluster in Kubernetes zu betreiben.
- Zur Installation supertubes CLI nutzen Sie die angegebenen Anweisungen hier.

EKS-Cluster

Bereiten Sie das EKS-Cluster mit dedizierten Arbeitsknoten c5.4xlarge in verschiedenen Verfügbarkeitszonen für Pods mit Kafka-Brokern vor, sowie dedizierte Knoten für den Lastgenerator und die Überwachungsinfrastruktur.

banzai cluster create -f https://raw.githubusercontent.com/banzaicloud/kafka-operator/master/docs/benchmarks/infrastructure/cluster_eks_202001.json

Sobald das EKS-Cluster läuft, aktivieren Sie seinen integrierten Überwachungsdienst — es wird Prometheus und Grafana im Cluster bereitstellen.

Systemkomponenten von Kafka

Installieren Sie die Systemkomponenten von Kafka (Zookeeper, kafka-operator) in EKS mit supertubes CLI:

supertubes install -a --no-democluster --kubeconfig

Kafka-Cluster

Standardmäßig verwendet EKS EBS-Typvolumes gp2, daher muss eine separate Speicherklasse auf Basis der Volumes io1 für das Kafka-Cluster erstellt werden:

kubectl create -f - <<EOF
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: fast-ssd
provisioner: kubernetes.io/aws-ebs
parameters:
  type: io1
  iopsPerGB: "50"
  fsType: ext4
volumeBindingMode: WaitForFirstConsumer
EOF

Setzen Sie den Parameter für Broker min.insync.replicas=3 und stellen Sie die Pods der Broker auf Knoten in drei verschiedenen Verfügbarkeitszonen bereit:

supertubes cluster create -n kafka --kubeconfig  -f https://raw.githubusercontent.com/banzaicloud/kafka-operator/master/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml --wait --timeout 600

Themen

Wir haben gleichzeitig drei Instanzen des Lastgenerators gestartet. Jede davon schreibt in ihr eigenes Thema, das heißt, insgesamt benötigen wir drei Themen:

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
  name: perftest1
spec:
  name: perftest1
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
    name: perftest2
spec:
  name: perftest2
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
  name: perftest3
spec:
  name: perftest3
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

Für jedes Thema beträgt der Replikationsfaktor 3 — der minimal empfohlene Wert für hochverfügbare Produktionssysteme.

Lastgenerator

Wir haben drei Exemplare des Lasttestgenerators gestartet (jeder schrieb in ein separates Topic). Für die Pods des Lasttestgenerators muss die Node-Affinität festgelegt werden, damit sie nur auf den dafür vorgesehenen Knoten geplant werden.

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  labels:
    app: loadtest
  name: perf-load1
  namespace: kafka
spec:
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: loadtest
  strategy:
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
    type: RollingUpdate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: loadtest
    spec:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: nodepool.banzaicloud.io/name
                operator: In
                values:
                - loadgen
      containers:
      - args:
        - -brokers=kafka-0:29092,kafka-1:29092,kafka-2:29092,kafka-3:29092
        - -topic=perftest1
        - -required-acks=all
        - -message-size=512
        - -workers=20
        image: banzaicloud/perfload:0.1.0-blog
        imagePullPolicy: Always
        name: sangrenel
        resources:
          limits:
            cpu: 2
            memory: 1Gi
          requests:
            cpu: 2
            memory: 1Gi
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30

Einige Punkte, die beachtet werden sollten:

Der Lasttestgenerator erzeugt Nachrichten mit einer Größe von 512 Byte und veröffentlicht diese in Kafka in Paketen von 500 Nachrichten.
Durch das Argument -required-acks=all wird die Veröffentlichung als erfolgreich angesehen, wenn alle synchronisierten Replikate der Nachricht von den Kafka-Brokern empfangen und bestätigt wurden. Das bedeutet, dass wir im Benchmark nicht nur die Geschwindigkeit der Leader, die die Nachrichten empfangen, sondern auch die ihrer Nachfolger, die die Nachrichten replizieren, gemessen haben. Ziel dieses Tests ist es nicht, die Geschwindigkeit der Verbraucher (consumers) sofort nach dem Empfang der Nachrichten zu bewerten, die sich noch im Seiten-Cache des Betriebssystems befinden, und diese mit der Geschwindigkeit des Lesens von Nachrichten auf der Festplatte zu vergleichen.
Der Lasttestgenerator startet parallel 20 Worker (-workers=20). Jeder Worker enthält 5 Producer, die sich die Verbindung des Workers zum Kafka-Cluster gemeinsam nutzen. Insgesamt hat jeder Generator somit 100 Producer, die alle Nachrichten in das Kafka-Cluster senden.

Überwachung des Zustands des Clusters

Während der Lasttestung des Kafka-Clusters haben wir auch auf dessen Gesundheit geachtet, um sicherzustellen, dass es keine Neustarts von Pods, desynchronisierte Replikate und maximale Durchsatzrate mit minimalen Schwankungen gibt.

Der Lastgenerator protokolliert Standardstatistiken über die Anzahl der veröffentlichten Nachrichten und den Fehlerraten. Der Fehleranteil sollte bei 0,00%.
Cruise Control, bereitgestellt durch den kafka-operator, bietet ein Dashboard, auf dem wir auch den Zustand des Clusters überwachen können. Um dieses Dashboard anzusehen, führen Sie aus:
```
supertubes cluster cruisecontrol show -n kafka --kubeconfig
```
Der ISR-Wert (Anzahl der „in-sync“ Replikate) ist gleich 0.

Die Messergebnisse

3 Broker, Nachrichtengröße – 512 Byte

Mit Partitionen, die gleichmäßig auf drei Broker verteilt sind, konnten wir eine Leistung von ~500 MB/s (ungefähr 990.000 Nachrichten pro Sekunde) erreichen.:

Der Speicherverbrauch der JVM überstieg nicht 2 GB:

Der Festplattendurchsatz erreichte die maximale I/O-Durchsatzrate des Knotens auf allen drei Instanzen, auf denen die Broker betrieben wurden:

Aus den Speicherverbrauchsdaten der Knoten geht hervor, dass das Systempuffern und Caching etwa ~10-15 GB beansprucht haben:

3 Broker, Nachrichtengröße – 100 Byte

Mit der Verringerung der Nachrichtengröße fällt der Durchsatz um etwa 15-20 %: Das ist die Folge der Zeit, die für die Verarbeitung jeder Nachricht aufgewendet wird. Außerdem hat sich die CPU-Auslastung fast verdoppelt.

Da die Broker-Knoten weiterhin ungenutzte Kerne haben, kann die Leistung durch eine Anpassung der Kafka-Konfiguration verbessert werden. Dies ist eine anspruchsvolle Aufgabe, daher ist es besser, die Durchsatzrate durch die Verarbeitung größerer Nachrichten zu steigern.

4 Broker, Nachrichtengröße – 512 Byte

Es lässt sich leicht die Leistung des Kafka-Clusters steigern, indem einfach neue Broker hinzugefügt und das Gleichgewicht der Partitionen gewahrt bleibt (was eine gleichmäßige Lastverteilung zwischen den Brokern sicherstellt). In unserem Fall stieg die Durchsatzrate des Clusters nach dem Hinzufügen eines Brokers auf ~580 MB/s (~1,1 Millionen Nachrichten pro Sekunde). Der Anstieg war geringer als erwartet: Dies ist hauptsächlich auf das Ungleichgewicht der Partitionen zurückzuführen (nicht alle Broker arbeiten am Limit ihrer Möglichkeiten).

Der Speicherverbrauch der JVM-Maschine liegt weiterhin unter 2 GB:

Die Arbeit der Broker mit Speichern wurde durch das Ungleichgewicht der Partitionen beeinflusst:

Das DBMS Tarantool ist ein attraktives, zukunftsträchtiges Produkt zur Erstellung von hochbelasteten Anwendungen.

Der oben dargestellte iterative Ansatz kann erweitert werden, um komplexere Szenarien abzudecken, die Hunderte von Consumer, Repartitionierungen, Rollouts von Updates, Neustarts von Pods usw. umfassen. All dies ermöglicht es uns, die Grenzen der Fähigkeiten des Kafka-Clusters unter verschiedenen Bedingungen zu bewerten, Engpässe in seiner Leistung zu identifizieren und Wege zu finden, um diese zu überwinden.

Wir haben Supertubes entwickelt, um das schnelle und einfache Bereitstellen von Clustern, deren Konfiguration, das Hinzufügen/Entfernen von Brokern und Themen, das Reagieren auf Benachrichtigungen und die Gewährleistung des ordnungsgemäßen Betriebs von Kafka in Kubernetes im Allgemeinen zu ermöglichen. Unser Ziel ist es, Ihnen zu helfen, sich auf die Hauptaufgabe zu konzentrieren („Nachrichten generieren“ und „Nachrichten konsumieren“), während die gesamte schwere Arbeit Supertubes und dem Kafka-Operator überlassen bleibt.

Wenn Sie an den Technologien und Open Source-Projekten von Banzai Cloud interessiert sind, abonnieren Sie das Unternehmen auf GitHub, LinkedIn oder Twitter.