Пользователи 2ГИС каждый день помогают нам поддерживать точность данных: сообщают о новых компаниях, добавляют дорожные события, загружают фото и пишут отзывы. Раньше мы могли поблагодарить их только словами или устроить розыгрыш подарков. Но слова со временем забываются, а подарки достаются далеко не каждому. Поэтому мы решили сделать так, чтобы все неравнодушные к 2ГИС люди видели свой вклад в продукт и нашу за это благодарность.

Так появились награды — виртуальные медальки, которые мы начисляем за разного рода задания: загрузить фото в карточки кафе, написать отзывы о театрах, уточнить время работы организаций и так далее. Заработанные награды пользователи видят в личном профиле 2ГИС и на вкладке «Мой 2ГИС» в мобильном приложении. Там же мы показываем, сколько осталось до следующего достижения.

Чтобы реализовать эту фичу, мы научились обрабатывать поток событий объемом 500 тысяч записей в час (местами — до 50 тысяч в секунду) и анализировать данные из нескольких сервисов. А еще — добавили немного метапрограммирования для того, чтобы упростить конфигурирование при разработке новых наград.

Вместе с Rapter расскажем, что находится под капотом процесса выдачи наград.

Концепция

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

Требования в тезисах

Требования — вещь довольно скучная, поэтому все нюансы расписывать не будем, сконцентрируемся на самых важных вещах:

• награды выдаются только авторизованным пользователям;
• обновление прогресса по награде должно быть максимально быстрым;
• награда — результат выполнения пользователем набора действий в продукте: загрузка фото, написание отзыва, поиск проезда и т. п. Систем-источников данных много.

Архитектурная идея

Идея реализации не очень сложна. Ее можно выразить тезисно:

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

Основные сущности

На схеме ниже представлены основные сущности предметной области и их взаимосвязи:



На схеме выделены две зоны:

• Схема — зона описания структуры наград и правил их начисления;
• Данные — зона наград конкретных пользователей и данных, связанных с их текущим состоянием

Сущности на схеме:

• Achieve — данные о награде, которую можно получить. Включает метаинформацию и описание способа объединения результатов заданий — стратегию.
• Objective — задание, условия которого нужно выполнить для продвижения к получению награды.
• UserAchieve — текущее состояние награды конкретного пользователя.
• UserObjective — текущее состояние задания награды пользователя.
• User — информация о пользователе, необходимая для уведомлений и понимания его текущего статуса (удаленным и забаненным награды не нужны).
• ProcessingLog — лог начислений по заданиям. Содержит информацию о том, как конкретное действие повлияло на прогресс по заданию.
• Event — минимально необходимая информация о событии, которое как-то повлияло на прогресс по заданиям пользователя.

Структура сервиса

Теперь рассмотрим основные компоненты сервиса и их зависимости:

• Events Bus — шина событий, которые могут использоваться для выполнения заданий. Мы используем Apache Kafka.
• Master и Slave DBs — основное хранилище данных. В данном случае кластер PostgreSQL.
• ConsumingWorkers — обработчики событий из шины. Основная задача — читать события из определенного источника (фото, отзывов и т. п.), применять их к заданиям пользователей и сохранять результат.
• AchievesWorker — пересчитывает прогресс по наградам пользователя согласно состоянию заданий.
• NotificationWorkers — набор обработчиков для планирования и отправки уведомлений о получении награды, анонсов новых возможных достижений и т. п.
• Public API — публичный REST-интерфейс для Web и мобильных приложений.
• Private API — REST-интерфейс для админки, которая помогает в расследовании инцидентов и поддержке сервиса. Она доступна для разработчиков и команды поддержки.

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

Обработка событий

Контент

Награды — это прежде всего сервис агрегации данных. Каждая мастер-система генерирует несколько типов событий. Как правило, каждый тип события тесно связан с состоянием контента, его статусной моделью. Так, фотография может быть на модерации, удалена, заблокирована, скрыта или активна. Все это — разные события, которые обрабатываются отдельным воркером, специализирующемся на конкретном источнике. На данный момент времени происходит взаимодействие со следующими источниками (мастер-системами):

