Введение

Микросервисная архитектура — это подход к разработке программного обеспечения, при котором приложение строится как набор небольших, независимо развёртываемых сервисов. Каждый сервис реализует конкретную бизнес-функцию, имеет собственное хранилище данных и взаимодействует с другими сервисами через чётко определённые API.

Этот архитектурный стиль стал доминирующим в крупных технологических компаниях — Netflix, Amazon, Uber, Spotify — и позволил им масштабировать не только инфраструктуру, но и команды разработки. Однако микросервисы — не серебряная пуля. Понимание их преимуществ, недостатков и альтернатив критически важно для принятия правильных архитектурных решений.

---

1. Монолит vs Микросервисы vs Модульный монолит

Монолитная архитектура

Монолит — это приложение, развёртываемое как единый артефакт. Все компоненты — UI, бизнес-логика, доступ к данным — живут в одной кодовой базе и одном процессе.

┌─────────────────────────────────────────────────────┐
│                       МОНОЛИТ                       │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌────────┐  │
│  │  Users  │  │ Orders  │  │Products │  │Payments│  │
│  └────┬────┘  └────┬────┘  └────┬────┘  └───┬────┘  │
│       └────────────┴────────────┴───────────┘       │
│                          │                          │
│                  ┌───────▼───────┐                  │
│                  │   Database    │                  │
│                  └───────────────┘                  │
└─────────────────────────────────────────────────────┘

Преимущества монолита:

• Простота разработки и отладки
• Единый deployment pipeline
• Отсутствие сетевых вызовов между компонентами
• Простота транзакций (ACID из коробки)
• Низкий операционный overhead

Недостатки монолита:

• Сложность масштабирования отдельных компонентов
• Долгое время сборки и развёртывания
• Технологический lock-in (один стек для всего)
• Риск "большого комка грязи" (Big Ball of Mud)
• Сложность onboarding новых разработчиков в большую кодовую базу

Микросервисная архитектура

Микросервисы — это архитектурный стиль, при котором система разбивается на небольшие автономные сервисы, каждый из которых:

• Реализует одну бизнес-функцию
• Имеет собственную базу данных
• Развёртывается независимо
• Взаимодействует через сеть (HTTP/gRPC/messaging)

┌──────────────────────────────────────────────────────────────┐
│                        МИКРОСЕРВИСЫ                          │
│                                                              │
│  ┌──────────┐   ┌──────────┐   ┌─────────┐   ┌──────────┐    │
│  │  Users   │   │  Orders  │   │ Products│   │ Payments │    │
│  │ Service  │   │ Service  │   │ Service │   │ Service  │    │
│  └────┬─────┘   └────┬─────┘   └────┬────┘   └────┬─────┘    │
│       │              │              │             │          │
│  ┌────▼─────┐    ┌───▼────┐    ┌────▼────┐   ┌────▼─────┐    │
│  │PostgreSQL│    │  MySQL │    │  Mongo  │   │PostgreSQL│    │
│  └──────────┘    └────────┘    └─────────┘   └──────────┘    │
└──────────────────────────────────────────────────────────────┘

Преимущества микросервисов:

• Независимое масштабирование сервисов
• Технологическая гибкость (polyglot persistence)
• Независимые релизы и команды
• Изоляция отказов (fault isolation)
• Возможность оптимизации каждого сервиса под его нагрузку

Недостатки микросервисов:

• Сложность распределённой системы
• Сетевая латентность и частичные отказы
• Сложность обеспечения консистентности данных
• Операционный overhead (мониторинг, логирование, tracing)
• Необходимость зрелой DevOps-культуры

Модульный монолит

Модульный монолит — это золотая середина: приложение развёртывается как единый артефакт, но внутри строго разделено на модули с чёткими границами и контрактами.

┌─────────────────────────────────────────────────────┐
│                  МОДУЛЬНЫЙ МОНОЛИТ                  │
│   ┌─────────────┐ ┌─────────────┐ ┌─────────────┐   │
│   │   Module:   │ │   Module:   │ │   Module:   │   │
│   │   Users     │ │   Orders    │ │  Products   │   │
│   │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │   │
│   │ │   API   │ │ │ │   API   │ │ │ │   API   │ │   │
│   │ ├─────────┤ │ │ ├─────────┤ │ │ ├─────────┤ │   │
│   │ │ Domain  │ │ │ │ Domain  │ │ │ │ Domain  │ │   │
│   │ ├─────────┤ │ │ ├─────────┤ │ │ ├─────────┤ │   │
│   │ │  Data   │ │ │ │  Data   │ │ │ │  Data   │ │   │
│   │ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │   │
│   └──────┬──────┘ └──────┬──────┘ └──────┬──────┘   │
│          └───────────────┼───────────────┘          │
│                   ┌──────▼──────┐                   │
│                   │  Database   │                   │
│                   │  (schemas)  │                   │
│                   └─────────────┘                   │
└─────────────────────────────────────────────────────┘

Ключевые принципы:

• Модули общаются только через публичные интерфейсы
• Каждый модуль имеет свою схему в БД
• Прямой доступ к таблицам другого модуля запрещён
• Циклические зависимости запрещены на уровне архитектуры

Преимущества модульного монолита:

• Простота deployment (один артефакт)
• Чёткие границы модулей подготавливают к возможному split
• Легко извлечь модуль в микросервис при необходимости
• Транзакции остаются простыми
• Нет сетевого overhead между модулями

Когда что выбирать

Правило от Мартина Фаулера: "Monolith First" — начинайте с монолита, разбивайте на сервисы когда появится реальная необходимость и понимание границ доменов.

---

2. Decomposition Patterns

Правильное разбиение системы на сервисы — одна из самых сложных задач в микросервисной архитектуре. Неверные границы приводят к распределённому монолиту — системе с недостатками и монолита, и микросервисов.

Decomposition by Business Capability

Разбиение по бизнес-возможностям — наиболее рекомендуемый подход. Каждый сервис соответствует конкретной бизнес-функции организации.

Пример для e-commerce:

┌────────────────────────────────────────────────────────┐
│                   БИЗНЕС-ВОЗМОЖНОСТИ                   │
│                                                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │   Product    │  │    Order     │  │   Customer   │  │
│  │  Management  │  │  Management  │  │  Management  │  │
│  └──────────────┘  └──────────────┘  └──────────────┘  │
│                                                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │   Payment    │  │   Shipping   │  │  Inventory   │  │
│  │  Processing  │  │              │  │  Management  │  │
│  └──────────────┘  └──────────────┘  └──────────────┘  │
└────────────────────────────────────────────────────────┘

Decomposition by Subdomain (DDD)

Domain-Driven Design предлагает разбиение на основе ограниченных контекстов (Bounded Contexts).

Типы поддоменов:

• Core Domain — ключевое конкурентное преимущество (максимум инвестиций)
• Supporting Subdomain — необходимо для бизнеса, но не является уникальным
• Generic Subdomain — стандартная функциональность (можно купить/аутсорсить)

