Ciao a tutti. In questo articolo parlerò del motivo per cui, nove mesi fa, abbiamo scelto Kafka in Avito e di cosa si tratta. Condividerò uno dei casi d'uso: il broker di messaggi. E infine, discuteremo dei vantaggi che abbiamo ottenuto adottando l'approccio Kafka as a Service.

Problema

Per cominciare, un po' di contesto. Un po' di tempo fa abbiamo iniziato a allontanarci dall'architettura monolitica e ora in Avito ci sono già diverse centinaia di servizi. Hanno i loro depositi, il loro stack tecnologico e rispondono alla loro parte di logica aziendale.

Uno dei problemi con un gran numero di servizi è la comunicazione. Il servizio A spesso vuole conoscere informazioni in possesso del servizio B. In questo caso, il servizio A si rivolge al servizio B tramite un'API sincrona. Il servizio C vuole sapere cosa sta succedendo nei servizi D e E, e questi, a loro volta, sono interessati ai servizi A e B. Quando ci sono molti servizi "curiosi", le connessioni tra di loro si trasformano in un groviglio complicato.

Tuttavia, in qualsiasi momento il servizio A potrebbe diventare non disponibile. Cosa dovrebbe fare il servizio B e tutti gli altri servizi che dipendono da esso in questa situazione? Se per completare un'operazione aziendale è necessario effettuare una serie di chiamate sincrone, la probabilità che l'intera operazione fallisca aumenta ulteriormente (e tanto più aumenta quanto più lunga è questa catena).

Scelta della tecnologia

Ok, i problemi sono chiari. Possono essere risolti creando un sistema di messaggistica centralizzato tra i servizi. Ora ogni servizio deve sapere solo di questo sistema di messaggistica. Inoltre, il sistema stesso deve essere resiliente ai guasti e scalabile orizzontalmente, e in caso di emergenze deve accumulare un buffer di richieste per una successiva elaborazione.

Adesso scegliamo la tecnologia che utilizzeremo per la consegna dei messaggi. Per farlo, prima comprendiamo cosa ci aspettiamo da essa:

• i messaggi tra i servizi non devono andare persi;
• i messaggi possono essere duplicati;
• i messaggi possono essere memorizzati e letti per un periodo di diversi giorni (buffer persistente);
• i servizi possono iscriversi ai dati di loro interesse;
• alcuni servizi possono leggere gli stessi dati;
• i messaggi possono contenere payload dettagliati e voluminosi (event-carried state transfer);
• a volte è necessaria una garanzia nell'ordine dei messaggi.

In questa fase, era fondamentale scegliere un sistema altamente scalabile e affidabile con un'alta capacità di throughput (almeno 100k messaggi di qualche kilobyte al secondo).

A questo punto abbiamo detto addio a RabbitMQ (difficile da mantenere stabile a elevati rps), PGQ di SkyTools (non sufficientemente veloce e scarsamente scalabile) e NSQ (non persistente). Tutte queste tecnologie vengono utilizzate nella nostra azienda, ma non si adattavano al problema in questione.

Successivamente, abbiamo iniziato a esplorare tecnologie nuove per noi — Apache Kafka, Apache Pulsar e NATS Streaming.

Per prima cosa abbiamo scartato Pulsar. Abbiamo ritenuto che Kafka e Pulsar siano soluzioni piuttosto simili. E nonostante Pulsar sia stato testato da grandi aziende, sia più recente e teoricamente offra una latenza inferiore, abbiamo deciso di mantenere Kafka come standard de facto per queste esigenze. Probabilmente, torneremo ad Apache Pulsar in futuro.

