Шина сообщений и RPC

Сервисы общаются через Redis Pub/Sub. Канал — это адрес: у каждого сервиса есть прямой канал с его именем, плюс общие каналы broadcast для рассылки событий и enrich для обогащения. Сообщения — JSON-RPC 2.0 с двумя дополнительными полями to и from. Всё это реализовано в rock-me-ts классами RPCAdapterRedis (транспорт) и RPCAgnostic (протокол).

Каналы

При создании адаптера сервис передаёт список каналов, на которые подписывается:

const channels = [config.rpc.name, BROADCAST];        // слушатель: chwriter, kafkastream
const channels = [ENRICH, config.rpc.name];           // обогатитель: mmgeo

Отправка — это PUBLISH в канал с именем получателя. Чтобы написать сервису chwriter, достаточно знать его имя: rpc.notify('chwriter', 'method', params).

Формат сообщения

Запрос:

{
  "jsonrpc": "2.0",
  "id": "1k",
  "from": "front_1",
  "to": "mmgeo_1",
  "method": "mmgeo_1:enrich:front_1",
  "params": { "ip": "93.184.216.34", "ua": "Mozilla/5.0 ..." }
}

Ответ:

{
  "jsonrpc": "2.0",
  "id": "1k",
  "from": "mmgeo_1",
  "to": "front_1",
  "service_group": "mmgeo",
  "result": { "country_iso": "RU", "city_en": "Moscow" }
}

Особенности:

API

register

Регистрирует обработчик метода. Один метод — одна функция, повторная регистрация заменяет предыдущую:

this.rpc.register(BROADCAST, this.writer.write);
this.rpc.register(METHOD_STATUS, () => this.status());

Обработчик получает params (или пустой объект, если их нет) и возвращает результат. Результат уходит отправителю, только если тот ждёт ответа.

notify

Уведомление без ответа. Так Front рассылает события слушателям, а сервисы отправляют heartbeat:

this.rpc.notify(SERVICE_DIRECTOR, METHOD_IAMALIVE, { name: this.name });

request

Запрос с ожиданием ответа. Возвращает промис:

const status = await this.rpc.request<RPCAppStatus>('mmgeo_1', METHOD_STATUS, {});
const geo = await this.rpc.request('enrich', 'enrich', { ip }, {
  timeout: 300,
  services: ['mmgeo_1', 'uapc_1']
});

Опции:

Если result одиночного ответа — объект вида { type__: 'data', data: {...} } (ответ, собранный через response.data), промис резолвится содержимым data.

import { getAppDeps, TheIds } from '@rockstat/rock-me-ts';
getAppDeps().setDep('ids', new TheIds());

Регистрация в Front

Front должен знать, какие сервисы обогащают события и какие обрабатывают ключи. Для этого сервис раз в несколько секунд шлёт heartbeat, а Front в ответ запрашивает его статус:

1. Сервис: notify('director', '__iamalive', { name }). Канал director слушает Front.
2. Front: request(name, '__status') в прямой канал сервиса.
3. Сервис отвечает структурой RPCAppStatus со списком register.
4. Front пересобирает таблицы маршрутизации. Если heartbeat перестаёт приходить, регистрация через некоторое время удаляется, поэтому интервал должен быть короче тайм-аута на стороне Front: mmgeo шлёт первый heartbeat через 500 мс после старта и далее каждые 9 секунд.

import { RPCAppStatus, MethodRegistration, METHOD_STATUS, METHOD_IAMALIVE, SERVICE_DIRECTOR } from '@rockstat/rock-me-ts';

async status(): Promise<RPCAppStatus> {
  return {
    name: this.name,
    app_state: 'running',
    app_started: Number(this.appStarted),
    app_uptime: Date.now() - Number(this.appStarted),
    register: [{
      method: 'enrich',
      role: 'enricher',
      options: { keys: ['in.gen.track'], props: { ip: 'td.ip', ua: 'td.ua' } }
    }]
  };
}

async setup() {
  this.rpc.register(METHOD_STATUS, () => this.status());
  const aliver = () => this.rpc.notify(SERVICE_DIRECTOR, METHOD_IAMALIVE, { name: this.name });
  setTimeout(aliver, 500);
  setInterval(aliver, 9 * 1000);
}

Элемент register:

Для обработчика Front строит ключ in.gen.<alias или имя сервиса>.<method> и вызывает метод запросом с полным событием. Ответ — либо структура response.data, response.redirect, response.pixel, response.error из rock-me-ts, либо любой объект, который Front обернёт в response.data. Метод * подписывает обработчик на все события сервиса.

Метрики RPC

RPCAgnostic сам пишет в Meter время каждого запроса (rpc.request с тегами target и method) и счётчик ошибок rpc.error. Свои метрики добавляйте через meter.tick('kafka.write.success') или meter.timenote('ch.flush'), который возвращает функцию-завершитель замера.