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

agent = node1|node2|node3:9312:shard2

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

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

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

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

По умолчанию все запросы направляются к лучшему из зеркал. Лучшее зеркало выбирается на основе последних статистических данных, контролируемых директивой конфигурации ha_period_karma. Основной узел хранит метрики (общее количество запросов, количество ошибок, время отклика и т.д.) для каждого агента и группирует их по временным интервалам. Карма — это длина временного интервала. Лучшее зеркало агента затем определяется динамически на основе последних двух таких временных интервалов. Конкретный алгоритм выбора зеркала можно настроить с помощью директивы ha_strategy.

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

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

Пример:

# sharding table over 4 servers total
# in just 2 shards but with 2 failover mirrors for each shard
# 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

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

ha_strategy = {random|nodeads|noerrors|roundrobin}

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

Стратегия, используемая для выбора зеркала, или другими словами, выбора конкретного зеркального агента в распределённой таблице, контролируется этой директивой. По сути, эта директива управляет тем, как мастер выполняет балансировку нагрузки между настроенными зеркальными агентскими узлами. Реализованы следующие стратегии:

ha_strategy = random

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

Адаптивные стратегии, основанные на вероятностях, взвешенных по задержкам, работают следующим образом:

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

ha_strategy = noerrors

ha_strategy = roundrobin

ha_period_karma = 2m

ha_period_karma определяет размер окна статистики агентских зеркал, в секундах (или с суффиксом времени). Опционально, по умолчанию 60.

Для распределённой таблицы с агентскими зеркалами сервер отслеживает несколько различных счётчиков для каждого зеркала. Эти счётчики затем используются для фейловера и балансировки. (Сервер выбирает лучшее зеркало на основе этих счётчиков.) Счётчики накапливаются блоками по ha_period_karma секунд.

После начала нового блока мастер всё ещё может использовать накопленные значения из предыдущего блока, пока новый блок не заполнится наполовину. Таким образом, любое предыдущая история перестаёт влиять на выбор зеркала спустя максимум 1.5 периода ha_period_karma.

Хотя для выбора зеркала используется не более 2 блоков, для целей инструментирования фактически хранится до 15 последних блоков. Их можно просмотреть с помощью оператора SHOW AGENT STATUS.

ha_ping_interval = 3s

Директива ha_ping_interval задаёт интервал между ping-запросами, отправляемыми агентским зеркалам, в миллисекундах (или с суффиксом времени). Эта опция является необязательной, по умолчанию значение равно 1000.

Для распределённой таблицы с агентскими зеркалами сервер шлёт всем зеркалам команду ping в периоды простоя, чтобы отслеживать их текущий статус (живы они или мертвы, время сетевого отклика и т. п.). Интервал между ping-запросами определяется настройкой ha_ping_interval.

Если вы хотите отключить ping-запросы, установите ha_ping_interval в 0.

С 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 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 задаёт каталог данных для кэша набора записей для репликации и для входящих таблиц с других узлов. Это значение должно быть уникальным среди всех кластеров системы и указываться как относительный путь к каталогу data_dir. По умолчанию оно совпадает со значением data_dir.

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

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

INSERT INTO posts:weekly_index VALUES ( 'iphone case' )
TRUNCATE RTINDEX click_query:weekly_index
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_index",
  "doc":
  {
    "title" : "iphone case",
    "price" : 19.85
  }
}'
POST /delete -d '
{
  "cluster":"posts",
  "table": "weekly_index",
  "id":1
}'

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

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

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

res = await indexApi.insert({"cluster":"posts","table":"weekly_index","doc":{"title":"iphone case","price":19.85}});
 res = await indexApi.delete({"cluster":"posts","table":"weekly_index","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.index("weekly_index").cluster("posts").id(1L).setDoc(doc);
sqlresult = indexApi.insert(newdoc);
DeleteDocumentRequest deleteRequest = new DeleteDocumentRequest();
deleteRequest.index("weekly_index").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_index", cluster:posts, id: 1, doc: doc);
var sqlresult = indexApi.Insert(newdoc);
DeleteDocumentRequest deleteDocumentRequest = new DeleteDocumentRequest(table: "weekly_index", 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_index"),
    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_index"),
    cluster: serde_json::json!("posts"),
    id: serde_json::json!(1),
};
index_api.delete(delete_req).await;

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

POST /search -d '
{
  "cluster":"posts",
  "table":"weekly_index",
  "query":{"match":{"title":"keyword"}}
}'
POST /search -d '
{
  "table":"weekly_index",
  "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"
  }
}'

$index->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.index("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(index: "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;

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

CREATE CLUSTER posts
CREATE CLUSTER click_query '/var/data/click_query/' as path
CREATE CLUSTER click_query '/var/data/click_query/' as path, 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312' as nodes

POST /cli -d "
CREATE CLUSTER posts
"
POST /cli -d "
CREATE CLUSTER click_query '/var/data/click_query/' as path
"
POST /cli -d "
CREATE CLUSTER click_query '/var/data/click_query/' as path, 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312' as nodes
"

$params = [
    'cluster' => 'posts',
    ]
];
$response = $client->cluster()->create($params);
$params = [
    'cluster' => 'click_query',
    'body' => [
        'path' => '/var/data/click_query/'
    ]
    ]
];
$response = $client->cluster()->create($params);
$params = [
    'cluster' => 'click_query',
    'body' => [
        'path' => '/var/data/click_query/',
        'nodes' => 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312'
    ]
    ]
];
$response = $client->cluster()->create($params);

utilsApi.sql('CREATE CLUSTER posts')
utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path')
utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path, \'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312\' as nodes')

await utilsApi.sql('CREATE CLUSTER posts')
await utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path')
await utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path, \'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312\' as nodes')

res = await utilsApi.sql('CREATE CLUSTER posts');
res = await utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path');
res = await utilsApi.sql('CREATE CLUSTER click_query \'/var/data/click_query/\' as path, \'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312\' as nodes');

utilsApi.sql("CREATE CLUSTER posts");
utilsApi.sql("CREATE CLUSTER click_query '/var/data/click_query/' as path");
utilsApi.sql("CREATE CLUSTER click_query '/var/data/click_query/' as path, 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312' as nodes");

utilsApi.Sql("CREATE CLUSTER posts");
utilsApi.Sql("CREATE CLUSTER click_query '/var/data/click_query/' as path");
utilsApi.Sql("CREATE CLUSTER click_query '/var/data/click_query/' as path, 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312' as nodes");

utils_api.sql("CREATE CLUSTER posts", Some(true)).await;
utils_api.sql("CREATE CLUSTER click_query '/var/data/click_query/' as path", Some(true)).await;
utils_api.sql("CREATE CLUSTER click_query '/var/data/click_query/' as path, 'clicks_mirror1:9312,clicks_mirror2:9312,clicks_mirror3:9312' as nodes", Some(true)).await;

Если опция nodes не указана при создании кластера, первый узел, присоединившийся к кластеру, будет сохранен как опция nodes.