Cluster health call to throw decommissioned exception for local decom…

…missioned nodes Signed-off-by: Rishab Nahata <[email protected]>
opensearch-project · Jan 25, 2023 · bd1df17 · bd1df17
1 parent 715ff72
commit bd1df17
Show file tree

Hide file tree

Showing 8 changed files with 44 additions and 5 deletions.
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/cluster.health.json b/rest-api-spec/src/main/resources/rest-api-spec/api/cluster.health.json
@@ -111,6 +111,10 @@
       "awareness_attribute":{
         "type":"string",
         "description":"The awareness attribute for which the health is required"
+      },
+      "ensureLocalNodeCommissioned":{
+        "type":"boolean",
+        "description": "Checks whether local node is commissioned or not. If set to true on a local call it will throw exception if node is decommissioned (default: false)"
       }
     }
   }

diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/health/ClusterHealthRequest.java b/server/src/main/java/org/opensearch/action/admin/cluster/health/ClusterHealthRequest.java
@@ -321,6 +321,9 @@ public ActionRequestValidationException validate() {
         } else if (!level.equals(Level.AWARENESS_ATTRIBUTES) && awarenessAttribute != null) {
             return addValidationError("level=awareness_attributes is required with awareness_attribute parameter", null);
         }
+        if (ensureLocalNodeCommissioned && local == false) {
+            return addValidationError("not a local request to ensure local node commissioned", null);
+        }
         return null;
     }
 

diff --git a/...rc/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java b/...rc/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java
@@ -46,6 +46,8 @@
 import org.opensearch.cluster.LocalClusterUpdateTask;
 import org.opensearch.cluster.NotClusterManagerException;
 import org.opensearch.cluster.block.ClusterBlockException;
+import org.opensearch.cluster.coordination.Coordinator;
+import org.opensearch.cluster.decommission.NodeDecommissionedException;
 import org.opensearch.cluster.health.ClusterHealthStatus;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.metadata.ProcessClusterEventTimeoutException;
@@ -57,6 +59,7 @@
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.CollectionUtils;
+import org.opensearch.discovery.Discovery;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.node.NodeClosedException;
 import org.opensearch.tasks.Task;
@@ -77,6 +80,7 @@ public class TransportClusterHealthAction extends TransportClusterManagerNodeRea
     private static final Logger logger = LogManager.getLogger(TransportClusterHealthAction.class);
 
     private final AllocationService allocationService;
