🥇Definire la dimensione adeguata per un cluster Kafka in Kubernetes

Nota di traduzione.: In questo articolo, Banzai Cloud condivide un esempio di utilizzo dei suoi strumenti speciali per semplificare l'operatività di Kafka all'interno di Kubernetes. Le istruzioni fornite illustrano come determinare la dimensione ottimale dell'infrastruttura e configurare Kafka per raggiungere la capacità richiesta.

Apache Kafka è una piattaforma di streaming distribuita per la creazione di sistemi di streaming real-time affidabili, scalabili e ad alte prestazioni. Le sue impressionanti capacità possono essere ampliate con Kubernetes. Per questo abbiamo sviluppato l'operatore Kafka Open Source e uno strumento chiamato Supertubes. Questi strumenti consentono di eseguire Kafka su Kubernetes e di sfruttare le sue varie funzionalità, come la fine regolazione della configurazione del broker, la scalabilità basata su metriche con ribilanciamento, la consapevolezza del rack, l'aggiornamento 'graceful' (graceful) delle versioni e così via.

Prova Supertubes nel tuo cluster:
curl https://getsupertubes.sh | sh e supertubes install -a --no-democluster --kubeconfig
Oppure contattaci per. documentazionePuoi anche leggere alcune delle funzionalità di Kafka, la cui gestione è automatizzata tramite Supertubes e l'operatore Kafka, di cui abbiamo già parlato nel blog:

Oh no! Un altro operatore Kafka per Kubernetes;

Monitora e gestisci Kafka basato su metriche Prometheus;

Consapevolezza del rack di Kafka su Kubernetes;

Esecuzione di Apache Kafka su Istio — benchmark;

Cluster autenticati e controllati dall'accesso con l'operatore Kafka;

Aggiornamento rolling di Kafka e configurazione dinamica su Kubernetes;

Filtro di protocollo Envoy per Kafka, in mesh.

Decidendo di implementare un cluster Kafka su Kubernetes, dovrai sicuramente affrontare il problema di determinare la dimensione ottimale dell'infrastruttura di base e la necessità di una fine regolazione della configurazione di Kafka per soddisfare i requisiti di capacità. Le massime prestazioni di ciascun broker sono determinati dalle prestazioni dei componenti dell'infrastruttura su cui si basa, come la memoria, il processore, la velocità del disco, la larghezza di banda della rete, ecc.

Idealmente, la configurazione del broker dovrebbe essere tale che tutti gli elementi dell'infrastruttura vengano utilizzati al massimo delle loro capacità. Tuttavia, nella vita reale, tale impostazione è piuttosto complessa. È più probabile che gli utenti configurino i broker in modo da massimizzare l'utilizzo di uno o due componenti (disco, memoria o processore). In generale, un broker mostra prestazioni massime quando la sua configurazione consente di utilizzare a pieno il componente più lento. Così possiamo avere un'idea approssimativa del carico che un singolo broker è in grado di gestire.

Teoricamente, possiamo anche stimare il numero di broker necessari per gestire un determinato carico. Tuttavia, nella pratica, le opzioni di configurazione a vari livelli sono così numerose che valutare le prestazioni potenziali di una configurazione è molto complesso (se non impossibile). In altre parole, è molto difficile pianificare una configurazione basandosi su una prestazione predefinita.

Per gli utenti di Supertubes, di solito adottiamo il seguente approccio: partiamo da una certa configurazione (infrastruttura + impostazioni), quindi misuriamo le sue prestazioni, correggiamo le impostazioni del broker e ripetiamo il processo ancora una volta. Questo avviene fino a quando il potenziale del componente più lento dell'infrastruttura non viene completamente utilizzato.

In questo modo otteniamo un'idea più chiara di quanti broker sono necessari nel cluster per gestire un determinato carico (il numero di broker dipende anche da altri fattori, come il numero minimo di repliche dei messaggi per garantire la resilienza, il numero di leader delle partizioni e così via). Inoltre, otteniamo un'idea su quale componente infrastrutturale sarebbe preferibile scalare verticalmente.

In questo articolo parleremo dei passaggi che intraprendiamo per "estrarre tutto" dai componenti più lenti nelle configurazioni iniziali e misurare la larghezza di banda del cluster Kafka. Una configurazione altamente resiliente richiede almeno tre broker attivi (min.insync.replicas=3), distribuiti su tre diverse zone di disponibilità. Per la configurazione, scalabilità e monitoraggio dell'infrastruttura Kubernetes, utilizziamo la nostra piattaforma di gestione dei container per cloud ibridi — Pipeline. Supporta on-premise (bare metal, VMware) e cinque tipi di cloud (Alibaba, AWS, Azure, Google, Oracle), nonché qualsiasi combinazione di essi.

