Для понима как добата табли с уда агents, важно first иметь базовое understanding о distributed tables In this article, we will focus на как использовать distributed table как basis для созда кластер Мanticore instances.

table mydist {
          type  = distributed
          agent = box1:9312:shard1
          agent = box2:9312:shard2
          agent = box3:9312:shard3
          agent = box4:9312:shard4
}

In event server failure, distributed table will still work, but results from failed shard will be missing.

table mydist {
          type  = distributed
          agent = box1:9312|box5:9312:shard1
          agent = box2:9312:|box6:9312:shard2
          agent = box3:9312:|box7:9312:shard3
          agent = box4:9312:|box8:9312:shard4
}

Балансировка нагрузки включена по умолчанию для любой распределенной таблицы, использующей зеркалирование. По умолчанию запросы распределяются случайным образом между зеркалами. Это поведение можно изменить с помощью ha_strategy.

ha_strategy = {random|nodeads|noerrors|roundrobin}

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

ha_strategy может быть установлена глобально в конфигурации searchd или для каждой распределенной таблицы отдельно. Эта директива управляет стратегией выбора зеркала, или, другими словами, выбором конкретного зеркала агента в распределенной таблице. На практике эта директива контролирует, как мастер балансирует запросы между зеркальными узлами агентов. Реализованы следующие стратегии:

ha_strategy = random

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='random';

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

Адаптивные стратегии, основанные на вероятностях, взвешенных по задержке, работают следующим образом. Размер окна статистики контролируется параметром ha_period_karma, в то время как проверка работоспособности неактивных зеркал и отслеживание времени кругового оборота управляются параметром ha_ping_interval. Оба параметра описаны далее на этой странице и в разделе Настройки Searchd.

1. Статистика задержек накапливается блоками по ha_period_karma секунд.
2. Вероятности, взвешенные по задержкам, пересчитываются один раз за период кармы.
3. Флаг "живой или мёртвый" корректируется на каждый запрос, включая ping-запросы.

Изначально вероятности равны. На каждом шаге они масштабируются обратными значениями задержек, наблюдаемых в течение последнего периода кармы, а затем нормализуются заново. Например, если в первые 60 секунд после запуска мастера 4 зеркала показали задержки 10 мс, 5 мс, 30 мс и 3 мс соответственно, то первый шаг корректировки будет следующим:

1. Исходные проценты: 0.25, 0.25, 0.25, 0.25.
2. Наблюдаемые задержки: 10 мс, 5 мс, 30 мс, 3 мс.
3. Обратные значения задержек: 0.1, 0.2, 0.0333, 0.333.
4. Масштабированные проценты: 0.025, 0.05, 0.008333, 0.0833.
5. Нормализованные проценты: 0.15, 0.30, 0.05, 0.50.

Это означает, что первое зеркало будет иметь 15% шанс быть выбранным в следующий период кармы, второе — 30%, третье (самое медленное, с задержкой 30 мс) — только 5%, а четвёртое и самое быстрое (3 мс) — 50%. После этого периода второй шаг корректировки снова обновит эти шансы и так далее.

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

ha_strategy = nodeads

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='nodeads';

ha_strategy = noerrors

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='noerrors';

ha_strategy = roundrobin

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

Предположим, что зеркальные распределенные таблицы уже существуют, например:

CREATE TABLE products_random type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products';
CREATE TABLE products_rr type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

SELECT node FROM products_random;
SELECT node FROM products_random;
SELECT node FROM products_rr;
SELECT node FROM products_rr;

+-------+
| node  |
+-------+
| node2 |
+-------+
+-------+
| node  |
+-------+
| node1 |
+-------+
+-------+
| node  |
+-------+
| node1 |
+-------+
+-------+
| node  |
+-------+
| node2 |
+-------+

SHOW CREATE TABLE products_rr;
SHOW AGENT STATUS;
SHOW AGENT STATUS LIKE '%errorsarow%';
SHOW AGENT STATUS LIKE '%1periods_%';