┌───────────────────────────────────────────────────────────┐
│                     E-COMMERCE ДОМЕНЫ                     │
│                                                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │                     CORE DOMAIN                     │  │
│  │   ┌─────────────────┐         ┌─────────────────┐   │  │
│  │   │ Recommendation  │         │     Pricing     │   │  │
│  │   │     Engine      │         │     Strategy    │   │  │
│  │   └─────────────────┘         └─────────────────┘   │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │                SUPPORTING SUBDOMAIN                 │  │
│  │  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐ │  │
│  │  │ Orders  │  │Inventory│  │Shipping │  │ Reviews │ │  │
│  │  └─────────┘  └─────────┘  └─────────┘  └─────────┘ │  │
│  └─────────────────────────────────────────────────────┘  │
│                                                           │
│  ┌─────────────────────────────────────────────────────┐  │
│  │                  GENERIC SUBDOMAIN                  │  │
│  │        ┌─────────┐  ┌─────────┐  ┌─────────┐        │  │
│  │        │  Auth   │  │Payments │  │  Email  │        │  │
│  │        │ (Auth0) │  │(Stripe) │  │(SendGrid│        │  │
│  │        └─────────┘  └─────────┘  └─────────┘        │  │
│  └─────────────────────────────────────────────────────┘  │
└───────────────────────────────────────────────────────────┘

Strangler Fig Pattern

Паттерн для постепенной миграции монолита на микросервисы. Название происходит от фикуса-душителя, который обвивает дерево и постепенно заменяет его.

        Шаг 1                    Шаг 2                 Шаг 3
    ┌───────────┐           ┌───────────┐           ┌───────────┐
    │           │           │  Facade/  │           │  Facade/  │
    │  Монолит  │           │   Proxy   │           │   Proxy   │
    │           │           └─────┬─────┘           └─────┬─────┘
    │           │                 │                       │
    └───────────┘           ┌─────┴─────┐           ┌─────▼─────┐
                            │           │           │           │
                      ┌─────▼─────┬─────▼─────┐     │           │
                      │  Монолит  │  Новый    │     │  Сервис   │
                      │    (-)    │  Сервис   │     │  A  B  C  │
                      └───────────┴───────────┘     └───────────┘
                                                   (монолит удалён)

Процесс миграции:

1. Определить функциональность для извлечения
2. Создать фасад/прокси перед монолитом
3. Реализовать новый сервис
4. Переключить трафик на новый сервис
5. Удалить старый код из монолита
6. Повторить для следующей функциональности

Anti-Corruption Layer (ACL)

ACL защищает новые сервисы от "заражения" моделями данных legacy-системы.

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Новый сервис  │────▶│       ACL       │────▶│  Legacy System  │
│                 │     │                 │     │                 │
│  Чистая модель  │     │  Трансляция     │     │  Устаревшая     │
│  домена         │     │  между моделями │     │  модель данных  │
└─────────────────┘     └─────────────────┘     └─────────────────┘

Database per Service

Каждый сервис владеет своими данными и не имеет прямого доступа к данным других сервисов.

Варианты реализации:

• Отдельные базы данных
• Отдельные схемы в одной БД
• Отдельные таблицы с чётким ownership

# Правильно: получение данных через API
class OrderService:
    def create_order(self, user_id: str, product_ids: List[str]):
        # Получаем данные о пользователе через User Service API
        user = self.user_client.get_user(user_id)

        # Получаем данные о продуктах через Product Service API
        products = self.product_client.get_products(product_ids)

        # Создаём заказ в своей базе
        order = Order(user_id=user_id, items=products)
        self.order_repository.save(order)

# Неправильно: прямой доступ к чужой базе
class OrderService:
    def create_order(self, user_id: str):
        # Антипаттерн: прямой запрос в базу другого сервиса
        user = self.db.execute("SELECT * FROM users WHERE id = ?", user_id)

---

11.3. Service Discovery

В динамической среде микросервисов экземпляры сервисов постоянно создаются и уничтожаются. Service Discovery решает задачу: "Как клиент узнаёт адрес нужного сервиса?"

Client-Side Discovery

Клиент самостоятельно запрашивает Service Registry и выбирает экземпляр.

┌────────────┐         ┌──────────────────┐
│   Client   │◀───────▶│ Service Registry │
└─────┬──────┘         │   (Consul/etcd)  │
      │                └──────────────────┘
      │                        ▲
      ▼                        │ регистрация
┌────────────┐         ┌───────┴────────┐
│  Service   │         │    Service     │
│ Instance A │         │  Instance B    │
└────────────┘         └────────────────┘

Преимущества:

• Клиент может реализовать умную балансировку
• Меньше hop'ов (нет дополнительного прокси)

Недостатки:

• Логика discovery размазана по клиентам
• Каждый язык требует своей реализации

Server-Side Discovery

Клиент обращается к load balancer или роутеру, который знает о доступных экземплярах.

┌────────────┐         ┌──────────────────┐
│   Client   │────────▶│   Load Balancer  │
└────────────┘         │    / Router      │
                       └────────┬─────────┘
                                │
              ┌─────────────────┼─────────────────┐
              ▼                 ▼                 ▼
       ┌────────────┐    ┌────────────┐    ┌────────────┐
       │  Service   │    │  Service   │    │  Service   │
       │ Instance A │    │ Instance B │    │ Instance C │
       └────────────┘    └────────────┘    └────────────┘

Преимущества:

• Клиенты остаются простыми
• Централизованное управление

Недостатки:

• Дополнительный компонент в системе
• Дополнительная латентность

Consul

HashiCorp Consul — полнофункциональное решение для service discovery с дополнительными возможностями health checking, KV storage и service mesh.

┌─────────────────────────────────────────────────────┐
│                    CONSUL CLUSTER                   │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  │
│  │   Server    │  │   Server    │  │   Server    │  │
│  │   (Leader)  │◀▶│  (Follower) │◀▶│  (Follower) │  │
│  └─────────────┘  └─────────────┘  └─────────────┘  │
└─────────────────────────────────────────────────────┘
       ▲                   ▲                   ▲
       │                   │                   │
       ▼                   ▼                   ▼
┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ Consul Agent│     │ Consul Agent│     │ Consul Agent│
│ + Service A │     │ + Service B │     │ + Service C │
└─────────────┘     └─────────────┘     └─────────────┘

Регистрация сервиса в Consul:

{
  "service": {
    "name": "order-service",
    "port": 8080,
    "tags": ["v1", "production"],
    "check": {
      "http": "http://localhost:8080/health",
      "interval": "10s",
      "timeout": "2s"
    }
  }
}

DNS-интерфейс Consul:

# Запрос адресов сервиса
dig @127.0.0.1 -p 8600 order-service.service.consul

# Результат
order-service.service.consul. 0 IN A 10.0.0.5
order-service.service.consul. 0 IN A 10.0.0.6

etcd

etcd — распределённое key-value хранилище, часто используемое для service discovery.

// Регистрация сервиса в etcd
func registerService(client *clientv3.Client, serviceName, addr string) error {
    // Создаём lease (аренду) на 10 секунд
    lease, err := client.Grant(context.Background(), 10)
    if err != nil {
        return err
    }

    // Регистрируем сервис с привязкой к lease
    key := fmt.Sprintf("/services/%s/%s", serviceName, addr)
    _, err = client.Put(context.Background(), key, addr,
        clientv3.WithLease(lease.ID))
    if err != nil {
        return err
    }

    // Поддерживаем lease alive
    ch, err := client.KeepAlive(context.Background(), lease.ID)
    if err != nil {
        return err
    }

    go func() {
        for range ch {
            // lease продлевается автоматически
        }
    }()

    return nil
}

Kubernetes DNS

В Kubernetes service discovery встроен на уровне платформы.

# Service definition
apiVersion: v1
kind: Service
metadata:
  name: order-service
  namespace: production
spec:
  selector:
    app: order
  ports:
    - port: 80
      targetPort: 8080

DNS-имена в Kubernetes:

# Внутри того же namespace
order-service

# Полное имя (FQDN)
order-service.production.svc.cluster.local

# Headless service (получить все pod IP)
order-service.production.svc.cluster.local → 10.0.0.5, 10.0.0.6, 10.0.0.7

Пример использования:

import requests

# Простой вызов сервиса по DNS имени
response = requests.get("http://order-service/api/orders/123")

---

11.4. Saga Pattern для распределённых транзакций

В монолите с единой базой данных транзакции реализуются через ACID-гарантии СУБД. В микросервисах, где каждый сервис имеет свою базу, это невозможно. Saga — паттерн для реализации распределённых транзакций через последовательность локальных транзакций с компенсирующими действиями.

Проблема

Рассмотрим процесс создания заказа:

1. Order Service создаёт заказ
2. Payment Service списывает деньги
3. Inventory Service резервирует товар
4. Shipping Service создаёт доставку

Что если Inventory Service не может зарезервировать товар после того, как Payment Service уже списал деньги?

Choreography-Based Saga

Сервисы координируются через события без центрального оркестратора.

┌─────────────────────────────────────────────────────────────────────┐
│                         CHOREOGRAPHY SAGA                           │
│                                                                     │
│    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐     │
│    │  Order   │───▶│ Payment  │───▶│Inventory │───▶│ Shipping │     │
│    │ Service  │    │ Service  │    │ Service  │    │ Service  │     │
│    └──────────┘    └──────────┘    └──────────┘    └──────────┘     │
│         │               │               │               │           │
│         ▼               ▼               ▼               ▼           │
│   ┌─────────────────────────────────────────────────────────────┐   │
│   │                    MESSAGE BROKER (Kafka)                   │   │
│   │  OrderCreated → PaymentCompleted → InventoryReserved → ...  │   │
│   └─────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

# Order Service
class OrderService:
    def create_order(self, order_data):
        order = Order.create(order_data)
        order.status = "PENDING"
        self.repository.save(order)

        # Публикуем событие
        self.event_bus.publish(OrderCreatedEvent(
            order_id=order.id,
            user_id=order.user_id,
            amount=order.total_amount
        ))

# Payment Service подписан на OrderCreated
class PaymentEventHandler:
    def handle_order_created(self, event: OrderCreatedEvent):
        try:
            payment = self.payment_service.process_payment(
                event.user_id,
                event.amount
            )
            self.event_bus.publish(PaymentCompletedEvent(
                order_id=event.order_id,
                payment_id=payment.id
            ))
        except PaymentFailedException:
            self.event_bus.publish(PaymentFailedEvent(
                order_id=event.order_id,
                reason="Insufficient funds"
            ))

# Order Service подписан на PaymentFailed для компенсации
class OrderCompensationHandler:
    def handle_payment_failed(self, event: PaymentFailedEvent):
        order = self.repository.get(event.order_id)
        order.status = "CANCELLED"
        self.repository.save(order)

Преимущества:

• Слабая связанность между сервисами
• Нет единой точки отказа

Недостатки:

• Сложно понять общий flow, читая код одного сервиса
• Риск циклических зависимостей
• Сложность отладки

Orchestration-Based Saga

Центральный оркестратор управляет последовательностью шагов и компенсациями.

┌────────────────────────────────────────────────────────┐
│                   ORCHESTRATION SAGA                   │
│                  ┌──────────────────┐                  │
│                  │   Order Saga     │                  │
│                  │   Orchestrator   │                  │
│                  └─────────┬────────┘                  │
│                            │                           │
│         ┌──────────────────┼──────────────────┐        │
│         ▼                  ▼                  ▼        │
│  ┌──────────┐        ┌──────────┐        ┌──────────┐  │
│  │ Payment  │        │Inventory │        │ Shipping │  │
│  │ Service  │        │ Service  │        │ Service  │  │
│  └──────────┘        └──────────┘        └──────────┘  │
└────────────────────────────────────────────────────────┘

class OrderSagaOrchestrator:
    def __init__(self):
        self.steps = [
            SagaStep(
                name="reserve_inventory",
                action=self.reserve_inventory,
                compensation=self.release_inventory
            ),
            SagaStep(
                name="process_payment",
                action=self.process_payment,
                compensation=self.refund_payment
            ),
            SagaStep(
                name="create_shipment",
                action=self.create_shipment,
                compensation=self.cancel_shipment
            )
        ]

    def execute(self, order_data: OrderData) -> SagaResult:
        context = SagaContext(order_data=order_data)
        completed_steps = []

        for step in self.steps:
            try:
                result = step.action(context)
                context.add_result(step.name, result)
                completed_steps.append(step)
            except Exception as e:
                # Откатываем выполненные шаги в обратном порядке
                for completed_step in reversed(completed_steps):
                    try:
                        completed_step.compensation(context)
                    except Exception as comp_error:
                        # Логируем ошибку компенсации
                        self.log_compensation_failure(
                            completed_step, comp_error
                        )

                return SagaResult.failure(str(e))

        return SagaResult.success(context)

    def reserve_inventory(self, ctx: SagaContext):
        response = self.inventory_client.reserve(
            ctx.order_data.product_id,
            ctx.order_data.quantity
        )
        return {"reservation_id": response.reservation_id}

    def release_inventory(self, ctx: SagaContext):
        reservation_id = ctx.get_result("reserve_inventory")["reservation_id"]
        self.inventory_client.release(reservation_id)

    def process_payment(self, ctx: SagaContext):
        response = self.payment_client.charge(
            ctx.order_data.user_id,
            ctx.order_data.amount
        )
        return {"payment_id": response.payment_id}

    def refund_payment(self, ctx: SagaContext):
        payment_id = ctx.get_result("process_payment")["payment_id"]
        self.payment_client.refund(payment_id)

Преимущества:

• Логика saga в одном месте
• Легче понять и отладить
• Проще управлять сложными workflow

Недостатки:

• Оркестратор — потенциальная точка отказа
• Более сильная связанность

State Machine подход

Каждая saga имеет состояние, которое сохраняется в базе данных:

┌──────────────────────────────────────────────────────────────┐
│                   ORDER SAGA STATE MACHINE                   │
│                                                              │
│  ┌─────────┐     ┌─────────┐     ┌─────────┐     ┌────────┐  │
│  │ PENDING │────▶│RESERVING│────▶│ PAYING  │────▶│SHIPPING│  │
│  └────┬────┘     └────┬────┘     └────┬────┘     └───┬────┘  │
│       │               │               │              │       │
│       ▼               ▼               ▼              ▼       │
│  ┌─────────┐     ┌─────────┐     ┌─────────┐    ┌────────┐   │
│  │CANCELLED│◀────│RELEASING│◀────│REFUNDING│◀───│ FAILED │   │
│  └─────────┘     └─────────┘     └─────────┘    └────────┘   │
│                                       │                      │
│                                       ▼                      │
│                                  ┌─────────┐                 │
│                                  │COMPLETED│                 │
│                                  └─────────┘                 │
└──────────────────────────────────────────────────────────────┘

CREATE TABLE order_saga (
    id UUID PRIMARY KEY,
    order_id UUID NOT NULL,
    state VARCHAR(50) NOT NULL,
    context JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW(),
    version INT DEFAULT 0  -- optimistic locking
);

---

5. Circuit Breaker, Bulkhead, Retry Patterns

Эти паттерны обеспечивают устойчивость микросервисной системы к частичным отказам.

Circuit Breaker

Circuit Breaker предотвращает каскадные отказы, прекращая вызовы к неработающему сервису.

┌────────────────────────────────────────────────────────────────┐
│                    CIRCUIT BREAKER STATES                      │
│                                                                │
│  ┌──────────┐      failures >= threshold      ┌──────────┐     │
│  │  CLOSED  │ ──────────────────────────────▶ │   OPEN   │     │
│  │(normal)  │                                 │(failing) │     │
│  └────┬─────┘                                 └────┬─────┘     │
│       ▲                                            │           │
│       │                                            │ timeout   │
│       │ success                                    ▼           │
│       │                                      ┌───────────┐     │
│       └────────────────────────────────────  │ HALF-OPEN │     │
│                                              │  (probe)  │     │
│                                              └───────────┘     │
└────────────────────────────────────────────────────────────────┘

Состояния:

• CLOSED — нормальная работа, все запросы проходят
• OPEN — сервис недоступен, запросы немедленно отклоняются
• HALF-OPEN — пробный режим, пропускается ограниченное число запросов

from enum import Enum
from threading import Lock
import time

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 30,
        half_open_requests: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_requests = half_open_requests

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.lock = Lock()

    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                else:
                    raise CircuitOpenException(
                        "Circuit is OPEN, request rejected"
                    )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        with self.lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.half_open_requests:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
            else:
                self.failure_count = 0

    def _on_failure(self):
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
            elif self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

    def _should_attempt_reset(self) -> bool:
        return (
            time.time() - self.last_failure_time
            >= self.recovery_timeout
        )

Использование:

payment_circuit = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=30
)

