Kafka-консьюмеры не всегда работают так стабильно, как хотелось бы. Иногда они просто зависают — без ошибок, без падений, но и без обработки сообщений. LivenessProbe в Kubernetes помогает автоматически перезапускать зависшие сервисы, но с Kafka-консьюмерами всё не так просто: стандартного решения для них нет. В этой статье разберём, как правильно реализовать livenessProbe для консьюмеров с помощью паттерна Heartbeat, чтобы не перезапускать их вручную.
Содержание
Зачем нужен livenessProbe
В Kubernetes livenessProbe периодически опрашивает под (приложение) на предмет работоcпособности и перезапускает его, если что-то не работает. Если бы это был контейнер с API на FastAPI, то мы бы удостоверились, что контейнер возвращает правильный HTTP код 200 с одного из эндпоинтов. Допустим, приложение зависло и перестало отдавать ответ клиенту – livenessProbe перезапустил бы контейнер.
В чем проблема livenessProbe для Kafka-консьюмера
Для Kafka Consumer нет единого описанного подхода к реализации такого механизма, из-за чего разработчики сходу не могут реализовать эту функциональность правильно. Зачастую можно наткнуться на статьи, где предлагают следить за Сonsumer lag, что делает такой подход некорректным, так-как наличие лага не говорит, есть ли проблема у конкретного консьюмера или он просто не справляется с нагрузкой.
В моей голове родилось несколько подходов к проверке (увы, не все из них жизнеспособны):
Через consumer lag (неточный);
Через unix сигналы (проверяет, не завис ли процесс ОС, неточный);
Через внутренние метрики кафки (сложно);
Через кастомный механизм heartbeat в консьюмере (оптимально и точно);
Через отправку сообщения в health топики и его вычитку в livenessProbe (точно, но громоздко)
Реализация livenessProbe через паттерн HeartBeat ❤️
Простым решением проверки работы консьюмера может стать периодическая отправка хартбитов в любое хранилище, а затем сверка текущего времени и времени последнего бита.
Допустим, мы примерно знаем, что если задача выполняется больше 10 секунд, то скорее всего консьюмер завис и нам стоит его перезапустить. Каждый цикл получения и обработки сообщений из топика сохраняем данные по времени в некое хранилище. Самый простой вариант — в файл. Kubernetes затем может посмотреть в файл, сравнить время последнего бита с текущим и перезапустить консьюмер, если разница больше 10 секунд.
Реализация на Python
Для использования этого подхода необходимы изменения на стороне консьюмера, а для Kubernetes необходим скрипт.
Абстрактный класс HeartBeat
Опишем абстрактный класс HeartBeat, наследуя и реализуя который мы сможем вызывать методы save для сохранения heartbeat консьюмера. Это позволит нам в дальнейшем подменять реализацию – сохранять данные в файл, в редис или БД.
import abc class HeartBeat(abc.ABC): @abc.abstractmethod def save(self) -> None: """Save timestamp data to mark that worker is alive.""" @abc.abstractmethod def is_alive(self) -> bool: """Check that worker is alive."""
Структура данных
Опишем структуру данных о heartbeat. В нашем случае важно хранить идентификатор хоста и время. И должно это сохраняться в JSON.
{ "host_id": "00000000-0000-0000-0000-0242ac160007", "last_beat": "2023-12-08T09:38:01.255909+00:00" }
class HeartBeatState(BaseModel): host_id: str last_beat: datetime
FileHeartBeat – сохранение в файл
Для простоты сохраним данные в файл – напишем реализацию FileHeartBeat. Придумаем файлу имя, соберем к нему путь, затем сериализуем HeartBeat в JSON и сохраним в файле.
HEART_BEAT_FILE_NAME = "heartbeat.json" logger = logging.getLogger(__name__) class FileHeartBeat(HeartBeat): """ Save heartbeats to a file to check if consumer is alive by livenessProbe. """ def save(self): """Save heartbeat timestamp data to indicate that worker is alive.""" heartbeat_file_path = self._retrieve_heartbeat_file_path() host_id = uuid.UUID(int=uuid.getnode()) last_beat = now() state = HeartBeatState( host_id=str(host_id), last_beat=last_beat, ) self._save_state_to_file(heartbeat_file_path, state) def _save_state_to_file( self, heartbeat_file_path: Path, state: HeartBeatState, ) -> None: """Save heartbeat timestamp and other information in a file.""" logger.info("Saving heart beat to file: %s, %s", heartbeat_file_path, str(state.json())) with open(heartbeat_file_path, "w+") as worker_heartbeat_file: worker_heartbeat_file.write(state.json()) def _retrieve_heartbeat_file_path(self) -> Path: """Construct file path where heartbeat info is located.""" base_dir = Path(settings.BASE_DIR) heartbeat_file_name = HEART_BEAT_FILE_NAME heartbeat_file_path = base_dir.joinpath(heartbeat_file_name) return heartbeat_file_path
FileHeartBeat – проверка, жив ли консьюмер
Для проверки, что консьюмер жив, опишем метод is_alive. Прочтем файл, десериализуем данные из JSON в нашу модель и сравним, сколько прошло секунд с момент последнего бита. Если больше, чем мы ожидаем, то считаем консьюмер мертвым.
class FileHeartBeat(HeartBeat): """ Saves heartbeats to a file to check if consumer is alive by livenessProbe. """ def is_alive(self, max_time_after_worker_dead: int = 15) -> bool: """ Main livenessProbe method to check if consumer is healthy. """ heartbeat_file_path = self._retrieve_heartbeat_file_path() try: logger.info("Trying to check heartbeat file: %s", heartbeat_file_path) heart_beat_stats = self._read_heartbeat_file(heartbeat_file_path) except FileNotFoundError: logger.error("Heartbeat file not found: %s", heartbeat_file_path) return False logger.info("Got last heartbeat info: %s", str(heart_beat_stats.json())) time_after_last_beat = now() - heart_beat_stats.last_beat if time_after_last_beat.seconds > max_time_after_worker_dead: logger.error( "Worker is dead. Time since last beat %s > %s", time_after_last_beat.seconds, time_offset, ) return False logger.info( "Worker is alive. Time since last beat %s < %s", time_after_last_beat.seconds, time_offset, ) return True def _read_heartbeat_file(self, heartbeat_file_path: Path) -> HeartBeatState: """Read heartbeat timestamp and other information from a file.""" with open(heartbeat_file_path) as worker_heartbeat_file: heart_beat_stats_json: dict = json.load(worker_heartbeat_file) heart_beat_state = HeartBeatState(**heart_beat_stats_json) return heart_beat_state def _retrieve_heartbeat_file_path(self) -> Path: """Construct file path where heartbeat info is located.""" base_dir = Path(settings.BASE_DIR) heartbeat_file_name = HEART_BEAT_FILE_NAME heartbeat_file_path = base_dir.joinpath(heartbeat_file_name) return heartbeat_file_path
FileHeartBeat – интегрируем в консьюмер
Интегрируем этот код в код консьюмера. В данном случае используется датакласс для возможности подмены реализации. В начале каждого цикла сохраняем данные, что живы.
@dataclasses.dataclass class ConfluentKafkaConsumer: heart_beat: HeartBeat def run(self): consumer = ... .... while True: message: ConfluentKafkaMessage = consumer.poll(2.0) self.heart_beat.save() if not message: # No message received continue self._process_message(message) confluent_kafka_consumer = ConfluentKafkaConsumer( heart_beat=FileHeartBeat(), )
FileHeartBeat - интегрируем в CLI
Дальше необходимо написать CLI-скрипт, который сообщит Kubernetes о том, что воркер жив или умер. Для этого необходима подобная функция, в зависимости от вашего фреймворка:
ERROR_CODE = 1 ... # Your CLI def health_check(max_time: Optional[int]): heart_beat = FileHeartBeat() is_alive = heart_beat.is_alive(max_time_after_worker_dead=max_time) if not is_alive: sys.exit(ERROR_CODE)
FileHeartBeat – интегрируем в livenessProbe
В helm пропишем livenessProbe проверять состояние консьюмера каждые 30 секунд. У нас используется Django, поэтому запускаем через него.
livenessProbe: exec: command: - ./manage.py - consumer - is_alive - --max_time - "30"
На этом всё. Мы готовы проверять консьюмеры. Код приведен концептуальный.
Что дальше?
Вы можете реализовать сохранение Heartbeat в БД или Redis, а затем в админке смотреть состояние ваших консьюмеров.
Вы можете пойти ещё дальше и написать команды для приостановки вашего консьюмера прямо в коде через подобную концепцию проверки состояния в цикле перед обработкой сообщения. Это может быть удобно, когда нужно сбросить оффсеты в критических случаях (консьюмер группа должна быть остановлена)
Помимо вышеописанного, можно попробовать выделить код в отдельную библиотеку и переиспользовать.
Выводы ✍️
Из опыта использования этого подхода могу выделить следующее:
После реализации механизма Heartbeat мы перестали перезапускать консьюмеры руками и полностью автоматизировали этот процесс.
Среди возможных реализаций данный способ оказался наиболее простым и точным.
Данный подход позволяет расширять функциональность по отслеживанию и управлению консьюмерами – если использовать Redis, то можно отследить в админке консьюмеры, а также реализовать функциональность по управлению без перезапуска и девопсов прямо из админки.
Дополнительные материалы 📚
Вот, что ещё можно почитать на эту тему:
