[apache#80][Part-3] feat: add REST API for decommisson (apache#684)

### What changes were proposed in this pull request?

Add REST API for decommisson

### Why are the changes needed?

Support shuffle server decommission. It is a part of apache#80

### Does this PR introduce _any_ user-facing change?

Env:

* Server IP: 127.0.0.1
* HTTP port: 19998
* RPC port: 19999

Decommission example:

```shell
curl -XPOST -H "Content-type:application/json" "http://127.0.0.1:19998/api/server/decommission" -d '{"serverIds:": ["127.0.0.1:19999"]}'
```

Cancel decommission example:

```shell
curl -XPOST -H "Content-type:application/json" "http://127.0.0.1:19998/api/server/cancelDecommission" -d '{"serverIds:": ["127.0.0.1:19999"]}'
```

Get server list:

```shell
# path: /api/server/nodes[?id={serverId}][?status={serverStatus}]
curl  "http://127.0.0.1:19998/api/server/nodes?status=DECOMMISSIONING"
curl  "http://127.0.0.1:19998/api/server/nodes?status=ACTIVE"
```

### How was this patch tested?

UT

common/src/test/java/org/apache/uniffle/common/ServerStatusTest.java

@@ -55,7 +55,7 @@ public void test() throws Exception {
         fail(e.getMessage());
       }
     }