def process_payment(order):
    return payment_circuit.call(
        payment_service.charge,
        order.user_id,
        order.amount
    )

Bulkhead Pattern

Bulkhead (переборка) изолирует компоненты друг от друга, предотвращая распространение отказа.

           ┌────────────────────────────────────────────────┐
           │               BULKHEAD PATTERN                 │
           │                                                │
           │  ┌──────────────────┐    ┌──────────────────┐  │
           │  │  Thread Pool A   │    │  Thread Pool B   │  │
           │  │  (Payment API)   │    │  (Inventory API) │  │
           │  │  ┌──┐ ┌──┐ ┌──┐  │    │  ┌──┐ ┌──┐ ┌──┐  │  │
           │  │  │T1│ │T2│ │T3│  │    │  │T1│ │T2│ │T3│  │  │
           │  │  └──┘ └──┘ └──┘  │    │  └──┘ └──┘ └──┘  │  │
           │  │  Max: 10 threads │    │  Max: 5 threads  │  │
           │  └──────────────────┘    └──────────────────┘  │
           │           │                       │            │
           │           ▼                       ▼            │
           │  ┌──────────────────┐    ┌──────────────────┐  │
           │  │  Payment Service │    │ Inventory Service│  │
           │  │    (slow/down)   │    │    (healthy)     │  │
           │  └──────────────────┘    └──────────────────┘  │
           │                                                │
           │      Отказ Payment не влияет на Inventory!     │
           └────────────────────────────────────────────────┘

Реализация через Semaphore:

from threading import Semaphore
from concurrent.futures import ThreadPoolExecutor

class BulkheadedClient:
    def __init__(self, service_name: str, max_concurrent: int):
        self.service_name = service_name
        self.semaphore = Semaphore(max_concurrent)
        self.executor = ThreadPoolExecutor(max_workers=max_concurrent)

    def call(self, func, *args, timeout: float = 5.0, **kwargs):
        acquired = self.semaphore.acquire(timeout=timeout)
        if not acquired:
            raise BulkheadFullException(
                f"Bulkhead for {self.service_name} is full"
            )

        try:
            future = self.executor.submit(func, *args, **kwargs)
            return future.result(timeout=timeout)
        finally:
            self.semaphore.release()

# Использование
payment_client = BulkheadedClient("payment", max_concurrent=10)
inventory_client = BulkheadedClient("inventory", max_concurrent=5)

# Если payment зависнет, inventory продолжит работать
payment_result = payment_client.call(payment_api.charge, amount=100)
inventory_result = inventory_client.call(inventory_api.reserve, item_id=123)

Retry Pattern

Автоматический повтор неудачных операций с экспоненциальной задержкой.

import time
import random
from typing import Callable, TypeVar

T = TypeVar('T')

def retry_with_exponential_backoff(
    func: Callable[[], T],
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True,
    retryable_exceptions: tuple = (Exception,)
) -> T:
    """
    Выполняет функцию с повторами и экспоненциальной задержкой.
    """
    last_exception = None

    for attempt in range(max_retries + 1):
        try:
            return func()
        except retryable_exceptions as e:
            last_exception = e

            if attempt == max_retries:
                break

            # Вычисляем задержку
            delay = min(
                base_delay * (exponential_base ** attempt),
                max_delay
            )

            # Добавляем jitter для предотвращения thundering herd
            if jitter:
                delay = delay * (0.5 + random.random())

            time.sleep(delay)

    raise last_exception

# Использование
def get_user_data(user_id: str):
    return retry_with_exponential_backoff(
        lambda: user_service.get(user_id),
        max_retries=3,
        base_delay=0.5,
        retryable_exceptions=(ConnectionError, TimeoutError)
    )

Важно: не все ошибки следует ретраить:

• Retryable: таймауты, временные сетевые ошибки (502, 503, 504)
• Non-retryable: ошибки валидации (400), аутентификации (401, 403), not found (404)

Комбинирование паттернов

class ResilientServiceClient:
    def __init__(self, service_name: str):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=30
        )
        self.bulkhead = BulkheadedClient(service_name, max_concurrent=10)

    def call(self, func, *args, **kwargs):
        # Внешний слой: Bulkhead (ограничивает параллелизм)
        # Средний слой: Circuit Breaker (защищает от каскадных отказов)
        # Внутренний слой: Retry (повторяет временные ошибки)

        def with_retry():
            return retry_with_exponential_backoff(
                lambda: func(*args, **kwargs),
                max_retries=3
            )

        def with_circuit_breaker():
            return self.circuit_breaker.call(with_retry)

        return self.bulkhead.call(with_circuit_breaker)

---

6. Sidecar и Service Mesh

Sidecar Pattern

Sidecar — это отдельный контейнер, работающий рядом с основным приложением и предоставляющий вспомогательную функциональность.

        ┌────────────────────────────────────────────────────┐
        │                        POD                         │
        │  ┌───────────────────┐      ┌───────────────────┐  │
        │  │    Application    │◀────▶│      Sidecar      │  │
        │  │    Container      │      │     Container     │  │
        │  │                   │      │                   │  │
        │  │  - Business logic │      │  - Logging        │  │
        │  │                   │      │  - Monitoring     │  │
        │  │                   │      │  - Proxy          │  │
        │  │                   │      │  - TLS termination│  │
        │  └───────────────────┘      └───────────────────┘  │
        │      localhost:8080            localhost:15001     │
        └────────────────────────────────────────────────────┘

Типичные использования sidecar:

• Envoy proxy для traffic management
• Log collectors (Fluentd, Filebeat)
• Configuration agents (Consul agent)
• Secrets injectors (Vault agent)

Service Mesh

Service Mesh — это инфраструктурный слой, обеспечивающий безопасную, надёжную и наблюдаемую коммуникацию между сервисами.

    ┌──────────────────────────────────────────────────────────────────┐
    │                          SERVICE MESH                            │
    │                                                                  │
    │  ┌────────────────────────────────────────────────────────────┐  │
    │  │                       CONTROL PLANE                        │  │
    │  │  ┌─────────┐  ┌─────────────┐  ┌───────────┐  ┌─────────┐  │  │
    │  │  │  Pilot  │  │   Citadel   │  │   Mixer   │  │  Galley │  │  │
    │  │  │(config) │  │(security)   │  │(telemetry)│  │(config) │  │  │
    │  │  └─────────┘  └─────────────┘  └───────────┘  └─────────┘  │  │
    │  └────────────────────────────────────────────────────────────┘  │
    │                                │                                 │
    │                                ▼ конфигурация                    │
    │  ┌────────────────────────────────────────────────────────────┐  │
    │  │                         DATA PLANE                         │  │
    │  │                                                            │  │
    │  │      ┌──────────────────┐        ┌──────────────────┐      │  │
    │  │      │       POD A      │        │      POD B       │      │  │
    │  │      │ ┌──────┐ ┌─────┐ │  mTLS  │ ┌─────┐ ┌──────┐ │      │  │
    │  │      │ │App A │ │Envoy│◀┼────────┼▶│Envoy│ │App B │ │      │  │
    │  │      │ └──────┘ └─────┘ │        │ └─────┘ └──────┘ │      │  │
    │  │      └──────────────────┘        └──────────────────┘      │  │
    │  │                                                            │  │
    │  └────────────────────────────────────────────────────────────┘  │
    └──────────────────────────────────────────────────────────────────┘

Возможности Service Mesh

Traffic Management:

• Load balancing
• Traffic splitting (canary deployments)
• Circuit breaker
• Retries и timeouts
• Fault injection (для тестирования)

Security:

• mTLS между сервисами
• Authorization policies
• Certificate rotation

Observability:

• Distributed tracing
• Metrics collection
• Access logging

Istio

Istio — наиболее популярный service mesh для Kubernetes.

Traffic routing (VirtualService):

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - match:
        - headers:
            x-canary:
              exact: "true"
      route:
        - destination:
            host: order-service
            subset: v2
    - route:
        - destination:
            host: order-service
            subset: v1
          weight: 90
        - destination:
            host: order-service
            subset: v2
          weight: 10

Canary Deployment:

# Постепенное увеличение трафика на новую версию
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: payment-service
spec:
  hosts:
    - payment-service
  http:
    - route:
        - destination:
            host: payment-service
            subset: v1
          weight: 80
        - destination:
            host: payment-service
            subset: v2
          weight: 20

Circuit Breaker (DestinationRule):

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: payment-service
spec:
  host: payment-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50

mTLS Policy:

apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: production
spec:
  mtls:
    mode: STRICT  # Все коммуникации требуют mTLS

Linkerd

Linkerd — легковесная альтернатива Istio, фокусирующаяся на простоте.

# Добавление в mesh через аннотацию
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  annotations:
    linkerd.io/inject: enabled
spec:
  # ...

ServiceProfile для retry и timeout:

apiVersion: linkerd.io/v1alpha2
kind: ServiceProfile
metadata:
  name: payment-service.production.svc.cluster.local
spec:
  routes:
    - name: POST /charge
      condition:
        method: POST
        pathRegex: /charge
      responseClasses:
        - condition:
            status:
              min: 500
              max: 599
          isFailure: true
      timeout: 5s
      retryBudget:
        retryRatio: 0.2
        minRetriesPerSecond: 10
        ttl: 10s

Сравнение Istio и Linkerd

Когда использовать Service Mesh

Используйте Service Mesh, когда:

• У вас десятки или сотни микросервисов
• Нужна строгая безопасность (mTLS везде)
• Требуется глубокая observability
• Нужны продвинутые traffic management фичи

Не используйте Service Mesh, когда:

• У вас меньше 10 сервисов
• Команда не готова к операционной сложности
• Латентность критична (mesh добавляет ~1-3ms)

---

7. API Gateway