+---------------------------------+-------+
| Key                             | Value |
+---------------------------------+-------+
| ag_0_pingtripmsec               | 0.233 |
| ag_0_errorsarow                 | 0     |
| ag_0_1periods_network_errors    | 0     |
| ag_0_1periods_warnings          | 0     |
| ag_0_1periods_succeeded_queries | 11    |
| ag_0_1periods_msecsperquery     | n/a   |
+---------------------------------+-------+

Практическая интерпретация:

• Если errorsarow превышает 3, nodeads начинает считать это зеркало неработоспособным.
• Если у зеркала нет недавних succeeded_queries, noerrors полностью пропустит его.
• Если у нескольких зеркал низкий коэффициент ошибок, noerrors все равно предпочтет те, у которых меньше задержка, потому что обычная перебалансировка весов остается в силе.
• pingtripmsec помогает отличить "доступные, но медленные" зеркала от "сбойных".

Опции, связанные с зеркалированием, таймаутами и повторными попытками, поддерживаются не во всех областях видимости одновременно. Эта страница фокусируется на том, как они влияют на балансировку нагрузки. Полный синтаксис и поведение на уровне демона смотрите на связанных справочных страницах.

ha_strategy = random
table products_rr {
  type = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products
  ha_strategy = roundrobin
}

CREATE TABLE products_rr type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

ha_period_karma = 2m

ha_ping_interval = 3s

agent_connect_timeout = 300ms

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  agent_connect_timeout='300ms';

agent_query_timeout = 500ms

SELECT * FROM products_dist OPTION agent_query_timeout=750;

table products_dist {
  type = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products[retry_count=2]
}

SELECT * FROM products_dist OPTION retry_count=1;

agent_retry_delay = 500ms

SELECT * FROM products_dist OPTION retry_count=2, retry_delay=300;

Агенты реплики могут использоваться взаимозаменяемо при обработке поискового запроса. Экземпляр Manticore, на котором размещена распределённая таблица с определёнными реплицированными агентами, отслеживает статус реплик (работают или нет) и время отклика, а также выполняет автоматическое переключение при сбое и балансировку нагрузки на основе этой информации.

agent = node1|node2|node3:9312:shard2

В приведённом выше примере объявляется, что node1:9312, node2:9312 и node3:9312 все имеют таблицу с именем shard2 и могут использоваться как взаимозаменяемые реплики. Если любой из этих серверов выйдет из строя, запросы будут распределены между оставшимися. Когда сервер вернётся в онлайн, мастер обнаружит это и снова начнёт направлять запросы ко всем репликам.

Зеркало также может включать индивидуальный список таблиц, как показано ниже:

agent = node1:9312:node1shard2|node2:9312:node2shard2

Это работает аналогично предыдущему примеру, но при запросе к разным серверам будут использоваться разные имена таблиц. Например, node1shard2 будет использоваться при запросе к node1:9312, а node2shard2 — при запросе к node2:9312.

По умолчанию выбор реплики использует глобальную или табличную стратегию ha_strategy. Если вы не зададите её явно, стратегия по умолчанию — random. Мастер хранит метрики, такие как общее количество запросов, количество ошибок и время отклика для каждого агента, и группирует их во временные интервалы, контролируемые параметром ha_period_karma. Затем эти статистические данные используются стратегиями балансировки, описанными в разделе Балансировка нагрузки.

Период кармы указывается в секундах и по умолчанию равен 60 секундам. Мастер хранит до 15 интервалов кармы со статистикой по агентам для целей инструментирования (см. SHOW AGENT STATUS). Однако для логики HA/LB используются только последние два интервала.

