🥇Determiniamo la dimensione adatta per il cluster Kafka in Kubernetes

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

Apache Kafka è una piattaforma di streaming distribuita per creare sistemi di streaming in tempo reale affidabili, scalabili e ad alte prestazioni. Le sue impressionanti funzionalità possono essere ampliate tramite Kubernetes. Per questo abbiamo sviluppato l'operatore Kafka Open Source e uno strumento chiamato Supertubes. Questi strumenti consentono di eseguire Kafka in Kubernetes e di sfruttare le sue varie funzionalità, come la personalizzazione della configurazione del broker, il ridimensionamento basato su metriche con bilanciamento del carico, la consapevolezza delle rack, il rilascio degli aggiornamenti in modo "soft" (graceful) e così via.

Prova Supertubes nel tuo cluster:
curl https://getsupertubes.sh | sh e supertubes install -a --no-democluster --kubeconfig
Oppure contatta documentazione. Puoi anche leggere alcune funzionalità di Kafka, la cui gestione è automatizzata tramite Supertubes e Kafka operator. Di essi abbiamo già parlato nel blog:

Oh no! Un altro Kafka operator per Kubernetes;

Monitora e gestisci Kafka basato su metriche Prometheus;

Consapevolezza rack di Kafka su Kubernetes;

Esecuzione di Apache Kafka su Istio — benchmark;

Cluster con accesso controllato e autenticato dagli utenti con il Kafka operator;

Aggiornamento continuo di Kafka e configurazione dinamica su Kubernetes;

Filtro protocollo Envoy per Kafka, in mesh.

Decidendo di distribuire un cluster Kafka in Kubernetes, ti troverai sicuramente di fronte al problema di determinare la dimensione ottimale dell'infrastruttura di base e la necessità di affinare la configurazione di Kafka per soddisfare i requisiti di capacità. Le massime prestazioni di ogni broker sono determinate dalle prestazioni dei componenti infrastrutturali sottostanti, come memoria, CPU, velocità del disco, 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 potenzialità. Tuttavia, nella vita reale, tale impostazione è piuttosto complessa. È più probabile che gli utenti configurino i broker in modo da massimizzare l'uso di uno o due componenti (disco, memoria o processore). In generale, un broker mostra la massima performance quando la sua configurazione consente di sfruttare appieno il componente più lento. Così possiamo avere un'idea approssimativa del carico che un singolo broker può gestire.

Teoricamente, possiamo anche stimare il numero di broker necessari per gestire un determinato carico. Tuttavia, nella pratica, ci sono così tante varianti di configurazione a vari livelli che valutare la potenziale performance di una certa configurazione è piuttosto difficile (se non impossibile). In altre parole, è molto complesso pianificare una configurazione basandosi su una certa performance prestabilita.

Per gli utenti di Supertubes, di solito adottiamo il seguente approccio: iniziamo con una configurazione (infrastruttura + impostazioni), quindi misuriamo le sue prestazioni, regoliamo 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 è completamente sfruttato.

In questo modo otteniamo una visione più chiara di quanti broker siano necessari al 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, ecc.). Inoltre, otteniamo un'idea di quale componente infrastrutturale sarebbe preferibile scalare verticalmente.

In questo articolo si parlerà dei passi che intraprendiamo per "estrarre il massimo" dai componenti più lenti nelle configurazioni iniziali e misurare la capacità di throughput 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. Essa supporta on-premise (bare metal, VMware) e cinque tipi di cloud (Alibaba, AWS, Azure, Google, Oracle), così come tutte le loro combinazioni.

Riflessioni sull'infrastruttura e configurazione del cluster Kafka

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

Disco

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

Tipi di istanze

Le prestazioni di Kafka dipendono fortemente dalla cache di pagina del sistema operativo, quindi abbiamo bisogno di istanze con un'adeguata quantità 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 ottimizzato per lavorare con EBS. Il suo svantaggio è che può garantire massime prestazioni per non più di 30 minuti ogni 24 ore. Se il carico di lavoro richiede prestazioni massime per un periodo di tempo più lungo, è opportuno considerare altri tipi di istanze. È esattamente ciò che abbiamo fatto, optando per c5.4xlarge. Offre una massima larghezza di banda di 593,75 MB/s. La massima larghezza di banda del volume EBS io1 è superiore a quella dell'istanza c5.4xlarge, quindi l'elemento più lento dell'infrastruttura sembra essere la larghezza di banda I/O di questo tipo di istanza (che dovrebbero anche confermare i risultati dei nostri test di carico).

