[CCR] Add endpoints for fetching and creating follower indices (#27646)

* Add GET /follower_indices endpoint with deserialization logic and tests.

* Add POST /follower_indices endpoint with serialization logic and tests.

x-pack/plugins/cross_cluster_replication/fixtures/follower_index.js

@@ -0,0 +1,116 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+const Chance = require('chance'); // eslint-disable-line import/no-extraneous-dependencies
+const chance = new Chance();
+
+export const getFollowerIndexMock = (
+  name = chance.string(),
+  shards = [{
+    id: chance.string(),
+    remoteCluster: chance.string(),
+    leaderIndex: chance.string(),
+    leaderGlobalCheckpoint: chance.integer(),
+    leaderMaxSequenceNum: chance.integer(),
+    followerGlobalCheckpoint: chance.integer(),
+    followerMaxSequenceNum: chance.integer(),
+    lastRequestedSequenceNum: chance.integer(),
+    outstandingReadRequestsCount: chance.integer(),
+    outstandingWriteRequestsCount: chance.integer(),
+    writeBufferOperationsCount: chance.integer(),
+    writeBufferSizeBytes: chance.integer(),
+    followerMappingVersion: chance.integer(),
+    followerSettingsVersion: chance.integer(),
+    totalReadTimeMs: chance.integer(),
+    totalReadRemoteExecTimeMs: chance.integer(),
+    successfulReadRequestCount: chance.integer(),
+    failedReadRequestsCount: chance.integer(),
+    operationsReadCount: chance.integer(),
+    bytesReadCount: chance.integer(),
+    totalWriteTimeMs: chance.integer(),
+    successfulWriteRequestsCount: chance.integer(),
+    failedWriteRequestsCount: chance.integer(),
+    operationsWrittenCount: chance.integer(),
+    readExceptions: [ chance.string() ],
+    timeSinceLastReadMs: chance.integer(),
+  }]
+) => {
+  const serializeShard = ({
+    id,
+    remoteCluster,
+    leaderIndex,
+    leaderGlobalCheckpoint,
+    leaderMaxSequenceNum,
+    followerGlobalCheckpoint,
+    followerMaxSequenceNum,
+    lastRequestedSequenceNum,
+    outstandingReadRequestsCount,
+    outstandingWriteRequestsCount,
+    writeBufferOperationsCount,
+    writeBufferSizeBytes,
+    followerMappingVersion,
+    followerSettingsVersion,
+    totalReadTimeMs,
+    totalReadRemoteExecTimeMs,
+    successfulReadRequestCount,
+    failedReadRequestsCount,
+    operationsReadCount,
+    bytesReadCount,
+    totalWriteTimeMs,
+    successfulWriteRequestsCount,
+    failedWriteRequestsCount,
+    operationsWrittenCount,
+    readExceptions,
+    timeSinceLastReadMs,
+  }) => ({
+    shard_id: id,
+    remote_cluster: remoteCluster,
+    leader_index: leaderIndex,
+    leader_global_checkpoint: leaderGlobalCheckpoint,
+    leader_max_seq_no: leaderMaxSequenceNum,
+    follower_global_checkpoint: followerGlobalCheckpoint,
+    follower_max_seq_no: followerMaxSequenceNum,
+    last_requested_seq_no: lastRequestedSequenceNum,
+    outstanding_read_requests: outstandingReadRequestsCount,
+    outstanding_write_requests: outstandingWriteRequestsCount,
+    write_buffer_operation_count: writeBufferOperationsCount,
+    write_buffer_size_in_bytes: writeBufferSizeBytes,
+    follower_mapping_version: followerMappingVersion,
+    follower_settings_version: followerSettingsVersion,
+    total_read_time_millis: totalReadTimeMs,
+    total_read_remote_exec_time_millis: totalReadRemoteExecTimeMs,
+    successful_read_requests: successfulReadRequestCount,
+    failed_read_requests: failedReadRequestsCount,
+    operations_read: operationsReadCount,
+    bytes_read: bytesReadCount,
+    total_write_time_millis: totalWriteTimeMs,
+    successful_write_requests: successfulWriteRequestsCount,
+    failed_write_requests: failedWriteRequestsCount,
+    operations_written: operationsWrittenCount,
+    read_exceptions: readExceptions,
+    time_since_last_read_millis: timeSinceLastReadMs,
+  });
+
+  return {
+    index: name,
+    shards: shards.map(serializeShard),
+  };
+};
+
+export const getFollowerIndexListMock = (total = 3) => {
+  const list = {
+    follow_stats: {
+      indices: [],
+    },
+  };
+
+  let i = total;
+  while(i--) {
+    list.follow_stats.indices.push(getFollowerIndexMock());
+  }
+
+  return list;
+};

x-pack/plugins/cross_cluster_replication/fixtures/index.js

@@ -10,3 +10,8 @@ export {
 } from './auto_follow_pattern';
 
 export { esErrors } from './es_errors';
+
+export {
+  getFollowerIndexMock,
+  getFollowerIndexListMock,
+} from './follower_index';

x-pack/plugins/cross_cluster_replication/server/client/elasticsearch_ccr.js

@@ -62,4 +62,28 @@ export const elasticsearchJsPlugin = (Client, config, components) => {
     needBody: true,
     method: 'DELETE'
   });