Когда запросов нет, мастер отправляет регулярную команду ping каждые ha_ping_interval для сбора статистики и проверки, жив ли ещё удалённый хост. ha_ping_interval по умолчанию равен 1000 мс. Установка значения 0 отключает пинги, и статистика будет накапливаться только на основе фактических запросов. Вместе с ha_period_karma это управляет тем, насколько быстро изменения в состоянии и задержке реплик влияют на балансировку нагрузки.

Это различие важно:

• Одна запись agent='host1|host2:table' означает один удалённый шард с реплицированными бэкендами.
• Несколько записей agent='...' означают несколько удалённых шардов.

Например:

# one shard, two mirrors
agent = node1|node2:9312:products
# two shards
agent = node1:9312:products_a
agent = node2:9312:products_b

table products_dist {
  type = distributed
  agent = 127.0.0.1:9312|127.0.0.1:9313:products
  ha_strategy = roundrobin
}

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

CREATE TABLE products_dist type='distributed'
  agent='127.0.0.1:9312:products|127.0.0.1:9313:products'
  ha_strategy='roundrobin';

SELECT id, title, node FROM products_dist;
SHOW CREATE TABLE products_dist;
SHOW AGENT STATUS;

+------+------------+-------+
| id   | title      | node  |
+------+------------+-------+
|    1 | same title | node1 |
+------+------------+-------+
+---------------+----------------------------------------------------------------------------------+
| Table         | Create Table                                                                     |
+---------------+----------------------------------------------------------------------------------+
| products_dist | CREATE TABLE products_dist type='distributed' agent='127.0.0.1:9312:products|... |
+---------------+----------------------------------------------------------------------------------+

# node1, node2 carry shard1 as local
# node3, node4 carry shard2 as local
# config on node1, node2
agent = node3:9312|node4:9312:shard2
# config on node3, node4
agent = node1:9312|node2:9312:shard1

С Manticore транзакции записи (такие как INSERT, REPLACE, DELETE, TRUNCATE, UPDATE, COMMIT) могут реплицироваться на другие узлы кластера до полного применения транзакции на текущем узле. В настоящее время репликация поддерживается для таблиц percolate, rt и distributed в Linux и macOS.

Нативные бинарные файлы Windows для Manticore не поддерживают репликацию. Мы рекомендуем устанавливать Manticore через WSL (Подсистема Windows для Linux).

На macOS репликация имеет ограниченную поддержку и рекомендуется только для целей разработки.

Репликация Manticore базируется на библиотеке Galera и обладает несколькими впечатляющими возможностями:

• Истинный multi-master: чтение и запись на любой узел в любое время.
• Почти синхронная репликация — отсутствие задержек ведомых и потери данных после сбоя узла.
• Горячий резерв: отсутствие времени простоя при переключении (так как переключения нет).
• Плотно связанные узлы: все узлы содержат одно и то же состояние, расхождений данных между узлами не допускается.
• Автоматическое добавление узлов: нет необходимости вручную создавать резервные копии базы и восстанавливать их на новом узле.
• Простота использования и развёртывания.
• Обнаружение и автоматическое удаление ненадёжных узлов.
• Репликация на основе сертификации.

Для настройки репликации в Manticore Search:

• Опция data_dir должна быть задана в разделе "searchd" конфигурационного файла. Репликация не поддерживается в plain-режиме.
• Директива listen должна содержать IP-адрес, доступный для других узлов, либо должен быть указан node_address с доступным IP-адресом.
• По желанию, можно задать уникальные значения для server_id на каждом узле кластера. Если значение не задано, узел попытается использовать MAC-адрес или случайное число для генерации server_id.

Если включены аутентификация и авторизация, для операций с кластером требуется разрешение replication. Выдайте его пользователю, который должен владеть операциями репликации для кластера:

GRANT replication ON 'posts' TO 'repl_user';

CREATE CLUSTER и JOIN CLUSTER могут указывать такого пользователя через '<user>' AS user. Пользователь должен существовать с совпадающими сохраненными данными аутентификации на узлах, участвующих в операции. Просто создать одно и то же имя пользователя и пароль независимо на каждом узле недостаточно, потому что сохраненные данные аутентификации могут отличаться. Если вы изменили данные аутентификации вне демона, выполните RELOAD AUTH на затронутых узлах перед использованием операции с кластером.

