Topology change resilience for incremental repair (#1235)

Allows incremental repair to survive nodes changing IP address during the repair.
The hostID is now stored in each segment and the ip address is recomputed from it when the segment runs.

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

@@ -44,6 +44,9 @@ public final class RepairSegment {
   private final DateTime startTime;
   private final DateTime endTime;
   private final Map<String, String> replicas;
+  // hostID field is only ever populated for incremental repairs. For full repairs it is always null.
+  private final UUID hostID;
+
 
   private RepairSegment(Builder builder, @Nullable UUID id) {
     this.id = id;
@@ -58,6 +61,7 @@ private RepairSegment(Builder builder, @Nullable UUID id) {
     this.replicas = builder.replicas != null
         ? ImmutableMap.copyOf(builder.replicas)
         : null;
+    this.hostID = builder.hostID;
   }
 
   public static Builder builder(Segment tokenRange, UUID repairUnitId) {
@@ -129,6 +133,10 @@ public boolean hasEndTime() {
     return null != endTime;
   }
 
+  public UUID getHostID() {
+    return hostID;
+  }
+
   /** Reset to NOT_STARTED state, with nulled startTime and endTime. */
   public Builder reset() {
     Builder builder = new Builder(this);
@@ -159,6 +167,7 @@ public static final class Builder {
     private DateTime startTime;
     private DateTime endTime;
     private Map<String, String> replicas;
+    private UUID hostID;
 
     private Builder() {}
 
@@ -170,6 +179,7 @@ private Builder(Segment tokenRange, UUID repairUnitId) {
       this.failCount = 0;
       this.state = State.NOT_STARTED;
       this.replicas = tokenRange.getReplicas();
+      this.hostID = null;
     }
 
     private Builder(RepairSegment original) {
@@ -183,6 +193,7 @@ private Builder(RepairSegment original) {
       startTime = original.startTime;
       endTime = original.endTime;
       replicas = original.replicas;
+      hostID = original.hostID;
     }
 
     public Builder withRunId(UUID runId) {
@@ -242,6 +253,12 @@ public Builder withReplicas(Map<String, String> replicas) {
       return this;
     }
 
+    public Builder withHostID(UUID hostID) {
+      this.hostID = hostID;
+      return this;
+    }
+
+
     public RepairSegment build() {
       // a null segmentId is a special case where the storage uses a sequence for it
       Preconditions.checkNotNull(runId);

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

@@ -30,14 +30,17 @@
 import io.cassandrareaper.jmx.JmxProxy;
 
 import java.math.BigInteger;
+
 import java.util.Arrays;
+
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
@@ -124,7 +127,7 @@ public RepairRun registerRepairRun(
 
     // the last preparation step is to generate actual repair segments
     List<RepairSegment.Builder> segmentBuilders = repairUnit.getIncrementalRepair()
-        ? createRepairSegmentsForIncrementalRepair(nodes, repairUnit)
+        ? createRepairSegmentsForIncrementalRepair(nodes, repairUnit, cluster, clusterFacade)
         : createRepairSegments(tokenSegments, repairUnit);
 
     RepairRun repairRun = context.storage.addRepairRun(runBuilder, segmentBuilders);
@@ -333,23 +336,27 @@ private static List<RepairSegment.Builder> createRepairSegments(
   @VisibleForTesting
   static List<RepairSegment.Builder> createRepairSegmentsForIncrementalRepair(
       Map<String, RingRange> nodes,
-      RepairUnit repairUnit) {
+      RepairUnit repairUnit,
+      Cluster cluster,
+      ClusterFacade clusterFacade) throws ReaperException {
+
+    Map<String, String> endpointHostIdMap = clusterFacade.getEndpointToHostId(cluster);
 
     List<RepairSegment.Builder> repairSegmentBuilders = Lists.newArrayList();
 
     nodes
         .entrySet()
-        .forEach(
-            range ->
-                repairSegmentBuilders.add(
-                    RepairSegment.builder(
-                            Segment.builder()
-                                .withTokenRanges(Arrays.asList(range.getValue()))
-                                .build(),
-                            repairUnit.getId())
-                        .withReplicas(Collections.emptyMap())
-                        .withCoordinatorHost(range.getKey())));
-
+        .forEach(range -> {
+          RepairSegment.Builder segment = RepairSegment.builder(
+                  Segment.builder()
+                      .withTokenRanges(Arrays.asList(range.getValue()))
+                      .build(),
+                  repairUnit.getId())
+              .withReplicas(Collections.emptyMap())
+              .withCoordinatorHost(range.getKey())
+              .withHostID(UUID.fromString(endpointHostIdMap.get(range.getKey())));
+          repairSegmentBuilders.add(segment);
+        });
     return repairSegmentBuilders;
   }
 

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

@@ -35,7 +35,6 @@
 import io.cassandrareaper.storage.IDistributedStorage;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -462,13 +461,24 @@ private void startNextSegment() throws ReaperException, InterruptedException {
                 repairRunId);
 
     Optional<RepairSegment> nextRepairSegment = Optional.empty();
-    Collection<String> potentialReplicas = new HashSet<>();
+    final Collection<String> potentialReplicas = new HashSet<>();
     for (RepairSegment segment : nextRepairSegments) {
       Map<String, String> potentialReplicaMap = this.repairRunService.getDCsByNodeForRepairSegment(
           cluster, segment.getTokenRange(), repairUnit.getKeyspaceName(), repairUnit);
-      potentialReplicas = repairUnit.getIncrementalRepair()
-          ? Collections.singletonList(segment.getCoordinatorHost())
-          : potentialReplicaMap.keySet();
+      if (repairUnit.getIncrementalRepair()) {
+        Map<String, String> endpointHostIdMap = clusterFacade.getEndpointToHostId(cluster);
+        if (segment.getHostID() == null) {
+          throw new ReaperException(
+            String.format("No host ID for repair segment %s", segment.getId().toString())
+          );
+        }
+        endpointHostIdMap.entrySet().stream()
+            .filter(entry -> entry.getValue().equals(segment.getHostID().toString()))
+            .forEach(entry -> potentialReplicas.add(entry.getKey()));
+      } else {
+        potentialReplicas.addAll(potentialReplicaMap.keySet());
+      }
+      LOG.debug("Potential replicas for segment {}: {}", segment.getId(), potentialReplicas);
       JmxProxy coordinator = clusterFacade.connect(cluster, potentialReplicas);
       if (nodesReadyForNewRepair(coordinator, segment, potentialReplicaMap, repairRunId)) {
         nextRepairSegment = Optional.of(segment);
@@ -533,6 +543,7 @@ private boolean nodesReadyForNewRepair(
       UUID segmentId) {
 
     Collection<String> nodes = getNodesInvolvedInSegment(dcByNode);
+    LOG.debug("Nodes involved in segment {}: {}", segmentId, nodes);
     String dc = EndpointSnitchInfoProxy.create(coordinator).getDataCenter();
     boolean requireAllHostMetrics = DatacenterAvailability.LOCAL != context.config.getDatacenterAvailability();
     boolean allLocalDcHostsChecked = true;
@@ -577,6 +588,7 @@ private boolean nodesReadyForNewRepair(
       LOG.debug("Ok to repair segment '{}' on repair run with id '{}'", segment.getId(), segment.getRunId());
       return true;
     } else {
+      LOG.debug("Couldn't get metrics for hosts {}, will retry later", unreachableNodes);
       String msg = String.format(
           "Postponed repair segment %s on repair run with id %s because we couldn't get %shosts metrics on %s",
           segment.getId(),
@@ -635,7 +647,7 @@ private boolean repairSegment(final UUID segmentId, Segment segment, Collection<
     String keyspace = repairUnit.getKeyspaceName();
     LOG.debug("preparing to repair segment {} on run with id {}", segmentId, repairRunId);
 
-    List<String> potentialCoordinators;
+    List<String> potentialCoordinators = Lists.newArrayList();
     if (!repairUnit.getIncrementalRepair()) {
       // full repair
       try {
@@ -675,7 +687,7 @@ private boolean repairSegment(final UUID segmentId, Segment segment, Collection<
       Thread.sleep(ThreadLocalRandom.current().nextInt(10, 100) * 100);
       Optional<RepairSegment> rs = context.storage.getRepairSegment(repairRunId, segmentId);
       if (rs.isPresent()) {
-        potentialCoordinators = Arrays.asList(rs.get().getCoordinatorHost());
+        potentialCoordinators.addAll(segmentReplicas);
       } else {
         // the segment has been removed. should only happen in tests on backends that delete repair segments.
         return false;

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

@@ -444,34 +444,34 @@ private void prepareStatements() {
         .prepare(
             "INSERT INTO repair_run"
                 + "(id,segment_id,repair_unit_id,start_token,end_token,"
-                + " segment_state,fail_count, token_ranges, replicas)"
-                + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)")
+                + " segment_state,fail_count, token_ranges, replicas,host_id)"
+                + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
         .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
     insertRepairSegmentIncrementalPrepStmt = session
         .prepare(
             "INSERT INTO repair_run"
                 + "(id,segment_id,repair_unit_id,start_token,end_token,"
-                + "segment_state,coordinator_host,fail_count,replicas)"
-                + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)")
+                + "segment_state,coordinator_host,fail_count,replicas,host_id)"
+                + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
         .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
     updateRepairSegmentPrepStmt = session
         .prepare(
             "INSERT INTO repair_run"
-                + "(id,segment_id,segment_state,coordinator_host,segment_start_time,fail_count)"
-                + " VALUES(?, ?, ?, ?, ?, ?)")
+                + "(id,segment_id,segment_state,coordinator_host,segment_start_time,fail_count,host_id)"
+                + " VALUES(?, ?, ?, ?, ?, ?, ?)")
         .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
     insertRepairSegmentEndTimePrepStmt = session
         .prepare("INSERT INTO repair_run(id, segment_id, segment_end_time) VALUES(?, ?, ?)")
         .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
     getRepairSegmentPrepStmt = session
             .prepare(
                 "SELECT id,repair_unit_id,segment_id,start_token,end_token,segment_state,coordinator_host,"
-                    + "segment_start_time,segment_end_time,fail_count, token_ranges, replicas"
+                    + "segment_start_time,segment_end_time,fail_count, token_ranges, replicas, host_id"
                     + " FROM repair_run WHERE id = ? and segment_id = ?")
             .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
     getRepairSegmentsByRunIdPrepStmt = session.prepare(
         "SELECT id,repair_unit_id,segment_id,start_token,end_token,segment_state,coordinator_host,segment_start_time,"
-            + "segment_end_time,fail_count, token_ranges, replicas FROM repair_run WHERE id = ?");
+            + "segment_end_time,fail_count, token_ranges, replicas, host_id FROM repair_run WHERE id = ?");
     getRepairSegmentCountByRunIdPrepStmt = session.prepare("SELECT count(*) FROM repair_run WHERE id = ?");
     prepareScheduleStatements();
     prepareLeaderElectionStatements(timeUdf);
@@ -504,7 +504,7 @@ private void prepareStatements() {
       try {
         getRepairSegmentsByRunIdAndStatePrepStmt = session.prepare(
             "SELECT id,repair_unit_id,segment_id,start_token,end_token,segment_state,coordinator_host,"
-                + "segment_start_time,segment_end_time,fail_count, token_ranges, replicas FROM repair_run "
+                + "segment_start_time,segment_end_time,fail_count, token_ranges, replicas, host_id FROM repair_run "
                 + "WHERE id = ? AND segment_state = ? ALLOW FILTERING");
         getRepairSegmentCountByRunIdAndStatePrepStmt = session.prepare(
             "SELECT count(segment_id) FROM repair_run WHERE id = ? AND segment_state = ? ALLOW FILTERING");
@@ -817,7 +817,10 @@ public RepairRun addRepairRun(Builder repairRun, Collection<RepairSegment.Builde
               segment.getState().ordinal(),
               segment.getCoordinatorHost(),
               segment.getFailCount(),
-              segment.getReplicas()));
+              segment.getReplicas(),
+              segment.getHostID()
+            )
+        );
       } else {
         try {
           repairRunBatch.add(
@@ -830,7 +833,10 @@ public RepairRun addRepairRun(Builder repairRun, Collection<RepairSegment.Builde
                   segment.getState().ordinal(),
                   segment.getFailCount(),
                   objectMapper.writeValueAsString(segment.getTokenRange().getTokenRanges()),
-                  segment.getReplicas()));
+                  segment.getReplicas(),
+                  segment.getHostID()
+              )
+          );
         } catch (JsonProcessingException e) {
           throw new IllegalStateException(e);
         }
@@ -1134,7 +1140,10 @@ public boolean updateRepairSegmentUnsafe(RepairSegment segment) {
             segment.getState().ordinal(),
             segment.getCoordinatorHost(),
             segment.hasStartTime() ? segment.getStartTime().toDate() : null,
-            segment.getFailCount()));
+            segment.getFailCount(),
+            segment.getHostID()
+        )
+    );
 
     if (null != segment.getEndTime() || State.NOT_STARTED == segment.getState()) {
 
@@ -1228,6 +1237,10 @@ private static RepairSegment createRepairSegmentFromRow(Row segmentRow) {
       builder = builder.withReplicas(segmentRow.getMap("replicas", String.class, String.class));
     }
 
+    if (null != segmentRow.getUUID("host_id")) {
+      builder = builder.withHostID(segmentRow.getUUID("host_id"));
+    }
+
     return builder.withId(segmentRow.getUUID("segment_id")).build();
   }
 

src/server/src/main/resources/db/cassandra/031_add_hostID.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.
+--
+-- Store the percent unrepaired threshold to trigger incremental repair schedules
+
+ALTER TABLE repair_run ADD host_id uuid;

src/server/src/test/java/io/cassandrareaper/service/RepairRunServiceTest.java

@@ -38,11 +38,13 @@
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.Semaphore;
 
 import com.datastax.driver.core.utils.UUIDs;
@@ -493,6 +495,12 @@ public void failIncrRepairRunCreationTest() throws ReaperException, UnknownHostE
         .thenReturn((Map)ImmutableMap.of(Lists.newArrayList("0", "100"), Lists.newArrayList(NODES)));
     when(clusterFacade.getCassandraVersion(any())).thenReturn("3.11.6");
     when(clusterFacade.getTokens(any())).thenReturn(TOKENS);
+    when(clusterFacade.getEndpointToHostId(any(Cluster.class))).thenReturn(Collections.emptyMap());
+    Map<String, String> endpointToHostIDMap = new HashMap<String, String>();
+    endpointToHostIDMap.put("127.0.0.1", UUID.randomUUID().toString());
+    endpointToHostIDMap.put("127.0.0.2", UUID.randomUUID().toString());
+    endpointToHostIDMap.put("127.0.0.3", UUID.randomUUID().toString());
+    when(clusterFacade.getEndpointToHostId(any(Cluster.class))).thenReturn(endpointToHostIDMap);
 
     RepairRunService repairRunService = RepairRunService.create(context, () -> clusterFacade);
 
@@ -634,7 +642,7 @@ public JmxProxy connectImpl(Node host) throws ReaperException {
   }
 
   @Test
-  public void createRepairSegmentsForIncrementalRepairTest() {
+  public void createRepairSegmentsForIncrementalRepairTest() throws ReaperException {
     final String KS_NAME = "reaper";
     final Set<String> CF_NAMES = Sets.newHashSet("reaper");
     final boolean INCREMENTAL_REPAIR = false;
@@ -643,7 +651,7 @@ public void createRepairSegmentsForIncrementalRepairTest() {
     final Set<String> BLACKLISTED_TABLES = Collections.emptySet();
     final int REPAIR_THREAD_COUNT = 1;
     final int segmentTimeout = 30;
-    Cluster cluster = Cluster.builder()
+    final Cluster cluster = Cluster.builder()
         .withName("test_" + RandomStringUtils.randomAlphabetic(12))
         .withSeedHosts(ImmutableSet.of("127.0.0.1", "127.0.0.2", "127.0.0.3"))
         .withState(Cluster.State.ACTIVE)
@@ -654,6 +662,12 @@ public void createRepairSegmentsForIncrementalRepairTest() {
     nodes.put("127.0.0.1", new RingRange("1", "2"));
     nodes.put("127.0.0.2", new RingRange("3", "4"));
 
+    Map<String, String> endpointToHostIDMap = new HashMap<String, String>();
+    endpointToHostIDMap.put("127.0.0.1", UUID.randomUUID().toString());
+    endpointToHostIDMap.put("127.0.0.2", UUID.randomUUID().toString());
+    endpointToHostIDMap.put("127.0.0.3", UUID.randomUUID().toString());
+    ClusterFacade clusterFacade = mock(ClusterFacade.class);
+    when(clusterFacade.getEndpointToHostId(any(Cluster.class))).thenReturn(endpointToHostIDMap);
     RepairUnit repairUnit = RepairUnit.builder()
         .clusterName(cluster.getName())
         .keyspaceName(KS_NAME)
@@ -665,9 +679,8 @@ public void createRepairSegmentsForIncrementalRepairTest() {
         .repairThreadCount(REPAIR_THREAD_COUNT)
         .incrementalRepair(true)
         .timeout(segmentTimeout).build(UUIDs.timeBased());
-
     List<RepairSegment.Builder> segmentBuilders
-        = RepairRunService.createRepairSegmentsForIncrementalRepair(nodes, repairUnit);
+        = RepairRunService.createRepairSegmentsForIncrementalRepair(nodes, repairUnit, cluster, clusterFacade);
     assertEquals("Not enough segment builders were created", 2, segmentBuilders.size());
   }