Improve status reporting in UI

src/server/src/main/java/io/cassandrareaper/core/RepairRun.java

@@ -180,12 +180,15 @@ public String toString() {
     return String.format("%s[%s] for %s", getClass().getSimpleName(), id.toString(), clusterName);
   }
 
+  // The values in this enum are declared in order of "interestingness",
+  // this is used to order RepairRuns in the UI so that e.g. RUNNING runs come first.
   public enum RunState {
-    NOT_STARTED,
     RUNNING,
+    PAUSED,
+    NOT_STARTED,
+
     ERROR,
     DONE,
-    PAUSED,
     ABORTED,
     DELETED;
 
@@ -341,4 +344,5 @@ public RepairRun build(UUID id) {
       return new RepairRun(this, id);
     }
   }
+
 }

src/server/src/main/java/io/cassandrareaper/resources/RepairRunResource.java

@@ -63,6 +63,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.lang.Math.min;
+
 
 @Path("/repair_run")
 @Produces(MediaType.APPLICATION_JSON)
@@ -608,7 +610,9 @@ public Response getRepairRunsForCluster(
       @QueryParam("limit") Optional<Integer> limit) {
 
     LOG.debug("get repair run for cluster called with: cluster_name = {}", clusterName);
-    final Collection<RepairRun> repairRuns = context.storage.getRepairRunsForCluster(clusterName, limit);
+    final Collection<RepairRun> repairRuns = context
+        .storage
+        .getRepairRunsForClusterPrioritiseRunning(clusterName, limit);
     final Collection<RepairRunStatus> repairRunViews = new ArrayList<>();
     for (final RepairRun repairRun : repairRuns) {
       repairRunViews.add(getRepairRunStatus(repairRun));
@@ -661,17 +665,18 @@ public Response listRepairRuns(
             : context.storage.getClusters();
 
       List<RepairRunStatus> runStatuses = Lists.newArrayList();
-      for (final Cluster clstr : clusters) {
-        Collection<RepairRun> runs = context.storage.getRepairRunsForCluster(clstr.getName(), limit);
-
-        runStatuses.addAll(
-            (List<RepairRunStatus>) getRunStatuses(runs, desiredStates)
-                .stream()
-                .filter((run) -> !keyspace.isPresent()
-                    || ((RepairRunStatus)run).getKeyspaceName().equals(keyspace.get()))
-                .collect(Collectors.toList()));
-      }
-
+      List<RepairRun> repairRuns = Lists.newArrayList();
+      clusters.forEach(clstr -> repairRuns.addAll(
+          context.storage.getRepairRunsForClusterPrioritiseRunning(clstr.getName(), limit))
+      );
+      RepairRunService.sortByRunState(repairRuns);
+      runStatuses.addAll(
+          (List<RepairRunStatus>) getRunStatuses(
+              repairRuns.subList(0, min(repairRuns.size(), limit.orElse(1000))), desiredStates)
+              .stream()
+              .filter((run) -> !keyspace.isPresent()
+                  || ((RepairRunStatus)run).getKeyspaceName().equals(keyspace.get()))
+              .collect(Collectors.toList()));
       return Response.ok().entity(runStatuses).build();
     } catch (IllegalArgumentException e) {
       return Response.serverError().entity("Failed find cluster " + cluster.get()).build();

src/server/src/main/java/io/cassandrareaper/service/RepairRunService.java

@@ -35,6 +35,7 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -85,6 +86,22 @@ public static RepairRunService create(AppContext context) {
     return new RepairRunService(context, () -> ClusterFacade.create(context));
   }
 
+  public static void sortByRunState(List<RepairRun> repairRunCollection) {
+    Comparator<RepairRun> comparator = new Comparator<RepairRun>() {
+      @Override
+      public int compare(RepairRun o1, RepairRun o2) {
+        if (!o1.getRunState().isTerminated() && o2.getRunState().isTerminated()) {
+          return -1; // o1 appears first.
+        }  else if (o1.getRunState().isTerminated() && !o2.getRunState().isTerminated()) {
+          return 1; // o2 appears first.
+        } else { // Both RunStates have equal isFinished() values; compare on time instead.
+          return o1.getId().compareTo(o2.getId());
+        }
+      }
+    };
+    Collections.sort(repairRunCollection, comparator);
+  }
+
   /**
    * Creates a repair run but does not start it immediately.
    *

src/server/src/main/java/io/cassandrareaper/storage/CassandraStorage.java

@@ -47,6 +47,7 @@
 import java.io.IOException;
 import java.math.BigInteger;
 import java.time.LocalDate;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
@@ -114,6 +115,8 @@
 import systems.composable.dropwizard.cassandra.pooling.PoolingOptionsFactory;
 import systems.composable.dropwizard.cassandra.retry.RetryPolicyFactory;
 
+import static java.lang.Math.min;
+
 public final class CassandraStorage implements IStorage, IDistributedStorage {
 
   private static final int METRICS_PARTITIONING_TIME_MINS = 10;
@@ -157,6 +160,7 @@ public RepairUnit load(UUID repairUnitId) throws Exception {
   private PreparedStatement insertRepairRunUnitIndexPrepStmt;
   private PreparedStatement getRepairRunPrepStmt;
   private PreparedStatement getRepairRunForClusterPrepStmt;
+  private PreparedStatement getRepairRunForClusterWhereStatusPrepStmt;
   private PreparedStatement getRepairRunForUnitPrepStmt;
   private PreparedStatement deleteRepairRunPrepStmt;
   private PreparedStatement deleteRepairRunByClusterPrepStmt;
@@ -422,6 +426,8 @@ private void prepareStatements() {
         .setConsistencyLevel(ConsistencyLevel.QUORUM);
     getRepairRunForClusterPrepStmt = session.prepare(
         "SELECT * FROM repair_run_by_cluster_v2 WHERE cluster_name = ? limit ?");
+    getRepairRunForClusterWhereStatusPrepStmt = session.prepare(
+        "SELECT id FROM repair_run_by_cluster_v2 WHERE cluster_name = ? AND repair_run_state = ? limit ?");
     getRepairRunForUnitPrepStmt = session.prepare("SELECT * FROM repair_run_by_unit WHERE repair_unit_id = ?");
     deleteRepairRunPrepStmt = session.prepare("DELETE FROM repair_run WHERE id = ?");
     deleteRepairRunByClusterPrepStmt
@@ -957,6 +963,69 @@ public Collection<RepairRun> getRepairRunsForCluster(String clusterName, Optiona
     return getRepairRunsAsync(repairRunFutures);
   }
 
+  @Override
+  public List<RepairRun> getRepairRunsForClusterPrioritiseRunning(String clusterName, Optional<Integer> limit) {
+    List<ResultSetFuture> repairUuidFuturesByState = Lists.<ResultSetFuture>newArrayList();
+    // We've set up the RunState enum so that values are declared in order of "interestingness",
+    // we iterate over the table via the secondary index according to that ordering.
+    for (String state:Arrays.asList("RUNNING", "PAUSED", "NOT_STARTED")) {
+      repairUuidFuturesByState.add(
+          // repairUUIDFutures will be a List of resultSetFutures, each of which contains a ResultSet of
+          // UUIDs for one status.
+          session
+              .executeAsync(getRepairRunForClusterWhereStatusPrepStmt
+                  .bind(clusterName, state.toString(), limit.orElse(MAX_RETURNED_REPAIR_RUNS)
+                  )
+              )
+      );
+    }
+    ResultSetFuture repairUuidFuturesNoState = session
+            .executeAsync(getRepairRunForClusterPrepStmt
+                .bind(clusterName, limit.orElse(MAX_RETURNED_REPAIR_RUNS)
+                )
+            );
+
+    List<UUID> flattenedUuids = Lists.<UUID>newArrayList();
+    // Flatten the UUIDs from each status down into a single array.
+    for (ResultSetFuture idResSetFuture : repairUuidFuturesByState) {
+      idResSetFuture
+          .getUninterruptibly()
+          .forEach(
+              row -> flattenedUuids.add(row.getUUID("id"))
+        );
+    }
+    // Merge the two lists and trim.
+    repairUuidFuturesNoState.getUninterruptibly().forEach(row -> {
+          UUID uuid = row.getUUID("id");
+          if (!flattenedUuids.contains(uuid)) {
+            flattenedUuids.add(uuid);
+          }
+        }
+    );
+    flattenedUuids.subList(0, min(flattenedUuids.size(), limit.orElse(MAX_RETURNED_REPAIR_RUNS)));
+
+    // Run an async query on each UUID in the flattened list, against the main repair_run table with
+    // all columns required as an input to `buildRepairRunFromRow`.
+    List<ResultSetFuture> repairRunFutures = Lists.<ResultSetFuture>newArrayList();
+    flattenedUuids.forEach(uuid ->
+        repairRunFutures.add(
+            session
+                .executeAsync(getRepairRunPrepStmt.bind(uuid)
+                )
+        )
+    );
+    // Defuture the repair_run rows and build the strongly typed RepairRun objects from the contents.
+    return repairRunFutures
+        .stream()
+        .map(
+            row -> {
+              Row extractedRow = row.getUninterruptibly().one();
+              return buildRepairRunFromRow(extractedRow, extractedRow.getUUID("id"));
+            }
+        ).collect(Collectors.toList());
+  }
+
+
   @Override
   public Collection<RepairRun> getRepairRunsForUnit(UUID repairUnitId) {
     List<ResultSetFuture> repairRunFutures = Lists.<ResultSetFuture>newArrayList();

src/server/src/main/java/io/cassandrareaper/storage/IStorage.java

@@ -71,6 +71,8 @@ public interface IStorage extends Managed {
   /** return all the repair runs in a cluster, in reverse chronological order, with default limit is 1000 */
   Collection<RepairRun> getRepairRunsForCluster(String clusterName, Optional<Integer> limit);
 
+  Collection<RepairRun> getRepairRunsForClusterPrioritiseRunning(String clusterName, Optional<Integer> limit);
+
   Collection<RepairRun> getRepairRunsForUnit(UUID repairUnitId);
 
   Collection<RepairRun> getRepairRunsWithState(RepairRun.RunState runState);

src/server/src/main/java/io/cassandrareaper/storage/MemoryStorage.java

@@ -27,12 +27,14 @@
 import io.cassandrareaper.core.Snapshot;
 import io.cassandrareaper.resources.view.RepairRunStatus;
 import io.cassandrareaper.resources.view.RepairScheduleStatus;
+import io.cassandrareaper.service.RepairRunService;
 
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
@@ -53,6 +55,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.lang.Math.min;
+
 /**
  * Implements the StorageAPI using transient Java classes.
  */
@@ -193,6 +197,18 @@ public List<RepairRun> getRepairRunsForCluster(String clusterName, Optional<Inte
     return foundRepairRuns;
   }
 
+  @Override
+  public List<RepairRun> getRepairRunsForClusterPrioritiseRunning(String clusterName, Optional<Integer> limit) {
+    List<RepairRun> foundRepairRuns = repairRuns
+        .values()
+        .stream()
+        .filter(
+            row -> row.getClusterName().equals(clusterName.toLowerCase(Locale.ROOT))).collect(Collectors.toList()
+        );
+    RepairRunService.sortByRunState(foundRepairRuns);
+    return foundRepairRuns.subList(0, min(foundRepairRuns.size(), limit.orElse(1000)));
+  }
+
   @Override
   public Collection<RepairRun> getRepairRunsForUnit(UUID repairUnitId) {
     List<RepairRun> foundRepairRuns = new ArrayList<>();

src/server/src/main/resources/db/cassandra/032_add_2i_status.cql

@@ -0,0 +1,18 @@
+--
+--  Copyright 2021-2021 Datastax inc.
+--
+--  Licensed 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.
+--
+-- Add a secondary index on `state` to the `repair_run` table.
+
+CREATE INDEX IF NOT EXISTS state2i ON repair_run_by_cluster_v2 (repair_run_state);