Позднее ALTER CLUSTER ... ADD, ALTER CLUSTER ... DROP, ALTER CLUSTER ... UPDATE nodes и DELETE CLUSTER используют сохраненного пользовательского кластера. Когда аутентификация включена, успешный JOIN CLUSTER заменяет все локальные данные аутентификации на присоединяющемся узле данными аутентификации кластера-источника.

Если директива replication listen не задана, Manticore использует первые два свободных порта из диапазона в 200 портов после порта прослушивания протокола по умолчанию для каждого созданного кластера. Для ручного задания портов репликации необходимо определить диапазон портов в директиве listen (типа replication), при этом пары адрес/диапазон портов не должны пересекаться между разными узлами на одном сервере. Как правило, диапазон портов должен содержать как минимум два порта на кластер. Когда вы определяете слушатель репликации с диапазоном портов (например, listen = 192.168.0.1:9320-9328:replication), Manticore не начинает прослушивать эти порты сразу. Вместо этого система будет выбирать случайные свободные порты из указанного диапазона только при начале использования репликации.

Репликационный кластер — это группа узлов, в которой реплицируются транзакции записи. Репликация настраивается на уровне таблицы, то есть одна таблица может принадлежать только одному кластеру. Нет ограничений на количество таблиц в кластере. Все операции INSERT, REPLACE, DELETE, TRUNCATE для любой percolate или real-time таблицы, входящей в кластер, реплицируются на все другие узлы этого кластера. В репликацию также могут быть включены distributed таблицы. Репликация является multi-master, поэтому записи на любой узел или несколько узлов одновременно работают корректно.

Для создания кластера обычно используется команда create cluster с CREATE CLUSTER <имя кластера>, а для присоединения к кластеру — команда join cluster с JOIN CLUSTER <имя кластера> at 'host:port'. Однако в некоторых редких случаях может понадобиться тонкая настройка поведения CREATE/JOIN CLUSTER. Доступные опции:

Эта опция задаёт имя кластера. Оно должно быть уникальным среди всех кластеров в системе.

Примечание: Максимальная длина имени хоста для команды JOIN составляет 253 символа. Если предел превышен, searchd выдаст ошибку.

Параметр path задаёт каталог данных для репликации кэша write-set и других файлов провайдера кластера. Он не влияет на место хранения реплицируемых таблиц. Входящие реплицируемые таблицы сохраняются в обычном каталоге таблицы внутри data_dir. Это значение должно быть уникальным среди всех кластеров в системе и задаваться как относительный путь к каталогу data_dir. По умолчанию оно равно значению data_dir.

Критическое изменение: В более старых версиях входящие файлы реплицируемых таблиц тоже сохранялись по пути кластера. Если вы использовали собственный path кластера, обновляйтесь осторожно, потому что реплицируемые таблицы, полученные старыми версиями, может потребоваться переместить или повторно синхронизировать в обычную структуру data_dir/<table>.

Опция nodes — это список адресов и портов всех узлов в кластере, разделённых запятыми. Этот список должен быть получен через API узла и может включать адрес текущего узла. Он используется для присоединения узла к кластеру и для повторного присоединения после перезапуска.

Опция options позволяет передавать дополнительные параметры непосредственно плагину репликации Galera, как описано в Galera Documentation Parameters

INSERT INTO posts:weekly_table VALUES ( 'iphone case' )
TRUNCATE TABLE click_query:weekly_table
UPDATE INTO posts:rt_tags SET tags=(101, 302, 304) WHERE MATCH ('use') AND id IN (1,101,201)
DELETE FROM clicks:rt WHERE MATCH ('dumy') AND gid>206