• Photo — генерирует разные события, которые касаются операций, производимых пользователями над фотографиями.
• Reviews — события, касающиеся операций над отзывами пользователей.
• Datafeedback — события, касающиеся операций над уточнениями. Уточнение — это изменение информации об объекте на карте, будь то фирма или памятник.
• Check — события, которые касаются приложения 2ГИС Чек.
• BSS — это события аналитики, которые генерируют приложения 2ГИС. Например, открытие определенной фирмы, поездки по навигатору и т. д.

Сгенерированные мастер-системой события попадают в топик Kafka в порядке изменения их статусов, что дает возможность двигать прогресс по награде для пользователя не только вперед, но и откатывать его назад. Например, если фотография находилась в статусе «активна», а потом по какой-то причине приобрела статус «заблокирована», прогресс по награде должен измениться в меньшую сторону. Прогресс награды — это интерпретация внутренних объектов, именуемых счетчиками контента.

Счетчики могут отличаться для разных данных. Например, для событий о фото они следующие: количество одобренных, количество на модерации, количество заблокированных, а для событий открытий карточек необходимо считать только количество карточек, открытых пользователем. Исходя из текущих значений счетчиков контента, для конкретного пользователя в рамках конкретной награды определяются ответы на следующие вопросы:

• началось ли выполнение награды?
• каков прогресс?
• награда полностью выполнена?

Фильтры и правила

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

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

Фильтр — это некоторое ограничение, которое накладывается на контент. Он занимается только тем, что отвечает на вопрос: «Подходит ли новое событие под данное условие или нет?»
Правило — это особый фильтр, цель которого сказать: «Если событие подошло под условие, то каким образом у него должны измениться счетчики?» Правило включает в себя алгоритм изменения счетчиков. Каждая награда содержит только одно правило.

Реализация фильтров и правил находится в коде проекта, а описание того, какие фильтры (правило) относятся к конкретной награде — в БД в формате JSON. К такому решению мы пришли не сразу. Изначально фильтры и правила нельзя было задать с помощью конфигурации через базу, награда была полностью описана в коде, в таблице же хранился лишь ее идентификатор. Такое решение давало ряд существенных недостатков:

• Проблема поддержки нескольких окружений. Если нужно выкатить на тестовую среду одно состояние списка наград, а в бой отправить другое, появляется необходимость знать в коде проекта об окружении или иметь конфигурационный файл со списком наград. При этом нет возможности использовать разные БД для этой задачи, хотя они уже есть для каждого окружения.
• Возможность конфигурировать фильтрацию только разработчиком. Так как все описано в коде, изменения мог делать только человек, который знает проект и язык программирования, хотелось же, чтобы это можно было сделать просто через Private API или БД.
• Неудобство просмотра. Наград много, иногда нужно посмотреть фильтры, которые они используют. Каждый раз делать это через просмотр кода — довольно утомительно.

На старте приложения мы производим матчинг по имени фильтров, подгруженных из БД и подкладываем их в конкретную награду. Пример описания фильтра:

[
   {
       "name":"SourceFilter",
       "config":{
           "sources":["reviews"]
       }
   },
   {
       "name": "ReviewsLengthFilter",
       "config": {
           "allowed_length": 100
       }
   }
]

В данном случае мы будем брать только те отзывы (об этом говорит первый объект-описание из массива фильтров), текст которых содержит более 100 символов (второй фильтр в списке).

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

{"name": "ReviewUniqueByObjectRule","config":{}}

Данное правило позволит изменить счетчики только в том случае, если пользователь написал отзыв для объекта, при этом к одному объекту будет учитываться только один отзыв.

BSS

Отдельно остановимся на работе с потоком BSS-событий. На это есть как минимум три причины:

