Cześć wszystkim. W tym artykule opowiem, dlaczego dziewięć miesięcy temu w Avito wybraliśmy Kafka i czym ona jest. Podzielę się jednym z przypadków użycia — brokerem wiadomości. Na koniec porozmawiamy o tym, jakie korzyści przyniosło nam zastosowanie podejścia Kafka as a Service.

Problem

Na początek trochę kontekstu. Jakiś czas temu zaczęliśmy odchodzić od monolitycznej architektury i obecnie w Avito mamy już kilka setek różnych usług. Mają one swoje własne bazy danych, swoje stosy technologiczne i odpowiadają za swoją część logiki biznesowej.

Jednym z problemów przy dużej liczbie usług są komunikacje. Usługa A często chce uzyskać informację, którą dysponuje usługa B. W takim przypadku usługa A zwraca się do usługi B za pośrednictwem synchronizowanego API. Usługa B chce wiedzieć, co się dzieje w usługach G i D, a te z kolei interesują się usługami A i B. Gdy takich 'ciekawskich' usług jest wiele, połączenia między nimi stają się splątanym węzłem.

W każdej chwili usługa A może stać się niedostępna. Co w takim przypadku zrobi usługa B oraz wszystkie inne usługi zależne od niej? A jeśli do wykonania operacji biznesowej konieczne jest wykonanie łańcucha kolejnych synchronizowanych wywołań, prawdopodobieństwo awarii całej operacji staje się jeszcze wyższe (im dłuższy jest ten łańcuch, tym wyższe to prawdopodobieństwo).

Wybór technologii

Ok, problemy są zrozumiałe. Można je rozwiązać, tworząc scentralizowany system wymiany wiadomości między usługami. Teraz każda usługa musi znać tylko ten system wymiany wiadomości. Dodatkowo sam system musi być odporny na awarie i skalowalny poziomo, a także w przypadku awarii gromadzić w sobie bufor odwołań do późniejszego przetworzenia.

Teraz wybierzmy technologię, na której zostanie zrealizowana dostawa wiadomości. Na początku zrozummy, czego od niej oczekujemy:

• wiadomości między usługami nie powinny być tracone;
• wiadomości mogą być duplikowane;
• wiadomości można przechowywać i odczytywać na głębokość kilku dni (bufor trwały);
• usługi mogą subskrybować interesujące je dane;
• kilka usług może odczytywać te same dane;
• wiadomości mogą zawierać szczegółowy, obszerne payload (event-carried state transfer);
• czasami potrzebna jest gwarancja kolejności wiadomości.

Również krytycznie ważne było dla nas wybranie maksymalnie skalowalnego i niezawodnego systemu o wysokiej przepustowości (nie mniej niż 100k wiadomości na kilka kilobajtów na sekundę).

Na tym etapie pożegnaliśmy się z RabbitMQ (trudno utrzymać stabilność przy wysokich rps), PGQ od SkyTools (niewystarczająco szybki i źle skalowalny) oraz NSQ (niepersistentny). Wszystkie te technologie są używane w naszej firmie, ale do rozwiązywanego zadania nie były odpowiednie.

Następnie zaczęliśmy przyglądać się nowym dla nas technologiom — Apache Kafka, Apache Pulsar i NATS Streaming.

Najpierw odrzuciliśmy Pulsar. Stwierdziliśmy, że Kafka i Pulsar to dość podobne rozwiązania. I mimo że Pulsar jest sprawdzony przez duże firmy, nowszy i teoretycznie oferuje niższe opóźnienia, postanowiliśmy pozostawić Kafka jako de facto standard dla takich zadań. Prawdopodobnie w przyszłości wrócimy do Apache Pulsar.

Pozostały więc dwa kandydaci: NATS Streaming i Apache Kafka. Dość szczegółowo zbadaliśmy oba rozwiązania i oba nadawały się do zadania. Ale ostatecznie obawialiśmy się względnej młodości NATS Streaming (i tego, że jeden z głównych programistów, Tyler Treat, zdecydował się opuścić projekt i rozpocząć własny — Liftbridge). W dodatku tryb klastrowania NATS Streaming nie pozwalał na silne poziome skalowanie (prawdopodobnie nie jest to już problem po dodaniu trybu partycjonowania w 2017 roku).

