Service Layer changes for Recommission API

CHANGELOG.md

@@ -12,6 +12,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Support for HTTP/2 (server-side) ([#3847](https://github.com/opensearch-project/OpenSearch/pull/3847))
 - BWC version 2.2.2 ([#4383](https://github.com/opensearch-project/OpenSearch/pull/4383))
 - Support for labels on version bump PRs, skip label support for changelog verifier ([#4391](https://github.com/opensearch-project/OpenSearch/pull/4391))
+- Recommission API changes for service layer ([#4320](https://github.com/opensearch-project/OpenSearch/pull/4320))
 ### Dependencies
 - Bumps `org.gradle.test-retry` from 1.4.0 to 1.4.1
 

server/src/main/java/org/opensearch/cluster/decommission/DecommissionAttribute.java

@@ -0,0 +1,92 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.decommission;
+
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.io.stream.Writeable;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * {@link DecommissionAttribute} encapsulates information about decommissioned node attribute like attribute name, attribute value.
+ *
+ * @opensearch.internal
+ */
+public final class DecommissionAttribute implements Writeable {
+    private final String attributeName;
+    private final String attributeValue;
+
+    /**
+     * Constructs new decommission attribute name value pair
+     *
+     * @param attributeName   attribute name
+     * @param attributeValue attribute value
+     */
+    public DecommissionAttribute(String attributeName, String attributeValue) {
+        this.attributeName = attributeName;
+        this.attributeValue = attributeValue;
+    }
+
+    /**
+     * Returns attribute name
+     *
+     * @return attributeName
+     */
+    public String attributeName() {
+        return this.attributeName;
+    }
+
+    /**
+     * Returns attribute value
+     *
+     * @return attributeValue
+     */
+    public String attributeValue() {
+        return this.attributeValue;
+    }
+
+    public DecommissionAttribute(StreamInput in) throws IOException {
+        attributeName = in.readString();
+        attributeValue = in.readString();
+    }
+
+    /**
+     * Writes decommission attribute name value to stream output
+     *
+     * @param out stream output
+     */
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(attributeName);
+        out.writeString(attributeValue);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        DecommissionAttribute that = (DecommissionAttribute) o;
+
+        if (!attributeName.equals(that.attributeName)) return false;
+        return attributeValue.equals(that.attributeValue);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(attributeName, attributeValue);
+    }
+
+    @Override
+    public String toString() {
+        return "DecommissionAttribute{" + "attributeName='" + attributeName + '\'' + ", attributeValue='" + attributeValue + '\'' + '}';
+    }
+}

server/src/main/java/org/opensearch/cluster/decommission/DecommissionFailedException.java

@@ -0,0 +1,54 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.decommission;
+
+import org.opensearch.OpenSearchException;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/**
+ * This exception is thrown whenever a failure occurs in decommission request @{@link DecommissionService}
+ *
+ * @opensearch.internal
+ */
+public class DecommissionFailedException extends OpenSearchException {
+
+    private final DecommissionAttribute decommissionAttribute;
+
+    public DecommissionFailedException(DecommissionAttribute decommissionAttribute, String msg) {
+        this(decommissionAttribute, msg, null);
+    }
+
+    public DecommissionFailedException(DecommissionAttribute decommissionAttribute, String msg, Throwable cause) {
+        super("[" + (decommissionAttribute == null ? "_na" : decommissionAttribute.toString()) + "] " + msg, cause);
+        this.decommissionAttribute = decommissionAttribute;
+    }
+
+    public DecommissionFailedException(StreamInput in) throws IOException {
+        super(in);
+        decommissionAttribute = new DecommissionAttribute(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        decommissionAttribute.writeTo(out);
+    }
+
+    /**
+     * Returns decommission attribute
+     *
+     * @return decommission attribute
+     */
+    public DecommissionAttribute decommissionAttribute() {
+        return decommissionAttribute;
+    }
+}

server/src/main/java/org/opensearch/cluster/decommission/DecommissionService.java

@@ -0,0 +1,80 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.decommission;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.opensearch.action.ActionListener;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.ClusterStateUpdateTask;
+import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
+import org.opensearch.cluster.metadata.DecommissionAttributeMetadata;
+import org.opensearch.cluster.metadata.Metadata;
+import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.Priority;
+import org.opensearch.common.inject.Inject;
+
+/**
+ * Service responsible for entire lifecycle of decommissioning and recommissioning an awareness attribute.
+ * <p>
+ * Whenever a cluster manager initiates operation to decommission an awareness attribute,
+ * the service makes the best attempt to perform the following task -
+ * <ul>
+ * <li>Remove cluster-manager eligible nodes from voting config [TODO - checks to avoid quorum loss scenarios]</li>
+ * <li>Initiates nodes decommissioning by adding custom metadata with the attribute and state as {@link DecommissionStatus#DECOMMISSION_INIT}</li>
+ * <li>Triggers weigh away for nodes having given awareness attribute to drain. This marks the decommission status as {@link DecommissionStatus#DECOMMISSION_IN_PROGRESS}</li>
+ * <li>Once weighed away, the service triggers nodes decommission</li>
+ * <li>Once the decommission is successful, the service clears the voting config and marks the status as {@link DecommissionStatus#DECOMMISSION_SUCCESSFUL}</li>
+ * <li>If service fails at any step, it would mark the status as {@link DecommissionStatus#DECOMMISSION_FAILED}</li>
+ * </ul>
+ *
+ * @opensearch.internal
+ */
+public class DecommissionService {
+
+    private static final Logger logger = LogManager.getLogger(DecommissionService.class);
+
+    private final ClusterService clusterService;
+
+    @Inject
+    public DecommissionService(ClusterService clusterService) {
+        this.clusterService = clusterService;
+    }
+
+    public void clearDecommissionStatus(final ActionListener<ClusterStateUpdateResponse> listener) {
+        clusterService.submitStateUpdateTask("delete_decommission_state", new ClusterStateUpdateTask(Priority.URGENT) {
+            @Override
+            public ClusterState execute(ClusterState currentState) {
+                return deleteDecommissionAttribute(currentState);
+            }
+
+            @Override
+            public void onFailure(String source, Exception e) {
+                logger.error(() -> new ParameterizedMessage("Failed to clear decommission attribute."), e);
+                listener.onFailure(e);
+            }
+
+            @Override
+            public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                // Once the cluster state is processed we can try to recommission nodes by setting the weights for the zone.
+                // TODO Set the weights for the recommissioning zone.
+                listener.onResponse(new ClusterStateUpdateResponse(true));
+            }
+        });
+    }
+
+    ClusterState deleteDecommissionAttribute(final ClusterState currentState) {
+        logger.info("Delete decommission request received");
+        Metadata metadata = currentState.metadata();
+        Metadata.Builder mdBuilder = Metadata.builder(metadata);
+        mdBuilder.removeCustom(DecommissionAttributeMetadata.TYPE);
+        return ClusterState.builder(currentState).metadata(mdBuilder).build();
+    }
+}

server/src/main/java/org/opensearch/cluster/decommission/DecommissionStatus.java

@@ -0,0 +1,80 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.decommission;
+
+/**
+ * An enumeration of the states during decommissioning and recommissioning.
+ */
+public enum DecommissionStatus {
+    /**
+     * Decommission process is initiated
+     */
+    DECOMMISSION_INIT("decommission_init"),
+    /**
+     * Decommission process has started, decommissioned nodes should be weighed away
+     */
+    DECOMMISSION_IN_PROGRESS("decommission_in_progress"),
+    /**
+     * Decommissioning awareness attribute completed
+     */
+    DECOMMISSION_SUCCESSFUL("decommission_successful"),
+    /**
+     * Decommission request failed
+     */
+    DECOMMISSION_FAILED("decommission_failed"),
+    /**
+     * Recommission request received, recommissioning process has started
+     */
+    RECOMMISSION_IN_PROGRESS("recommission_in_progress"),
+    /**
+     * Recommission request failed. No nodes should fail to join the cluster with decommission exception
+     */
+    RECOMMISSION_FAILED("recommission_failed");
+
+    private final String status;
+
+    DecommissionStatus(String status) {
+        this.status = status;
+    }
+
+    /**
+     * Returns status that represents the decommission state
+     *
+     * @return status
+     */
+    public String status() {
+        return status;
+    }
+
+    /**
+     * Generate decommission status from given string
+     *
+     * @param status status in string
+     * @return status
+     */
+    public static DecommissionStatus fromString(String status) {
+        if (status == null) {
+            throw new IllegalArgumentException("decommission status cannot be null");
+        }
+        if (status.equals(DECOMMISSION_INIT.status())) {
+            return DECOMMISSION_INIT;
+        } else if (status.equals(DECOMMISSION_IN_PROGRESS.status())) {
+            return DECOMMISSION_IN_PROGRESS;
+        } else if (status.equals(DECOMMISSION_SUCCESSFUL.status())) {
+            return DECOMMISSION_SUCCESSFUL;
+        } else if (status.equals(DECOMMISSION_FAILED.status())) {
+            return DECOMMISSION_FAILED;
+        } else if (status.equals(RECOMMISSION_IN_PROGRESS.status())) {
+            return RECOMMISSION_IN_PROGRESS;
+        } else if (status.equals(RECOMMISSION_FAILED.status())) {
+            return RECOMMISSION_FAILED;
+        }
+        throw new IllegalStateException("Decommission status [" + status + "] not recognized.");
+    }
+}

server/src/main/java/org/opensearch/cluster/decommission/NodeDecommissionedException.java

@@ -0,0 +1,31 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.decommission;
+
+import org.opensearch.OpenSearchException;
+import org.opensearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+
+/**
+ * This exception is thrown if the node is decommissioned by @{@link DecommissionService}
+ * and this nodes needs to be removed from the cluster
+ *
+ * @opensearch.internal
+ */
+public class NodeDecommissionedException extends OpenSearchException {
+
+    public NodeDecommissionedException(String msg, Object... args) {
+        super(msg, args);
+    }
+
+    public NodeDecommissionedException(StreamInput in) throws IOException {
+        super(in);
+    }
+}