API Gateway — единая точка входа для всех клиентов микросервисной системы. Он принимает входящие запросы, маршрутизирует их к нужным сервисам, агрегирует ответы и реализует сквозную функциональность.

Зачем нужен API Gateway

Без API Gateway каждый клиент (мобильное приложение, веб-интерфейс, партнёрская интеграция) должен знать адреса всех сервисов и самостоятельно реализовывать аутентификацию, rate limiting, SSL-терминацию. Это приводит к дублированию логики и хрупкой связанности.

      БЕЗ API GATEWAY                        С API GATEWAY

 ┌──────────┐ ┌──────────┐            ┌──────────┐  ┌──────────┐
 │   Web    │ │  Mobile  │            │   Web    │  │  Mobile  │
 │  Client  │ │  Client  │            │  Client  │  │  Client  │
 └──┬──┬──┬─┘ └─┬──┬──┬──┘            └────┬─────┘  └────┬─────┘
    │  │  │     │  │  │                    │             │
    │  │  │     │  │  │                    ▼             ▼
    │  │  │     │  │  │               ┌───────────────────────┐
    │  │  │     │  │  │               │      API Gateway      │
    │  │  │     │  │  │               │  - Auth               │
    │  │  │     │  │  │               │  - Rate Limiting      │
    │  │  │     │  │  │               │  - SSL Termination    │
    │  │  │     │  │  │               │  - Request Routing    │
    ▼  ▼  ▼     ▼  ▼  ▼               └────┬──────┬──────┬────┘
  ┌─────┐ ┌─────┐ ┌─────┐                  │      │      │
  │Svc A│ │Svc B│ │Svc C│                  ▼      ▼      ▼
  └─────┘ └─────┘ └─────┘               ┌─────┐┌─────┐┌─────┐
                                        │Svc A││Svc B││Svc C│
                                        └─────┘└─────┘└─────┘ 

Основные функции API Gateway

Backend for Frontend (BFF)

Паттерн BFF предлагает создавать отдельный API Gateway для каждого типа клиента. Это позволяет оптимизировать API под нужды конкретного фронтенда.

        ┌──────────┐   ┌──────────┐   ┌──────────┐
        │   Web    │   │  Mobile  │   │  Partner │
        │  Client  │   │  Client  │   │   API    │
        └────┬─────┘   └────┬─────┘   └────┬─────┘
             │              │              │
             ▼              ▼              ▼
        ┌─────────┐   ┌──────────┐   ┌─────────────┐
        │ Web BFF │   │  Mobile  │   │ Partner     │
        │         │   │    BFF   │   │ API Gateway │
        └────┬────┘   └─────┬────┘   └──────┬──────┘
             │              │               │
             └──────────────┼───────────────┘
                            │
            ┌───────────────┼───────────────┐
            ▼               ▼               ▼
        ┌────────┐    ┌──────────┐    ┌──────────┐
        │ Users  │    │  Orders  │    │ Products │
        │Service │    │ Service  │    │ Service  │
        └────────┘    └──────────┘    └──────────┘

Почему BFF лучше единого Gateway:

• Web-клиенту нужен полный набор полей — Mobile-клиенту облегчённый
• Разные стратегии кэширования и pagination
• Разные требования к авторизации (OAuth для web, API key для партнёров)
• Независимые релизные циклы для каждого BFF

Пример: один и тот же эндпоинт, разные BFF:

# Web BFF — возвращает полные данные
class WebOrderBFF:
    def get_order(self, order_id: str):
        order = self.order_service.get(order_id)
        user = self.user_service.get(order.user_id)
        products = self.product_service.get_many(order.product_ids)
        reviews = self.review_service.get_for_products(order.product_ids)

        return {
            "order": order,
            "user": user,
            "products": products,
            "reviews": reviews,  # Полные отзывы для web
            "recommendations": self.recommendation_service.get(order.user_id)
        }

# Mobile BFF — возвращает облегчённые данные
class MobileOrderBFF:
    def get_order(self, order_id: str):
        order = self.order_service.get(order_id)
        products = self.product_service.get_summary(order.product_ids)

        return {
            "order_id": order.id,
            "status": order.status,
            "total": order.total,
            "items": [{"name": p.name, "image_url": p.thumbnail} for p in products]
            # Без отзывов и рекомендаций — экономим трафик
        }

Популярные решения

Kong:

# Kong Declarative Configuration
_format_version: "3.0"

services:
  - name: order-service
    url: http://order-service:8080
    routes:
      - name: order-route
        paths:
          - /api/v1/orders
        strip_path: true
    plugins:
      - name: rate-limiting
        config:
          minute: 100
          policy: redis
          redis_host: redis
      - name: jwt
        config:
          key_claim_name: kid
      - name: correlation-id
        config:
          header_name: X-Request-ID
          generator: uuid

NGINX как API Gateway:

upstream order_service {
    server order-service-1:8080 weight=3;
    server order-service-2:8080 weight=2;
    server order-service-3:8080 backup;
}

server {
    listen 443 ssl;
    server_name api.example.com;

    # Rate limiting
    limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;

    location /api/v1/orders {
        limit_req zone=api burst=20 nodelay;

        # JWT validation
        auth_jwt "API";
        auth_jwt_key_file /etc/nginx/jwt_key.pem;

        proxy_pass http://order_service;
        proxy_set_header X-Request-ID $request_id;
        proxy_set_header X-Real-IP $remote_addr;

        # Circuit breaker через passive health checks
        proxy_next_upstream error timeout http_502 http_503;
        proxy_next_upstream_tries 2;
        proxy_connect_timeout 5s;
        proxy_read_timeout 10s;
    }

    location /api/v1/products {
        proxy_pass http://product_service;
        proxy_cache api_cache;
        proxy_cache_valid 200 5m;
        proxy_cache_key $request_uri;
    }
}

API Gateway Anti-Patterns

---

11.8. Inter-Service Communication

Выбор способа коммуникации между сервисами — одно из ключевых архитектурных решений, влияющих на надёжность, производительность и связанность системы.

Синхронная vs Асинхронная коммуникация

        СИНХРОННАЯ                                 АСИНХРОННАЯ

   Client ──────▶ Server             Producer ──────▶ Broker ──────▶ Consumer
         ◀──────                              (fire & forget)
       (ждёт ответ)                                 или
                                    Producer ──────▶ Broker ──────▶ Consumer
                                                            ──────▶ Consumer
                                                            ──────▶ Consumer
                                                           (fan-out)

REST (HTTP/JSON)

Наиболее распространённый способ синхронной коммуникации.

# Клиент (Order Service вызывает Product Service)
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ProductServiceClient:
    def __init__(self, base_url: str):
        self.client = httpx.Client(
            base_url=base_url,
            timeout=httpx.Timeout(5.0, connect=2.0),
            headers={"Content-Type": "application/json"}
        )

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=0.5, max=10)
    )
    def get_product(self, product_id: str) -> dict:
        response = self.client.get(f"/api/v1/products/{product_id}")
        response.raise_for_status()
        return response.json()

    def get_products_batch(self, product_ids: list[str]) -> list[dict]:
        """Batch-запрос для уменьшения числа сетевых вызовов."""
        response = self.client.post(
            "/api/v1/products/batch",
            json={"ids": product_ids}
        )
        response.raise_for_status()
        return response.json()["products"]

Преимущества REST:

• Универсальность, поддержка любого языка
• Простая отладка (curl, Postman)
• Кэширование через HTTP-заголовки
• Широкая экосистема инструментов

Недостатки REST:

• Текстовый JSON — неэффективная сериализация
• Нет строгого контракта (без OpenAPI)
• Over-fetching / Under-fetching (решается GraphQL)
• Overhead HTTP/1.1 заголовков

gRPC

Высокопроизводительный RPC-фреймворк от Google на основе Protocol Buffers и HTTP/2.

// product_service.proto
syntax = "proto3";

package ecommerce;

service ProductService {
    // Унарный вызов
    rpc GetProduct(GetProductRequest) returns (Product);

    // Server streaming — получить список продуктов потоком
    rpc ListProducts(ListProductsRequest) returns (stream Product);

    // Client streaming — batch-загрузка продуктов
    rpc UploadProducts(stream Product) returns (UploadResponse);

    // Bidirectional streaming — real-time обновления цен
    rpc PriceUpdates(stream PriceQuery) returns (stream PriceUpdate);
}

message GetProductRequest {
    string product_id = 1;
}

message Product {
    string id = 1;
    string name = 2;
    string description = 3;
    int64 price_cents = 4;           // Цена в центах для точности
    int32 stock_quantity = 5;
    repeated string category_ids = 6;
    ProductStatus status = 7;
    google.protobuf.Timestamp created_at = 8;
}

enum ProductStatus {
    PRODUCT_STATUS_UNSPECIFIED = 0;
    PRODUCT_STATUS_ACTIVE = 1;
    PRODUCT_STATUS_INACTIVE = 2;
    PRODUCT_STATUS_OUT_OF_STOCK = 3;
}

message ListProductsRequest {
    int32 page_size = 1;
    string page_token = 2;
    string category_id = 3;
}

message UploadResponse {
    int32 uploaded_count = 1;
    repeated string failed_ids = 2;
}

message PriceQuery {
    string product_id = 1;
}

message PriceUpdate {
    string product_id = 1;
    int64 old_price_cents = 2;
    int64 new_price_cents = 3;
    google.protobuf.Timestamp updated_at = 4;
}

Python gRPC-сервер:

import grpc
from concurrent import futures
import product_pb2
import product_pb2_grpc

class ProductServicer(product_pb2_grpc.ProductServiceServicer):
    def GetProduct(self, request, context):
        product = self.repository.get(request.product_id)
        if not product:
            context.abort(grpc.StatusCode.NOT_FOUND, "Product not found")

        return product_pb2.Product(
            id=product.id,
            name=product.name,
            price_cents=product.price_cents,
            stock_quantity=product.stock
        )

    def ListProducts(self, request, context):
        """Server streaming — отправляем продукты по одному."""
        products = self.repository.list(
            category_id=request.category_id,
            page_size=request.page_size
        )
        for product in products:
            yield product_pb2.Product(
                id=product.id,
                name=product.name,
                price_cents=product.price_cents
            )