Rete

La larghezza di banda di rete dovrebbe risultare sufficientemente ampia 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 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 (programmati in Kubernetes) su nodi dedicati per evitare conflitti con altri processi per le risorse di CPU, memoria, rete e disco.

Versione Java

La scelta più logica è Java 11, dato che è compatibile con Docker nel senso che la JVM riconosce correttamente i processori e la memoria disponibili per il container in cui è in esecuzione il broker. Essendo i limiti della CPU importanti, la JVM imposta internamente e automaticamente il numero di thread GC e di thread JIT compiler. 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, ti invitiamo a consultare le seguenti nostre pubblicazioni:

Perché la mia applicazione Java è stata OOMKilled;

Come dimensionare correttamente i container per le applicazioni Java 10.

Impostazioni di memoria del broker

Ci sono due aspetti chiave nella configurazione della memoria del broker: le impostazioni per la JVM e quelle per il pod Kubernetes. Il limite di memoria impostato per il pod deve essere superiore alla dimensione massima dell'heap, affinché la JVM abbia spazio per lo spazio dei metadati Java, che si trova nella propria memoria, e per la cache delle pagine del sistema operativo, che Kafka utilizza attivamente. Nei nostri test abbiamo eseguito broker Kafka con le seguenti impostazioni -Xmx4G -Xms2G, e il limite di memoria per il pod era 10 Gi. Si prega di notare che le impostazioni di memoria per la JVM possono essere ottenute automaticamente usando -XX:MaxRAMPercentage e -X:MinRAMPercentage, in base al limite di memoria del pod.

Impostazioni del processore del broker

In generale, è possibile aumentare le prestazioni incrementando il parallelismo attraverso l'aumento del numero di thread utilizzati da Kafka. Maggiore è il numero di processori disponibili per Kafka, migliore è la prestazione. Nel nostro test abbiamo iniziato con un limite di 6 processori e lo abbiamo aumentato gradualmente (in iterazioni) fino a 15. Inoltre, abbiamo impostato num.network.threads=12 nelle impostazioni del broker, per aumentare il numero di thread che ricevono dati dalla rete e li inviano. Subito dopo aver scoperto che i broker follower non potevano ricevere le repliche abbastanza velocemente, abbiamo aumentato num.replica.fetchers a 4, per incrementare la velocità con cui i broker follower replicavano i messaggi dai leader.

Strumento di generazione del carico

È importante assicurarsi che il potenziale del generatore di carico scelto non si esaurisca prima che il cluster Kafka (il cui benchmark viene eseguito) raggiunga il suo massimo carico. In altre parole, è necessario effettuare una valutazione preliminare delle capacità dello strumento di generazione del carico e scegliere per esso tipi di istanze con un numero adeguato di processori e memoria. In questo modo, il nostro strumento produrrà più carico di quanto il cluster Kafka possa gestire. Dopo diversi esperimenti, abbiamo optato per tre istanze c5.4xlarge, in ognuna delle quali è stato avviato il 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 del carico per un certo periodo per filtrare le deviazioni casuali nei dati delle prestazioni raccolti;
ottimizzazione dell'infrastruttura e configurazione del broker in base ai dati delle prestazioni osservate;
ripetizione del processo fino a raggiungere il livello richiesto di capacità del cluster Kafka. Questo deve essere riproducibile in modo stabile e mostrare minime variazioni nella capacità.

Nella sezione successiva sono descritti i passaggi eseguiti nel processo di benchmark del cluster di test.

Strumenti

Per un rapido deploy della configurazione di base, generazione del 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 di metriche Kafka e infrastruttura) e Grafana (per visualizzare queste metriche). Abbiamo utilizzato integrati in Pipeline servizi che offrono monitoraggio federativo, raccolta centralizzata dei log, scansione delle vulnerabilità, recupero dopo guasti, sicurezza a livello aziendale e molto altro.
Sangrenel — strumento per il test delle performance del cluster Kafka.
Pannelli Grafana per visualizzare le metriche di Kafka e dell'infrastruttura: Kubernetes Kafka, Node Exporter.
Supertubes CLI per una configurazione estremamente semplice del cluster Kafka in Kubernetes. Zookeeper, kafka operator, Envoy e molti altri componenti vengono installati e configurati correttamente per avviare un cluster Kafka pronto per la produzione in Kubernetes.
- Per installare supertubes CLI utilizza le istruzioni fornite qui.

