RepairRunResource { @POST public Response addRepairRun( @Context UriInfo uriInfo, @QueryParam("clusterName") Optional<String> clusterName, @QueryParam("keyspace") Optional<String> keyspace, @QueryParam("tables") Optional<String> tableNamesParam, @QueryParam("owner") Optional<String> owner, @QueryParam("cause") Optional<String> cause, @QueryParam("segmentCount") Optional<Integer> segmentCountPerNode, @QueryParam("repairParallelism") Optional<String> repairParallelism, @QueryParam("intensity") Optional<String> intensityStr, @QueryParam("incrementalRepair") Optional<String> incrementalRepairStr, @QueryParam("nodes") Optional<String> nodesToRepairParam, @QueryParam("datacenters") Optional<String> datacentersToRepairParam, @QueryParam("blacklistedTables") Optional<String> blacklistedTableNamesParam, @QueryParam("repairThreadCount") Optional<Integer> repairThreadCountParam) { try { final Response possibleFailedResponse = RepairRunResource.checkRequestForAddRepair( context, clusterName, keyspace, tableNamesParam, owner, segmentCountPerNode, repairParallelism, intensityStr, incrementalRepairStr, nodesToRepairParam, datacentersToRepairParam, blacklistedTableNamesParam, repairThreadCountParam); if (null != possibleFailedResponse) { return possibleFailedResponse; } Double intensity; if (intensityStr.isPresent()) { intensity = Double.parseDouble(intensityStr.get()); } else { intensity = context.config.getRepairIntensity(); LOG.debug("no intensity given, so using default value: {}", intensity); } boolean incrementalRepair; if (incrementalRepairStr.isPresent()) { incrementalRepair = Boolean.parseBoolean(incrementalRepairStr.get()); } else { incrementalRepair = context.config.getIncrementalRepair(); LOG.debug("no incremental repair given, so using default value: {}", incrementalRepair); } int segments = context.config.getSegmentCountPerNode(); if (!incrementalRepair) { if (segmentCountPerNode.isPresent()) { LOG.debug( "using given segment count {} instead of configured value {}", segmentCountPerNode.get(), context.config.getSegmentCount()); segments = segmentCountPerNode.get(); } } else { segments = -1; } final Cluster cluster = context.storage.getCluster(Cluster.toSymbolicName(clusterName.get())); Set<String> tableNames; try { tableNames = repairRunService.getTableNamesBasedOnParam(cluster, keyspace.get(), tableNamesParam); } catch (IllegalArgumentException ex) { LOG.error(ex.getMessage(), ex); return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build(); } Set<String> blacklistedTableNames; try { blacklistedTableNames = repairRunService.getTableNamesBasedOnParam(cluster, keyspace.get(), blacklistedTableNamesParam); } catch (IllegalArgumentException ex) { LOG.error(ex.getMessage(), ex); return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build(); } final Set<String> nodesToRepair; try { nodesToRepair = repairRunService.getNodesToRepairBasedOnParam(cluster, nodesToRepairParam); } catch (IllegalArgumentException ex) { LOG.error(ex.getMessage(), ex); return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build(); } final Set<String> datacentersToRepair; try { datacentersToRepair = RepairRunService .getDatacentersToRepairBasedOnParam(cluster, datacentersToRepairParam); } catch (IllegalArgumentException ex) { LOG.error(ex.getMessage(), ex); return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build(); } RepairUnit.Builder builder = RepairUnit.builder() .clusterName(cluster.getName()) .keyspaceName(keyspace.get()) .columnFamilies(tableNames) .incrementalRepair(incrementalRepair) .nodes(nodesToRepair) .datacenters(datacentersToRepair) .blacklistedTables(blacklistedTableNames) .repairThreadCount(repairThreadCountParam.orElse(context.config.getRepairThreadCount())); final RepairUnit theRepairUnit = repairUnitService.getOrCreateRepairUnit(cluster, builder); if (theRepairUnit.getIncrementalRepair() != incrementalRepair) { String msg = String.format( "A repair unit %s already exist for the same cluster/keyspace/tables" + " but with a different incremental repair value. Requested value %s | Existing value: %s", theRepairUnit.getId(), incrementalRepair, theRepairUnit.getIncrementalRepair()); return Response.status(Response.Status.BAD_REQUEST).entity(msg).build(); } RepairParallelism parallelism = context.config.getRepairParallelism(); if (repairParallelism.isPresent()) { LOG.debug( "using given repair parallelism {} instead of configured value {}", repairParallelism.get(), context.config.getRepairParallelism()); parallelism = RepairParallelism.valueOf(repairParallelism.get().toUpperCase()); } if (incrementalRepair) { parallelism = RepairParallelism.PARALLEL; } final RepairRun newRepairRun = repairRunService.registerRepairRun( cluster, theRepairUnit, cause, owner.get(), 0, segments, parallelism, intensity); return Response.created(buildRepairRunUri(uriInfo, newRepairRun)) .entity(new RepairRunStatus(newRepairRun, theRepairUnit, 0)) .build(); } catch (ReaperException e) { LOG.error(e.getMessage(), e); return Response.serverError().entity(e.getMessage()).build(); } } RepairRunResource(AppContext context); @POST Response addRepairRun(
      @Context UriInfo uriInfo,
      @QueryParam("clusterName") Optional<String> clusterName,
      @QueryParam("keyspace") Optional<String> keyspace,
      @QueryParam("tables") Optional<String> tableNamesParam,
      @QueryParam("owner") Optional<String> owner,
      @QueryParam("cause") Optional<String> cause,
      @QueryParam("segmentCount") Optional<Integer> segmentCountPerNode,
      @QueryParam("repairParallelism") Optional<String> repairParallelism,
      @QueryParam("intensity") Optional<String> intensityStr,
      @QueryParam("incrementalRepair") Optional<String> incrementalRepairStr,
      @QueryParam("nodes") Optional<String> nodesToRepairParam,
      @QueryParam("datacenters") Optional<String> datacentersToRepairParam,
      @QueryParam("blacklistedTables") Optional<String> blacklistedTableNamesParam,
      @QueryParam("repairThreadCount") Optional<Integer> repairThreadCountParam); @PUT @Path("/{id}/state/{state}") Response modifyRunState(
      @Context UriInfo uriInfo,
      @PathParam("id") UUID repairRunId,
      @PathParam("state") Optional<String> stateStr); @PUT @Path("/{id}/intensity/{intensity}") Response modifyRunIntensity(
      @Context UriInfo uriInfo,
      @PathParam("id") UUID repairRunId,
      @PathParam("intensity") Optional<String> intensityStr); @PUT @Path("/{id}") @Deprecated Response oldModifyRunState(
      @Context UriInfo uriInfo,
      @PathParam("id") UUID repairRunId,
      @QueryParam("state") Optional<String> stateStr); @GET @Path("/{id}") Response getRepairRun(
      @PathParam("id") UUID repairRunId); @GET @Path("/{id}/segments") Response getRepairRunSegments(@PathParam("id") UUID repairRunId); @POST @Path("/{id}/segments/abort/{segment_id}") Response abortRepairRunSegment(@PathParam("id") UUID repairRunId, @PathParam("segment_id") UUID segmentId); @GET @Path("/cluster/{cluster_name}") Response getRepairRunsForCluster(
      @PathParam("cluster_name") String clusterName,
      @QueryParam("limit") Optional<Integer> limit); @GET Response listRepairRuns(
      @QueryParam("state") Optional<String> state,
      @QueryParam("cluster_name") Optional<String> cluster,
      @QueryParam("keyspace_name") Optional<String> keyspace,
      @QueryParam("limit") Optional<Integer> limit); @DELETE @Path("/{id}") Response deleteRepairRun(
      @PathParam("id") UUID runId,
      @QueryParam("owner") Optional<String> owner); @POST @Path("/purge") Response purgeRepairRuns();  }