def serve():
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ('grpc.max_send_message_length', 50 * 1024 * 1024),
            ('grpc.max_receive_message_length', 50 * 1024 * 1024),
            ('grpc.keepalive_time_ms', 10000),
        ]
    )
    product_pb2_grpc.add_ProductServiceServicer_to_server(
        ProductServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

gRPC-клиент:

import grpc
import product_pb2
import product_pb2_grpc

class ProductGrpcClient:
    def __init__(self, target: str):
        # Connection pooling через channel
        self.channel = grpc.insecure_channel(
            target,
            options=[
                ('grpc.lb_policy_name', 'round_robin'),
                ('grpc.enable_retries', 1),
                ('grpc.service_config', json.dumps({
                    "retryPolicy": {
                        "maxAttempts": 3,
                        "initialBackoff": "0.1s",
                        "maxBackoff": "1s",
                        "backoffMultiplier": 2,
                        "retryableStatusCodes": ["UNAVAILABLE"]
                    }
                }))
            ]
        )
        self.stub = product_pb2_grpc.ProductServiceStub(self.channel)

    def get_product(self, product_id: str) -> product_pb2.Product:
        request = product_pb2.GetProductRequest(product_id=product_id)
        return self.stub.GetProduct(
            request,
            timeout=5.0,
            metadata=[("x-request-id", str(uuid.uuid4()))]
        )

Сравнение REST vs gRPC:

Асинхронная коммуникация через Message Broker

    ┌──────────────────────────────────────────────────────────────────┐
    │                     MESSAGE BROKER PATTERNS                      │
    │                                                                  │
    │  1. Point-to-Point (Queue)                                       │
    │     Producer ──▶ [Queue] ──▶ Consumer                            │
    │     (один получатель)                                            │
    │                                                                  │
    │  2. Publish/Subscribe (Topic)                                    │
    │     Producer ──▶ [Topic] ──▶ Consumer A                          │
    │                          ──▶ Consumer B                          │
    │                          ──▶ Consumer C                          │
    │     (все подписчики получают копию)                              │
    │                                                                  │
    │  3. Consumer Group (Kafka-стиль)                                 │
    │     Producer ──▶ [Topic, partition 0] ──▶ Consumer A (Group 1)   │
    │              ──▶ [Topic, partition 1] ──▶ Consumer B (Group 1)   │
    │              ──▶ [Topic, partition 2] ──▶ Consumer C (Group 1)   │
    │     (каждое сообщение — одному consumer в группе)                │
    └──────────────────────────────────────────────────────────────────┘

Apache Kafka — пример Producer/Consumer:

# Producer: Order Service публикует события
from confluent_kafka import Producer
import json

class OrderEventProducer:
    def __init__(self, bootstrap_servers: str):
        self.producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'acks': 'all',                    # Ждать подтверждения от всех реплик
            'retries': 3,
            'enable.idempotence': True,       # Exactly-once семантика
            'max.in.flight.requests.per.connection': 5,
            'compression.type': 'snappy'
        })

    def publish_order_created(self, order: Order):
        event = {
            "event_type": "OrderCreated",
            "event_id": str(uuid.uuid4()),
            "timestamp": datetime.utcnow().isoformat(),
            "data": {
                "order_id": order.id,
                "user_id": order.user_id,
                "items": [item.to_dict() for item in order.items],
                "total_amount": order.total_amount
            }
        }
        self.producer.produce(
            topic="order-events",
            key=order.id.encode(),    # Ключ гарантирует порядок для одного заказа
            value=json.dumps(event).encode(),
            callback=self._delivery_report
        )
        self.producer.flush()

    def _delivery_report(self, err, msg):
        if err:
            logger.error(f"Delivery failed: {err}")
        else:
            logger.info(f"Delivered to {msg.topic()} [{msg.partition()}]")

# Consumer: Inventory Service слушает события
from confluent_kafka import Consumer

class InventoryEventConsumer:
    def __init__(self, bootstrap_servers: str, group_id: str):
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,    # Ручной commit после обработки
            'max.poll.interval.ms': 300000
        })

    def start(self):
        self.consumer.subscribe(['order-events'])

        while True:
            msg = self.consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                logger.error(f"Consumer error: {msg.error()}")
                continue

            try:
                event = json.loads(msg.value().decode())
                self._handle_event(event)
                # Commit только после успешной обработки
                self.consumer.commit(message=msg)
            except Exception as e:
                logger.error(f"Failed to process message: {e}")
                # Не делаем commit — сообщение будет redelivered

    def _handle_event(self, event: dict):
        if event["event_type"] == "OrderCreated":
            self.inventory_service.reserve_items(
                order_id=event["data"]["order_id"],
                items=event["data"]["items"]
            )

RabbitMQ — пример с Dead Letter Queue:

import pika

class RabbitMQSetup:
    def setup_with_dlq(self, channel):
        # Dead Letter Exchange для необработанных сообщений
        channel.exchange_declare(
            exchange='order.dlx',
            exchange_type='direct'
        )
        channel.queue_declare(
            queue='order.dead-letter',
            durable=True
        )
        channel.queue_bind(
            queue='order.dead-letter',
            exchange='order.dlx',
            routing_key='order.created'
        )

        # Основная очередь с привязкой к DLX
        channel.queue_declare(
            queue='order.created',
            durable=True,
            arguments={
                'x-dead-letter-exchange': 'order.dlx',
                'x-dead-letter-routing-key': 'order.created',
                'x-message-ttl': 60000,           # TTL 60 секунд
                'x-max-length': 100000             # Максимум 100K сообщений
            }
        )

Идемпотентность в асинхронной коммуникации

При at-least-once доставке сообщение может быть обработано повторно. Идемпотентность гарантирует, что повторная обработка не приведёт к ошибкам.

class IdempotentEventHandler:
    def __init__(self, redis_client):
        self.redis = redis_client

    def handle(self, event: dict):
        event_id = event["event_id"]

        # Проверяем, обрабатывали ли мы это событие
        if self.redis.exists(f"processed:{event_id}"):
            logger.info(f"Event {event_id} already processed, skipping")
            return

        # Атомарная проверка + установка через SET NX
        acquired = self.redis.set(
            f"processed:{event_id}",
            "1",
            nx=True,        # SET if Not eXists
            ex=86400        # TTL 24 часа
        )
        if not acquired:
            return  # Другой consumer уже обрабатывает

        try:
            self._process(event)
        except Exception:
            # Откатываем маркер, чтобы повторить обработку
            self.redis.delete(f"processed:{event_id}")
            raise

Outbox Pattern

Outbox гарантирует атомарность записи бизнес-данных и публикации событий. Без него возможна ситуация, когда данные сохранены, а событие не отправлено (или наоборот).

        ┌─────────────────────────────────────────────────────┐
        │                    OUTBOX PATTERN                   │
        │                                                     │
        │  1. Одна транзакция: сохраняем данные + событие     │
        │  в outbox таблицу                                   │ 
        │   ┌─────────────────────────────────────────────┐   │
        │   │              TRANSACTION                    │   │
        │   │  INSERT INTO orders (...)                   │   │
        │   │  INSERT INTO outbox (event_type, payload)   │   │
        │   │  COMMIT                                     │   │
        │   └─────────────────────────────────────────────┘   │
        │  2. Relay читает outbox и публикует в брокер        │
        │  ┌─────────┐     ┌──────────┐     ┌──────────────┐  │
        │  │  Outbox │────▶│  Relay   │────▶│ Message      │  │
        │  │  Table  │     │(Debezium)│     │ Broker       │  │
        │  └─────────┘     └──────────┘     └──────────────┘  │
        └─────────────────────────────────────────────────────┘

class OrderService:
    def create_order(self, order_data: OrderData):
        with self.db.transaction() as tx:
            # Сохраняем заказ
            order = Order.create(order_data)
            tx.execute(
                "INSERT INTO orders (id, user_id, total, status) VALUES (?, ?, ?, ?)",
                (order.id, order.user_id, order.total, "PENDING")
            )

            # Сохраняем событие в outbox — в той же транзакции!
            tx.execute(
                """INSERT INTO outbox
                   (id, aggregate_type, aggregate_id, event_type, payload, created_at)
                   VALUES (?, ?, ?, ?, ?, NOW())""",
                (
                    str(uuid.uuid4()),
                    "Order",
                    order.id,
                    "OrderCreated",
                    json.dumps(order.to_event_payload())
                )
            )
            # COMMIT — атомарно сохранены и данные, и событие

# Outbox Relay (отдельный процесс или CDC через Debezium)
class OutboxRelay:
    def poll_and_publish(self):
        while True:
            with self.db.transaction() as tx:
                events = tx.execute(
                    """SELECT * FROM outbox
                       WHERE published = FALSE
                       ORDER BY created_at
                       LIMIT 100
                       FOR UPDATE SKIP LOCKED"""
                )

                for event in events:
                    self.kafka_producer.produce(
                        topic=f"{event.aggregate_type.lower()}-events",
                        key=event.aggregate_id,
                        value=event.payload
                    )

                    tx.execute(
                        "UPDATE outbox SET published = TRUE WHERE id = ?",
                        (event.id,)
                    )

            time.sleep(0.5)  # Poll interval

-- Outbox таблица
CREATE TABLE outbox (
    id UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id VARCHAR(100) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    published BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT NOW(),

    INDEX idx_outbox_unpublished (published, created_at)
        WHERE published = FALSE
);

---

9. CQRS и Event Sourcing в микросервисах

CQRS (Command Query Responsibility Segregation)

CQRS разделяет модель чтения и модель записи, позволяя оптимизировать каждую независимо.

        ┌────────────────────────────────────────────────────┐
        │                        CQRS                        │
        │                  ┌──────────────┐                  │
        │                  │   API Layer  │                  │
        │                  └──────┬───────┘                  │
        │                         │                          │
        │           ┌─────────────┴─────────────┐            │
        │           │                           │            │
        │   ┌───────▼───────┐          ┌────────▼────────┐   │
        │   │   Command     │          │    Query        │   │
        │   │   Service     │          │    Service      │   │
        │   │               │          │                 │   │
        │   │  - Validation │          │ - Denormalized  │   │
        │   │  - Business   │          │ - Optimized for │   │
        │   │    rules      │          │   read patterns │   │
        │   └───────┬───────┘          └────────▲────────┘   │
        │           │                           │            │
        │    ┌──────▼──────┐    Events   ┌──────┴───────┐    │
        │    │   Write DB  │────────────▶│   Read DB    │    │
        │    │ (PostgreSQL)│             │(Elasticsearch│    │
        │    │ normalized  │             │ / Redis /    │    │
        │    └─────────────┘             │ MongoDB)     │    │
        │                                └──────────────┘    │
        └────────────────────────────────────────────────────┘

Пример реализации:

# Command side — нормализованная модель, бизнес-правила
class CreateOrderCommand:
    user_id: str
    items: list[OrderItem]
    shipping_address: Address

class OrderCommandHandler:
    def handle_create(self, cmd: CreateOrderCommand):
        # Валидация
        user = self.user_repo.get(cmd.user_id)
        if not user.is_active:
            raise BusinessRuleViolation("User account is inactive")

        for item in cmd.items:
            product = self.product_repo.get(item.product_id)
            if product.stock < item.quantity:
                raise BusinessRuleViolation(f"Insufficient stock for {product.name}")

        # Создание заказа
        order = Order.create(cmd)
        self.order_repo.save(order)

        # Публикация события для обновления read model
        self.event_bus.publish(OrderCreatedEvent(
            order_id=order.id,
            user_id=order.user_id,
            items=order.items,
            total=order.total,
            created_at=order.created_at
        ))

# Query side — денормализованная модель, оптимизированная для чтения
class OrderQueryService:
    def __init__(self, elasticsearch_client):
        self.es = elasticsearch_client

    def search_orders(self, user_id: str, status: str = None,
                      page: int = 1, size: int = 20) -> OrderListResponse:
        query = {"bool": {"must": [{"term": {"user_id": user_id}}]}}
        if status:
            query["bool"]["must"].append({"term": {"status": status}})

        result = self.es.search(
            index="orders",
            body={
                "query": query,
                "sort": [{"created_at": "desc"}],
                "from": (page - 1) * size,
                "size": size
            }
        )
        return OrderListResponse.from_es(result)

    def get_order_detail(self, order_id: str) -> OrderDetailResponse:
        """Все данные в одном документе — без JOIN'ов."""
        doc = self.es.get(index="orders", id=order_id)
        return OrderDetailResponse.from_es(doc)

