Repository: thelastpickle/cassandra-reaper
Issue: millisSinceLastRepairForSchedule metric isn't cleaned up in distributed mode

Unlike schedules, which are persisted into the database, metrics only exist in memory for all Reaper instances.
When a schedule gets deleted, a cleanup is done to remove the corresponding `millisSinceLastRepairForSchedule` metric, but in distributed mode, only a single instance will get the deletion call. This means that we need to clean up this metric on a regular basis by checking if there's a corresponding schedule for all registered metrics.



┆Issue is synchronized with this [Jira Task](https://k8ssandra.atlassian.net/browse/K8SSAND-1629) by [Unito](https://www.unito.io)
┆friendlyId: K8SSAND-1629
┆priority: Medium


Changed files:
src/server/src/main/java/io/cassandrareaper/service/SchedulingManager.java, src/server/src/main/java/io/cassandrareaper/service/RepairScheduleService.java, src/server/src/test/java/io/cassandrareaper/service/SchedulingManagerTest.java

diff --git a/src/server/src/main/java/io/cassandrareaper/service/RepairScheduleService.java b/src/server/src/main/java/io/cassandrareaper/service/RepairScheduleService.java
index 2ee21e94..f5a9bdea 100644
--- a/src/server/src/main/java/io/cassandrareaper/service/RepairScheduleService.java
+++ b/src/server/src/main/java/io/cassandrareaper/service/RepairScheduleService.java
@@ -39,6 +39,7 @@ import static io.cassandrareaper.metrics.MetricNameUtils.cleanName;
 
 public final class RepairScheduleService {
 
+  public static final String MILLIS_SINCE_LAST_REPAIR_METRIC_NAME = "millisSinceLastRepairForSchedule";
   private final AppContext context;
   private final RepairUnitService repairUnitService;
 
@@ -143,7 +144,7 @@ public final class RepairScheduleService {
   private void registerScheduleMetrics(UUID repairScheduleId) {
     RepairSchedule schedule = context.storage.getRepairSchedule(repairScheduleId).get();
     RepairUnit repairUnit = context.storage.getRepairUnit(schedule.getRepairUnitId());
-    String metricName = metricName("millisSinceLastRepairForSchedule",
+    String metricName = metricName(MILLIS_SINCE_LAST_REPAIR_METRIC_NAME,
             repairUnit.getClusterName(),
             repairUnit.getKeyspaceName(),
             schedule.getId());
@@ -157,7 +158,7 @@ public final class RepairScheduleService {
     Optional<RepairSchedule> schedule = context.storage.getRepairSchedule(repairScheduleId);
     schedule.ifPresent(sched -> {
       RepairUnit repairUnit = context.storage.getRepairUnit(sched.getRepairUnitId());
-      String metricName = metricName("millisSinceLastRepairForSchedule",
+      String metricName = metricName(MILLIS_SINCE_LAST_REPAIR_METRIC_NAME,
           repairUnit.getClusterName(),
           repairUnit.getKeyspaceName(),
           sched.getId());
diff --git a/src/server/src/main/java/io/cassandrareaper/service/SchedulingManager.java b/src/server/src/main/java/io/cassandrareaper/service/SchedulingManager.java
index 5b995133..fdddb438 100644
--- a/src/server/src/main/java/io/cassandrareaper/service/SchedulingManager.java
+++ b/src/server/src/main/java/io/cassandrareaper/service/SchedulingManager.java
@@ -106,8 +106,11 @@ public final class SchedulingManager extends TimerTask {
       LOG.debug("Checking for repair schedules...");
       UUID lastId = null;
       try {
+        Collection<RepairSchedule> schedules = context.storage.getAllRepairSchedules();
+        // Cleanup metric registry from deleted schedules
+        cleanupMetricsRegistry(schedules);
+        // Start repairs for schedules that require it
         if (currentReaperIsSchedulingLeader()) {
-          Collection<RepairSchedule> schedules = context.storage.getAllRepairSchedules();
           boolean anyRunStarted = false;
           for (RepairSchedule schedule : schedules) {
             lastId = schedule.getId();
@@ -142,6 +145,19 @@ public final class SchedulingManager extends TimerTask {
     }
   }
 
+  // Cleanup metric registry from deleted schedules
+  // Such metrics are named after the following pattern:
+  //   "millisSinceLastRepairForSchedule.<cluster>.<keyspace>.<schedule id>"
+  @VisibleForTesting
+  void cleanupMetricsRegistry(Collection<RepairSchedule> schedules) {
+    // Cycle through the metrics registry and delete any metrics that are not in the current schedules
+    context.metricRegistry.getMetrics().keySet().stream()
+        .filter(key -> key.startsWith(RepairScheduleService.MILLIS_SINCE_LAST_REPAIR_METRIC_NAME))
+        .filter(key -> !schedules.stream().anyMatch(
+            schedule -> schedule.getId().toString().equals(key.split("\\\\.")[3])))
+        .forEach(context.metricRegistry::remove);
+  }
+
   /**
    * Manage, i.e. check whether a new repair run should be started with this schedule.
    *
diff --git a/src/server/src/test/java/io/cassandrareaper/service/SchedulingManagerTest.java b/src/server/src/test/java/io/cassandrareaper/service/SchedulingManagerTest.java
index a29acca3..be17df9f 100644
--- a/src/server/src/test/java/io/cassandrareaper/service/SchedulingManagerTest.java
+++ b/src/server/src/test/java/io/cassandrareaper/service/SchedulingManagerTest.java
@@ -28,13 +28,18 @@ import io.cassandrareaper.core.RepairUnit;
 import io.cassandrareaper.storage.CassandraStorage;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricRegistry;
 import com.datastax.driver.core.utils.UUIDs;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 
 import org.apache.cassandra.repair.RepairParallelism;
 import org.joda.time.DateTime;
@@ -362,4 +367,38 @@ public final class SchedulingManagerTest {
     // We're below the threshold and the repair shouldn't start
     Mockito.verify(context.repairManager, Mockito.times(0)).startRepairRun(any());
   }
+
+  @Test
+  public void cleanupMetricsRegistryTest() {
+    AppContext context = new AppContext();
+    context.storage = mock(CassandraStorage.class);
+    context.config = new ReaperApplicationConfiguration();
+    context.config.setPercentRepairedCheckIntervalMinutes(10);
+    context.metricRegistry = mock(MetricRegistry.class);
+    List<UUID> scheduleIds = Lists.newArrayList();
+    IntStream.range(0, 4).forEach(i -> scheduleIds.add(UUIDs.timeBased()));
+    HashMap<String, Metric> metrics = Maps.newHashMap();
+    scheduleIds.stream().forEach(scheduleId ->
+        metrics.put(MetricRegistry.name(RepairScheduleService.MILLIS_SINCE_LAST_REPAIR_METRIC_NAME,
+          "test", "test", scheduleId.toString()), null));
+
+    List<RepairSchedule> repairSchedules = scheduleIds.stream().map(scheduleId ->
+        RepairSchedule.builder(scheduleId)
+            .daysBetween(1)
+            .nextActivation(DateTime.now().plusDays(1))
+            .repairParallelism(RepairParallelism.PARALLEL)
+            .intensity(1)
+            .segmentCountPerNode(10)
+            .lastRun(UUIDs.timeBased())
+            .percentUnrepairedThreshold(5)
+            .state(RepairSchedule.State.ACTIVE)
+            .build(scheduleId)).collect(Collectors.toList());
+
+    // Removing a schedule should trigger the removal of one metric
+    repairSchedules.remove(0);
+    when(context.metricRegistry.getMetrics()).thenReturn(metrics);
+    SchedulingManager schedulingManager = SchedulingManager.create(context, () -> null);
+    schedulingManager.cleanupMetricsRegistry(repairSchedules);
+    Mockito.verify(context.metricRegistry, Mockito.times(1)).remove(any());
+  }
 }