+
+  ccr.followerIndices = ca({
+    urls: [
+      {
+        fmt: '/_ccr/stats',
+      }
+    ],
+    method: 'GET'
+  });
+
+  ccr.saveFollowerIndex = ca({
+    urls: [
+      {
+        fmt: '/<%=name%>/_ccr/follow',
+        req: {
+          name: {
+            type: 'string'
+          }
+        }
+      }
+    ],
+    needBody: true,
+    method: 'PUT'
+  });
 };

x-pack/plugins/cross_cluster_replication/server/lib/follower_index_serialization.js

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+/* eslint-disable camelcase */
+export const deserializeShard = ({
+  remote_cluster,
+  leader_index,
+  shard_id,
+  leader_global_checkpoint,
+  leader_max_seq_no,
+  follower_global_checkpoint,
+  follower_max_seq_no,
+  last_requested_seq_no,
+  outstanding_read_requests,
+  outstanding_write_requests,
+  write_buffer_operation_count,
+  write_buffer_size_in_bytes,
+  follower_mapping_version,
+  follower_settings_version,
+  total_read_time_millis,
+  total_read_remote_exec_time_millis,
+  successful_read_requests,
+  failed_read_requests,
+  operations_read,
+  bytes_read,
+  total_write_time_millis,
+  successful_write_requests,
+  failed_write_requests,
+  operations_written,
+  read_exceptions,
+  time_since_last_read_millis,
+}) => ({
+  id: shard_id,
+  remoteCluster: remote_cluster,
+  leaderIndex: leader_index,
+  leaderGlobalCheckpoint: leader_global_checkpoint,
+  leaderMaxSequenceNum: leader_max_seq_no,
+  followerGlobalCheckpoint: follower_global_checkpoint,
+  followerMaxSequenceNum: follower_max_seq_no,
+  lastRequestedSequenceNum: last_requested_seq_no,
+  outstandingReadRequestsCount: outstanding_read_requests,
+  outstandingWriteRequestsCount: outstanding_write_requests,
+  writeBufferOperationsCount: write_buffer_operation_count,
+  writeBufferSizeBytes: write_buffer_size_in_bytes,
+  followerMappingVersion: follower_mapping_version,
+  followerSettingsVersion: follower_settings_version,
+  totalReadTimeMs: total_read_time_millis,
+  totalReadRemoteExecTimeMs: total_read_remote_exec_time_millis,
+  successfulReadRequestCount: successful_read_requests,
+  failedReadRequestsCount: failed_read_requests,
+  operationsReadCount: operations_read,
+  bytesReadCount: bytes_read,
+  totalWriteTimeMs: total_write_time_millis,
+  successfulWriteRequestsCount: successful_write_requests,
+  failedWriteRequestsCount: failed_write_requests,
+  operationsWrittenCount: operations_written,
+  // This is an array of exception objects
+  readExceptions: read_exceptions,
+  timeSinceLastReadMs: time_since_last_read_millis,
+});
+/* eslint-enable camelcase */
+
+export const deserializeFollowerIndex = ({ index, shards }) => ({
+  name: index,
+  shards: shards.map(deserializeShard),
+});
+
+export const deserializeListFollowerIndices = followerIndices =>
+  followerIndices.map(deserializeFollowerIndex);
+
+export const serializeFollowerIndex = ({ remoteCluster, leaderIndex }) => ({
+  remote_cluster: remoteCluster,
+  leader_index: leaderIndex,
+});