Adds cluster state size to /_cluster/state response (#23440)

This commit adds the size of the cluster state to the response for the
get cluster state API call (GET /_cluster/state).  The size that is
returned is the size of the full cluster state in bytes when compressed.
This is the same size of the full cluster state when serialized to
transmit over the network.  Specifying the ?human flag displays the
compressed size in a more human friendly manner.  Note that even if the
cluster state request filters items from the cluster state (so a subset
of the cluster state is returned), the size that is returned is the
compressed size of the entire cluster state.

Closes #3415

core/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateResponse.java

@@ -19,40 +19,75 @@
 
 package org.elasticsearch.action.admin.cluster.state;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.unit.ByteSizeValue;
 
 import java.io.IOException;
 
+/**
+ * The response for getting the cluster state.
+ */
 public class ClusterStateResponse extends ActionResponse {
 
     private ClusterName clusterName;
     private ClusterState clusterState;
+    // the total compressed size of the full cluster state, not just
+    // the parts included in this response
+    private ByteSizeValue totalCompressedSize;
 
     public ClusterStateResponse() {
     }
 
-    public ClusterStateResponse(ClusterName clusterName, ClusterState clusterState) {
+    public ClusterStateResponse(ClusterName clusterName, ClusterState clusterState, long sizeInBytes) {
         this.clusterName = clusterName;
         this.clusterState = clusterState;
+        this.totalCompressedSize = new ByteSizeValue(sizeInBytes);
     }
 
+    /**
+     * The requested cluster state.  Only the parts of the cluster state that were
+     * requested are included in the returned {@link ClusterState} instance.
+     */
     public ClusterState getState() {
         return this.clusterState;
     }
 
+    /**
+     * The name of the cluster.
+     */
     public ClusterName getClusterName() {
         return this.clusterName;
     }
 
+    /**
+     * The total compressed size of the full cluster state, not just the parts
+     * returned by {@link #getState()}.  The total compressed size is the size
+     * of the cluster state as it would be transmitted over the network during
+     * intra-node communication.
+     */
+    public ByteSizeValue getTotalCompressedSize() {
+        return totalCompressedSize;
+    }
+
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         clusterName = new ClusterName(in);
         clusterState = ClusterState.readFrom(in, null);
+        if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            totalCompressedSize = new ByteSizeValue(in);
+        } else {
+            // in a mixed cluster, if a pre 6.0 node processes the get cluster state
+            // request, then a compressed size won't be returned, so just return 0;
+            // its a temporary situation until all nodes in the cluster have been upgraded,
+            // at which point the correct cluster state size will always be reported
+            totalCompressedSize = new ByteSizeValue(0L);
+        }
     }
 
     @Override
@@ -60,5 +95,8 @@ public class ClusterStateResponse extends ActionResponse {
         super.writeTo(out);
         clusterName.writeTo(out);
         clusterState.writeTo(out);
+        if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            totalCompressedSize.writeTo(out);
+        }
     }
 }

core/src/main/java/org/elasticsearch/action/admin/cluster/state/TransportClusterStateAction.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.action.admin.cluster.state;
 
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
@@ -36,6 +37,10 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.io.IOException;
+
+import static org.elasticsearch.discovery.zen.PublishClusterStateAction.serializeFullClusterState;
+
 public class TransportClusterStateAction extends TransportMasterNodeReadAction<ClusterStateRequest, ClusterStateResponse> {
 
 
@@ -66,7 +71,8 @@ public class TransportClusterStateAction extends TransportMasterNodeReadAction<C
     }
 
     @Override
-    protected void masterOperation(final ClusterStateRequest request, final ClusterState state, ActionListener<ClusterStateResponse> listener) {
+    protected void masterOperation(final ClusterStateRequest request, final ClusterState state,
+                                   final ActionListener<ClusterStateResponse> listener) throws IOException {
         ClusterState currentState = clusterService.state();
         logger.trace("Serving cluster state request using version {}", currentState.version());
         ClusterState.Builder builder = ClusterState.builder(currentState.getClusterName());
@@ -122,7 +128,8 @@ public class TransportClusterStateAction extends TransportMasterNodeReadAction<C
         if (request.customs()) {
             builder.customs(currentState.customs());
         }
-        listener.onResponse(new ClusterStateResponse(currentState.getClusterName(), builder.build()));
+        listener.onResponse(new ClusterStateResponse(currentState.getClusterName(), builder.build(),
+                                                        serializeFullClusterState(currentState, Version.CURRENT).length()));
     }
 
 

core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestClusterStateAction.java

@@ -90,6 +90,7 @@ public class RestClusterStateAction extends BaseRestHandler {
             public RestResponse buildResponse(ClusterStateResponse response, XContentBuilder builder) throws Exception {
                 builder.startObject();
                 builder.field(Fields.CLUSTER_NAME, response.getClusterName().value());
+                builder.byteSizeField(Fields.CLUSTER_STATE_SIZE_IN_BYTES, Fields.CLUSTER_STATE_SIZE, response.getTotalCompressedSize());
                 response.getState().toXContent(builder, request);
                 builder.endObject();
                 return new BytesRestResponse(RestStatus.OK, builder);
@@ -118,5 +119,7 @@ public class RestClusterStateAction extends BaseRestHandler {
 
     static final class Fields {
         static final String CLUSTER_NAME = "cluster_name";
+        static final String CLUSTER_STATE_SIZE = "compressed_size";
+        static final String CLUSTER_STATE_SIZE_IN_BYTES = "compressed_size_in_bytes";
     }
 }

core/src/test/java/org/elasticsearch/action/search/RemoteClusterConnectionTests.java

@@ -96,7 +96,7 @@ public class RemoteClusterConnectionTests extends ESTestCase {
                         builder.add(node);
                     }
                     ClusterState build = ClusterState.builder(ClusterName.DEFAULT).nodes(builder.build()).build();
-                    channel.sendResponse(new ClusterStateResponse(ClusterName.DEFAULT, build));
+                    channel.sendResponse(new ClusterStateResponse(ClusterName.DEFAULT, build, 0L));
                 });
             newService.start();
             newService.acceptIncomingRequests();

core/src/test/java/org/elasticsearch/client/transport/FailAndRetryMockTransport.java

@@ -96,7 +96,7 @@ abstract class FailAndRetryMockTransport<Response extends TransportResponse> imp
                     } else if (ClusterStateAction.NAME.equals(action)) {
                         TransportResponseHandler transportResponseHandler = transportServiceAdapter.onResponseReceived(requestId);
                         ClusterState clusterState = getMockClusterState(node);
-                        transportResponseHandler.handleResponse(new ClusterStateResponse(clusterName, clusterState));
+                        transportResponseHandler.handleResponse(new ClusterStateResponse(clusterName, clusterState, 0L));
                     } else {
                         throw new UnsupportedOperationException("Mock transport does not understand action " + action);
                     }

