[7.x] Deprecate returning 408 for a server timeout on _cluster/health (#78180) (#78940)

Backports #78180 to 7.x.

Add a deprecation warning and a system property es.cluster_health.request_timeout_200 to opt in for returning 200 which will be the default in 8.0.0

Fixes #70849

client/rest-high-level/src/test/java/org/elasticsearch/client/ClusterClientIT.java

@@ -304,6 +304,9 @@ public class ClusterClientIT extends ESRestHighLevelClientTestCase {
         assertThat(response.status(), equalTo(RestStatus.REQUEST_TIMEOUT));
         assertThat(response.getStatus(), equalTo(ClusterHealthStatus.RED));
         assertNoIndices(response);
+        assertWarnings("The HTTP status code for a cluster health timeout will be changed from 408 to 200 in a " +
+            "future version. Set the [es.cluster_health.request_timeout_200] system property to [true] to suppress this message and " +
+            "opt in to the future behaviour now.");
     }
 
     public void testRemoteInfo() throws Exception {

server/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthResponse.java

@@ -13,6 +13,8 @@ import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.health.ClusterIndexHealth;
 import org.elasticsearch.cluster.health.ClusterStateHealth;
+import org.elasticsearch.common.logging.DeprecationCategory;
+import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.xcontent.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -24,6 +26,7 @@ import org.elasticsearch.common.xcontent.StatusToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.rest.action.search.RestSearchAction;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -55,6 +58,7 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
     private static final String INITIALIZING_SHARDS = "initializing_shards";
     private static final String UNASSIGNED_SHARDS = "unassigned_shards";
     private static final String INDICES = "indices";
+    private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(RestSearchAction.class);
 
     private static final ConstructingObjectParser<ClusterHealthResponse, Void> PARSER =
             new ConstructingObjectParser<>("cluster_health_response", true,
@@ -98,6 +102,10 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
 
     private static final ObjectParser.NamedObjectParser<ClusterIndexHealth, Void> INDEX_PARSER =
         (XContentParser parser, Void context, String index) -> ClusterIndexHealth.innerFromXContent(parser, index);
+    private static final String ES_CLUSTER_HEALTH_REQUEST_TIMEOUT_200_KEY = "es.cluster_health.request_timeout_200";
+    static final String CLUSTER_HEALTH_REQUEST_TIMEOUT_DEPRECATION_MSG = "The HTTP status code for a cluster health timeout " +
+        "will be changed from 408 to 200 in a future version. Set the [" + ES_CLUSTER_HEALTH_REQUEST_TIMEOUT_200_KEY + "] " +
+        "system property to [true] to suppress this message and opt in to the future behaviour now.";
 
     static {
         // ClusterStateHealth fields
@@ -130,8 +138,15 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
     private boolean timedOut = false;
     private ClusterStateHealth clusterStateHealth;
     private ClusterHealthStatus clusterHealthStatus;
+    private boolean esClusterHealthRequestTimeout200 = readEsClusterHealthRequestTimeout200FromProperty();
 
-    public ClusterHealthResponse() {}
+    public ClusterHealthResponse() {
+    }
+
+    /** For the testing of opting in for the 200 status code without setting a system property */
+    ClusterHealthResponse(boolean esClusterHealthRequestTimeout200) {
+        this.esClusterHealthRequestTimeout200 = esClusterHealthRequestTimeout200;
+    }
 
     public ClusterHealthResponse(StreamInput in) throws IOException {
         super(in);
@@ -299,7 +314,16 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
 
     @Override
     public RestStatus status() {
-        return isTimedOut() ? RestStatus.REQUEST_TIMEOUT : RestStatus.OK;
+        if (isTimedOut() == false) {
+            return RestStatus.OK;
+        }
+        if (esClusterHealthRequestTimeout200) {
+            return RestStatus.OK;
+        } else {
+            deprecationLogger.critical(DeprecationCategory.API,"cluster_health_request_timeout",
+                CLUSTER_HEALTH_REQUEST_TIMEOUT_DEPRECATION_MSG);
+            return RestStatus.REQUEST_TIMEOUT;
+        }
     }
 
     @Override
@@ -359,4 +383,17 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
         return Objects.hash(clusterName, numberOfPendingTasks, numberOfInFlightFetch, delayedUnassignedShards, taskMaxWaitingTime,
                 timedOut, clusterStateHealth, clusterHealthStatus);
     }
+
+    private static boolean readEsClusterHealthRequestTimeout200FromProperty() {
+        String property = System.getProperty(ES_CLUSTER_HEALTH_REQUEST_TIMEOUT_200_KEY);
+        if (property == null) {
+            return false;
+        }
+        if (Boolean.parseBoolean(property)) {
+            return true;
+        } else {
+            throw new IllegalArgumentException(ES_CLUSTER_HEALTH_REQUEST_TIMEOUT_200_KEY + " can only be unset or [true] but was ["
+                + property + "]");
+        }
+    }
 }

server/src/test/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthResponsesTests.java

@@ -48,12 +48,21 @@ public class ClusterHealthResponsesTests extends AbstractSerializingTestCase<Clu
             res.setTimedOut(randomBoolean());
             if (res.isTimedOut()) {
                 assertEquals(RestStatus.REQUEST_TIMEOUT, res.status());
+                assertWarnings(ClusterHealthResponse.CLUSTER_HEALTH_REQUEST_TIMEOUT_DEPRECATION_MSG);
             } else {
                 assertEquals(RestStatus.OK, res.status());
             }
         }
     }
 
+    public void testTimeoutReturns200IfOptedIn() {
+        ClusterHealthResponse res = new ClusterHealthResponse(true);
+        for (int i = 0; i < 5; i++) {
+            res.setTimedOut(randomBoolean());
+            assertEquals(RestStatus.OK, res.status());
+        }
+    }
+
     public void testClusterHealth() throws IOException {
         ClusterState clusterState = ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.getDefault(Settings.EMPTY)).build();
         int pendingTasks = randomIntBetween(0, 200);