Add endpoint for rebuild (#161)

This adds an endpoint for rebuild operations that are performed asynchronously similar to cleanup and decommission.

Author: jsanda

management-api-agent-common/src/main/java/com/datastax/mgmtapi/NodeOpsProvider.java

@@ -171,6 +171,17 @@ public String decommission(@RpcParam(name="force") boolean force, @RpcParam(name
         return jobPair.left;
     }
 
+    public String rebuild(@RpcParam(name = "srcDatacenter") String srcDatacenter)
+    {
+        logger.debug("Starting rebuild");
+
+        Pair<String, CompletableFuture<Void>> jobPair = service.submit("rebuild",
+                () -> ShimLoader.instance.get().rebuild(srcDatacenter)
+        );
+
+        return jobPair.left;
+    }
+
     @Rpc(name = "setCompactionThroughput")
     public void setCompactionThroughput(@RpcParam(name="value") int value)
     {

management-api-common/src/main/java/com/datastax/mgmtapi/shims/CassandraAPI.java

@@ -48,6 +48,11 @@ default public boolean isFullQueryLogEnabled()
 
     void decommission(boolean force) throws InterruptedException;
 
+    default void rebuild(String srcDc)
+    {
+        getStorageService().rebuild(srcDc);
+    }
+
     Map<List<Long>, List<String>> checkConsistencyLevel(String consistencyLevelName, Integer rfPerDc);
 
     SeedProvider getK8SeedProvider();

management-api-server/src/main/java/com/datastax/mgmtapi/resources/models/FeatureSet.java

@@ -12,7 +12,8 @@ public class FeatureSet implements Serializable {
 
     public enum Feature {
         ASYNC_SSTABLE_TASKS("async_sstable_tasks"),
-        FULL_QUERY_LOGGING("full_query_logging");
+        FULL_QUERY_LOGGING("full_query_logging"),
+        REBUILD("rebuild");
 
         @JsonValue
         private String featureName;

management-api-server/src/main/java/com/datastax/mgmtapi/resources/v1/NodeOpsResources.java

@@ -35,4 +35,16 @@ public Response decommission(@QueryParam(value="force")boolean force)
                         ).build());
     }
 
+    @POST
+    @Path("/rebuild")
+    @Operation(summary = "Rebuild data by streaming data from other nodes. This operation returns immediately with a job id.")
+    @Produces(MediaType.TEXT_PLAIN)
+    public Response rebuild(@QueryParam("src_dc") String srcDatacenter)
+    {
+        return com.datastax.mgmtapi.resources.NodeOpsResources.handle(() ->
+                Response.accepted(
+                        ResponseTools.getSingleRowStringResponse(app.dbUnixSocketFile, cqlService, "CALL NodeOps.rebuild(?)", srcDatacenter)
+                ).build());
+    }
+
 }

management-api-server/src/test/java/com/datastax/mgmtapi/K8OperatorResourcesTest.java

@@ -175,6 +175,30 @@ public void testDecommissionMissingValue() throws Exception {
         verify(context.cqlService).executePreparedStatement(any(), eq("CALL NodeOps.decommission(?, ?)"), eq(false), eq(false));
     }
 
+    @Test
+    public void testRebuild() throws Exception {
+        Context context = setup();
+        String srcDc = "dc1";
+        String jobId = "0fe65b47-98c2-47d8-9c3c-5810c9988e10";
+
+        ResultSet mockResultSet = mock(ResultSet.class);
+        Row mockRow = mock(Row.class);
+
+        when(context.cqlService.executePreparedStatement(any(), anyString(), eq(srcDc))).thenReturn(mockResultSet);
+
+        when(mockResultSet.one()).thenReturn(mockRow);
+
+        when(mockRow.getString(0)).thenReturn(jobId);
+
+        MockHttpRequest request = MockHttpRequest.post("/api/v1/ops/node/rebuild?src_dc=dc1");
+        MockHttpResponse response = context.invoke(request);
+
+        Assert.assertEquals(HttpStatus.SC_ACCEPTED, response.getStatus());
+        assertEquals(jobId, response.getContentAsString());
+
+        verify(context.cqlService, timeout(500)).executePreparedStatement(any(), eq("CALL NodeOps.rebuild(?)"), eq(srcDc));
+    }
+
     @Test
     public void testSetCompactionThroughput() throws Exception {
         int value = 1;