• Cобытия аналитики не могут откатиться, статусная модель в них отсутствует, что, в общем-то, логично, т. к. проезд по навигатору или построение маршрута нельзя отменить. Действие либо было, либо нет.
• Объемы. Напомню, что общая аудитория 2ГИС составляет 50+ млн пользователей в месяц. Вместе они совершают более 1,5 млрд поисковых запросов, а также множество других действий: запуск приложения, просмотр карточки объекта и т. д. В пике число событий может достигать 50 000 в секунду. Всю эту информацию мы должны пропустить через фильтры, чтобы выдать награду пользователю.
• События аналитики имеют особенности: несколько форматов, большое разнообразие типов.

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

Для сокращения описанных различий был создан отдельный сервис, который занимается предобработкой таких событий. Сервис умеет работать со всем многообразием форматов сообщений, приходящих с аналитики. Суть его работы заключается в следующем: читается весь BSS-поток событий, из которого берутся лишь те, которые нужны для Наград. Такой сервис-фильтр значительно снижает нагрузку (после фильтрации скорость потока составляет ?300 событий в секунду) с обработчика BSS-потока Наград, а также генерирует события в едином формате, нивелируя недостаток, связанный с историей развития внутренней аналитики.

Выдача наград

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

Первый вопрос, который возникает: зачем выделять выдачу в отдельный worker, разве нельзя пересчитывать при обработке каждого события? Ответ: можно, но не стоит.

Причин выделения выдачи в отдельный процесс несколько:

1. Перенеся пересчет в каждый ConsumingWorker, получим Race Condition на операции обновления прогресса по награде, т. к. каждый обработчик будет пытаться обновить прогресс, опираясь на известные ему состояния заданий, а другие будут активно это состояние менять.
2. Каждый ConsumingWorker пакетно обрабатывает события из Kafka в транзакции. Добавив вставку в таблицу наград пользователя, вызовем лишние блокировки на уровне БД, которые будут тормозить другие обработчики.
3. В процессе выдачи наград есть логика отправки уведомлений, которая будет только тормозить обработку потока событий, что нежелательно.

С причинами появления отдельного AchievesWorker (обработчика для выдачи наград) разобрались. Теперь нужно разобраться с двумя важными частями обработки:

1. Есть набор заданий в награде. Есть набор счетчиков по этим заданиями. Как понять, насколько выполнена награда и как это выразить коде?
   Пример: Нужно написать 3 отзыва или загрузить 3 фото. У пользователя 1 отзыв и 2 фото. Какой прогресс по награде? Ответ: 3, т. к. пользователь точно будет уверен, что нужно 3 в сумме.
2. У нас появился отдельный обработчик для выдачи наград. Каждый раз пересчитывать несколько десятков наград по каждому авторизованному пользователю, т. е. несколько десятков миллионов, вряд ли получится быстро. Как ему узнать о том, прогресс каких именно пользователей и по каким заданиям менялся с момента последней обработки?

Каждую часть рассмотрим отдельно.

Перетекание прогресса

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

«Выполни одно задание на X единиц». Пример: проехать 10 км по навигатору.



«Выполни несколько заданий на X единиц каждое». Пример: загрузить 5 фото и написать 5 отзывов в карточки — всего 10 единиц контента.



«Выполни несколько заданий на X единиц в сумме». Пример: написать 5 отзывов или загрузить 5 фото.



«Выполни несколько заданий, сгруппированных по типам». Пример: загрузить 5 единиц контента (фото или отзывов) и проехать 10 км по навигатору.



Теоретически могут быть более сложные вложенные комбинации. Однако в реальных условиях объяснить пользователю в двух-трёх предложениях сложную логическую комбинацию, которую нужно выполнить для получения награды не представляется возможным. Поэтому в большинстве случаев этих вариантов достаточно.