+    private final Discovery discovery;
 
     @Inject
     public TransportClusterHealthAction(
@@ -85,7 +89,8 @@ public TransportClusterHealthAction(
         ThreadPool threadPool,
         ActionFilters actionFilters,
         IndexNameExpressionResolver indexNameExpressionResolver,
-        AllocationService allocationService
+        AllocationService allocationService,
+        Discovery discovery
     ) {
         super(
             ClusterHealthAction.NAME,
@@ -98,6 +103,7 @@ public TransportClusterHealthAction(
             indexNameExpressionResolver
         );
         this.allocationService = allocationService;
+        this.discovery = discovery;
     }
 
     @Override
@@ -134,7 +140,9 @@ protected void clusterManagerOperation(
         final ClusterState unusedState,
         final ActionListener<ClusterHealthResponse> listener
     ) {
-
+        if (request.ensureLocalNodeCommissioned() && discovery instanceof Coordinator && ((Coordinator) discovery).localNodeCommissioned() == false) {
+            listener.onFailure(new NodeDecommissionedException("local node is decommissioned"));
+        }
         final int waitCount = getWaitCount(request);
 
         if (request.waitForEvents() != null) {

diff --git a/...main/java/org/opensearch/action/support/clustermanager/ClusterManagerNodeReadRequest.java b/...main/java/org/opensearch/action/support/clustermanager/ClusterManagerNodeReadRequest.java
@@ -46,18 +46,21 @@ public abstract class ClusterManagerNodeReadRequest<Request extends ClusterManag
     ClusterManagerNodeRequest<Request> {
 
     protected boolean local = false;
+    protected boolean ensureLocalNodeCommissioned = false;
 
     protected ClusterManagerNodeReadRequest() {}
 
     protected ClusterManagerNodeReadRequest(StreamInput in) throws IOException {
         super(in);
         local = in.readBoolean();
+        ensureLocalNodeCommissioned = in.readBoolean();
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(local);
+        out.writeBoolean(ensureLocalNodeCommissioned);
     }
 
     @SuppressWarnings("unchecked")
@@ -66,6 +69,12 @@ public final Request local(boolean local) {
         return (Request) this;
     }
 
+    @SuppressWarnings("unchecked")
+    public final Request ensureLocalNodeCommissioned(boolean ensureLocalNodeCommissioned) {
+        this.ensureLocalNodeCommissioned = ensureLocalNodeCommissioned;
+        return (Request) this;
+    }
+
     /**
      * Return local information, do not retrieve the state from cluster-manager node (default: false).
      * @return <code>true</code> if local information is to be returned;
@@ -74,4 +83,13 @@ public final Request local(boolean local) {
     public final boolean local() {
         return local;
     }
+
+    /**
+     * For a given local request, checks if the local node is commissioned or not (default: false).
+     * @return <code>true</code> if local information is to be returned only when local node is also commissioned
+     * <code>false</code> to not check local node if commissioned or not for a local request
+     */
+    public final boolean ensureLocalNodeCommissioned() {
+        return ensureLocalNodeCommissioned;
+    }
 }
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
@@ -1437,8 +1437,7 @@ synchronized void onNodeCommissionStatusChange(boolean localNodeCommissioned) {
         peerFinder.onNodeCommissionStatusChange(localNodeCommissioned);
     }
 
-    // package-visible for testing
-    boolean localNodeCommissioned() {
+    public boolean localNodeCommissioned() {
         return localNodeCommissioned;
     }
 

diff --git a/server/src/main/java/org/opensearch/cluster/decommission/NodeDecommissionedException.java b/server/src/main/java/org/opensearch/cluster/decommission/NodeDecommissionedException.java
@@ -10,6 +10,7 @@
 
 import org.opensearch.OpenSearchException;
 import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.rest.RestStatus;
 
 import java.io.IOException;
 
@@ -28,4 +29,9 @@ public NodeDecommissionedException(String msg, Object... args) {
     public NodeDecommissionedException(StreamInput in) throws IOException {
         super(in);
     }
+
+    @Override
+    public RestStatus status() {
+        return RestStatus.UNPROCESSABLE_ENTITY;
+    }
 }
diff --git a/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterHealthAction.java b/server/src/main/java/org/opensearch/rest/action/admin/cluster/RestClusterHealthAction.java
@@ -89,6 +89,7 @@ public static ClusterHealthRequest fromRequest(final RestRequest request) {
         final ClusterHealthRequest clusterHealthRequest = clusterHealthRequest(Strings.splitStringByCommaToArray(request.param("index")));
         clusterHealthRequest.indicesOptions(IndicesOptions.fromRequest(request, clusterHealthRequest.indicesOptions()));
         clusterHealthRequest.local(request.paramAsBoolean("local", clusterHealthRequest.local()));
+        clusterHealthRequest.ensureLocalNodeCommissioned(request.paramAsBoolean("ensureLocalNodeCommissioned", clusterHealthRequest.ensureLocalNodeCommissioned()));
         clusterHealthRequest.clusterManagerNodeTimeout(
             request.paramAsTime("cluster_manager_timeout", clusterHealthRequest.clusterManagerNodeTimeout())
         );

diff --git a/server/src/test/java/org/opensearch/cluster/health/ClusterStateHealthTests.java b/server/src/test/java/org/opensearch/cluster/health/ClusterStateHealthTests.java
@@ -177,7 +177,7 @@ public void testClusterHealthWaitsForClusterStateApplication() throws Interrupte
             threadPool,
             new ActionFilters(new HashSet<>()),
             indexNameExpressionResolver,
-            new AllocationService(null, new TestGatewayAllocator(), null, null, null)
+            new AllocationService(null, new TestGatewayAllocator(), null, null, null), null
         );
         PlainActionFuture<ClusterHealthResponse> listener = new PlainActionFuture<>();
         action.execute(new ClusterHealthRequest().waitForGreenStatus(), listener);