POST /insert -d '
{
  "cluster":"posts",
  "table":"weekly_table",
  "doc":
  {
    "title" : "iphone case",
    "price" : 19.85
  }
}'
POST /delete -d '
{
  "cluster":"posts",
  "table": "weekly_table",
  "id":1
}'

$table->addDocuments([
        1, ['title' => 'iphone case', 'price' => 19.85]
]);
$table->deleteDocument(1);

indexApi.insert({"cluster":"posts","table":"weekly_table","doc":{"title":"iphone case","price":19.85}})
indexApi.delete({"cluster":"posts","table":"weekly_table","id":1})

await indexApi.insert({"cluster":"posts","table":"weekly_table","doc":{"title":"iphone case","price":19.85}})
await indexApi.delete({"cluster":"posts","table":"weekly_table","id":1})

res = await indexApi.insert({"cluster":"posts","table":"weekly_table","doc":{"title":"iphone case","price":19.85}});
 res = await indexApi.delete({"cluster":"posts","table":"weekly_table","id":1});

InsertDocumentRequest newdoc = new InsertDocumentRequest();
HashMap<String,Object> doc = new HashMap<String,Object>(){{
    put("title","Crossbody Bag with Tassel");
    put("price",19.85);
}};
newdoc.table("weekly_table").cluster("posts").id(1L).setDoc(doc);
sqlresult = indexApi.insert(newdoc);
DeleteDocumentRequest deleteRequest = new DeleteDocumentRequest();
deleteRequest.table("weekly_table").cluster("posts").setId(1L);
indexApi.delete(deleteRequest);

Dictionary<string, Object> doc = new Dictionary<string, Object>();
doc.Add("title", "Crossbody Bag with Tassel");
doc.Add("price", 19.85);
InsertDocumentRequest newdoc = new InsertDocumentRequest(table: "weekly_table", cluster:posts, id: 1, doc: doc);
var sqlresult = indexApi.Insert(newdoc);
DeleteDocumentRequest deleteDocumentRequest = new DeleteDocumentRequest(table: "weekly_table", cluster: "posts", id: 1);
indexApi.Delete(deleteDocumentRequest);

let mut doc = HashMap::new();
doc.insert("title".to_string(), serde_json::json!("Crossbody Bag with Tassel"));
doc.insert("price".to_string(), serde_json::json!(19.85));
let insert_req = InsertDocumentRequest {
    table: serde_json::json!("weekly_table"),
    doc: serde_json::json!(doc),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(1),
};
let insert_res = index_api.insert(insert_req).await;
let delete_req = DeleteDocumentRequest {
    table: serde_json::json!("weekly_table"),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(1),
};
index_api.delete(delete_req).await;

SELECT * FROM weekly_table
CALL PQ('posts:weekly_table', 'document is here')

POST /search -d '
{
  "cluster":"posts",
  "table":"weekly_table",
  "query":{"match":{"title":"keyword"}}
}'
POST /search -d '
{
  "table":"weekly_table",
  "query":{"match":{"title":"keyword"}}
}'

SET CLUSTER click_query GLOBAL 'pc.bootstrap' = 1

POST /cli -d "
SET CLUSTER click_query GLOBAL 'pc.bootstrap' = 1
"

SET CLUSTER posts GLOBAL 'pc.bootstrap' = 1

POST /cli -d "
SET CLUSTER posts GLOBAL 'pc.bootstrap' = 1
"

searchd {
  listen   = 9312
  listen   = 192.168.1.101:9360-9370:replication
  data_dir = /var/lib/manticore/
  ...
 }

CREATE CLUSTER posts

POST /cli -d "
CREATE CLUSTER posts
"

$params = [
    'cluster' => 'posts'
    ]
];
$response = $client->cluster()->create($params);

utilsApi.sql('CREATE CLUSTER posts')

await utilsApi.sql('CREATE CLUSTER posts')

res = await utilsApi.sql('CREATE CLUSTER posts');

