Adjust /_cat/templates not to request all metadata

Today `GET /_cat/templates` retrieves the whole cluster metadata from
the master, which includes all sorts of unnecessary junk and consumes
significant resources. This commit reimplements these endpoints using
`GetIndexTemplatesAction` and `GetComposableIndexTemplateAction` which
are much more efficient.

The docs for this API indicate that it accepts a comma-separated list of
template names/patterns of the form `GET /_cat/templates/name1,name2`
but in fact today it only accepts a single name or pattern. This commit
also adds support for multiple names/patterns as the docs claim.

Backport of elastic#78812

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/cat.templates/20_matching.yml

@@ -0,0 +1,179 @@
+setup:
+    - do:
+        indices.put_template:
+          name: test-legacy-1
+          body:
+            order: 12
+            version: 3
+            index_patterns: foo*
+
+    - do:
+        indices.put_template:
+          name: test-legacy-2
+          body:
+            order: 45
+            version: 6
+            index_patterns:
+              - bar*
+              - baz*
+
+    - do:
+        cluster.put_component_template:
+          name: test-component-template
+          body:
+            template:
+              settings:
+                number_of_shards: 1
+                number_of_replicas: 0
+
+    - do:
+        indices.put_index_template:
+          name: test-composable-1
+          body:
+            index_patterns:
+              - quux*
+            priority: 78
+            version: 9
+            composed_of:
+              - test-component-template
+
+    - do:
+        indices.put_index_template:
+          name: test-composable-2
+          body:
+            index_patterns:
+              - gruly*
+            priority: 99
+            version: 1
+            composed_of:
+              - test-component-template
+
+---
+"Matching all templates":
+
+    - do:
+        cat.templates:
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /test-composable-1\ntest-composable-2\ntest-legacy-1\ntest-legacy-2\n/
+
+    - do:
+        cat.templates:
+            name: "*"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /test-composable-1\ntest-composable-2\ntest-legacy-1\ntest-legacy-2\n/
+
+---
+"Matching all templates with other patterns":
+    - skip:
+        version: " - 7.15.99"
+        reason: "support for multiple patterns added in 7.16.0"
+
+    - do:
+        cat.templates:
+            name: "nonexistent*,*,other-name"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /test-composable-1\ntest-composable-2\ntest-legacy-1\ntest-legacy-2\n/
+
+---
+"Matching no templates":
+
+    - do:
+        cat.templates:
+            name: "nonexistent"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^$/
+
+---
+"Matching single names":
+
+    - do:
+        cat.templates:
+            name: "test-legacy-1"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-legacy-1\n$/
+
+
+    - do:
+        cat.templates:
+            name: "test-composable-2"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-composable-2\n$/
+
+---
+"Matching single patterns":
+
+    - do:
+        cat.templates:
+            name: "test-legacy-*"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-legacy-1\ntest-legacy-2\n$/
+
+
+    - do:
+        cat.templates:
+            name: "test-*-2"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-composable-2\ntest-legacy-2\n$/
+
+---
+"Matching lists of names":
+    - skip:
+        version: " - 7.15.99"
+        reason: "support for multiple patterns added in 7.16.0"
+
+    - do:
+        cat.templates:
+            name: "test-legacy-1,test-composable-2"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-composable-2\ntest-legacy-1\n$/
+
+---
+"Matching names and wildcards":
+    - skip:
+        version: " - 7.15.99"
+        reason: "support for multiple patterns added in 7.16.0"
+
+    - do:
+        cat.templates:
+            name: "test-legacy-1,test-composable-*"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-composable-1\ntest-composable-2\ntest-legacy-1\n$/
+
+    - do:
+        cat.templates:
+            name: "test-legacy-*,test-composable-2"
+            h: [name]
+            s: [name]
+
+    - match:
+        $body: /^test-composable-2\ntest-legacy-1\ntest-legacy-2\n$/

server/src/main/java/org/elasticsearch/rest/action/cat/RestTemplatesAction.java

@@ -8,21 +8,27 @@
 
 package org.elasticsearch.rest.action.cat;
 
-import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
-import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
-import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.StepListener;
+import org.elasticsearch.action.admin.indices.template.get.GetComposableIndexTemplateAction;
+import org.elasticsearch.action.admin.indices.template.get.GetIndexTemplatesRequest;
+import org.elasticsearch.action.admin.indices.template.get.GetIndexTemplatesResponse;
 import org.elasticsearch.client.node.NodeClient;
-import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
-import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.Table;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestResponseListener;
 
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
 
 import static java.util.Arrays.asList;
 import static java.util.Collections.unmodifiableList;
