Синхронизация различных хранилищ данных

Всем привет! Я Станислав Бушуев, Software Engineer в Semrush. Сегодня хочу поделиться идеями, как можно реализовать синхронизацию данных между различными хранилищами. Такие задачи иногда возникают в работе, например, при удалении пользовательских данных в рамках General Data Protection Regulation (GDPR) и California Consumer Privacy Act (CCPA).

Следовать этим законам нетрудно, если вы поддерживаете среднестатистический сайт или небольшой продукт. Скорее всего, в данных случаях используется один из популярных фреймворков, и таблица пользователей хранится в общеизвестной базе данных (MySQL, PostgreSQL).

Пришел запрос на удаление всех данных пользователя? Не проблема! Удаляем строку в таблице, и готово. Не считая логов, трейсов ошибок, бэкапов и других интересных мест. Подразумеваем, конечно, что все настроено верно и личные данные пользователей скрываются звездочками или еще как-то.

В нашем случае ситуация чуть сложнее. За 13 лет в Semrush создали более 50 инструментов, и все они поддерживаются десятками команд разработки. Несколько лет назад каждая команда вынесла код своего инструмента в отдельный микросервис, данные пользователя мы получаем из сервиса пользователей. Но так или иначе мы храним всю информацию, которая необходима для работы инструмента.

Таким образом, перед нами встал вопрос: как синхронизировать хранилища данных микросервиса и сервиса пользователей.

Способы синхронизации 

1. Периодически мы можем сверяться с сервисом пользователей по REST API:

