[CCR] Add auto follow stats api (#33801)

GET /_ccr/auto_follow/stats

Returns:

```
{
   "number_of_successful_follow_indices": ...
   "number_of_failed_follow_indices": ...
   "number_of_failed_remote_cluster_state_requests": ...
   "recent_auto_follow_errors": [
      ...
   ]
}
```

Relates to #33007

Author: martijnvg

x-pack/plugin/ccr/qa/multi-cluster/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexIT.java

@@ -118,6 +118,10 @@ public void testAutoFollowPatterns() throws Exception {
         }
 
         assertBusy(() -> {
+            Request statsRequest = new Request("GET", "/_ccr/auto_follow/stats");
+            Map<String, ?> response = toMap(client().performRequest(statsRequest));
+            assertThat(response.get("number_of_successful_follow_indices"), equalTo(1));
+
             ensureYellow("logs-20190101");
             verifyDocuments("logs-20190101", 5);
         });

x-pack/plugin/ccr/qa/rest/src/test/resources/rest-api-spec/test/ccr/auto_follow_stats.yml

@@ -0,0 +1,10 @@
+---
+"Test autofollow stats":
+  - do:
+      ccr.auto_follow_stats: {}
+
+  - match: { number_of_successful_follow_indices: 0 }
+  - match: { number_of_failed_follow_indices: 0 }
+  - match: { number_of_failed_remote_cluster_state_requests: 0 }
+  - length: { recent_auto_follow_errors: 0 }
+

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java

@@ -40,6 +40,9 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.ccr.action.AutoFollowCoordinator;
+import org.elasticsearch.xpack.ccr.action.TransportAutoFollowStatsAction;
+import org.elasticsearch.xpack.ccr.rest.RestAutoFollowStatsAction;
+import org.elasticsearch.xpack.core.ccr.action.AutoFollowStatsAction;
 import org.elasticsearch.xpack.core.ccr.action.DeleteAutoFollowPatternAction;
 import org.elasticsearch.xpack.core.ccr.action.PutAutoFollowPatternAction;
 import org.elasticsearch.xpack.ccr.action.ShardChangesAction;
@@ -153,6 +156,7 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(ClusterServic
                 new ActionHandler<>(ShardChangesAction.INSTANCE, ShardChangesAction.TransportAction.class),
                 // stats action
                 new ActionHandler<>(CcrStatsAction.INSTANCE, TransportCcrStatsAction.class),
+                new ActionHandler<>(AutoFollowStatsAction.INSTANCE, TransportAutoFollowStatsAction.class),
                 // follow actions
                 new ActionHandler<>(CreateAndFollowIndexAction.INSTANCE, TransportCreateAndFollowIndexAction.class),
                 new ActionHandler<>(FollowIndexAction.INSTANCE, TransportFollowIndexAction.class),
@@ -173,6 +177,7 @@ public List<RestHandler> getRestHandlers(Settings settings, RestController restC
         return Arrays.asList(
                 // stats API
                 new RestCcrStatsAction(settings, restController),
+                new RestAutoFollowStatsAction(settings, restController),
                 // follow APIs
                 new RestCreateAndFollowIndexAction(settings, restController),
                 new RestFollowIndexAction(settings, restController),

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/AutoFollowCoordinator.java

@@ -9,6 +9,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.client.Client;
@@ -114,7 +115,7 @@ synchronized void updateStats(List<AutoFollowResult> results) {
                     if (entry.getValue() != null) {
                         numberOfFailedIndicesAutoFollowed++;
                         recentAutoFollowErrors.put(result.clusterAlias + ":" + entry.getKey().getName(),
-                            new ElasticsearchException(entry.getValue()));
+                            ExceptionsHelper.convertToElastic(entry.getValue()));
                         LOGGER.warn(new ParameterizedMessage("failure occurred while auto following index [{}] in leader cluster [{}]",
                             entry.getKey(), result.clusterAlias), entry.getValue());
                     } else {

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportAutoFollowStatsAction.java

@@ -0,0 +1,94 @@
+/*
+ * 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.
+ */
+
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.LicenseUtils;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.Ccr;
+import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
+import org.elasticsearch.xpack.core.ccr.AutoFollowStats;
+import org.elasticsearch.xpack.core.ccr.action.AutoFollowStatsAction;
+
+import java.util.Objects;
+
+public class TransportAutoFollowStatsAction
+    extends TransportMasterNodeAction<AutoFollowStatsAction.Request, AutoFollowStatsAction.Response> {
+
+    private final CcrLicenseChecker ccrLicenseChecker;
+    private final AutoFollowCoordinator autoFollowCoordinator;
+
+    @Inject
+    public TransportAutoFollowStatsAction(
+            Settings settings,
+            TransportService transportService,
+            ClusterService clusterService,
+            ThreadPool threadPool,
+            ActionFilters actionFilters,
+            IndexNameExpressionResolver indexNameExpressionResolver,
+            AutoFollowCoordinator autoFollowCoordinator,
+            CcrLicenseChecker ccrLicenseChecker
+    ) {
+        super(
+            settings,
+            AutoFollowStatsAction.NAME,
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            AutoFollowStatsAction.Request::new,
+            indexNameExpressionResolver
+        );
+        this.ccrLicenseChecker = Objects.requireNonNull(ccrLicenseChecker);
+        this.autoFollowCoordinator = Objects.requireNonNull(autoFollowCoordinator);
+    }
+
+    @Override
+    protected String executor() {
+        return Ccr.CCR_THREAD_POOL_NAME;
+    }
+
+    @Override
+    protected AutoFollowStatsAction.Response newResponse() {
+        return new AutoFollowStatsAction.Response();
+    }
+
+    @Override
+    protected void doExecute(Task task, AutoFollowStatsAction.Request request, ActionListener<AutoFollowStatsAction.Response> listener) {
+        if (ccrLicenseChecker.isCcrAllowed() == false) {
+            listener.onFailure(LicenseUtils.newComplianceException("ccr"));
+            return;
+        }
+        super.doExecute(task, request, listener);
+    }
+
+    @Override
+    protected void masterOperation(
+            AutoFollowStatsAction.Request request,
+            ClusterState state,
+            ActionListener<AutoFollowStatsAction.Response> listener
+    ) throws Exception {
+        AutoFollowStats stats = autoFollowCoordinator.getStats();
+        listener.onResponse(new AutoFollowStatsAction.Response(stats));
+    }
+
+    @Override
+    protected ClusterBlockException checkBlock(AutoFollowStatsAction.Request request, ClusterState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
+    }
+}

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestAutoFollowStatsAction.java

@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+package org.elasticsearch.xpack.ccr.rest;
+
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestController;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.core.ccr.action.AutoFollowStatsAction;
+
+import java.io.IOException;
+
+public class RestAutoFollowStatsAction extends BaseRestHandler {
+
+    public RestAutoFollowStatsAction(final Settings settings, final RestController controller) {
+        super(settings);
+        controller.registerHandler(RestRequest.Method.GET, "/_ccr/auto_follow/stats", this);
+    }
+
+    @Override
+    public String getName() {
+        return "ccr_auto_follow_stats";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
+        final AutoFollowStatsAction.Request request = new AutoFollowStatsAction.Request();
+        return channel -> client.execute(AutoFollowStatsAction.INSTANCE, request, new RestToXContentListener<>(channel));
+    }
+
+}

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowStatsResponseTests.java

@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+package org.elasticsearch.xpack.ccr.action;
+
+import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.xpack.core.ccr.AutoFollowStats;
+import org.elasticsearch.xpack.core.ccr.action.AutoFollowStatsAction;
+
+import static org.elasticsearch.xpack.ccr.action.AutoFollowStatsTests.randomReadExceptions;
+
+public class AutoFollowStatsResponseTests extends AbstractStreamableTestCase<AutoFollowStatsAction.Response> {
+
+    @Override
+    protected AutoFollowStatsAction.Response createBlankInstance() {
+        return new AutoFollowStatsAction.Response();
+    }
+
+    @Override
+    protected AutoFollowStatsAction.Response createTestInstance() {
+        AutoFollowStats autoFollowStats = new AutoFollowStats(
+            randomNonNegativeLong(),
+            randomNonNegativeLong(),
+            randomNonNegativeLong(),
+            randomReadExceptions()
+        );
+        return new AutoFollowStatsAction.Response(autoFollowStats);
+    }
+}

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowStatsTests.java

@@ -38,7 +38,7 @@ protected AutoFollowStats createTestInstance() {
         );
     }
 
-    private static NavigableMap<String, ElasticsearchException> randomReadExceptions() {
+    static NavigableMap<String, ElasticsearchException> randomReadExceptions() {
         final int count = randomIntBetween(0, 16);
         final NavigableMap<String, ElasticsearchException> readExceptions = new TreeMap<>();
         for (int i = 0; i < count; i++) {

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/AutoFollowTests.java

@@ -15,6 +15,8 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.xpack.ccr.LocalStateCcr;
+import org.elasticsearch.xpack.core.ccr.AutoFollowStats;
+import org.elasticsearch.xpack.core.ccr.action.AutoFollowStatsAction;
 import org.elasticsearch.xpack.core.ccr.action.DeleteAutoFollowPatternAction;
 import org.elasticsearch.xpack.core.ccr.action.PutAutoFollowPatternAction;
 
@@ -58,6 +60,9 @@ public void testAutoFollow() throws Exception {
         });
         createIndex("transactions-201901", leaderIndexSettings, "_doc");
         assertBusy(() -> {
+            AutoFollowStats autoFollowStats = getAutoFollowStats();
+            assertThat(autoFollowStats.getNumberOfSuccessfulFollowIndices(), equalTo(2L));
+
             IndicesExistsRequest request = new IndicesExistsRequest("copy-transactions-201901");
             assertTrue(client().admin().indices().exists(request).actionGet().isExists());
         });
@@ -82,9 +87,8 @@ public void testAutoFollowManyIndices() throws Exception {
         }
         int expectedVal1 = numIndices;
         assertBusy(() -> {
-            MetaData metaData = client().admin().cluster().prepareState().get().getState().metaData();
-            int count = (int) Arrays.stream(metaData.getConcreteAllIndices()).filter(s -> s.startsWith("copy-")).count();
-            assertThat(count, equalTo(expectedVal1));
+            AutoFollowStats autoFollowStats = getAutoFollowStats();
+            assertThat(autoFollowStats.getNumberOfSuccessfulFollowIndices(), equalTo((long) expectedVal1));
         });
 
         deleteAutoFollowPatternSetting();
@@ -188,4 +192,9 @@ private void deleteAutoFollowPatternSetting() {
         assertTrue(client().execute(DeleteAutoFollowPatternAction.INSTANCE, request).actionGet().isAcknowledged());
     }
 
+    private AutoFollowStats getAutoFollowStats() {
+        AutoFollowStatsAction.Request request = new AutoFollowStatsAction.Request();
+        return client().execute(AutoFollowStatsAction.INSTANCE, request).actionGet().getStats();
+    }
+
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/action/AutoFollowStatsAction.java

@@ -0,0 +1,103 @@
+/*
+ * 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.
+ */
+
+package org.elasticsearch.xpack.core.ccr.action;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.support.master.MasterNodeRequest;
+import org.elasticsearch.action.Action;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.ccr.AutoFollowStats;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class AutoFollowStatsAction extends Action<AutoFollowStatsAction.Response> {
+
+    public static final String NAME = "cluster:monitor/ccr/auto_follow_stats";
+    public static final AutoFollowStatsAction INSTANCE = new AutoFollowStatsAction();
+
+    private AutoFollowStatsAction() {
+        super(NAME);
+    }
+
+    @Override
+    public Response newResponse() {
+        return new Response();
+    }
+
+    public static class Request extends MasterNodeRequest<Request> {
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+        }
+
+        public Request() {
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+        }
+    }
+
+    public static class Response extends ActionResponse implements ToXContentObject {
+
+        private AutoFollowStats stats;
+
+        public Response(AutoFollowStats stats) {
+            this.stats = stats;
+        }
+
+        public Response() {
+        }
+
+        public AutoFollowStats getStats() {
+            return stats;
+        }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            super.readFrom(in);
+            stats = new AutoFollowStats(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            stats.writeTo(out);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            stats.toXContent(builder, params);
+            return builder;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Response response = (Response) o;
+            return Objects.equals(stats, response.stats);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(stats);
+        }
+    }
+
+}

x-pack/plugin/src/test/resources/rest-api-spec/api/ccr.auto_follow_stats.json

@@ -0,0 +1,12 @@
+{
+  "ccr.auto_follow_stats": {
+    "documentation": "http://www.elastic.co/guide/en/elasticsearch/reference/current",
+    "methods": [ "GET" ],
+    "url": {
+      "path": "/_ccr/auto_follow/stats",
+      "paths": [ "/_ccr/auto_follow/stats" ],
+      "parts": {},
+      "body": null
+    }
+  }
+}