Riflessioni sull'infrastruttura e la configurazione del cluster Kafka

Per gli esempi riportati di seguito, abbiamo scelto AWS come fornitore di servizi cloud e EKS come distribuzione di Kubernetes. Una configurazione simile può essere realizzata utilizzando PKE — una distribuzione di Kubernetes di Banzai Cloud, certificata da CNCF.

Disco

Amazon offre diversi tipi di volumi EBS. Alla base di gp2 e io1 ci sono dischi SSD, ma per garantire un'elevata larghezza di banda gp2 consuma crediti di I/O (I/O credits), quindi abbiamo preferito il tipo io1, che offre una larghezza di banda stabile e alta.

Tipi di istanze

Le prestazioni di Kafka dipendono fortemente dalla cache di pagina del sistema operativo, quindi abbiamo bisogno di istanze con una quantità adeguata di memoria per i broker (JVM) e la cache di pagina. L'istanza c5.2xlarge è un buon inizio, poiché ha 16 GB di memoria e ottimizzata per lavorare con EBS. Il suo svantaggio è che è in grado di fornire prestazioni massime per non più di 30 minuti ogni 24 ore. Se il carico di lavoro necessita di prestazioni massime per un periodo di tempo più lungo, è opportuno considerare altri tipi di istanze. È proprio quello che abbiamo fatto, optando per c5.4xlarge. Garantisce una larghezza di banda massima di 593,75 MB/s. La larghezza di banda massima del volume EBS io1 è superiore a quella dell'istanza c5.4xlarge, quindi il collo di bottiglia dell'infrastruttura sembra essere la larghezza di banda I/O di questo tipo di istanza (cosa che dovrebbero confermare anche i risultati dei nostri test di carico).

Rete

La larghezza di banda della rete deve rivelarsi sufficientemente grande rispetto alle prestazioni dell'istanza VM e del disco, altrimenti la rete diventa un collo di bottiglia. Nel nostro caso, l'interfaccia di rete c5.4xlarge supporta una velocità fino a 10 Gb/s, che è significativamente superiore alla larghezza di banda I/O dell'istanza VM.

Distribuzione dei broker

I broker devono essere distribuiti (pianificati in Kubernetes) su nodi dedicati per evitare la competizione con altri processi per le risorse di CPU, memoria, rete e disco.

Versione Java

La scelta logica è Java 11, poiché è compatibile con Docker nel senso che la JVM riconosce correttamente i processori e la memoria disponibili nel contenitore in cui opera il broker. Sapendo che i limiti della CPU sono importanti, la JVM imposta internamente e in modo trasparente il numero di thread GC e thread del compilatore JIT. Abbiamo utilizzato l'immagine Kafka banzaicloud/kafka:2.13-2.4.0, che include la versione Kafka 2.4.0 (Scala 2.13) su Java 11.

Se desideri saperne di più su Java/JVM su Kubernetes, dai un'occhiata ai nostri seguenti articoli:

Why my Java application is OOMKilled;

How to correctly size containers for Java 10 applications.

Impostazioni di memoria del broker

Ci sono due aspetti chiave nella configurazione della memoria del broker: le impostazioni per la JVM e per il pod Kubernetes. Il limite di memoria impostato per il pod deve essere maggiore della dimensione massima dell'heap, per garantire che la JVM abbia spazio per il metaspace di Java, che risiede nella propria memoria, e per la cache di pagina del sistema operativo, che Kafka utilizza attivamente. Nei nostri test abbiamo eseguito i broker Kafka con i seguenti parametri -Xmx4G -Xms2G, mentre il limite di memoria per il pod era di 10 Gi. Tieni presente che le impostazioni di memoria per la JVM possono essere ottenute automaticamente utilizzando -XX:MaxRAMPercentage e -X:MinRAMPercentage, in base al limite di memoria per il pod.

Impostazioni della CPU del broker

In generale, è possibile aumentare le prestazioni aumentando il parallelismo utilizzando un maggior numero di thread da parte di Kafka. Più processori sono disponibili per Kafka, meglio è. Nel nostro test, siamo partiti con un limite di 6 processori e, progressivamente (a iterazioni), abbiamo aumentato il numero fino a 15. Inoltre, abbiamo impostato num.network.threads=12 nelle impostazioni del broker, per aumentare il numero di thread che ricevono i dati dalla rete e li inviano. Notando immediatamente che i broker followers non potevano ricevere repliche abbastanza rapidamente, abbiamo aumentato num.replica.fetchers fino a 4, per aumentare la velocità con cui i broker followers replicavano i messaggi dai leader.