# Event handler для обновления read model
class OrderReadModelUpdater:
    def handle_order_created(self, event: OrderCreatedEvent):
        # Собираем денормализованный документ
        user = self.user_client.get(event.user_id)
        products = self.product_client.get_many(
            [item.product_id for item in event.items]
        )

        doc = {
            "order_id": event.order_id,
            "user_id": event.user_id,
            "user_name": user.name,
            "user_email": user.email,
            "status": "PENDING",
            "items": [{
                "product_id": item.product_id,
                "product_name": products[item.product_id].name,
                "quantity": item.quantity,
                "price": item.price,
                "image_url": products[item.product_id].image_url
            } for item in event.items],
            "total": event.total,
            "created_at": event.created_at.isoformat()
        }

        self.es.index(index="orders", id=event.order_id, body=doc)

Event Sourcing

Вместо хранения текущего состояния — хранение всех событий, которые к нему привели. Текущее состояние восстанавливается воспроизведением событий.

    ┌───────────────────────────────────────────────────────────────────┐
    │                          EVENT SOURCING                           │
    │                                                                   │
    │                  Event Store (append-only log):                   │
    │  ┌─────────────────────────────────────────────────────────────┐  │
    │  │ Event 1: OrderCreated    {order_id: "123", items: [...]}    │  │
    │  │ Event 2: PaymentReceived {order_id: "123", amount: 100}     │  │
    │  │ Event 3: ItemShipped     {order_id: "123", tracking: "XY"}  │  │
    │  │ Event 4: OrderCompleted  {order_id: "123"}                  │  │
    │  └─────────────────────────────────────────────────────────────┘  │
    │                                 │                                 │
    │                             ▼ replay                              │
    │                         ┌───────────────┐                         │
    │                         │   Current     │                         │
    │                         │   State:      │                         │
    │                         │   COMPLETED   │                         │
    │                         └───────────────┘                         │
    └───────────────────────────────────────────────────────────────────┘

# Базовый event-sourced агрегат
class OrderAggregate:
    def __init__(self):
        self.id = None
        self.status = None
        self.items = []
        self.total = 0
        self.events = []          # Новые события текущей сессии
        self.version = 0          # Для optimistic concurrency

    # --- Команды (изменяют состояние через события) ---

    def create(self, order_id: str, user_id: str, items: list):
        if self.id is not None:
            raise InvalidOperation("Order already exists")

        self._apply(OrderCreated(
            order_id=order_id,
            user_id=user_id,
            items=items,
            total=sum(i.price * i.quantity for i in items)
        ))

    def confirm_payment(self, payment_id: str, amount: Decimal):
        if self.status != "PENDING":
            raise InvalidOperation(f"Cannot pay order in status {self.status}")
        if amount != self.total:
            raise InvalidOperation("Payment amount mismatch")

        self._apply(PaymentConfirmed(
            order_id=self.id,
            payment_id=payment_id,
            amount=amount
        ))

    def cancel(self, reason: str):
        if self.status in ("COMPLETED", "CANCELLED"):
            raise InvalidOperation(f"Cannot cancel order in status {self.status}")

        self._apply(OrderCancelled(
            order_id=self.id,
            reason=reason
        ))

    # --- Event handlers (обновляют внутреннее состояние) ---

    def _apply(self, event):
        self._handle(event)
        self.events.append(event)
        self.version += 1

    def _handle(self, event):
        handler = getattr(self, f"_on_{type(event).__name__}", None)
        if handler:
            handler(event)

    def _on_OrderCreated(self, event: OrderCreated):
        self.id = event.order_id
        self.status = "PENDING"
        self.items = event.items
        self.total = event.total

    def _on_PaymentConfirmed(self, event: PaymentConfirmed):
        self.status = "PAID"

    def _on_OrderCancelled(self, event: OrderCancelled):
        self.status = "CANCELLED"

    # --- Восстановление из истории ---

    @classmethod
    def from_events(cls, events: list) -> "OrderAggregate":
        aggregate = cls()
        for event in events:
            aggregate._handle(event)
            aggregate.version += 1
        return aggregate

# Event Store
class EventStore:
    def save(self, aggregate_id: str, events: list, expected_version: int):
        """Сохраняет события с optimistic concurrency check."""
        with self.db.transaction() as tx:
            current_version = tx.execute(
                "SELECT MAX(version) FROM events WHERE aggregate_id = ?",
                (aggregate_id,)
            ).scalar() or 0

            if current_version != expected_version:
                raise ConcurrencyConflict(
                    f"Expected version {expected_version}, got {current_version}"
                )

            for i, event in enumerate(events):
                tx.execute(
                    """INSERT INTO events
                       (aggregate_id, version, event_type, data, timestamp)
                       VALUES (?, ?, ?, ?, NOW())""",
                    (
                        aggregate_id,
                        expected_version + i + 1,
                        type(event).__name__,
                        json.dumps(event.to_dict())
                    )
                )

    def get_events(self, aggregate_id: str) -> list:
        rows = self.db.execute(
            "SELECT event_type, data FROM events WHERE aggregate_id = ? ORDER BY version",
            (aggregate_id,)
        )
        return [self._deserialize(row) for row in rows]

Snapshot Optimization:

При большом количестве событий восстановление агрегата замедляется. Снепшоты решают эту проблему:

class SnapshotStore:
    def save_snapshot(self, aggregate_id: str, state: dict, version: int):
        self.db.execute(
            """INSERT INTO snapshots (aggregate_id, state, version, created_at)
               VALUES (?, ?, ?, NOW())
               ON CONFLICT (aggregate_id)
               DO UPDATE SET state = ?, version = ?, created_at = NOW()""",
            (aggregate_id, json.dumps(state), version,
             json.dumps(state), version)
        )

    def load_aggregate(self, aggregate_id: str) -> OrderAggregate:
        # 1. Загружаем snapshot (если есть)
        snapshot = self.snapshot_store.get(aggregate_id)
        if snapshot:
            aggregate = OrderAggregate.from_snapshot(snapshot.state)
            start_version = snapshot.version
        else:
            aggregate = OrderAggregate()
            start_version = 0

        # 2. Применяем события после snapshot
        events = self.event_store.get_events_after(
            aggregate_id, start_version
        )
        for event in events:
            aggregate._handle(event)
            aggregate.version += 1

        # 3. Создаём snapshot если накопилось много событий
        if aggregate.version - start_version > 100:
            self.snapshot_store.save_snapshot(
                aggregate_id, aggregate.to_snapshot(), aggregate.version
            )

        return aggregate

Когда использовать Event Sourcing:

---

11.10. Observability: Трейсинг, Логирование, Метрики

В микросервисной архитектуре отладка и диагностика проблем радикально усложняются. Один пользовательский запрос может пройти через десятки сервисов. Observability — это три столпа: логи, метрики и трейсы.

Три столпа Observability

    ┌─────────────────────────────────────────────────────────────────────┐
    │                    THREE PILLARS OF OBSERVABILITY                   │
    │  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐   │
    │  │      LOGS        │  │     METRICS      │  │     TRACES       │   │
    │  │                  │  │                  │  │                  │   │
    │  │ "Что случилось?" │  │ "Сколько/как     │  │ "Как запрос      │   │
    │  │                  │  │  быстро?"        │  │  прошёл через    │   │
    │  │ Текстовые записи │  │ Числовые         │  │  систему?"       │   │
    │  │ событий с        │  │ значения с       │  │                  │   │
    │  │ контекстом       │  │ временными       │  │ Путь запроса     │   │
    │  │                  │  │ метками          │  │ через сервисы    │   │
    │  │ ELK, Loki,       │  │ Prometheus,      │  │ Jaeger, Zipkin,  │   │
    │  │ CloudWatch       │  │ Datadog, Grafana │  │ Tempo            │   │
    │  └──────────────────┘  └──────────────────┘  └──────────────────┘   │
    └─────────────────────────────────────────────────────────────────────┘

Structured Logging

import structlog
import uuid

# Настройка structured logging
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.JSONRenderer()
    ]
)

logger = structlog.get_logger()

class OrderService:
    def create_order(self, request):
        # Создаём correlation_id для сквозного трейсинга
        correlation_id = request.headers.get(
            "X-Correlation-ID", str(uuid.uuid4())
        )

        log = logger.bind(
            correlation_id=correlation_id,
            user_id=request.user_id,
            service="order-service"
        )

        log.info("order_creation_started",
                 items_count=len(request.items))

        try:
            order = self._process_order(request)
            log.info("order_created",
                     order_id=order.id,
                     total=order.total,
                     duration_ms=elapsed_ms)
            return order

        except InsufficientStockError as e:
            log.warning("order_creation_failed",
                        reason="insufficient_stock",
                        product_id=e.product_id)
            raise

        except Exception as e:
            log.error("order_creation_error",
                      error_type=type(e).__name__,
                      error_message=str(e),
                      exc_info=True)
            raise

Пример JSON-лога:

{
    "timestamp": "2026-02-20T14:30:00.000Z",
    "level": "info",
    "event": "order_created",
    "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
    "user_id": "user-123",
    "service": "order-service",
    "order_id": "order-456",
    "total": 150.00,
    "duration_ms": 45
}

Distributed Tracing (OpenTelemetry)

┌───────────────────────────────────────────────────────────────┐
│       DISTRIBUTED TRACE (один пользовательский запрос)        │
│                                                               │
│  Trace ID: abc123                                             │
│                                                               │
│  ├─ Span: API Gateway (15ms)                                  │
│  │   ├─ Span: Auth Service - validate token (3ms)             │
│  │   └─ Span: Order Service - create order (45ms)             │
│  │       ├─ Span: Product Service - get products (8ms)        │
│  │       ├─ Span: Inventory Service - reserve (12ms)          │
│  │       ├─ Span: Payment Service - charge (20ms)             │
│  │       │   └─ Span: Stripe API - external call (15ms)       │
│  │       └─ Span: DB - insert order (3ms)                     │
│  │                                                            │
│  │  Total: 60ms                                               │
│  │  Bottleneck: Payment Service (33% of total time)           │
└───────────────────────────────────────────────────────────────┘

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.flask import FlaskInstrumentor

# Настройка OpenTelemetry
provider = TracerProvider()
jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831
)
provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(provider)

# Автоинструментация HTTP-клиентов и Flask
RequestsInstrumentor().instrument()
FlaskInstrumentor().instrument_app(app)

tracer = trace.get_tracer("order-service")