-    for (int i = 0; i < serverStatuses.size() - 1; i++) {
+    for (int i = 0; i < serverStatuses.size(); i++) {
       assertEquals(protoServerStatuses.get(i), serverStatuses.get(i).toProto());
       assertEquals(ServerStatus.fromProto(protoServerStatuses.get(i)), serverStatuses.get(i));
     }

common/src/test/java/org/apache/uniffle/common/metrics/TestUtils.java

@@ -20,6 +20,7 @@
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.OutputStream;
 import java.net.HttpURLConnection;
 import java.net.URL;
 
@@ -28,18 +29,38 @@ public class TestUtils {
   private TestUtils() {
   }
 
-  public static String httpGetMetrics(String urlString) throws IOException {
+  public static String httpGet(String urlString) throws IOException {
     URL url = new URL(urlString);
     HttpURLConnection con = (HttpURLConnection) url.openConnection();
     con.setRequestMethod("GET");
-    BufferedReader in = new BufferedReader(
-        new InputStreamReader(con.getInputStream()));
-    String inputLine;
-    StringBuffer content = new StringBuffer();
-    while ((inputLine = in.readLine()) != null) {
-      content.append(inputLine);
+    StringBuilder content = new StringBuilder();
+    try (BufferedReader in = new BufferedReader(
+        new InputStreamReader(con.getInputStream()));) {
+      String inputLine;
+      while ((inputLine = in.readLine()) != null) {
+        content.append(inputLine);
+      }
     }
-    in.close();
+    return content.toString();
+  }
+
+  public static String httpPost(String urlString, String postData) throws IOException {
+    URL url = new URL(urlString);
+    HttpURLConnection con = (HttpURLConnection) url.openConnection();
+    con.setDoOutput(true);
+    con.setRequestMethod("POST");
+    StringBuilder content = new StringBuilder();
+    try (OutputStream outputStream = con.getOutputStream();) {
+      outputStream.write(postData.getBytes());
+      try (BufferedReader in = new BufferedReader(
+          new InputStreamReader(con.getInputStream()));) {
+        String inputLine;
+        while ((inputLine = in.readLine()) != null) {
+          content.append(inputLine);
+        }
+      }
+    }
+
     return content.toString();
   }
 }

common/src/test/java/org/apache/uniffle/common/rpc/StatusCodeTest.java

@@ -55,7 +55,7 @@ public void test() throws Exception {
         fail(e.getMessage());
       }
     }
-    for (int i = 0; i < statusCodes.size() - 1; i++) {
+    for (int i = 0; i < statusCodes.size(); i++) {
       assertEquals(protoStatusCode.get(i), statusCodes.get(i).toProto());
       assertEquals(StatusCode.fromProto(protoStatusCode.get(i)), statusCodes.get(i));
     }

coordinator/pom.xml

@@ -107,6 +107,7 @@
                 <includes>
                   <include>com.google.protobuf:protobuf-java-util</include>
                   <include>com.google.guava:guava</include>
+                  <include>com.google.guava:failureaccess</include>
                   <include>com.fasterxml.jackson.core:jackson-databind</include>
                   <include>com.fasterxml.jackson.core:jackson-core</include>
                 </includes>

coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorServer.java

@@ -41,6 +41,9 @@
 import org.apache.uniffle.coordinator.strategy.assignment.AssignmentStrategy;
 import org.apache.uniffle.coordinator.strategy.assignment.AssignmentStrategyFactory;
 import org.apache.uniffle.coordinator.util.CoordinatorUtils;
+import org.apache.uniffle.coordinator.web.servlet.CancelDecommissionServlet;
+import org.apache.uniffle.coordinator.web.servlet.DecommissionServlet;
+import org.apache.uniffle.coordinator.web.servlet.NodesServlet;
 
 import static org.apache.uniffle.common.config.RssBaseConf.RSS_SECURITY_HADOOP_KERBEROS_ENABLE;
 import static org.apache.uniffle.common.config.RssBaseConf.RSS_SECURITY_HADOOP_KERBEROS_KEYTAB_FILE;
@@ -153,6 +156,7 @@ private void initialization() throws Exception {
     id = ip + "-" + port;
     LOG.info("Start to initialize coordinator {}", id);
     jettyServer = new JettyServer(coordinatorConf);
+    registerRESTAPI();
     // register metrics first to avoid NPE problem when add dynamic metrics
     registerMetrics();
     coordinatorConf.setString(CoordinatorUtils.COORDINATOR_ID, id);
@@ -185,6 +189,19 @@ private void initialization() throws Exception {
     server = coordinatorFactory.getServer();
   }
 
+  private void registerRESTAPI() throws Exception {
+    LOG.info("Register REST API");
+    jettyServer.addServlet(
+        new NodesServlet(this),
+        "/api/server/nodes");
+    jettyServer.addServlet(
+        new DecommissionServlet(this),
+        "/api/server/decommission");
+    jettyServer.addServlet(
+        new CancelDecommissionServlet(this),
+        "/api/server/cancelDecommission");
+  }
+
   private void registerMetrics() throws Exception {
     LOG.info("Register metrics");
     CollectorRegistry coordinatorCollectorRegistry = new CollectorRegistry(true);

coordinator/src/main/java/org/apache/uniffle/coordinator/web/Response.java

@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.uniffle.coordinator.web;
+
+public class Response<T> {
+  private static final int SUCCESS_CODE = 0;
+  private static final int ERROR_CODE = -1;
+  private int code;
+  private T data;
+  private String errMsg;
+
+  public Response() {
+  }
+
+  public Response(int code, T data, String errMsg) {
+    this.code = code;
+    this.data = data;
+    this.errMsg = errMsg;
+  }
+
+  public static <T> Response<T> success(T data) {
+    return new Response<>(SUCCESS_CODE, data, null);
+  }
+
+  public static <T> Response<T> fail(String msg) {
+    return new Response<>(ERROR_CODE, null, msg);
+  }
+
+  public static <T> Response<T> fail(String msg, int code) {
+    return new Response<>(code, null, msg);
+  }
+
+  public int getCode() {
+    return code;
+  }
+
+  public void setCode(int code) {
+    this.code = code;
+  }
+
+  public T getData() {
+    return data;
+  }
+
+  public void setData(T data) {
+    this.data = data;
+  }
+
+  public String getErrMsg() {
+    return errMsg;
+  }
+
+  public void setErrMsg(String errMsg) {
+    this.errMsg = errMsg;
+  }
+}

coordinator/src/main/java/org/apache/uniffle/coordinator/web/request/CancelDecommissionRequest.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.uniffle.coordinator.web.request;
+
+import java.util.Set;
+
+public class CancelDecommissionRequest {
+  private Set<String> serverIds;
+
+  public Set<String> getServerIds() {
+    return serverIds;
+  }
+
+  public void setServerIds(Set<String> serverIds) {
+    this.serverIds = serverIds;
+  }
+}

coordinator/src/main/java/org/apache/uniffle/coordinator/web/request/DecommissionRequest.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.uniffle.coordinator.web.request;
+
+import java.util.Set;
+
+public class DecommissionRequest {
+  private Set<String> serverIds;
+
+  public Set<String> getServerIds() {
+    return serverIds;
+  }
+
+  public void setServerIds(Set<String> serverIds) {
+    this.serverIds = serverIds;
+  }
+}

coordinator/src/main/java/org/apache/uniffle/coordinator/web/servlet/BaseServlet.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.uniffle.coordinator.web.servlet;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.Callable;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.apache.uniffle.coordinator.web.Response;
+
+public abstract class BaseServlet extends HttpServlet {
+  public static final String JSON_MIME_TYPE = "application/json";
+  final ObjectMapper mapper = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);
+
+  @Override
+  protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+    writeJSON(resp, handlerRequest(() -> handleGet(req, resp)));
+  }
+
+  @Override
+  protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+    writeJSON(resp, handlerRequest(() -> handlePost(req, resp)));
+  }
+
+  private Response handlerRequest(
+      Callable<Response> function) {
+    Response response;
+    try {
+      // todo: Do something for authentication
+      response = function.call();
+    } catch (Exception e) {
+      response = Response.fail(e.getMessage());
+    }
+    return response;
+  }
+
+  protected Response handleGet(
+      HttpServletRequest req,
+      HttpServletResponse resp) throws ServletException, IOException {
+    throw new IOException("Method not support!");
+  }
+
+  protected Response handlePost(
+      HttpServletRequest req,
+      HttpServletResponse resp) throws ServletException, IOException {
+    throw new IOException("Method not support!");
+  }
+
+  protected void writeJSON(final HttpServletResponse resp, final Object obj)
+      throws IOException {
+    if (obj == null) {
+      return;
+    }
+    resp.setContentType(JSON_MIME_TYPE);
+    final OutputStream stream = resp.getOutputStream();
+    mapper.writeValue(stream, obj);
+  }
+
+  protected <T> T parseParamsFromJson(HttpServletRequest req, Class<T> clazz) throws IOException {
+    return mapper.readValue(req.getInputStream(), clazz);
+  }
+}

coordinator/src/main/java/org/apache/uniffle/coordinator/web/servlet/CancelDecommissionServlet.java

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.uniffle.coordinator.web.servlet;
+
+import java.io.IOException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import org.apache.uniffle.coordinator.ClusterManager;
+import org.apache.uniffle.coordinator.CoordinatorServer;
+import org.apache.uniffle.coordinator.web.Response;
+import org.apache.uniffle.coordinator.web.request.CancelDecommissionRequest;
+
+public class CancelDecommissionServlet extends BaseServlet {
+  private final CoordinatorServer coordinator;
+
+  public CancelDecommissionServlet(CoordinatorServer coordinator) {
+    this.coordinator = coordinator;
+  }
+
+  @Override
+  protected Response handlePost(HttpServletRequest req, HttpServletResponse resp) throws IOException {
+    CancelDecommissionRequest params = parseParamsFromJson(req, CancelDecommissionRequest.class);
+    if (CollectionUtils.isEmpty(params.getServerIds())) {
+      return Response.fail("Parameter[serverIds] should not be null!");
+    }
+    ClusterManager clusterManager = coordinator.getClusterManager();
+    params.getServerIds().forEach((serverId) -> {
+      clusterManager.cancelDecommission(serverId);
+    });
+    return Response.success(null);
+  }
+}