utilsApi.sql("CREATE CLUSTER posts");

utilsApi.Sql("CREATE CLUSTER posts");

utils_api.sql("CREATE CLUSTER posts", Some(true)).await;

ALTER CLUSTER posts ADD pq_title
ALTER CLUSTER posts ADD pq_clicks

POST /cli -d "
ALTER CLUSTER posts ADD pq_title
"
POST /cli -d "
ALTER CLUSTER posts ADD pq_clicks
"

$params = [
  'cluster' => 'posts',
  'body' => [
     'operation' => 'add',
     'table' => 'pq_title'
  ]
];
$response = $client->cluster()->alter($params);
$params = [
  'cluster' => 'posts',
  'body' => [
     'operation' => 'add',
     'table' => 'pq_clicks'
  ]
];
$response = $client->cluster()->alter($params);

utilsApi.sql('ALTER CLUSTER posts ADD pq_title')
utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks')

await utilsApi.sql('ALTER CLUSTER posts ADD pq_title')
await utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks')

res = await utilsApi.sql('ALTER CLUSTER posts ADD pq_title');
res = await utilsApi.sql('ALTER CLUSTER posts ADD pq_clicks');

utilsApi.sql("ALTER CLUSTER posts ADD pq_title");
utilsApi.sql("ALTER CLUSTER posts ADD pq_clicks");

utilsApi.Sql("ALTER CLUSTER posts ADD pq_title");
utilsApi.Sql("ALTER CLUSTER posts ADD pq_clicks");

utils_api.sql("ALTER CLUSTER posts ADD pq_title", Some(true)).await;
utils_api.sql("ALTER CLUSTER posts ADD pq_clicks", Some(true)).await;

JOIN CLUSTER posts AT '192.168.1.101:9312'

POST /cli -d "
JOIN CLUSTER posts AT '192.168.1.101:9312'
"

$params = [
  'cluster' => 'posts',
  'body' => [
      '192.168.1.101:9312'
  ]
];
$response = $client->cluster->join($params);

utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'')

await utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'')

res = await utilsApi.sql('JOIN CLUSTER posts AT \'192.168.1.101:9312\'');

utilsApi.sql("JOIN CLUSTER posts AT '192.168.1.101:9312'");

utilsApi.Sql("JOIN CLUSTER posts AT '192.168.1.101:9312'");

utils_api.sql("JOIN CLUSTER posts AT '192.168.1.101:9312'", Some(true)).await;

INSERT INTO posts:pq_title VALUES ( 3, 'test me' )

POST /insert -d '
{
  "cluster":"posts",
  "table":"pq_title",
  "id": 3
  "doc":
  {
    "title" : "test me"
  }
}'

$table->addDocuments([
        3, ['title' => 'test me']
]);

indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}})

await indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}})

res = await indexApi.insert({"cluster":"posts","table":"pq_title","id":3"doc":{"title":"test me"}});

InsertDocumentRequest newdoc = new InsertDocumentRequest();
HashMap<String,Object> doc = new HashMap<String,Object>(){{
    put("title","test me");
}};
newdoc.table("pq_title").cluster("posts").id(3L).setDoc(doc);
sqlresult = indexApi.insert(newdoc);

Dictionary<string, Object> doc = new Dictionary<string, Object>();
doc.Add("title", "test me");
InsertDocumentRequest newdoc = new InsertDocumentRequest(table: "pq_title", cluster: "posts", id: 3, doc: doc);
var sqlresult = indexApi.Insert(newdoc);

let mut doc = HashMap::new();
doc.insert("title".to_string(), serde_json::json!("test me"));
let insert_req = InsertDocumentRequest {
    table: serde_json::json!("pq_title"),
    doc: serde_json::json!(doc),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(3),
};
let insert_res = index_api.insert(insert_req).await;

Все запросы, которые изменяют таблицы в кластере, теперь реплицируются на все узлы в кластере.