class OrderService:
    def create_order(self, order_data):
        with tracer.start_as_current_span("create_order") as span:
            span.set_attribute("order.user_id", order_data.user_id)
            span.set_attribute("order.items_count", len(order_data.items))

            # Каждый вызов внешнего сервиса — отдельный span
            with tracer.start_as_current_span("validate_products"):
                products = self.product_client.validate(order_data.items)

            with tracer.start_as_current_span("reserve_inventory") as inv_span:
                try:
                    reservation = self.inventory_client.reserve(order_data.items)
                    inv_span.set_attribute("reservation.id", reservation.id)
                except Exception as e:
                    inv_span.set_status(
                        trace.StatusCode.ERROR, str(e)
                    )
                    inv_span.record_exception(e)
                    raise

            with tracer.start_as_current_span("process_payment"):
                payment = self.payment_client.charge(
                    order_data.user_id, order_data.total
                )

            with tracer.start_as_current_span("save_order"):
                order = self.repository.save(order_data)

            span.set_attribute("order.id", order.id)
            span.set_attribute("order.total", float(order.total))
            return order

Метрики (Prometheus)

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Определение метрик
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status_code', 'service']
)

REQUEST_LATENCY = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint', 'service'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

ORDERS_IN_PROGRESS = Gauge(
    'orders_in_progress',
    'Number of orders currently being processed'
)

ORDER_TOTAL = Counter(
    'orders_total',
    'Total orders created',
    ['status']  # success, failure, cancelled
)

# Middleware для автоматического сбора метрик
class MetricsMiddleware:
    def __call__(self, request, next_handler):
        method = request.method
        endpoint = request.path

        ORDERS_IN_PROGRESS.inc()
        start_time = time.time()

        try:
            response = next_handler(request)
            REQUEST_COUNT.labels(
                method=method,
                endpoint=endpoint,
                status_code=response.status_code,
                service="order-service"
            ).inc()
            return response
        finally:
            duration = time.time() - start_time
            REQUEST_LATENCY.labels(
                method=method,
                endpoint=endpoint,
                service="order-service"
            ).observe(duration)
            ORDERS_IN_PROGRESS.dec()

# Запуск metrics endpoint
start_http_server(9090)  # GET /metrics на порту 9090

Prometheus scrape config:

# prometheus.yml
scrape_configs:
  - job_name: 'order-service'
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: order-service
        action: keep
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_port]
        target_label: __address__
        regex: (.+)
        replacement: ${1}:9090

Ключевые метрики (RED method):

USE method для инфраструктурных метрик:

Alerting

# Prometheus alerting rules
groups:
  - name: order-service-alerts
    rules:
      - alert: HighErrorRate
        expr: |
          sum(rate(http_requests_total{service="order-service",status_code=~"5.."}[5m]))
          /
          sum(rate(http_requests_total{service="order-service"}[5m]))
          > 0.01
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "High error rate on order-service"
          description: "Error rate is {{ $value | humanizePercentage }} (threshold: 1%)"

      - alert: HighLatency
        expr: |
          histogram_quantile(0.99,
            sum(rate(http_request_duration_seconds_bucket{service="order-service"}[5m])) by (le)
          ) > 2.0
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P99 latency above 2 seconds"

      - alert: OrderSagaFailureRate
        expr: |
          sum(rate(orders_total{status="failure"}[15m]))
          /
          sum(rate(orders_total[15m]))
          > 0.05
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Order saga failure rate above 5%"

---

11.11. Тестирование микросервисов

Тестирование микросервисов отличается от тестирования монолита из-за распределённой природы системы. Необходимо тестировать как отдельные сервисы, так и их взаимодействие.

Пирамида тестирования для микросервисов