E alla fine sono rimasti due candidati: NATS Streaming e Apache Kafka. Abbiamo esaminato entrambi i sistemi in modo dettagliato, e entrambi erano adatti per il lavoro. Tuttavia, alla fine abbiamo avuto delle riserve sulla relativa giovinezza di NATS Streaming (e sul fatto che uno dei principali sviluppatori, Tyler Treat, ha deciso di lasciare il progetto per iniziarne uno proprio — Liftbridge). Inoltre, la modalità di clustering di NATS Streaming non permetteva un forte scaling orizzontale (probabilmente questo non è più un problema dopo l'aggiunta della modalità di partizionamento nel 2017).

Tuttavia, NATS Streaming è una tecnologia impressionante, scritta in Go e supportata dalla Cloud Native Computing Foundation. A differenza di Apache Kafka, non necessita di Zookeeper per funzionare (forse, presto si potrà dire lo stesso anche di Kafka), in quanto implementa RAFT al suo interno. Inoltre, NATS Streaming è più semplice da amministrare. Non escludiamo che in futuro potremmo tornare a questa tecnologia.

E tuttavia, oggi il nostro vincitore è Apache Kafka. Nei nostri test ha dimostrato di essere sufficientemente veloce (oltre un milione di messaggi al secondo in lettura e scrittura con dimensioni dei messaggi di 1 kilobyte), affidabile, scalabile e collaudata da grandi aziende in produzione. Inoltre, Kafka è supportato da diverse grandi aziende commerciali (noi, ad esempio, utilizziamo la versione di Confluent), e ha un ecosistema ben sviluppato.

Panoramica di Kafka

Prima di iniziare, consiglio subito un ottimo libro — «Kafka: The Definitive Guide» (disponibile anche in traduzione russa, ma i termini possono risultare un po' complicati). Qui si può trovare informazioni utili per una comprensione di base di Kafka e anche qualcosa in più. La documentazione di Apache e il blog di Confluent sono scritti egregiamente e facili da seguire.

Dunque, diamo un'occhiata a Kafka dall'alto. La topologia di base di Kafka è composta da producer, consumer, broker e zookeeper.

Broker

Il broker è responsabile dell'archiviazione dei tuoi dati. Tutti i dati sono memorizzati in formato binario e il broker sa poco riguardo a cosa rappresentano e quale sia la loro struttura.

Ogni tipo di evento logico si trova solitamente in un topic separato. Ad esempio, un evento di creazione di un annuncio può finire nel topic item.created, mentre un evento di modifica andrà in item.changed. I topic possono essere considerati come classificatori degli eventi. A livello di topic è possibile impostare parametri di configurazione come:

• la quantità di dati memorizzati e/o la loro età (retention.bytes, retention.ms);
• il fattore di ridondanza dei dati (replication factor);
• la dimensione massima di un singolo messaggio (max.message.bytes);
• il numero minimo di repliche allineate necessarie per poter registrare dati nel topic (min.insync.replicas);
• la possibilità di effettuare failover su una replica non sincronizzata con possibile perdita di dati (unclean.leader.election.enable);
• e molte altre ancora (https://kafka.apache.org/documentation/#topicconfigs).

A sua volta, ogni topic è suddiviso in una o più partizioni (partition). Sono nelle partizioni che alla fine finiscono gli eventi. Se nel cluster ci sono più broker, le partizioni verranno distribuite uniformemente su tutti i broker (per quanto possibile), consentendo di scalare il carico di scrittura e lettura su un topic su più broker.

Sul disco, i dati per ogni partizione sono memorizzati come segmenti di file, normalmente di un gigabyte (controllati tramite log.segment.bytes). Una caratteristica importante è che la rimozione dei dati dalle partizioni (al verificarsi della retention) avviene proprio per segmenti (non è possibile rimuovere un singolo evento da una partizione, si può rimuovere solo un intero segmento, e solo se non attivo).

Zookeeper

Zookeeper funge da archivio per i metadati e coordinatore. È lui in grado di dire se i broker sono attivi (può essere visto attraverso il comando zookeeper-shell ls /brokers/ids), quale broker è il controller (get /controller), se le partizioni sono in uno stato sincronizzato con le loro repliche (get /brokers/topics/topic_name/partitions/partition_number/state). Anche i producer e consumer inizialmente si connetteranno a zookeeper per sapere su quale broker sono memorizzati quali topic e partizioni. Nei casi in cui per un topic è impostato un fattore di replicazione maggiore di 1, zookeeper indicherà quali partizioni sono leader (in esse avverranno le scritture e da esse avverranno le letture). In caso di guasto di un broker, le informazioni sulle nuove partizioni leader saranno registrate proprio in zookeeper (a partire dalla versione 1.1.0 in modo asincrono, ed è importante).

Nelle versioni più vecchie di Kafka, zookeeper si occupava anche della memorizzazione degli offset, ma ora sono conservati in un topic speciale. __consumer_offsets sul broker (anche se puoi ancora utilizzare zookeeper per questi scopi).

Il modo più semplice per far diventare i tuoi dati una zucca è proprio perdere informazioni con zookeeper. In questo scenario, capire cosa e da dove leggere sarà molto difficile.

Producer

Il Producer è spesso un servizio che scrive direttamente dati in Apache Kafka. Il Producer sceglie il topic in cui verranno memorizzati i suoi messaggi tematici e inizia a scrivere informazioni. Ad esempio, un producer potrebbe essere un servizio di annunci. In tal caso, invierà nei topic tematici eventi come 'annuncio creato', 'annuncio aggiornato', 'annuncio cancellato', ecc. Ogni evento rappresenta una coppia chiave-valore.

Per impostazione predefinita, tutti gli eventi sono distribuiti nelle partizioni del topic con il metodo round-robin, se la chiave non è specificata (perdendo l'ordinamento), e tramite MurmurHash (chiave), se la chiave è presente (ordinamento all'interno di una singola partizione).

È importante notare che Kafka garantisce l'ordine degli eventi solo all'interno di una singola partizione. Tuttavia, spesso questo non rappresenta un problema. Ad esempio, è possibile garantire che tutte le modifiche a un annuncio specifico vengano aggiunte a una partizione (mantenendo così l'ordine di queste modifiche per l'annuncio). È possibile anche trasmettere un numero di sequenza in uno dei campi dell'evento.

Consumer

Il consumer è responsabile della ricezione dei dati da Apache Kafka. Tornando all'esempio di prima, un consumer può essere un servizio di moderazione. Questo servizio si iscriverà al topic del servizio di annunci e, quando verrà creato un nuovo annuncio, lo riceverà e lo analizzerà rispetto a determinate politiche prestabilite.

Apache Kafka memorizza quali sono stati gli ultimi eventi ricevuti dal consumer (per questo si utilizza un topic di sistema __consumer__offsets), garantendo così che, in caso di lettura riuscita, il consumer non riceva lo stesso messaggio due volte. Tuttavia, se si utilizza l'opzione enable.auto.commit = true e si delega completamente il tracciamento della posizione del consumer nel topic a Kafka, è possibile perdere dei dati.. Nel codice di produzione, la posizione del consumer è spesso controllata manualmente (lo sviluppatore gestisce il momento in cui deve necessariamente avvenire il commit dell'evento letto).

Nei casi in cui un solo consumer non è sufficiente (ad esempio, quando il flusso di nuovi eventi è molto elevato), è possibile aggiungere altri consumer, collegandoli insieme in un consumer group. Il consumer group rappresenta logicamente lo stesso consumer, ma con la distribuzione dei dati tra i membri del gruppo. Questo consente a ciascun partecipante di ricevere la propria quota di messaggi, aumentando così la velocità di lettura.

Risultati dei test

Qui non scriverò molto testo esplicativo, ma condividerò semplicemente i risultati ottenuti. I test sono stati condotti su 3 macchine fisiche (12 CPU, 384GB RAM, 15k SAS DISK, 10GBit/s Net), broker e zookeeper sono stati distribuiti in lxc.

Test delle prestazioni

Durante i test sono stati ottenuti i seguenti risultati.

• La velocità di scrittura di messaggi delle dimensioni di 1KB con 9 producer contemporaneamente — 1300000 eventi al secondo.
• La velocità di lettura di messaggi delle dimensioni di 1KB con 9 consumer contemporaneamente — 1500000 eventi al secondo.

Test di resilienza

Durante i test sono stati ottenuti i seguenti risultati (3 broker, 3 zookeeper).

• Il malfunzionamento di uno dei broker non porta all'arresto o all'inaccessibilità del cluster. Il funzionamento continua normalmente, ma i broker rimanenti subiscono un carico maggiore.
• Il malfunzionamento di due broker in un cluster di tre broker e min.isr = 2 porta all'inaccessibilità del cluster in scrittura, ma è accessibile in lettura. Se min.isr = 1, il cluster rimane accessibile sia in lettura che in scrittura. Tuttavia, questa modalità contraddice il requisito di alta integrità dei dati.
• Il malfunzionamento di uno dei server Zookeeper non porta all'arresto o all'inaccessibilità del cluster. Il funzionamento continua normalmente.
• Il malfunzionamento di due server Zookeeper porta all'inaccessibilità del cluster fino al ripristino di almeno uno dei server Zookeeper. Questa affermazione è valida per un cluster Zookeeper di 3 server. Di conseguenza, dopo le ricerche, è stato deciso di aumentare il cluster Zookeeper a 5 server per migliorare la resilienza.
  

Kafka as a service

Abbiamo constatato che Kafka è una tecnologia eccellente per affrontare la nostra sfida (la realizzazione di un broker di messaggi). Tuttavia, abbiamo deciso di vietare ai servizi di accedere direttamente a Kafka, chiudendola dietro un servizio di data-bus. Perché abbiamo fatto questo? In realtà, ci sono diverse ragioni.

• Il data-bus ha assunto tutte le responsabilità relative all'integrazione con Kafka (implementazione e configurazione di consumer e producer, monitoraggio, allerta, registrazione, scalabilità, ecc.). In questo modo, l'integrazione con il broker di messaggi avviene in modo molto semplice.

• Il data-bus ha consentito di astrarsi dal linguaggio specifico o dalla libreria utilizzata per lavorare con Kafka.

• Il data-bus ha permesso ad altri servizi di astrarsi dal livello di storage. Potremmo voler sostituire Kafka con Pulsar, e in tal caso nessuno se ne accorgerà (tutti i servizi conoscono solo l'API del data-bus).

• Il data-bus ha preso in carico la validazione degli schemi degli eventi.

• Con il data-bus è stata implementata l'autenticazione.

• Grazie al data-bus possiamo aggiornare le versioni di Kafka senza downtime, in modo invisibile, e gestire centralmente le configurazioni di producer, consumer, broker, ecc.

• Data-bus ci ha permesso di aggiungere funzionalità necessarie che mancano in Kafka (come l'audit dei topic, il controllo delle anomalie nel cluster, la creazione di DLQ, ecc.).

• Data-bus consente di implementare il failover in modo centralizzato per tutti i servizi.

Attualmente, per iniziare a inviare eventi al broker di messaggi, basta collegare una piccola libreria nel codice del proprio servizio. È tutto. Avete la possibilità di scrivere, leggere e scalare con una sola riga di codice. Tutta l'implementazione è nascosta da voi, emergono solo alcune maniglie come la dimensione del batch. Sotto il cofano, il servizio data-bus avvia in Kubernetes il numero necessario di istanze di producer e consumer, e fornisce loro la configurazione adatta, ma tutto questo è trasparente per il vostro servizio.

Naturalmente, non esiste una soluzione universale, e questo approccio ha le sue limitazioni.

• Data-bus deve essere gestito internamente, a differenza delle librerie di terze parti.
• Data-bus aumenta il numero di interazioni tra i servizi e il broker di messaggi, il che porta a una diminuzione delle prestazioni rispetto a Kafka nudo.
• Non tutto può essere così facilmente nascosto ai servizi e non vogliamo duplicare le funzionalità di KSQL o Kafka Streams nel data-bus, quindi talvolta è necessario consentire ai servizi di accedere direttamente.

Nel nostro caso, i vantaggi hanno superato gli svantaggi e la decisione di coprire il broker di messaggi con un servizio separato si è rivelata vantaggiosa. In un anno di utilizzo, non abbiamo avuto incidenti o problemi seri.

P.S. Grazie alla mia ragazza, Ekaterina Obalayeva, per le fantastiche illustrazioni di questo articolo. Se vi sono piaciute, qui ce ne saranno ancora di più.

Fonte: habr.com