Updated backup DirectorySize interface to get the file count for snap…

…shots/backups

priam/src/main/java/com/netflix/priam/backup/DirectorySize.java

@@ -2,9 +2,10 @@
 
 import com.google.inject.ImplementedBy;
 
-/** estimates the number of bytes remaining to upload in a snapshot */
-@ImplementedBy(SnapshotDirectorySize.class)
+/** estimates the number of bytes and files remaining to upload in a snapshot/backup */
 public interface DirectorySize {
- /** return the total bytes of all snapshot files south of location in the filesystem */
+ /** return the total bytes of all snapshot/backup files south of location in the filesystem */
  long getBytes(String location);
+ /** return the total files of all snapshot/backup files south of location in the filesystem */
+ int getFiles(String location);
 }

priam/src/main/java/com/netflix/priam/backup/IncrementalBackup.java

@@ -140,19 +140,4 @@ private Void deleteIfEmpty(File dir) {
  if (FileUtils.sizeOfDirectory(dir) == 0) FileUtils.deleteQuietly(dir);
  return null;
  }
-
- public static int countFilesInBackupDir(IConfiguration config) throws Exception {
- int totalFileCount = 0;
- Set<Path> backupDirectories =
- AbstractBackup.getBackupDirectories(config, INCREMENTAL_BACKUP_FOLDER);
- for (Path backupDir : backupDirectories) {
- try (Stream<Path> stream = Files.list(backupDir)) {
- totalFileCount += stream.filter(Files::isRegularFile).count();
- } catch (Exception e) {
- logger.error("Failed to get files in backups directory. {}", e.getMessage());
- e.printStackTrace();
- }
- }
- return totalFileCount;
- }
 }

priam/src/main/java/com/netflix/priam/backup/IncrementalBackupDirectorySize.java