Способ преобразования мы назвали стратегией и постарались сделать ее более-менее универсальной, выработав формальное описание в виде JSON-объекта. Можно было, конечно, подумать над записью в виде формулы, но тогда пришлось бы использовать подобия eval или описывать грамматику и реализовывать ее, а это явно переусложнение. Хранение в исходном коде стратегии для каждой награды не очень удобно, т. к. порвется описания награды (часть в БД, а часть в коде), а также не позволит в будущем собирать награды из готовых компонентов без участия разработки.

Стратегия представляется в виде дерева, где каждый узел:

• Ссылается на текущий прогресс по заданию или является группой других узлов.
• Может иметь ограничение сверху — по сути указание на необходимость применения min().
• Может иметь коэффициент нормализации. Нужен для простых преобразований путем домножения результата на число. Нам пригодился для преобразования метры в километры.

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

Вот пример описания стратегии для последней категории:

{
  "goal": 15,
  "operation": "sum",
  "strategy": [
    {
      "goal": 5,
      "operation": "sum",
      "strategy": [
        {
          "objective_id": "photo"
        },
        {
          "objective_id": "reviews"
        }
      ]
    },
    {
      "goal": 10,
      "operation": "sum",
      "strategy": [
        {
          "objective_id": "navi",
          "normalization_factor": 0.001
        }
      ]
    }
  ]
}

Необходимые обновления

Есть несколько обработчиков, которые неустанно анализируют события по пользователям и применяют изменения к прогрессу по заданиям. Регулярный перебор всех пользователей с каждой наградой приведет к анализу нескольких десятков миллионов наград — не очень радужно при условии того, что реальные обновления будут измеряться в тысячах. Как узнать только о тысячах и не тратить впустую CPU на миллионы?

Идея того, как пересчитывать прогресс только по тем наградам, которые на самом деле изменились, пришла довольно быстро. Она основана на использовании векторных часов.

Перед описанием напомню о сущностях:

• UserObjective — данные о прогрессе пользователя по заданию награды.
• UserAchieve — данные о прогрессе пользователя по награде.

Реализация выглядит так:

• Заводим поле version для UserObjective и UserAchieve и Sequence в PostgreSQL.
• Каждое обновление сущности UserObjective изменяет ее версию. Значение берется из последовательности (у нас она общая для всех записей).
• Значение version для UserAchieve будет определяться как максимум от версий связанных с ним UserObjective.
• На каждом цикле обработки AchievesWorker ищет такие UserObjective, для которых нет UserAchieve или UserAchieve.version < UserObjective.version. Задача решается одним запросом к БД.

Сразу стоит отметить, что в решении есть ограничения по количеству записей в таблицах наград и заданий, а также по частотности изменения прогресса по заданиям, но при количествах в пару десятков миллионов наград и количеством обновлений меньше тысячи в минуту жить с таким решением вполне можно. О том, как мы оптимизировали выдачу для конкурса «Агенты 2ГИС», как-нибудь расскажем отдельно.

Выводы

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

Какие выводы мы сделали благодаря Наградам:

• Принцип «разделяй и властвуй» в данном случае сыграл нам на руку. Выделение обработчиков событий на каждый источник помогает нам масштабироваться при необходимости. Их работа изолирована по данным и пересекается только в небольших зонах. Выделение логики выдачи наград позволяет уменьшить накладные расходы в обработчиках событий.
• Если нужно переваривать много данных и обработка довольно затратна, стоит сразу подумать, как отфильтровать то, что точно не нужно. Опыт с фильтрацией BSS-потока тому пример.
• Еще раз убедились в том, что интеграция сервисов через общую шину событий очень удобна и позволяет избежать лишней нагрузки на другие сервисы. Если бы сервис Наград получал данные из сервисов Photo, Reviews и т. д. через http-запросы, то пришлось бы несколько сервисов подготовить к дополнительной нагрузке.
• Капелька метапрограммирования может помочь сохранить целостность конфигурирования данных и разделить окружения произвольным образом. Хранение фильтров, правил и стратегий в БД упростило процесс разработки и релиза новых наград.