Cluster EKS

Prepara il cluster EKS con nodi di lavoro dedicati c5.4xlarge in diverse zone di disponibilità per i pod con 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, attiva il suo servizio di monitoraggio — installerà 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, EKS utilizza 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 il parametro per i broker min.insync.replicas=3 e distribuisci i pod dei broker sui 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 eseguito contemporaneamente tre istanze di un generatore di carico. Ognuna di esse scrive nel proprio argomento, quindi abbiamo bisogno di tre argomenti in totale:

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 i sistemi di produzione ad alta disponibilità.

Strumento di generazione del carico

Abbiamo avviato tre istanze del generatore di carico (ognuna scriveva in un argomento separato). Per i pod del generatore di carico è necessario specificare l'affinità del nodo, in modo che 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 punti da considerare:

Il generatore di carico genera messaggi di 512 byte e li pubblica in Kafka in pacchetti da 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. Questo significa che nei benchmark abbiamo misurato non solo la velocità di elaborazione dei leader nel ricevere i messaggi, ma anche quella dei loro follower nel replicarli. Questo test non ha come obiettivo la valutazione della velocità di lettura da parte dei consumatori (consumatori) di messaggi recentemente ricevuti, che rimangono nel cache della pagina del sistema operativo, e il suo confronto con la velocità di lettura dei messaggi memorizzati su disco.
Il generatore di carico avvia parallelamente 20 worker (-workers=20). Ogni worker contiene 5 producer che condividono la connessione del worker al cluster Kafka. In totale, ogni generatore conta 100 producer, tutti inviando 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 desincronizzate e massima larghezza di banda con minime fluttuazioni:

Il generatore di carico fornisce statistiche standard sul numero di messaggi pubblicati e sul livello di errori. La percentuale di errori deve rimanere a 0,00%.
Cruise Control, distribuito dal kafka-operator, offre 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 è pari a 0.

Risultati delle misurazioni

3 broker, dimensione dei messaggi — 512 byte

Con le partition distribuite uniformemente tra i tre broker, siamo riusciti a raggiungere una performance ~500 Mb/s (circa 990 mila messaggi al secondo):

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

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

Dai dati sull'uso della memoria da parte dei nodi, si evince che la memoria di sistema e la cache hanno occupato circa 10-15 Gb:

3 broker, dimensione dei messaggi — 100 byte

Con una riduzione della dimensione dei messaggi, la larghezza di banda diminuisce di circa il 15-20%: influisce il tempo necessario per elaborare ogni messaggio. Inoltre, il carico sulla CPU è quasi raddoppiato.

Poiché nei nodi broker ci sono ancora core inutilizzati, le prestazioni possono essere migliorate modificando la configurazione di Kafka. Questo è un compito arduo, quindi per aumentare la larghezza di banda è 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 garantisce una distribuzione uniforme del carico tra i broker). Nel nostro caso, dopo aver aggiunto un broker, la larghezza di banda del cluster è aumentata fino a ~580 MB/s (~1,1 milioni di messaggi al secondo). L'aumento è stato inferiore alle aspettative: principalmente a causa del disallineamento delle partizioni (non tutti i broker operano al massimo delle loro capacità).

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

Il funzionamento dei broker con gli storage è stato influenzato dal disallineamento delle partizioni:

Conclusioni

L'approccio iterativo presentato sopra può essere esteso per coprire scenari più complessi, che includono centinaia di consumer, repartitioning, aggiornamenti rolling, riavvii dei pod, ecc. Questo 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 un rapido e facile deployment del cluster, la sua configurazione, l'aggiunta/rimozione di broker e topic, la risposta a notifiche e la corretta operatività di Kafka in Kubernetes nel suo complesso. Il nostro obiettivo è aiutare a concentrarsi sul compito principale ("generare" e "consumare" messaggi Kafka), lasciando il lavoro pesante a Supertubes e all'operatore Kafka.

Se sei interessato alle tecnologie e ai progetti Open Source di Banzai Cloud, segui l'azienda su GitHub, LinkedIn o Twitter.