Паттерны Backpressure и Circuit Breaker

Метрики нагрузки

• DAU (Daily Active Users) — количество пользователей в день. Также часто используют метрику MAU — Monthly Active Users.
• RPS (Requests Per Second) — количество запросов, которые сервер приложения обрабатывает в секунду.
• QPS (Queries Per Second) — количество запросов в секунду к базе данных.

• DAU вырос до 100 000 активных пользователей, то есть в тысячу раз.
• Количество запросов обычно получают в результате мониторинга метрик. А пока мониторинг не настроен, предположим, что каждый пользователь делал 5 запросов в секунду. В таком случае показатель RPS вырос с 500 до 500 000 запросов в секунду.
• После грамотной настройки кеширования лишь один запрос из 100 пропускал кеш и шёл напрямую в базу данных. Значит, старое значение QPS было 5, а новое — 5 000.

• Пользователи могут столкнуться с долгим временем ответа сервера или вовсе с обрывом запроса по тайм-ауту. Никто не хочет ждать загрузки страницы по 5 минут. Это ухудшит их опыт — они могут перейти на использование продукта конкурентов.
• Пользователи могут потерять важные данные. Например, если запрос на запись в базу данных не был правильно обработан.
• Такую систему гораздо сложнее поддерживать: пользователи будут медленнее получать обновления и новые фичи.

Паттерны снижения нагрузки

• замедлять выпуск деталей,
• выбрасывать детали,
• складывать детали на склад,
• создавать новый сборочный цех.

10 ГБ / 200 МБ/c ~ 50 секунд
50 МБ/c x 50 ~ 2,4 ГБ — достаточно много 

Паттерн Backpressure

1. Производитель данных, или producer, отправляет данные в очередь потребителя (например, Kafka).
2. Потребитель данных, или consumer, обрабатывает данные из очереди.
3. Канал обратной связи, или feedback channel, обеспечивает коммуникацию между потребителем и производителем. У потребителя появляется возможность давать сигналы обратной связи производителю о способности обрабатывать поступающие данные.
4. Механизм регулирования, или flow control mechanism, управляет потоком данных от производителя к потребителю на основе полученной обратной связи. В ответ на сигналы обратного давления производитель корректирует скорость или поведение передачи данных в соответствии с возможностями потребителя. Механизм регулирования потока может включать такие методы, как ограничение скорости, буферизация или динамическая настройка обработки данных.
5. Механизм сигнализации, или pressure signal mechanism, — средства, с помощью которых потребитель отправляет сигналы об обратном давлении производителю данных: сообщения, управляющие сообщения, сигнальные флаги или индикаторы доступности ресурсов.

Реализация паттерна Backpressure

• есть общий buffer (размер 5) для производителя и потребителя;
• производитель генерирует случайное число и пытается поместить его в буфер;
• если буфер заполнен, производитель ожидает;
• потребитель пытается использовать данные из буфера;
• если буфер пуст, потребитель ожидает;
• потребитель «спит» случайное время, чтобы имитировать обратное давление, то есть потребитель обрабатывает сообщения медленнее, чем их производит производитель.

import queue
import threading
import time
import random

class Producer(threading.Thread):
    def __init__(self, buffer):
        super().__init__()
        self.buffer = buffer

    def run(self):
        while True:
            item = random.randint(1, 100)
            print(f"Producing {item}")
            #ждем буфер
            while self.buffer.full():
                print("Buffer full, waiting...")
                time.sleep(1)
            
            self.buffer.put(item)
            time.sleep(random.random())  # Спим случайное время

class Consumer(threading.Thread):
    def __init__(self, buffer):
        super().__init__()
        self.buffer = buffer

    def run(self):
        while True:
            # Ждём, если буфер пустой
            while self.buffer.empty():
                print("Buffer empty, waiting for items...")
                time.sleep(1)
            
            item = self.buffer.get()
            print(f"Consuming {item}")
            time.sleep(1 + random.random())  # спим для эмуляции backpressure

if __name__ == "__main__":
    buffer = queue.Queue(maxsize=5) 

    producer = Producer(buffer)
    consumer = Consumer(buffer)

    producer.start()
    consumer.start()

    producer.join()
    consumer.join() 

Consuming 35
Producing 53
Producing 66
Producing 90
Buffer full, waiting...
Consuming 93
Producing 25
Buffer full, waiting. 

Паттерн Curcuit Breaker

Реализация паттерна Curcuit Breaker

import time

def circuit_breaker(fn, failure_count, time_threshold):
    failures = 0
    time_since_last_failure = 0
    is_closed = False **#начальное состояние, закрыт, вызовы идут**

    def wrapper(*args):
        nonlocal failures, time_since_last_failure, is_closed

        # if service is closed
        if is_closed:
            diff = time.time() * 1000 - time_since_last_failure

            # if the time since last failure has exceeded 
            # the time threshold
            # open the service
            if diff > time_threshold:
                is_closed = False
            # else throw error
            else:
                print("Service unavailable")
                return

        # try running the function
        # if it passes reset the failure count
        try:
            result = fn(*args) **#вызываем удалённый сервис**
            failures = 0
            return result
        # if function throws error / fails
        # increase the failure count and 
        # time when it fails
        except Exception as error:
            failures += 1
            time_since_last_failure = time.time() * 1000
            if failures >= failure_count:
                is_closed = True #открыт

            print("Error")

    return wrapper

def test_function():
    count = 0

    def inner():
        nonlocal count
        count += 1
        if count < 4:
            raise Exception("failed")
        else:
            return "hello"

    return inner

t = test_function()
c = circuit_breaker(t, 3, 200)

c()  # "Error"
c()  # "Error"
c()  # "Error"

# service is closed for 200 MS
c()  # "Service unavailable" 
c()  # "Service unavailable"
c()  # "Service unavailable"
c()  # "Service unavailable"
c()  # "Service unavailable"

# service becomes available after 300ms
time.sleep(0.3)
print(c())  # "hello" 

python3 имя_файла.py 

Error - ошибка вызова сервиса
Error - ошибка вызова сервиса
Error - ошибка вызова сервиса
Service unavailable - полуоткрытый
Service unavailable - полуоткрытый
Service unavailable - полуоткрытый
Service unavailable - полуоткрытый
Service unavailable - полуоткрытый
hello - открыт 

ИТОГИ

• Чтобы стабилизировать систему, команде нужно провести диагностику системы и выявить «узкие места». Для измерения нагрузки нужно для каждого сервиса собрать метрики производительности: DAU (количество пользователей в день), RPS (количество запросов, которые сервер приложения обрабатывает в секунду) и QPS (количество запросов в секунду к базе данных).
• Следующим шагом можно внедрить в архитектуру паттерн Backpressure. На уровне API нужно настроить лимитирование скорости запросов. Это уменьшит нагрузку на серверы, ограничив количество запросов, поступающих в систему одновременно. А затем настроить приоритеты для обработки запросов. Например, можно в первую очередь обрабатывать запросы на свежие новости, а менее актуальные данные помещать в низкоприоритетную очередь.
• Для ключевых микросервисов можно настроить Circuit Breaker. Это позволит временно отключать или ограничивать взаимодействие с проблемными сервисами, предотвращая дальнейшее распространение сбоев. В случае ошибки Circuit Breaker будет отправлять пользователю сообщение-заглушку, например: «Новостей очень много, чтобы их донести, нам нужно чуть больше времени».
• В перспективе стоит масштабироваться — добавить больше серверов или ресурсов в облаке, чтобы справляться с возросшей нагрузкой.

Что посмотреть про паттерны отказоустойчивости архитектуры?