@@ -0,0 +1,66 @@
+package com.netflix.priam.backup;
+
+import java.io.IOException;
+import java.nio.file.*;
+import java.nio.file.attribute.BasicFileAttributes;
+
+/** Estimates remaining bytes or files to upload in a backup by looking at the file system */
+public class IncrementalBackupDirectorySize implements DirectorySize {
+
+ public long getBytes(String location) {
+ SummingFileVisitor fileVisitor = new SummingFileVisitor();
+ try {
+ Files.walkFileTree(Paths.get(location), fileVisitor);
+ } catch (IOException e) {
+ // BackupFileVisitor is happy with an estimate and won't produce these in practice.
+ }
+ return fileVisitor.getTotalBytes();
+ }
+
+ public int getFiles(String location) {
+ SummingFileVisitor fileVisitor = new SummingFileVisitor();
+ try {
+ Files.walkFileTree(Paths.get(location), fileVisitor);
+ } catch (IOException e) {
+ // BackupFileVisitor is happy with an estimate and won't produce these in practice.
+ }
+ return fileVisitor.getTotalFiles();
+ }
+
+ private static final class SummingFileVisitor implements FileVisitor<Path> {
+ private long totalBytes;
+ private int totalFiles;
+
+ @Override
+ public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
+ if (file.toString().contains(AbstractBackup.INCREMENTAL_BACKUP_FOLDER) && attrs.isRegularFile()) {
+ totalBytes += attrs.size();
+ totalFiles += 1;
+ }
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult visitFileFailed(Path file, IOException exc) {
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult postVisitDirectory(Path dir, IOException exc) {
+ return FileVisitResult.CONTINUE;
+ }
+
+ long getTotalBytes() {
+ return totalBytes;
+ }
+
+ int getTotalFiles() {
+ return totalFiles;
+ }
+ }
+}

priam/src/main/java/com/netflix/priam/backup/SnapshotDirectorySize.java

@@ -4,7 +4,7 @@
 import java.nio.file.*;
 import java.nio.file.attribute.BasicFileAttributes;
 
-/** Estimates remaining bytes to upload in a backup by looking at the file system */
+/** Estimates remaining bytes or files to upload in a backup by looking at the file system */
 public class SnapshotDirectorySize implements DirectorySize {
 
  public long getBytes(String location) {
@@ -17,8 +17,19 @@ public long getBytes(String location) {
  return fileVisitor.getTotalBytes();
  }
 
+ public int getFiles(String location) {
+ SummingFileVisitor fileVisitor = new SummingFileVisitor();
+ try {
+ Files.walkFileTree(Paths.get(location), fileVisitor);
+ } catch (IOException e) {
+ // BackupFileVisitor is happy with an estimate and won't produce these in practice.
+ }
+ return fileVisitor.getTotalFiles();
+ }
+
  private static final class SummingFileVisitor implements FileVisitor<Path> {
  private long totalBytes;
+ private int totalFiles;
 
  @Override
  public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
@@ -29,6 +40,7 @@ public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
  public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
  if (file.toString().contains(AbstractBackup.SNAPSHOT_FOLDER) && attrs.isRegularFile()) {
  totalBytes += attrs.size();
+ totalFiles += 1;
  }
  return FileVisitResult.CONTINUE;
  }
@@ -46,5 +58,9 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exc) {
  long getTotalBytes() {
  return totalBytes;
  }
+
+ int getTotalFiles() {
+ return totalFiles;
+ }
  }
 }

priam/src/main/java/com/netflix/priam/backupv2/BackupV2Service.java

@@ -17,6 +17,7 @@
 
 package com.netflix.priam.backupv2;
 
+import com.netflix.priam.backup.DirectorySize;
 import com.netflix.priam.backup.IncrementalBackup;
 import com.netflix.priam.config.IBackupRestoreConfig;
 import com.netflix.priam.config.IConfiguration;
@@ -29,6 +30,9 @@
 import java.util.Map;
 import javax.inject.Inject;
 import org.apache.commons.lang3.math.Fraction;
+import com.netflix.priam.backup.SnapshotDirectorySize;
+import com.netflix.priam.backup.IncrementalBackupDirectorySize;
+
 
 /**
  * Encapsulate the backup service 2.0 - Execute all the tasks required to run backup service.
@@ -41,6 +45,8 @@ public class BackupV2Service implements IService {
  private final SnapshotMetaTask snapshotMetaTask;
  private final CassandraTunerService cassandraTunerService;
  private final ITokenRetriever tokenRetriever;
+ private final DirectorySize snapshotDirectorySize = new SnapshotDirectorySize();
+ private final DirectorySize incrementalBackupDirectorySize = new IncrementalBackupDirectorySize();
 
  @Inject
  public BackupV2Service(
@@ -106,8 +112,8 @@ public void updateServicePost() throws Exception {}
 
  public Map<String, Integer> countPendingBackupFiles() throws Exception {
  Map<String, Integer> backupFiles = new HashMap<String, Integer>();
- backupFiles.put("snapshotFiles", snapshotMetaTask.countFilesInSnapshotDir(configuration));
- backupFiles.put("incrementalFiles", IncrementalBackup.countFilesInBackupDir(configuration));
+ backupFiles.put("totalFiles", (snapshotDirectorySize.getFiles(configuration.getDataFileLocation()) +
+  incrementalBackupDirectorySize.getFiles(configuration.getDataFileLocation())));
  return backupFiles;
  }
 }

priam/src/main/java/com/netflix/priam/backupv2/SnapshotMetaTask.java

@@ -443,26 +443,4 @@ public void onFailure(Throwable t) {
  };
  Futures.addCallback(future, callback, MoreExecutors.directExecutor());
  }
-
- public int countFilesInSnapshotDir(IConfiguration config) throws Exception {
- int totalFileCount = 0;
- Set<Path> snapshotDirectories =
- AbstractBackup.getBackupDirectories(config, SNAPSHOT_FOLDER);
- for (Path snapshotDir : snapshotDirectories) {
- try (DirectoryStream<Path> directoryStream =
- Files.newDirectoryStream(snapshotDir, Files::isDirectory)) {
- for (Path backupDir : directoryStream) {
- if (backupDir.toFile().getName().startsWith(SNAPSHOT_PREFIX)) {
- try (Stream<Path> stream = Files.list(backupDir)) {
- totalFileCount += stream.filter(Files::isRegularFile).count();
- }
- }
- }
- } catch (Exception e) {
- logger.error("Failed to get files in snapshot directory. {}", e.getMessage());
- e.printStackTrace();
- }
- }
- return totalFileCount;
- }
 }

priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java

@@ -25,7 +25,6 @@
 import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.backupv2.SnapshotMetaTask;
 import com.netflix.priam.config.IConfiguration;
-import com.netflix.priam.merics.BackupMetrics;
 import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.DateUtil;
 import com.netflix.priam.utils.DateUtil.DateRange;
@@ -62,7 +61,6 @@ public class BackupServletV2 {
  private final BackupNotificationMgr backupNotificationMgr;
  private final PriamServer priamServer;
 
- private final BackupMetrics backupMetrics;
  private static final String REST_SUCCESS = "[\"ok\"]";
 
  @Inject
@@ -77,8 +75,7 @@ public BackupServletV2(
  Provider<AbstractBackupPath> pathProvider,
  BackupV2Service backupService,
  BackupNotificationMgr backupNotificationMgr,
- PriamServer priamServer,
- BackupMetrics backupMetrics) {
+ PriamServer priamServer) {
  this.backupStatusMgr = backupStatusMgr;
  this.backupVerification = backupVerification;
  this.snapshotMetaService = snapshotMetaService;
@@ -89,7 +86,6 @@ public BackupServletV2(
  this.backupService = backupService;
  this.backupNotificationMgr = backupNotificationMgr;
  this.priamServer = priamServer;
- this.backupMetrics = backupMetrics;
  }
 
  @GET

priam/src/test/java/com/netflix/priam/backup/TestBackupDynamicRateLimiter.java

@@ -20,7 +20,8 @@
 public class TestBackupDynamicRateLimiter {
  private static final Instant NOW = Instant.ofEpochMilli(1 << 16);
  private static final Instant LATER = NOW.plusMillis(Duration.ofHours(1).toMillis());
- private static final int DIR_SIZE = 1 << 16;
+ private static final int DIR_SIZE_BYTES = 1 << 16;
+ private static final int DIR_SIZE_FILES = 10;
 
  private BackupDynamicRateLimiter rateLimiter;
  private FakeConfiguration config;
@@ -34,22 +35,22 @@ public void setUp() {
 
  @Test
  public void sunnyDay() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  Stopwatch timer = timePermitAcquisition(getBackupPath(), LATER, 21);
  Truth.assertThat(timer.elapsed(TimeUnit.MILLISECONDS)).isAtLeast(1_000);
  Truth.assertThat(timer.elapsed(TimeUnit.MILLISECONDS)).isAtMost(2_000);
  }
 
  @Test
  public void targetSetToEpoch() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  Stopwatch timer = timePermitAcquisition(getBackupPath(), Instant.EPOCH, 20);
  assertNoRateLimiting(timer);
  }
 
  @Test
  public void pathIsNotASnapshot() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  AbstractBackupPath path =
  getBackupPath(
  "target/data/Keyspace1/Standard1/backups/Keyspace1-Standard1-ia-4-Data.db");
@@ -59,47 +60,47 @@ public void pathIsNotASnapshot() {
 
  @Test
  public void targetIsNow() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  Stopwatch timer = timePermitAcquisition(getBackupPath(), NOW, 20);
  assertNoRateLimiting(timer);
  }
 
  @Test
  public void targetIsInThePast() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  Instant target = NOW.minus(Duration.ofHours(1L));
  Stopwatch timer = timePermitAcquisition(getBackupPath(), target, 20);
  assertNoRateLimiting(timer);
  }
 
  @Test
  public void noBackupThreads() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 0), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 0), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  assertIllegalState(() -> timePermitAcquisition(getBackupPath(), LATER, 20));
  }
 
  @Test
  public void negativeBackupThreads() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", -1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", -1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  assertIllegalState(() -> timePermitAcquisition(getBackupPath(), LATER, 20));
  }
 
  @Test
  public void noData() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, 0);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, 0, 0);
  Stopwatch timer = timePermitAcquisition(getBackupPath(), LATER, 20);
  assertNoRateLimiting(timer);
  }
 
  @Test
  public void noPermitsRequested() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  assertIllegalArgument(() -> timePermitAcquisition(getBackupPath(), LATER, 0));
  }
 
  @Test
  public void negativePermitsRequested() {
- rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE);
+ rateLimiter = getRateLimiter(ImmutableMap.of("Priam.backup.threads", 1), NOW, DIR_SIZE_BYTES, DIR_SIZE_FILES);
  assertIllegalArgument(() -> timePermitAcquisition(getBackupPath(), LATER, -1));
  }
 
@@ -123,12 +124,12 @@ private Stopwatch timePermitAcquisition(AbstractBackupPath path, Instant now, in
  }
 
  private BackupDynamicRateLimiter getRateLimiter(
- Map<String, Object> properties, Instant now, long directorySize) {
+ Map<String, Object> properties, Instant now, long directorySizeBytes, int directorySizeFiles) {
  properties.forEach(config::setFakeConfig);
  return new BackupDynamicRateLimiter(
  config,
  Clock.fixed(now, ZoneId.systemDefault()),
- new FakeDirectorySize(directorySize));
+ new FakeDirectorySize(directorySizeBytes, directorySizeFiles));
  }
 
  private void assertNoRateLimiting(Stopwatch timer) {
@@ -155,14 +156,21 @@ private void assertIllegalArgument(Runnable method) {
 
  private static final class FakeDirectorySize implements DirectorySize {
  private final long size;
+ private final int fileCount;
 
- FakeDirectorySize(long size) {
+ FakeDirectorySize(long size, int fileCount) {
  this.size = size;
+ this.fileCount = fileCount;
  }
 
  @Override
  public long getBytes(String location) {
  return size;
  }
+
+ @Override
+ public int getFiles(String location) {
+ return fileCount;
+ }
  }
 }