$ curl '<http://user-service.internal.net/api/v1/users/42>' [  {    "id": 42,    "registration_date": "2015-03-08 01:00:00",    "email": "john@example.com",    "name": "John", ... 

Можно даже сразу отправить список пользователей на проверку, но это как-то странно. Удаления пользователей происходят не так часто, чтобы совершать DDoS-атаки на сервис.

2. Событийно-ориентированная архитектура — еще один подход для решения нашей задачи. Здесь появляются две сущности: генератор сообщений (Publisher) и подписчик (Subscriber), который читает канал событий (topic).

Можно отказаться от сущности подписчика и всем микросервисам выставить endpoint API, на который сервис пользователей слал бы запросы при появлении нового события. В таком случае нужно согласовать протоколы взаимодействия API: REST, JSON-RPC, gRPC, GraphQL, OpenAPI или что там еще может быть. Кроме того, необходимо держать конфигурационные файлы микросервисов, куда слать запросы, а самое главное: что делать, когда запрос не доходит до микросервиса после третьего повтора?

Плюсы данной архитектуры:

• Асинхронная автоматическая синхронизация хранилищ данных.

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

• Синхронизация данных различных хранилищ находится отдельно от (и так нагруженного) сервиса пользователей.

Минусы:

• Минус, вытекающий из первого плюса: неконсистентность данных между сервисами пользователей и остальными микросервисами.

• Отсутствие транзакций: формируются простые сообщения.

• Необходимо учитывать, что сообщения в очереди могут повториться.

Вообще, все перечисленные преимущества и недостатки довольно условны и зависят от конкретных задач. Универсальных решений, на мой взгляд, нет. По этой теме полезно будет почитать о CAP-теореме.

Реализация событийно-ориентированной архитектуры на примере Pub/Sub от Google Cloud

Существует множество альтернатив: Kafka, RabbitMQ, но наша команда выбрала решение от Pub/Sub Google Cloud, так как мы уже используем Google Cloud (подробнее можно почитать в статье моего коллеги Никиты Шальнова: Что такое Immutable Infrastructure), и оно проще в настройке тех же Kafka или RabbitMQ.

В нашем случае Publisher — это сервис пользователей, а Subscriber — микросервис команды определенного инструмента. Subscriber’ов (как и Subscription) может быть сколько угодно. Учитывая, что в Semrush большое количество инструментов и команд, очередь с подписчиками для нас идеальна:

• Каждый читает очередь с необходимой для него частотой.

• Кто-то может выставить endpoint, и topic будет его вызывать немедленно при появлении сообщения (если нужно получать новые сообщения моментально).

• Даже экзотические варианты, например откат базы данных инструмента из старого бэкапа, не вызывают проблем: просто перечитаем сообщения из топика (стоит оговориться, что нужно учитывать идемпотентность запросов, и, возможно, вам стоит читать топик не с начала, а с какого-то момента).

• Subscriber предоставляет REST протокол, но для упрощения разработки существуют еще клиенты для различных языков программирования: Go, Java, Python, Node.js, C#, C++, PHP, Ruby.

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

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

Создание топика и подписчика:

gcloud pubsub topics create topic gcloud pubsub subscriptions create subscription --topic=topic

Подробнее можно узнать в документации.

Создание пользователя для чтения топика:

gcloud iam service-accounts create SERVICE_ACCOUNT_ID \    --description="DESCRIPTION" \    --display-name="DISPLAY_NAME" gcloud projects add-iam-policy-binding PROJECT_ID \    --member="serviceAccount:SERVICE_ACCOUNT_ID@PROJECT_ID.iam.gserviceaccount.com" \    --role="pubsub.subscriber" gcloud iam service-accounts keys create key-file \    --iam-account=sa-name@project-id.iam.gserviceaccount.com

Скачанный ключ в формате json нужно сохранить и пробросить сервису. Не забывайте про правила обращения с секретами! Об этом знают все и немного больше мои коллеги из Security-команды. Дайте знать в комментариях, если тема кажется вам полезной и интересной для нашей следующей статьи.

Пользователь для публикации сообщений создается аналогично, за исключением роли: —role=»pubsub.subscriber» → —role=»pubsub.publisher».

Для примера возьмем один из наших микросервисов, работающих на Python c Celery. Для сообщений из сервиса пользователей есть схема, описанная с помощью Protocol Buffers.

import json import os  import celery from google.cloud import pubsub_v1 from google.oauth2 import service_account  from user_pb2 import UserEventData  PUBSUB_SERVICE_ACCOUNT_INFO = json.loads(os.environ.get('PUBSUB_SERVICE_ACCOUNT', '{}')) PUBSUB_PROJECT = 'your project' PUBSUB_SUBSCRIBER = 'subscription'  @celery.shared_task def pubsub_synchronisation() -> None:    credentials = service_account.Credentials.from_service_account_info(        PUBSUB_SERVICE_ACCOUNT_INFO, scopes=['<https://www.googleapis.com/auth/pubsub>']    )     with pubsub_v1.SubscriberClient(credentials=credentials) as subscriber:        subscription_path = subscriber.subscription_path(PUBSUB_PROJECT, PUBSUB_SUBSCRIBER)        response = subscriber.pull(request={"subscription": subscription_path, "max_messages": 10000})         ack_ids, removed_user_ids = [], []        for msg in response.received_messages:            user_event_data = UserEventData()            user_event_data.ParseFromString(msg.message.data)             removed_user_ids.append(user_event_data.Id)            ack_ids.append(msg.ack_id)         # Here you can do everything with removed users :)         subscriber.acknowledge(request={"subscription": subscription_path, "ack_ids": ack_ids})

И запускаем задание раз в пять минут, так как удаление пользователей — не такая частая операция:

CELERY_BEAT_SCHEDULE = {    'pubsub_synchronisation': {        'task': 'tasks.pubsub_ubs_synchronisation',        'schedule': timedelta(minutes=5)    },

Пример публикации сообщений в топик на Python реализуется аналогичным образом. Используйте PublisherClient вместо SubscriberClient и вызывайте метод publish вместо pull.

В результате есть синхронизация удаления пользователей для соответствия законам GDPR/CCPA. Например, в 7:37 произошло массовое удаление аккаунтов из сервиса хранения учетных записей пользователей. В 7:40 сработала задача на получение данных из Topic’а. Все задачи выбраны, локальная база синхронизирована.

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

Надеюсь, этот материал окажется полезным. Спасибо за внимание!