@Test public void testAddRunMissingArgument() { RepairRunResource resource = new RepairRunResource(context); Response response = addRepairRun( resource, uriInfo, clustername, null, TABLES, OWNER, null, SEGMENT_CNT, NODES, BLACKLISTED_TABLES, REPAIR_THREAD_COUNT); assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); assertTrue(response.getEntity() instanceof String); }
@Test public void testTriggerRunMissingArgument() { RepairRunResource resource = new RepairRunResource(context); Response response = addRepairRun( resource, uriInfo, clustername, null, TABLES, OWNER, null, SEGMENT_CNT, NODES, BLACKLISTED_TABLES, REPAIR_THREAD_COUNT); assertEquals(Response.Status.BAD_REQUEST.getStatusCode(), response.getStatus()); assertTrue(response.getEntity() instanceof String); }
@Test public void testAddRepairRun() throws Exception { DateTimeUtils.setCurrentMillisFixed(TIME_CREATE); RepairRunResource resource = new RepairRunResource(context); Response response = addDefaultRepairRun(resource); assertEquals(201, response.getStatus()); assertTrue(response.getEntity() instanceof RepairRunStatus); assertEquals(1, context.storage.getClusters().size()); assertEquals(1, context.storage.getRepairRunsForCluster(clustername, Optional.of(2)).size()); assertEquals(1, context.storage.getRepairRunIdsForCluster(clustername, Optional.empty()).size()); UUID runId = context.storage.getRepairRunIdsForCluster(clustername, Optional.empty()).iterator().next(); RepairRun run = context.storage.getRepairRun(runId).get(); final RepairUnit unit = context.storage.getRepairUnit(run.getRepairUnitId()); assertEquals(RepairRun.RunState.NOT_STARTED, run.getRunState()); assertEquals(TIME_CREATE, run.getCreationTime().getMillis()); assertEquals(REPAIR_INTENSITY, run.getIntensity(), 0.0f); assertNull(run.getStartTime()); assertNull(run.getEndTime()); assertEquals(2, unit.getRepairThreadCount()); assertEquals( 20, context.storage.getSegmentAmountForRepairRunWithState( run.getId(), RepairSegment.State.NOT_STARTED)); response = addDefaultRepairRun(resource); assertEquals(201, response.getStatus()); assertTrue(response.getEntity() instanceof RepairRunStatus); assertEquals(1, context.storage.getClusters().size()); assertEquals(1, context.storage.getRepairRunsForCluster(clustername, Optional.of(1)).size()); assertEquals(2, context.storage.getRepairRunsForCluster(clustername, Optional.of(2)).size()); assertEquals(2, context.storage.getRepairRunsForCluster(clustername, Optional.of(3)).size()); assertEquals( context.storage.getRepairRunsForCluster(clustername, Optional.of(3)).iterator().next().getId(), context.storage.getRepairRunsForCluster(clustername, Optional.of(1)).iterator().next().getId()); assertEquals( context.storage.getRepairRunsForCluster(clustername, Optional.of(2)).iterator().next().getId(), context.storage.getRepairRunsForCluster(clustername, Optional.of(1)).iterator().next().getId()); }
@Test public void doesNotDisplayBlacklistedCompactionStrategies() throws Exception { context.config.setBlacklistTwcsTables(true); when(proxy.getKeyspaces()).thenReturn(Lists.newArrayList(keyspace)); when(proxy.getTablesForKeyspace(keyspace)).thenReturn( Sets.newHashSet( Table.builder().withName("table1") .withCompactionStrategy("org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy") .build(), Table.builder().withName("table2") .withCompactionStrategy("org.apache.cassandra.db.compaction.DateTieredCompactionStrategy") .build(), Table.builder().withName("table3") .withCompactionStrategy("org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy") .build())); RepairRunResource resource = new RepairRunResource(context); Response response = addRepairRun(resource, uriInfo, clustername, keyspace, Collections.EMPTY_SET, OWNER, "", SEGMENT_CNT, NODES, BLACKLISTED_TABLES, REPAIR_THREAD_COUNT); assertTrue(response.getEntity().toString(), response.getEntity() instanceof RepairRunStatus); RepairRunStatus repairRunStatus = (RepairRunStatus) response.getEntity(); Assertions.assertThat(repairRunStatus.getBlacklistedTables()).isEmpty(); assertFalse("Tables shouldn't contain 'table1'", repairRunStatus.getColumnFamilies().contains("table1")); assertFalse("Tables shouldn't contain 'table2'", repairRunStatus.getColumnFamilies().contains("table2")); assertTrue("Tables shouldn't contain 'table3'", repairRunStatus.getColumnFamilies().contains("table3")); }
@Test public void displaysExplicitBlacklistedCompactionStrategies() throws Exception { context.config.setBlacklistTwcsTables(true); when(proxy.getKeyspaces()).thenReturn(Lists.newArrayList(keyspace)); when(proxy.getTablesForKeyspace(keyspace)).thenReturn( Sets.newHashSet( Table.builder().withName("table1") .withCompactionStrategy("org.apache.cassandra.db.compaction.TimeWindowCompactionStrategy") .build(), Table.builder().withName("table2") .withCompactionStrategy("org.apache.cassandra.db.compaction.DateTieredCompactionStrategy") .build(), Table.builder().withName("table3") .withCompactionStrategy("org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy") .build())); RepairRunResource resource = new RepairRunResource(context); Response response = addRepairRun(resource, uriInfo, clustername, keyspace, Sets.newHashSet("table1"), OWNER, "", SEGMENT_CNT, NODES, BLACKLISTED_TABLES, REPAIR_THREAD_COUNT); assertTrue(response.getEntity().toString(), response.getEntity() instanceof RepairRunStatus); RepairRunStatus repairRunStatus = (RepairRunStatus) response.getEntity(); assertFalse("Blacklisted shouldn't contain 'table1'", repairRunStatus.getBlacklistedTables().contains("table1")); assertFalse("Blacklisted shouldn't contain 'table2'", repairRunStatus.getBlacklistedTables().contains("table2")); assertFalse("Blacklisted shouldn't contain 'table3'", repairRunStatus.getBlacklistedTables().contains("table3")); assertTrue("ColumnFamilies should contain 'table1'", repairRunStatus.getColumnFamilies().contains("table1")); assertFalse("ColumnFamilies shouldn't contain 'table2'", repairRunStatus.getColumnFamilies().contains("table2")); assertFalse("ColumnFamilies shouldn't contain 'table3'", repairRunStatus.getColumnFamilies().contains("table3")); }