@@ -49,18 +55,43 @@ protected void documentation(StringBuilder sb) {
 
     @Override
     protected RestChannelConsumer doCatRequest(final RestRequest request, NodeClient client) {
-        final String matchPattern = request.hasParam("name") ? request.param("name") : null;
-        final ClusterStateRequest clusterStateRequest = new ClusterStateRequest();
-        clusterStateRequest.clear().metadata(true);
-        clusterStateRequest.local(request.paramAsBoolean("local", clusterStateRequest.local()));
-        clusterStateRequest.masterNodeTimeout(request.paramAsTime("master_timeout", clusterStateRequest.masterNodeTimeout()));
-
-        return channel -> client.admin().cluster().state(clusterStateRequest, new RestResponseListener<ClusterStateResponse>(channel) {
-            @Override
-            public RestResponse buildResponse(ClusterStateResponse clusterStateResponse) throws Exception {
-                return RestTable.buildResponse(buildTable(request, clusterStateResponse, matchPattern), channel);
-            }
-        });
+        final String[] templateNames = Strings.splitStringByCommaToArray(request.param("name", ""));
+
+        final GetIndexTemplatesRequest getIndexTemplatesRequest = new GetIndexTemplatesRequest(templateNames);
+        getIndexTemplatesRequest.local(request.paramAsBoolean("local", getIndexTemplatesRequest.local()));
+        getIndexTemplatesRequest.masterNodeTimeout(request.paramAsTime("master_timeout", getIndexTemplatesRequest.masterNodeTimeout()));
+
+        final GetComposableIndexTemplateAction.Request getComposableTemplatesRequest
+            = new GetComposableIndexTemplateAction.Request();
+        getComposableTemplatesRequest.local(request.paramAsBoolean("local", getComposableTemplatesRequest.local()));
+        getComposableTemplatesRequest.masterNodeTimeout(
+            request.paramAsTime("master_timeout", getComposableTemplatesRequest.masterNodeTimeout()));
+
+        return channel -> {
+
+            final StepListener<GetIndexTemplatesResponse> getIndexTemplatesStep = new StepListener<>();
+            client.admin().indices().getTemplates(getIndexTemplatesRequest, getIndexTemplatesStep);
+
+            final StepListener<GetComposableIndexTemplateAction.Response> getComposableTemplatesStep = new StepListener<>();
+            client.execute(GetComposableIndexTemplateAction.INSTANCE, getComposableTemplatesRequest, getComposableTemplatesStep);
+
+            final ActionListener<Table> tableListener = new RestResponseListener<Table>(channel) {
+                @Override
+                public RestResponse buildResponse(Table table) throws Exception {
+                    return RestTable.buildResponse(table, channel);
+                }
+            };
+
+            getIndexTemplatesStep.whenComplete(getIndexTemplatesResponse ->
+                getComposableTemplatesStep.whenComplete(getComposableIndexTemplatesResponse ->
+                    ActionListener.completeWith(tableListener, () -> buildTable(
+                        request,
+                        getIndexTemplatesResponse,
+                        getComposableIndexTemplatesResponse,
+                        templateNames)
+                    ), tableListener::onFailure
+                ), tableListener::onFailure);
+        };
     }
 
     @Override
@@ -76,26 +107,30 @@ protected Table getTableWithHeader(RestRequest request) {
         return table;
     }
 
-    private Table buildTable(RestRequest request, ClusterStateResponse clusterStateResponse, String patternString) {
-        Table table = getTableWithHeader(request);
-        Metadata metadata = clusterStateResponse.getState().metadata();
-        for (ObjectObjectCursor<String, IndexTemplateMetadata> entry : metadata.templates()) {
-            IndexTemplateMetadata indexData = entry.value;
-            if (patternString == null || Regex.simpleMatch(patternString, indexData.name())) {
-                table.startRow();
-                table.addCell(indexData.name());
-                table.addCell("[" + String.join(", ", indexData.patterns()) + "]");
-                table.addCell(indexData.getOrder());
-                table.addCell(indexData.getVersion());
-                table.addCell("");
-                table.endRow();
-            }
+    private Table buildTable(
+        RestRequest request,
+        GetIndexTemplatesResponse getIndexTemplatesResponse,
+        GetComposableIndexTemplateAction.Response getComposableIndexTemplatesResponse,
+        String[] requestedNames
+    ) {
+        final Predicate<String> namePredicate = getNamePredicate(requestedNames);
+
+        final Table table = getTableWithHeader(request);
+        for (IndexTemplateMetadata indexData : getIndexTemplatesResponse.getIndexTemplates()) {
+            assert namePredicate.test(indexData.getName());
+            table.startRow();
+            table.addCell(indexData.name());
+            table.addCell("[" + String.join(", ", indexData.patterns()) + "]");
+            table.addCell(indexData.getOrder());
+            table.addCell(indexData.getVersion());
+            table.addCell("");
+            table.endRow();
         }
 
-        for (Map.Entry<String, ComposableIndexTemplate> entry : metadata.templatesV2().entrySet()) {
-            String name = entry.getKey();
-            ComposableIndexTemplate template = entry.getValue();
-            if (patternString == null || Regex.simpleMatch(patternString, name)) {
+        for (Map.Entry<String, ComposableIndexTemplate> entry : getComposableIndexTemplatesResponse.indexTemplates().entrySet()) {
+            final String name = entry.getKey();
+            if (namePredicate.test(name)) {
+                final ComposableIndexTemplate template = entry.getValue();
                 table.startRow();
                 table.addCell(name);
                 table.addCell("[" + String.join(", ", template.indexPatterns()) + "]");
@@ -105,6 +140,41 @@ private Table buildTable(RestRequest request, ClusterStateResponse clusterStateR
                 table.endRow();
             }
         }
+
         return table;
     }
+
+    private Predicate<String> getNamePredicate(String[] requestedNames) {
+        if (requestedNames.length == 0) {
+            return name -> true;
+        }
+
+        final Set<String> exactMatches = new HashSet<>();
+        final List<String> patterns = new ArrayList<>();
+        for (String requestedName : requestedNames) {
+            if (Regex.isMatchAllPattern(requestedName)) {
+                return name -> true;
+            } else if (Regex.isSimpleMatchPattern(requestedName)) {
+                patterns.add(requestedName);
+            } else {
+                exactMatches.add(requestedName);
+            }
+        }
+
+        if (patterns.isEmpty()) {
+            return exactMatches::contains;
+        }
+
+        return name -> {
+            if (exactMatches.contains(name)) {
+                return true;
+            }
+            for (String pattern : patterns) {
+                if (Regex.simpleMatch(pattern, name)) {
+                    return true;
+                }
+            }
+            return false;
+        };
+    }
 }