┌──────────────────────────────────────────────────────────────┐
│             TESTING PYRAMID                                  │
│                                                              │
│                   /\                                         │
│                  /  \         End-to-End Tests               │
│                 / E2E\        (несколько, медленные,         │
│                /──────\        хрупкие)                      │
│               /        \                                     │
│              / Contract \      Contract Tests                │
│             /   Tests    \     (проверяют контракты          │
│            /──────────────\     между сервисами)             │
│           /                \                                 │
│          /  Integration     \   Integration Tests            │
│         /    Tests           \  (БД, внешние API,            │
│        /──────────────────────\  message broker)             │
│       /                        \                             │
│      /       Unit Tests         \  Unit Tests                │
│     /          (много,           \ (быстрые, изолированные)  │
│    /──────────────────────────────\                          │
│                                                              │
└──────────────────────────────────────────────────────────────┘

Unit Tests

# Тестирование бизнес-логики в изоляции
import pytest
from unittest.mock import Mock, patch
from order_service import OrderService, InsufficientStockError

class TestOrderService:
    def setup_method(self):
        self.product_client = Mock()
        self.inventory_client = Mock()
        self.payment_client = Mock()
        self.order_repo = Mock()

        self.service = OrderService(
            product_client=self.product_client,
            inventory_client=self.inventory_client,
            payment_client=self.payment_client,
            order_repo=self.order_repo
        )

    def test_create_order_success(self):
        # Arrange
        self.product_client.get_products.return_value = [
            Product(id="p1", name="Widget", price=10.0, stock=100)
        ]
        self.inventory_client.reserve.return_value = Reservation(id="r1")
        self.payment_client.charge.return_value = Payment(id="pay1")

        # Act
        order = self.service.create_order(
            user_id="u1",
            items=[OrderItem(product_id="p1", quantity=2)]
        )

        # Assert
        assert order.status == "PENDING"
        assert order.total == 20.0
        self.inventory_client.reserve.assert_called_once()
        self.payment_client.charge.assert_called_once_with("u1", 20.0)

    def test_create_order_insufficient_stock(self):
        self.product_client.get_products.return_value = [
            Product(id="p1", name="Widget", price=10.0, stock=1)
        ]
        self.inventory_client.reserve.side_effect = InsufficientStockError("p1")

        with pytest.raises(InsufficientStockError):
            self.service.create_order(
                user_id="u1",
                items=[OrderItem(product_id="p1", quantity=5)]
            )

        # Payment не должен вызываться если нет товара
        self.payment_client.charge.assert_not_called()

Contract Tests (Pact)

Contract testing проверяет, что API-контракт между consumer и provider не нарушен. Это позволяет тестировать совместимость без запуска всех сервисов.

# Consumer side (Order Service — потребитель Product Service API)
import atexit
from pact import Consumer, Provider

pact = Consumer('OrderService').has_pact_with(
    Provider('ProductService'),
    pact_dir='./pacts'
)
pact.start_service()
atexit.register(pact.stop_service)

class TestProductServiceContract:
    def test_get_product(self):
        # Описываем ожидаемый контракт
        (pact
         .given('product with id p1 exists')
         .upon_receiving('a request for product p1')
         .with_request('GET', '/api/v1/products/p1')
         .will_respond_with(200, body={
             'id': 'p1',
             'name': Like('Widget'),         # Любая строка
             'price': Like(10.0),            # Любое число
             'stock': Like(100),             # Любое число
             'status': Term(r'active|inactive', 'active')
         }))

        with pact:
            # Act — вызываем реальный клиент против mock-сервера Pact
            client = ProductServiceClient(
                base_url=f"http://localhost:{pact.port}"
            )
            product = client.get_product("p1")

            # Assert
            assert product.id == "p1"
            assert product.price > 0

# Provider side (Product Service проверяет, что соответствует контракту)
from pact import Verifier

def test_provider_honors_contract():
    verifier = Verifier(
        provider='ProductService',
        provider_base_url='http://localhost:8080'
    )

    # Загружаем контракт и проверяем
    output, _ = verifier.verify_pacts(
        './pacts/orderservice-productservice.json',
        provider_states_setup_url='http://localhost:8080/_pact/provider-states'
    )
    assert output == 0

Integration Tests с Testcontainers

import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.kafka import KafkaContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:15") as pg:
        yield pg

@pytest.fixture(scope="session")
def kafka():
    with KafkaContainer() as k:
        yield k

@pytest.fixture(scope="session")
def redis():
    with RedisContainer() as r:
        yield r

class TestOrderIntegration:
    def test_create_order_persists_to_db(self, postgres):
        """Тест с реальной базой данных."""
        db_url = postgres.get_connection_url(
        repo = OrderRepository(db_url)
        repo.run_migrations()

        order = Order(user_id="u1", items=[...], total=50.0)
        repo.save(order)

        loaded = repo.get(order.id)
        assert loaded.user_id == "u1"
        assert loaded.total == 50.0

    def test_order_event_published_to_kafka(self, kafka):
        """Тест публикации события в Kafka."""
        bootstrap = kafka.get_bootstrap_server()
        producer = OrderEventProducer(bootstrap)
        consumer = TestConsumer(bootstrap, "test-group")

        order = Order(id="o1", user_id="u1", total=50.0)
        producer.publish_order_created(order)

        # Ждём сообщение
        messages = consumer.consume(timeout=10)
        assert len(messages) == 1
        assert messages[0]["event_type"] == "OrderCreated"
        assert messages[0]["data"]["order_id"] == "o1"

    def test_idempotent_handler_with_redis(self, redis):
        """Тест идемпотентности с реальным Redis."""
        redis_client = redis.get_client()
        handler = IdempotentEventHandler(redis_client)

        event = {"event_id": "e1", "data": {"order_id": "o1"}}

        handler.handle(event)     # Первая обработка
        handler.handle(event)     # Повторная — должна быть пропущена

        # Проверяем, что событие обработано только один раз
        assert handler.process_count == 1

End-to-End Tests

class TestOrderE2E:
    """
    E2E тесты запускаются против staging-окружения
    или docker-compose с полным набором сервисов.
    """

    def test_full_order_flow(self, api_client):
        # 1. Создаём пользователя
        user = api_client.post("/api/v1/users", json={
            "name": "Test User",
            "email": "test@example.com"
        }).json()

        # 2. Создаём заказ
        order = api_client.post("/api/v1/orders", json={
            "user_id": user["id"],
            "items": [{"product_id": "p1", "quantity": 2}]
        }).json()
        assert order["status"] == "PENDING"

        # 3. Ждём обработки (saga, async events)
        order = self.wait_for_status(
            api_client, order["id"], "PAID", timeout=30
        )
        assert order["status"] == "PAID"

        # 4. Проверяем, что inventory обновился
        product = api_client.get("/api/v1/products/p1").json()
        assert product["reserved_quantity"] >= 2

    def wait_for_status(self, client, order_id, expected_status, timeout):
        """Polling с таймаутом для eventually consistent систем."""
        start = time.time()
        while time.time() - start < timeout:
            order = client.get(f"/api/v1/orders/{order_id}").json()
            if order["status"] == expected_status:
                return order
            time.sleep(1)
        raise TimeoutError(
            f"Order {order_id} did not reach status {expected_status}"
        )

---

12. Deployment Strategies

Стратегии развёртывания определяют, как новая версия сервиса заменяет старую с минимальным влиянием на пользователей.

Обзор стратегий

┌───────────────────────────────────────────────────────────────────┐
│                      DEPLOYMENT STRATEGIES                        │
│                                                                   │
│  1. Rolling Update                                                │
│     v1 v1 v1 v1  →  v2 v1 v1 v1  →  v2 v2 v1 v1  →  v2 v2 v2 v2   │
│                                                                   │
│  2. Blue-Green                                                    │
│     [Blue: v1 v1 v1] ──switch──▶ [Green: v2 v2 v2]                │
│                                                                   │
│  3. Canary                                                        │
│     [v1 v1 v1 v1] + [v2]  →  [v1 v1 v1] + [v2 v2]  →  [v2 v2 v2]  │
│     95% traffic      5%       70%           30%          100%     │
│                                                                   │
│  4. Feature Flags                                                 │
│     v2 deployed everywhere, but new feature enabled for 5% users  │
└───────────────────────────────────────────────────────────────────┘

Rolling Update (Kubernetes)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
spec:
  replicas: 4
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxUnavailable: 1       # Максимум 1 pod недоступен
      maxSurge: 1             # Максимум 1 дополнительный pod
  template:
    spec:
      containers:
        - name: order-service
          image: order-service:v2
          readinessProbe:      # Pod принимает трафик только когда ready
            httpGet:
              path: /health/ready
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 5
          livenessProbe:       # Перезапуск если pod завис
            httpGet:
              path: /health/live
              port: 8080
            initialDelaySeconds: 15
            periodSeconds: 10

Blue-Green Deployment

┌───────────────────────────────────────────────────────────────────┐
│                       BLUE-GREEN DEPLOYMENT                       │
│                                                                   │
│  Шаг 1: Blue (v1) обслуживает трафик                              │
│                                                                   │
│  ┌────────────┐        ┌──────────────────┐                       │
│  │   Router   │───────▶│   Blue (v1)      │  ◀── active           │
│  │            │        │   ████████████   │                       │
│  └────────────┘        └──────────────────┘                       │
│                        ┌──────────────────┐                       │
│                        │   Green (v2)     │  ◀── idle (testing)   │
│                        │   ████████████   │                       │
│                        └──────────────────┘                       │
│                                                                   │
│  Шаг 2: Переключаем router на Green                               │
│                                                                   │
│  ┌────────────┐        ┌──────────────────┐                       │
│  │   Router   │        │   Blue (v1)      │  ◀── idle (rollback)  │
│  │            │        │   ████████████   │                       │
│  └─────┬──────┘        └──────────────────┘                       │
│        │               ┌──────────────────┐                       │
│        └──────────────▶│   Green (v2)     │  ◀── active           │
│                        │   ████████████   │                       │
│                        └──────────────────┘                       │
│                                                                   │
│  Rollback: переключаем обратно на Blue за секунды                 │
└───────────────────────────────────────────────────────────────────┘

# Скрипт Blue-Green deployment
class BlueGreenDeployer:
    def deploy(self, new_version: str):
        # 1. Определяем текущий и новый слот
        current = self.get_active_slot()        # "blue"
        target = "green" if current == "blue" else "blue"

        # 2. Деплоим новую версию в неактивный слот
        self.deploy_to_slot(target, new_version)

        # 3. Ждём readiness
        self.wait_for_healthy(target, timeout=120)

        # 4. Запускаем smoke tests против нового слота
        if not self.run_smoke_tests(target):
            self.rollback(target)
            raise DeploymentFailed("Smoke tests failed")

        # 5. Переключаем трафик
        self.switch_traffic(target)

        # 6. Мониторим ошибки
        if self.error_rate_exceeded(threshold=0.01, window=300):
            self.switch_traffic(current)    # Мгновенный rollback
            raise DeploymentFailed("Error rate exceeded after switch")

        # 7. Очищаем старый слот (опционально, оставляем для rollback)
        logger.info(f"Deployment successful: {current} → {target}")

Canary Deployment

# Istio VirtualService для canary
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    - route:
        - destination:
            host: order-service
            subset: stable    # v1
          weight: 95
        - destination:
            host: order-service
            subset: canary    # v2
          weight: 5
---
# Автоматический canary rollout с Argo Rollouts
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: order-service
spec:
  replicas: 10
  strategy:
    canary:
      steps:
        - setWeight: 5           # 5% трафика
        - pause: {duration: 5m}  # Наблюдаем 5 минут
        - setWeight: 20
        - pause: {duration: 5m}
        - setWeight: 50
        - pause: {duration: 10m}
        - setWeight: 100         # Полный rollout
      analysis:
        templates:
          - templateName: success-rate
        startingStep: 1          # Начинаем анализ с первого шага
        args:
          - name: service-name
            value: order-service
---
# Автоматический анализ метрик
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: success-rate
spec:
  metrics:
    - name: success-rate
      interval: 60s
      successCondition: result[0] >= 0.99   # 99% success rate
      provider:
        prometheus:
          address: http://prometheus:9090
          query: |
            sum(rate(http_requests_total{service="{{args.service-name}}",
                     status_code!~"5.."}[5m]))
            /
            sum(rate(http_requests_total{service="{{args.service-name}}"}[5m]))

Сравнение стратегий

---

Вопросы для самопроверки

Архитектура и декомпозиция (1–2)

1. Опишите ключевые различия между монолитом, модульным монолитом и микросервисами. В каких ситуациях вы бы выбрали каждый подход?

2. Что такое "распределённый монолит" и как его избежать при проектировании микросервисов?

3. Какие критерии вы бы использовали для определения границ микросервисов? Чем отличается Decomposition by Business Capability от Decomposition by Subdomain?

4. Опишите Strangler Fig pattern. Как бы вы применили его для миграции монолита на микросервисы? Какую роль играет Anti-Corruption Layer?

Service Discovery и Saga (3–4)

5. В чём разница между Client-Side и Server-Side Service Discovery? Какой подход используется в Kubernetes и почему?

6. Объясните разницу между Choreography-based и Orchestration-based Saga. Когда стоит предпочесть один подход другому?

7. Что такое State Machine подход для Saga? Как optimistic locking помогает при параллельной обработке?

Resilience Patterns (5)

8. Как работает Circuit Breaker? Опишите все три состояния и условия переходов между ними. Что произойдёт, если circuit breaker настроен слишком агрессивно?

9. Почему важен Bulkhead pattern? Приведите пример сценария, где его отсутствие приводит к каскадному отказу.

10. Как правильно реализовать retry с экспоненциальной задержкой? Почему важен jitter и как он предотвращает thundering herd?

11. В каком порядке следует комбинировать Circuit Breaker, Bulkhead и Retry? Почему этот порядок важен?

Service Mesh (6)

12. Что такое Service Mesh? Какие проблемы он решает и когда его использование не оправдано?

13. Сравните Istio и Linkerd. Какие компромиссы между функциональностью и операционной сложностью?

API Gateway (7)

14. Какие функции выполняет API Gateway? Чем он отличается от обычного load balancer?

15. Что такое паттерн Backend for Frontend (BFF)? Почему один API Gateway для всех клиентов может быть антипаттерном?

16. Какие anti-patterns существуют при использовании API Gateway? Как избежать "God Gateway"?

Inter-Service Communication (8)

17. Сравните REST и gRPC для внутренней коммуникации между сервисами. Когда стоит выбирать каждый из них?

18. Объясните разницу между Point-to-Point, Publish/Subscribe и Consumer Group моделями в message broker.

19. Что такое Outbox Pattern? Какую проблему он решает и почему нельзя просто записать данные и отправить событие в message broker последовательно?

20. Как обеспечить идемпотентность при обработке сообщений в at-least-once delivery? Приведите пример реализации.

CQRS и Event Sourcing (9)

21. Что такое CQRS? Когда разделение моделей чтения и записи оправдано, а когда это over-engineering?

22. Объясните Event Sourcing. Как восстановить текущее состояние агрегата из истории событий? Зачем нужны snapshot'ы?

23. Как Event Sourcing помогает с аудитом? Какие проблемы возникают при необходимости удалить данные (GDPR)?

Observability (10)

24. Назовите три столпа observability. Как логи, метрики и трейсы дополняют друг друга?

25. Что такое distributed tracing? Как OpenTelemetry помогает отследить путь запроса через десятки сервисов?

26. Объясните методологии RED и USE для метрик. Для чего подходит каждая из них?

27. Как настроить alerting так, чтобы он был полезным, а не шумным? Что значит "alert fatigue" и как его избежать?

Тестирование (11)

28. Как выглядит пирамида тестирования для микросервисов? Чем она отличается от тестирования монолита?

29. Что такое Contract Testing (Pact)? Как он позволяет тестировать совместимость сервисов без их совместного запуска?

30. Как Testcontainers помогают в integration testing? Приведите пример тестирования с реальной базой данных и Kafka.

Deployment Strategies (12)

31. Сравните Rolling Update, Blue-Green и Canary deployment. Какие trade-offs у каждого подхода в плане ресурсов, скорости rollback и риска?

32. Как Argo Rollouts автоматизирует canary deployment? Какие метрики можно использовать для автоматического решения о продолжении rollout?

Сквозные вопросы (дизайн-интервью)

33. Задача: Вам нужно спроектировать систему онлайн-заказа еды (аналог Delivery Club). Опишите: как вы определите границы сервисов, какой тип коммуникации выберете, как обеспечите reliability, какие паттерны используете для распределённых транзакций.

34. Задача: Ваш монолит e-commerce платформы обрабатывает 10K RPS и команда выросла до 50 человек. Руководство хочет перейти на микросервисы. Опишите стратегию миграции, порядок выделения сервисов, и как вы обеспечите отсутствие downtime.

35. Задача: Payment Service начал отвечать с задержкой 30 секунд вместо обычных 200ms. Опишите: какие паттерны защитят систему от каскадного отказа, как вы диагностируете проблему через observability инструменты, и какие alerting правила должны были сработать.

---

Дополнительные ресурсы

Книги

• Sam Newman, "Building Microservices" (2nd Edition) — основы микросервисной архитектуры
• Chris Richardson, "Microservices Patterns" — каталог паттернов с примерами
• Vaughn Vernon, "Domain-Driven Design Distilled" — DDD для декомпозиции
• Nora Jones, "Chaos Engineering" — тестирование устойчивости распределённых систем
• Betsy Beyer et al., "Site Reliability Engineering" (Google SRE Book) — observability, alerting, incident management

Papers

• "Microservices: A Definition of This New Architectural Term" — Martin Fowler
• "Design Patterns for Container-Based Distributed Systems" — Brendan Burns, David Oppenheimer
• "CQRS Documents" — Greg Young
• "Life Beyond Distributed Transactions: an Apostate's Opinion" — Pat Helland

Блоги и статьи

• Netflix Tech Blog: серия о microservices architecture, Circuit Breaker, Chaos Engineering
• Uber Engineering Blog: Domain-Oriented Microservice Architecture, DOMA
• Spotify Engineering: How We Build Microservices
• Martin Fowler: "Monolith First", "StranglerFigApplication", "CQRS"
• Confluent Blog: Event-Driven Architecture, Outbox Pattern, Kafka best practices

Инструменты для изучения

• Service Mesh: Istio documentation, Linkerd Getting Started guide
• Service Discovery: Consul tutorials, etcd documentation
• Messaging: Apache Kafka, RabbitMQ tutorials
• Resilience: Resilience4j (Java), Polly (.NET), Tenacity (Python)
• Observability: OpenTelemetry documentation, Jaeger, Prometheus + Grafana
• Contract Testing: Pact documentation и examples
• Deployment: Argo Rollouts, Flagger для progressive delivery
• API Gateway: Kong, NGINX, AWS API Gateway, Envoy