core/src/test/java/org/elasticsearch/client/transport/TransportClientHeadersTests.java

@@ -170,7 +170,7 @@ public class TransportClientHeadersTests extends AbstractClientHeadersTestCase {
                             address.address().getHostString(), address.getAddress(), address, Collections.emptyMap(),
                                 Collections.singleton(DiscoveryNode.Role.DATA), Version.CURRENT)));
                         ((TransportResponseHandler<ClusterStateResponse>) handler)
-                                .handleResponse(new ClusterStateResponse(cluster1, builder.build()));
+                                .handleResponse(new ClusterStateResponse(cluster1, builder.build(), 0L));
                         clusterStateLatch.countDown();
                     } else if (TransportService.HANDSHAKE_ACTION_NAME .equals(action)) {
                         ((TransportResponseHandler<TransportService.HandshakeResponse>) handler).handleResponse(

core/src/test/java/org/elasticsearch/cluster/GetClusterStateTests.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.cluster;
+
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.test.ESSingleNodeTestCase;
+
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+
+/**
+ * Tests for the get cluster state API.
+ *
+ * See: {@link org.elasticsearch.action.admin.cluster.state.TransportClusterStateAction}
+ *      {@link org.elasticsearch.rest.action.admin.cluster.RestClusterStateAction}
+ */
+public class GetClusterStateTests extends ESSingleNodeTestCase {
+
+    public void testGetClusterState() {
+        ClusterStateResponse response = client().admin().cluster().prepareState().get();
+        assertNotNull(response.getState());
+        assertNotNull(response.getClusterName());
+        // assume the cluster state size is 50 bytes or more, just so we aren't testing against size of 0
+        assertThat(response.getTotalCompressedSize().getBytes(), greaterThanOrEqualTo(50L));
+    }
+
+    public void testSizeDerivedFromFullClusterState() {
+        ClusterStateResponse response = client().admin().cluster().prepareState().get();
+        final ClusterState clusterState = response.getState();
+        final long totalCompressedSize = response.getTotalCompressedSize().getBytes();
+        // exclude the nodes from being returned, the total size returned should still be
+        // the same as when no filtering was applied to the cluster state retrieved
+        response = client().admin().cluster().prepareState().setNodes(false).get();
+        assertEquals(totalCompressedSize, response.getTotalCompressedSize().getBytes());
+        assertNotEquals(clusterState, response.getState());
+        assertEquals(0, response.getState().nodes().getSize());
+    }
+}

docs/reference/cluster/state.asciidoc

@@ -9,6 +9,11 @@ the whole cluster.
 $ curl -XGET 'http://localhost:9200/_cluster/state'
 --------------------------------------------------
 
+The response provides the cluster name, the total compressed size
+of the cluster state (its size when serialized for transmission over
+the network), and the cluster state itself, which can be filtered to
+only retrieve the parts of interest, as described below.
+
 By default, the cluster state request is routed to the master node, to
 ensure that the latest cluster state is returned.   
 For debugging purposes, you can retrieve the cluster state local to a

rest-api-spec/src/main/resources/rest-api-spec/test/cluster.state/10_basic.yaml

@@ -1,6 +1,20 @@
 ---
-"cluster state test":
+"get cluster state":
   - do:
       cluster.state: {}
   
   - is_true: master_node
+
+---
+"get cluster state returns cluster state size with human readable format":
+  - skip:
+      version:  " - 5.99.99"
+      reason:   "cluster state size is only available in v6.0.0 and higher"
+
+  - do:
+      cluster.state:
+        human: true
+
+  - is_true: master_node
+  - gte: { compressed_size_in_bytes: 50 }
+  - is_true: compressed_size