Strumento di generazione del carico

È fondamentale assicurarsi che il potenziale del generatore di carico scelto non si esaurisca prima che il cluster Kafka (il cui benchmark viene effettuato) raggiunga il suo carico massimo. In altre parole, è necessario eseguire una valutazione preliminare delle capacità dello strumento di generazione del carico e selezionare per esso tipi di istanze con un numero adeguato di processori e memoria. In questo caso, il nostro strumento produrrà più carico di quanto il cluster Kafka possa gestire. Dopo numerosi esperimenti, ci siamo concentrati su tre istanze c5.4xlarge, in cui è stato avviato un generatore.

Benchmarking

La misurazione delle prestazioni è un processo iterativo che comprende le seguenti fasi:

configurazione dell'infrastruttura (cluster EKS, cluster Kafka, strumento di generazione del carico, nonché Prometheus e Grafana);
generazione di carico per un determinato periodo per filtrare le deviazioni casuali nei parametri di prestazione raccolti;
ottimizzazione dell'infrastruttura e della configurazione del broker in base ai parametri di prestazione osservati;
ripetizione del processo fino a raggiungere il livello di capacità richiesto del cluster Kafka. Questo deve essere riproducibile in modo stabile e dimostrare variazioni minime della capacità.

Nella sezione successiva sono descritti i passaggi eseguiti durante il benchmark del cluster di test.

Strumenti

Per un rapido dispiegamento della configurazione di base, generazione di carico e misurazione delle prestazioni sono stati utilizzati i seguenti strumenti:

Banzai Cloud Pipeline per organizzare il cluster EKS di Amazon con Prometheus (per la raccolta delle metriche di Kafka e dell'infrastruttura) e Grafana (per la visualizzazione di tali metriche). Abbiamo utilizzato servizi in Pipeline integrati che forniscono monitoraggio federato, raccolta centralizzata dei log, scansione delle vulnerabilità, recovery da guasti, sicurezza a livello enterprise e molto altro.
Sangrenel è uno strumento per il testing del carico del cluster Kafka.
Pannelli Grafana per la visualizzazione delle metriche di Kafka e dell'infrastruttura: Kubernetes Kafka, Node Exporter.
Supertubes CLI per una configurazione semplicissima del cluster Kafka in Kubernetes. Zookeeper, Kafka operator, Envoy e molti altri componenti sono installati e configurati correttamente per far funzionare un cluster Kafka pronto per la produzione in Kubernetes.
- Per l'installazione supertubes CLI utilizza le istruzioni fornite qui.

Cluster EKS

Prepara un cluster EKS con nodi di lavoro dedicati c5.4xlarge in diverse zone di disponibilità per i pod con i broker Kafka, oltre a nodi dedicati per il generatore di carico e l'infrastruttura di monitoraggio.

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

Quando il cluster EKS è attivo, abilita la sua servizio di monitoraggio — distribuirà Prometheus e Grafana nel cluster.

Componenti di sistema Kafka

Installa i componenti di sistema Kafka (Zookeeper, kafka-operator) in EKS utilizzando supertubes CLI:

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

Cluster Kafka

Per impostazione predefinita, in EKS vengono utilizzati volumi EBS di tipo gp2, quindi è necessario creare una classe di archiviazione separata basata sui volumi io1 per il cluster Kafka:

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

Imposta per i broker il parametro min.insync.replicas=3 e distribuisci i pod broker su nodi in tre diverse zone di disponibilità:

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

Argomenti

Abbiamo avviato in parallelo tre istanze del generatore di carico. Ognuna di esse scrive nel proprio argomento, quindi abbiamo bisogno di tre argomenti:

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

Per ogni argomento, il fattore di replicazione è pari a 3 — il valore minimo raccomandato per sistemi di produzione altamente disponibili.

Strumento di generazione del carico

Abbiamo avviato tre istanze del generatore di carico (ognuna scriveva in un topic separato). Per i pod del generatore di carico è necessario specificare l'affinità del nodo, affinché vengano pianificati solo sui nodi a loro dedicati:

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

Alcuni aspetti da tenere a mente:

Il generatore di carico genera messaggi lunghi 512 byte e li pubblica in Kafka in batch di 500 messaggi.
Utilizzando l'argomento -required-acks=all la pubblicazione è considerata riuscita quando tutte le repliche sincronizzate del messaggio sono state ricevute e confermate dai broker Kafka. Ciò significa che nel benchmark abbiamo misurato non solo la velocità dei leader nel ricevere i messaggi, ma anche dei loro seguaci, che replicano i messaggi. Questo test non valuta la velocità di lettura dei consumatori (consumers) di messaggi recentemente inviati, che rimangono ancora nella cache di pagina del sistema operativo, e la sua comparazione con la velocità di lettura dei messaggi memorizzati su disco.
Il generatore di carico avvia parallelamente 20 worker (-workers=20). Ogni worker contiene 5 produttori, che condividono la connessione del worker al cluster Kafka. Alla fine, ogni generatore conta 100 produttori, e tutti inviano messaggi al cluster Kafka.

Monitoraggio dello stato del cluster

Durante il test di carico del cluster Kafka, abbiamo anche monitorato la sua salute per assicurarci che non ci fossero riavvii dei pod, repliche non sincronizzate e una massima capacità di throughput con minime fluttuazioni:

Il generatore di carico registra statistiche standard sul numero di messaggi pubblicati e sul tasso di errori. La percentuale di errori deve rimanere al di sotto di 0,00%.
Cruise Control, distribuito dal kafka-operator, fornisce un pannello di monitoraggio dove possiamo anche osservare lo stato del cluster. Per visualizzare questo pannello, eseguire:
```
supertubes cluster cruisecontrol show -n kafka --kubeconfig
```
Livello ISR (numero di repliche "in-sync") shrink e expansion sono pari a 0.

Risultati delle misurazioni

3 broker, dimensione dei messaggi — 512 byte

Con le partizioni equamente distribuite tra i tre broker, siamo riusciti a raggiungere una capacità di produzione ~500 Mb/s (circa 990.000 messaggi al secondo):

Il consumo di memoria dalla macchina virtuale JVM non ha superato i 2 Gb:

La capacità di I/O del disco ha raggiunto la massima capacità del nodo su tutte e tre le istanze in cui operavano i broker:

Dai dati sull'utilizzo della memoria dei nodi, si osserva che il buffering di sistema e la cache hanno occupato ~10-15 Gb:

3 broker, dimensione dei messaggi — 100 byte

Con la diminuzione della dimensione dei messaggi, la capacità di throughput scende di circa il 15-20%: ciò è dovuto al tempo impiegato per elaborare ogni messaggio. Inoltre, il carico sulla CPU è aumentato quasi di due volte.

Poiché sui nodi dei broker ci sono ancora core non utilizzati, è possibile aumentare le prestazioni modificando la configurazione di Kafka. Questa è un'operazione complessa, quindi per aumentare la capacità di throughput è meglio lavorare con messaggi di dimensioni maggiori.

4 broker, dimensione dei messaggi — 512 byte

È possibile aumentare facilmente le prestazioni del cluster Kafka semplicemente aggiungendo nuovi broker e mantenendo l'equilibrio delle partizioni (questo consente una distribuzione uniforme del carico tra i broker). Nel nostro caso, dopo aver aggiunto un broker, la capacità di throughput del cluster è aumentata a ~580 Mb/s (~1,1 milioni di messaggi al secondo). L'aumento è stato inferiore a quanto atteso: principalmente questo è dovuto al disequilibrio delle partizioni (non tutti i broker lavorano al massimo delle loro capacità).

Il consumo di memoria della macchina JVM è rimasto al di sotto di 2 GB:

Il lavoro dei broker con i repository è stato influenzato dal disallineamento delle partizioni:

Conclusioni

L'approccio iterativo presentato sopra può essere ampliato per coprire scenari più complessi che coinvolgono centinaia di consumer, repartitioning, aggiornamenti in-roll e riavvii dei pod, ecc. Tutto ciò ci consente di valutare i limiti delle capacità del cluster Kafka in diverse condizioni, identificare i colli di bottiglia nel suo funzionamento e trovare modi per affrontarli.

Abbiamo sviluppato Supertubes per il rapido e facile dispiegamento del cluster, la sua configurazione, l'aggiunta/rimozione di broker e argomenti, la gestione delle notifiche e l'assicurazione del corretto funzionamento di Kafka in Kubernetes nel suo complesso. Il nostro obiettivo è aiutarti a concentrarti sul compito principale ("generare" e "consumare" messaggi Kafka), mentre tutto il lavoro pesante è gestito da Supertubes e Kafka operator.

Se ti interessano le tecnologie e i progetti Open Source di Banzai Cloud, segui l'azienda su GitHub, LinkedIn o Twitter.