Niemniej jednak, NATS Streaming to świetna technologia napisana w Go i wspierana przez Cloud Native Computing Foundation. W przeciwieństwie do Apache Kafka, nie potrzebuje Zookeepera do działania (możliwe, wkrótce będzie można powiedzieć to samo o Kafka), ponieważ wewnętrznie implementuje RAFT. Ponadto, NATS Streaming jest prostszy w administracji. Nie wykluczamy, że w przyszłości wrócimy do tej technologii.

Jednak na dzień dzisiejszy naszym zwycięzcą została Apache Kafka. W naszych testach wykazała się wystarczającą szybkością (ponad milion wiadomości na sekundę przy odczycie i zapisie o objętości wiadomości 1 kilobajt), wystarczającą niezawodnością, dobrą skalowalnością i została sprawdzona w produkcji przez duże firmy. Dodatkowo, Kafka wspiera przynajmniej kilka dużych firm komercyjnych (my, na przykład, korzystamy z wersji Confluent), a także Kafka ma rozwiniętą ekosystem.

Przegląd Kafka

Zanim zaczniemy, od razu polecam świetną książkę — „Kafka: The Definitive Guide” (jest także w rosyjskim tłumaczeniu, ale terminy trochę łamią mózg). Można w niej znaleźć informacje niezbędne do podstawowego zrozumienia Kafki, a nawet trochę więcej. Sama dokumentacja od Apache i blog od Confluent są również doskonale napisane i łatwe do przeczytania.

Zatem przyjrzyjmy się, jak wygląda Kafka z lotu ptaka. Podstawowa topologia Kafki składa się z producenta, konsumenta, brokera i zookeepra.

Broker

Za przechowywanie Twoich danych odpowiada broker (broker). Wszystkie dane są przechowywane w formie binarnej, a broker ma mało wiedzy na temat tego, czym one są i jaka jest ich struktura.

Każdy logiczny typ zdarzeń zazwyczaj znajduje się w osobnym temacie (topic). Na przykład zdarzenie utworzenia ogłoszenia może trafić do tematu item.created, a zdarzenie jego zmiany — do item.changed. Tematy można traktować jako klasyfikatory zdarzeń. Na poziomie tematu można ustawić takie parametry konfiguracyjne, jak:

• objętość przechowywanych danych i/lub ich wiek (retention.bytes, retention.ms);
• współczynnik redundancji danych (replication factor);
• maksymalny rozmiar jednego komunikatu (max.message.bytes);
• minimalna liczba zgodnych replik, przy której dane będą mogły być zapisane w temacie (min.insync.replicas);
• możliwość przeprowadzenia failover na niesynchronizowanej opóźnionej replicie z potencjalną utratą danych (unclean.leader.election.enable);
• i wiele innych (https://kafka.apache.org/documentation/#topicconfigs).

Z kolei każdy temat dzieli się na jedną lub więcej partycji (partition). To właśnie w partycjach ostatecznie trafiają zdarzenia. Jeśli w klastrze jest więcej niż jeden broker, partycje będą równomiernie rozłożone pomiędzy wszystkimi brokerami (na ile to możliwe), co pozwoli na skalowanie obciążenia zapisu i odczytu w jednym temacie na kilka brokerów.

Na dysku dane dla każdej partycji przechowywane są w formie plików segmentów, domyślnie równych jednemu gigabajtowi (kontrolowane przez log.segment.bytes). Ważną cechą jest to, że usuwanie danych z partycji (po wygaśnięciu retention) odbywa się właśnie na poziomie segmentów (nie można usunąć jednego zdarzenia z partycji, można usunąć tylko cały segment, przy czym tylko nieaktywny).

Zookeeper

Zookeeper pełni rolę magazynu metadanych i koordynatora. To on potrafi powiedzieć, czy brokerzy są aktywni (można to zobaczyć oczami zookeepra przez zookeeper-shell komendą ls /brokers/ids), który z brokerów jest kontrolerem (get /controller), czy partycje są w stanie synchronizacji ze swoimi replikami (get /brokers/topics/topic_name/partitions/partition_number/state). To właśnie do zookeepera najpierw udają się producent i konsument, aby dowiedzieć się, na którym brokerze jakie tematy i partycje są przechowywane. W przypadkach, gdy dla tematu określono współczynnik replikacji większy niż 1, zookeeper wskaże, które partycje są liderami (to w nich będą dokonywane zapisy, a stamtąd również nastąpi odczyt). W przypadku awarii brokera informacje o nowych liderach partycji będą zapisane w zookeeperze (od wersji 1.1.0 asynchronicznie, i to jest istotne).

W starszych wersjach Kafka zookeeper odpowiadał również za przechowywanie offsetów, ale obecnie są one przechowywane w specjalnym temacie __consumer_offsets na brokerze (choć wciąż możesz używać zookeepera do tych celów).

Najprostszym sposobem na przekształcenie twoich danych w dynię jest właśnie utrata informacji z zookeepera. W takim scenariuszu zrozumienie, co i skąd należy odczytać, będzie bardzo trudne.

Producent

Producent — to najczęściej usługa dokonująca bezpośredniego zapisu danych do Apache Kafka. Producent wybiera temat, w którym będą przechowywane jego wiadomości, i zaczyna w nim zapisywać informacje. Na przykład producentem może być usługa ogłoszeń. W takim przypadku będzie on wysyłał do tematycznych tematów takie zdarzenia, jak „ogłoszenie utworzone”, „ogłoszenie zaktualizowane”, „ogłoszenie usunięte” itd. Każde zdarzenie to para klucz-wartość.

Domyślnie wszystkie zdarzenia rozdzielane są pomiędzy partycje tematu metodą round-robin, jeśli klucz nie jest określony (tracąc porządek), oraz przez MurmurHash (klucz), jeśli klucz jest obecny (zachowując porządek w ramach jednej partycji).

Tutaj od razu warto zauważyć, że Kafka gwarantuje porządek zdarzeń tylko w obrębie jednej partycji. Ale w rzeczywistości często nie jest to problemem. Na przykład można gwarantować dodawanie wszystkich zmian tego samego ogłoszenia do jednej partycji (w ten sposób zachowując porządek tych zmian w obrębie ogłoszenia). Można także przekazać numer porządkowy w jednym z pól zdarzenia.

Consumer

Consumer odpowiada za odbieranie danych z Apache Kafka. Jeśli wrócić do powyższego przykładu, consumerem może być serwis moderacji. Ten serwis będzie subskrybować temat serwisu ogłoszeń i przy pojawieniu się nowego ogłoszenia odbierze je oraz przeanalizuje pod kątem zgodności z określonymi politykami.

Apache Kafka zapamiętuje, jakie były ostatnie zdarzenia, które odebrał consumer (do tego służy serwisowy temat __consumer__offsets), co gwarantuje, że przy pomyślnym odczycie consumer nie otrzyma tego samego komunikatu dwa razy. Niemniej jednak, jeśli używać opcji enable.auto.commit = true i całkowicie powierzyć kontrolowanie pozycji consumer'a w temacie Kafce, można stracić dane. W kodzie produkcyjnym najczęściej kontrola pozycji consumer’a odbywa się ręcznie (programista sam decyduje, kiedy musi nastąpić commit odczytanego zdarzenia).

W przypadkach, gdy jeden consumer nie wystarcza (na przykład, gdy strumień nowych zdarzeń jest bardzo duży), można dodać kilku dodatkowych consumerów, łącząc je w consumer group. Consumer group logicznie stanowi taki sam consumer, ale z rozdzielonymi danymi między uczestnikami grupy. Umożliwia to każdemu z uczestników wzięcie swojej części komunikatów, co pozwala zwiększyć szybkość odczytu.

Wyniki testowania

Tutaj nie będę pisał dużo wyjaśniającego tekstu, po prostu podzielę się uzyskanymi wynikami. Testy były przeprowadzane na 3 fizycznych maszynach (12 CPU, 384GB RAM, 15k SAS DISK, 10GBit/s Net), brokerzy i zookeeper zostali uruchomieni w lxc.

Testowanie wydajności

W trakcie testów uzyskano następujące wyniki.

• Szybkość zapisu komunikatów o rozmiarze 1KB jednocześnie przez 9 producerów – 1300000 zdarzeń na sekundę.
• Szybkość odczytu komunikatów o rozmiarze 1KB jednocześnie przez 9 consumerów – 1500000 zdarzeń na sekundę.

Testowanie odporności na awarie

W trakcie testów uzyskano następujące wyniki (3 brokerzy, 3 zookeepery).

• Nieoczekiwane zakończenie pracy jednego z brokerów nie prowadzi do zatrzymania lub niedostępności klastra. Praca trwa w normalnym trybie, ale na pozostałe brokerzy przypada większe obciążenie.
• Nieprawidłowe zakończenie dwóch brokerów w przypadku klastra składającego się z trzech brokerów i min.isr = 2 prowadzi do braku dostępności klastra do zapisu, ale dostępność do odczytu pozostaje. W przypadku, gdy min.isr = 1, klaster nadal jest dostępny zarówno do odczytu, jak i do zapisu. Niemniej jednak, ten tryb jest sprzeczny z wymaganiem wysokiej integralności danych.
• Nieprawidłowe zakończenie jednego z serwerów Zookeeper nie prowadzi do zatrzymania ani braku dostępności klastra. Praca trwa w trybie normalnym.
• Nieprawidłowe zakończenie dwóch serwerów Zookeeper prowadzi do braku dostępności klastra do momentu przywrócenia działania przynajmniej jednego z serwerów Zookeeper. To stwierdzenie jest prawdziwe dla klastra Zookeeper z 3 serwerami. W wyniku badań zdecydowano o zwiększeniu klastra Zookeeper do 5 serwerów w celu zwiększenia odporności na awarie.
  

Kafka jako usługa

Przekonaliśmy się, że Kafka to doskonała technologia, która pozwala nam rozwiązać postawione przed nami zadanie (implementację brokera wiadomości). Niemniej jednak postanowiliśmy zabronić usługom bezpośredniego dostępu do Kafki i zamknęliśmy ją z góry usługą data-bus. Dlaczego to zrobiliśmy? W rzeczywistości jest kilka powodów.

• Data-bus przyjął na siebie wszystkie zadania związane z integracją z Kafką (implementacja i konfiguracja konsumentów oraz producentów, monitorowanie, alertowanie, logowanie, skalowanie itp.). Dzięki temu integracja z brokerem wiadomości odbywa się w maksymalnie prosty sposób.

• Data-bus umożliwił abstrahowanie od konkretnego języka czy biblioteki do pracy z Kafką.

• Data-bus umożliwił innym usługom abstrahowanie od warstwy przechowywania. Być może w pewnym momencie zmienimy Kafkę na Pulsara, a nikt niczego nie zauważy (wszystkie usługi znają tylko API data-bus).

• Data-bus przejął walidację schematów zdarzeń.

• Za pomocą data-bus zrealizowano autoryzację.

• Pod przykryciem data-bus możemy bez przestojów, niezauważalnie aktualizować wersje Kafki, centralnie prowadzić konfiguracje producentów, konsumentów, brokerów itp.

• Data-bus pozwolił na dodanie nam potrzebnych funkcji, które nie są dostępne w Kafce (takich jak audyt tematów, kontrola anomalii w klastrze, tworzenie DLQ itp.).

• Data-bus umożliwia realizację failover centralnie dla wszystkich usług.

Obecnie, aby rozpocząć wysyłanie zdarzeń do brokera wiadomości, wystarczy podłączyć małą bibliotekę w kodzie swojego serwisu. To wszystko. Zyskujesz możliwość pisania, odczytywania i skalowania za pomocą jednej linii kodu. Cała implementacja jest ukryta przed tobą, na zewnątrz wystają tylko kilka uchwytów typu rozmiar partii. Pod maską serwis data-bus uruchamia w Kubernetes odpowiednią liczbę instancji producerów i konsumentów, i dostarcza im wymaganą konfigurację, ale wszystko to jest przezroczyste dla twojego serwisu.

Oczywiście, nie ma złotego środka, a takie podejście ma swoje ograniczenia.

• Data-bus trzeba utrzymywać samodzielnie, w przeciwieństwie do zewnętrznych bibliotek.
• Data-bus zwiększa liczbę interakcji między serwisami a brokerem wiadomości, co prowadzi do spadku wydajności w porównaniu do surowej Kafki.
• Nie wszystko można tak łatwo ukryć przed serwisami, nie chcemy duplikować funkcjonalności KSQL lub Kafka Streams w data-bus, dlatego czasami musimy pozwolić serwisom na bezpośredni dostęp.

W naszym przypadku zalety przewyższyły wady i decyzja o ukryciu brokera wiadomości za pomocą osobnego serwisu okazała się uzasadniona. W ciągu roku użytkowania nie mieliśmy żadnych poważnych awarii ani problemów.

P.S. Dziękuję mojej dziewczynie, Ekaterinie Obalaiewej, za świetne grafiki do tego artykułu. Jeśli się wam spodobały, tutaj znajdą się jeszcze więcej ilustracji.

Źródło: habr.com