SnapshotService { public List<Pair<Node, String>> takeSnapshotClusterWide( String snapshotName, String clusterName, String owner, String cause, String... keyspace) throws ReaperException { try { List<Pair<Node, String>> snapshotResults = Lists.newArrayList(); Cluster cluster = context.storage.getCluster(clusterName); Snapshot snapshot = Snapshot.builder() .withClusterName(clusterName) .withName(snapshotName) .withOwner(owner) .withCause(cause) .withCreationDate(DateTime.now()) .build(); context.storage.saveSnapshot(snapshot); LOG.info("Cluster : {} ; Cluster obj : {}", clusterName, cluster); List<String> liveNodes = clusterFacade.getLiveNodes(cluster); List<Callable<Pair<Node, String>>> snapshotTasks = liveNodes .stream() .map(host -> Node.builder().withCluster(cluster).withHostname(host).build()) .map(node -> takeSnapshotTask(snapshotName, node, keyspace)) .collect(Collectors.toList()); List<Future<Pair<Node, String>>> futures = executor.invokeAll(snapshotTasks); for (Future<Pair<Node, String>> future : futures) { snapshotResults.add(future.get()); } return snapshotResults; } catch (InterruptedException | ExecutionException e) { LOG.error("Failed taking snapshot for cluster {}", clusterName, e); throw new ReaperException(e); } } private  SnapshotService(AppContext context, ExecutorService executor, Supplier<ClusterFacade> clusterFacadeSupplier); static SnapshotService create(AppContext context, ExecutorService executor); Pair<Node, String> takeSnapshot(String snapshotName, Node host, String... keyspaces); List<Pair<Node, String>> takeSnapshotClusterWide(
      String snapshotName,
      String clusterName,
      String owner,
      String cause,
      String... keyspace); Map<String, List<Snapshot>> listSnapshotsGroupedByName(Node host); List<Snapshot> listSnapshots(Node host); Map<String, Map<String, List<Snapshot>>> listSnapshotsClusterWide(String clusterName); void clearSnapshot(String snapshotName, Node host); void clearSnapshotClusterWide(String snapshotName, String clusterName); String formatSnapshotName(String snapshotName); static final String SNAPSHOT_PREFIX; }
@Test public void testTakeSnapshotClusterWide() throws InterruptedException, ReaperException, ClassNotFoundException, IOException { JmxProxy proxy = (JmxProxy) mock(Class.forName("io.cassandrareaper.jmx.JmxProxyImpl")); StorageServiceMBean storageMBean = Mockito.mock(StorageServiceMBean.class); JmxProxyTest.mockGetStorageServiceMBean(proxy, storageMBean); AppContext cxt = new AppContext(); cxt.config = TestRepairConfiguration.defaultConfig(); cxt.jmxConnectionFactory = mock(JmxConnectionFactory.class); when(cxt.jmxConnectionFactory.connectAny(any(Collection.class))).thenReturn(proxy); ClusterFacade clusterFacadeSpy = Mockito.spy(ClusterFacade.create(cxt)); Mockito.doReturn(Arrays.asList("127.0.0.1", "127.0.0.2", "127.0.0.3")).when(clusterFacadeSpy).getLiveNodes(any()); Mockito.doReturn(Arrays.asList("127.0.0.1", "127.0.0.2", "127.0.0.3")) .when(clusterFacadeSpy) .getLiveNodes(any(), any()); cxt.storage = mock(IStorage.class); Cluster cluster = Cluster.builder() .withName("testCluster") .withPartitioner("murmur3") .withSeedHosts(ImmutableSet.of("127.0.0.1")) .build(); when(cxt.storage.getCluster(anyString())).thenReturn(cluster); List<Pair<Node,String>> result = SnapshotService .create(cxt, SNAPSHOT_MANAGER_EXECUTOR, () -> clusterFacadeSpy) .takeSnapshotClusterWide("snapshot", "testCluster", "testOwner", "testCause"); Assertions.assertThat(result.size()).isEqualTo(3); for (int i = 0 ; i < 3 ; ++i) { Assertions.assertThat(result.get(i).getLeft().getClusterName()).isEqualTo("testcluster"); Assertions.assertThat(result.get(i).getLeft().getHostname()).isEqualTo("127.0.0." + (i + 1)); Assertions.assertThat(result.get(i).getRight()).isEqualTo("snapshot"); } verify(storageMBean, times(3)).takeSnapshot("snapshot"); }