Consoidate DirectorySize implementations into one, change interface t…

…o pass optional filters; Move methods to from BackupV2Service to BackupServletV2 where they are used. (#1106)

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

@@ -35,7 +35,7 @@
 /** Abstract Backup class for uploading files to backup location */
 public abstract class AbstractBackup extends Task {
     private static final Logger logger = LoggerFactory.getLogger(AbstractBackup.class);
-    static final String INCREMENTAL_BACKUP_FOLDER = "backups";
+    public static final String INCREMENTAL_BACKUP_FOLDER = "backups";
     public static final String SNAPSHOT_FOLDER = "snapshots";
 
     @Inject

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

@@ -38,7 +38,8 @@ public void acquire(AbstractBackupPath path, Instant target, int permits) {
         }
         int backupThreads = config.getBackupThreads();
         Preconditions.checkState(backupThreads > 0);
-        long bytesPerThread = this.dirSize.getBytes(config.getDataFileLocation()) / backupThreads;
+        long bytesPerThread =
+                this.dirSize.getBytes(config.getDataFileLocation(), AbstractBackup.SNAPSHOT_FOLDER) / backupThreads;
         if (bytesPerThread < 1) {
             return;
         }

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

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

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

@@ -1,14 +1,16 @@
 package com.netflix.priam.backup;
 
+import javax.annotation.Nonnull;
 import java.io.IOException;
 import java.nio.file.*;
 import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Arrays;
 
 /** Estimates remaining bytes or files to upload in a backup by looking at the file system */
-public class IncrementalBackupDirectorySize implements DirectorySize {
+public class DirectorySizeImpl implements DirectorySize {
 
-    public long getBytes(String location) {
-        SummingFileVisitor fileVisitor = new SummingFileVisitor();
+    public long getBytes(String location, String... filters) {
+        SummingFileVisitor fileVisitor = new SummingFileVisitor(filters);
         try {
             Files.walkFileTree(Paths.get(location), fileVisitor);
         } catch (IOException e) {
@@ -17,8 +19,8 @@ public long getBytes(String location) {
         return fileVisitor.getTotalBytes();
     }
 
-    public int getFiles(String location) {
-        SummingFileVisitor fileVisitor = new SummingFileVisitor();
+    public int getFiles(String location, String... filters) {
+        SummingFileVisitor fileVisitor = new SummingFileVisitor(filters);
         try {
             Files.walkFileTree(Paths.get(location), fileVisitor);
         } catch (IOException e) {
@@ -30,6 +32,11 @@ public int getFiles(String location) {
     private static final class SummingFileVisitor implements FileVisitor<Path> {
         private long totalBytes;
         private int totalFiles;
+        private final String[] filters;
+
+        public SummingFileVisitor(String... filters) {
+            this.filters = filters;
+        }
 
         @Override
         public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
@@ -38,18 +45,24 @@ public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
 
         @Override
         public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
-            if (file.toString().contains(AbstractBackup.INCREMENTAL_BACKUP_FOLDER) && attrs.isRegularFile()) {
+            if (!attrs.isRegularFile()) {
+                return FileVisitResult.CONTINUE;
+            }
+            // This will return 0 if no filter is provided.
+            if (Arrays.stream(filters).anyMatch(filter -> file.toString().contains(filter))) {
                 totalBytes += attrs.size();
                 totalFiles += 1;
             }
             return FileVisitResult.CONTINUE;
         }
 
+        @Nonnull
         @Override
         public FileVisitResult visitFileFailed(Path file, IOException exc) {
             return FileVisitResult.CONTINUE;
         }
 
+        @Nonnull
         @Override
         public FileVisitResult postVisitDirectory(Path dir, IOException exc) {
             return FileVisitResult.CONTINUE;

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

@@ -17,7 +17,6 @@
 
 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;
@@ -26,12 +25,9 @@
 import com.netflix.priam.scheduler.PriamScheduler;
 import com.netflix.priam.scheduler.TaskTimer;
 import com.netflix.priam.tuner.CassandraTunerService;
-import java.util.HashMap;
-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;
+
+import javax.inject.Inject;
 
 
 /**
@@ -45,8 +41,6 @@ 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(
@@ -110,10 +104,4 @@ public void updateServicePre() throws Exception {
     @Override
     public void updateServicePost() throws Exception {}
 
-    public Map<String, Integer> countPendingBackupFiles() throws Exception {
-        Map<String, Integer> backupFiles = new HashMap<String, Integer>();
-        backupFiles.put("totalFiles", (snapshotDirectorySize.getFiles(configuration.getDataFileLocation()) +
-                incrementalBackupDirectorySize.getFiles(configuration.getDataFileLocation())));
-        return backupFiles;
-    }
 }

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

@@ -76,7 +76,7 @@ public class SnapshotMetaTask extends AbstractBackup {
     public static final String JOBNAME = "SnapshotMetaService";
 
     private static final Logger logger = LoggerFactory.getLogger(SnapshotMetaTask.class);
-    private static final String SNAPSHOT_PREFIX = "snap_v2_";
+    public static final String SNAPSHOT_PREFIX = "snap_v2_";
     private static final String CASSANDRA_MANIFEST_FILE = "manifest.json";
     private static final String CASSANDRA_SCHEMA_FILE = "schema.cql";
     private static final TimeZone UTC = TimeZone.getTimeZone(ZoneId.of("UTC"));

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

@@ -50,6 +50,8 @@
 @Produces(MediaType.APPLICATION_JSON)
 public class BackupServletV2 {
     private static final Logger logger = LoggerFactory.getLogger(BackupServletV2.class);
+    private static final String INCREMENTALS = "/" + AbstractBackup.INCREMENTAL_BACKUP_FOLDER + "/";
+    private static final String SNAPSHOTS = "/" + SnapshotMetaTask.SNAPSHOT_PREFIX;
     private final BackupVerification backupVerification;
     private final IBackupStatusMgr backupStatusMgr;
     private final SnapshotMetaTask snapshotMetaService;
@@ -59,7 +61,8 @@ public class BackupServletV2 {
     private final Provider<AbstractBackupPath> pathProvider;
     private final BackupV2Service backupService;
     private final BackupNotificationMgr backupNotificationMgr;
-    private final PriamServer priamServer;
+    private final IConfiguration config;
+    private final DirectorySize directorySize;
 
     private static final String REST_SUCCESS = "[\"ok\"]";
 
@@ -69,13 +72,13 @@ public BackupServletV2(
             BackupVerification backupVerification,
             SnapshotMetaTask snapshotMetaService,
             BackupTTLTask backupTTLService,
-            IConfiguration configuration,
             IBackupFileSystem fileSystem,
             @Named("v2") IMetaProxy metaV2Proxy,
             Provider<AbstractBackupPath> pathProvider,
             BackupV2Service backupService,
             BackupNotificationMgr backupNotificationMgr,
-            PriamServer priamServer) {
+            IConfiguration config,
+            DirectorySize directorySize) {
         this.backupStatusMgr = backupStatusMgr;
         this.backupVerification = backupVerification;
         this.snapshotMetaService = snapshotMetaService;
@@ -85,7 +88,8 @@ public BackupServletV2(
         this.pathProvider = pathProvider;
         this.backupService = backupService;
         this.backupNotificationMgr = backupNotificationMgr;
-        this.priamServer = priamServer;
+        this.config = config;
+        this.directorySize = directorySize;
     }
 
     @GET
@@ -190,11 +194,8 @@ public Response backupState(@PathParam("hours") int hours) throws Exception {
         Map<String, Object> responseMap = new HashMap<>();
 
         responseMap.put("tasksQueued", fs.getUploadTasksQueued());
-        responseMap.put("queueSize", priamServer.getConfiguration().getBackupQueueSize());
-        for (Map.Entry<String, Integer> entry :
-                backupService.countPendingBackupFiles().entrySet()) {
-            responseMap.put(entry.getKey(), entry.getValue());
-        }
+        responseMap.put("queueSize", config.getBackupQueueSize());
+        responseMap.put("totalFiles", directorySize.getFiles(config.getDataFileLocation(), SNAPSHOTS, INCREMENTALS));
 
         List<BackupMetadata> latestBackupMetadata =
                 backupStatusMgr.getLatestBackupMetadata(

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

@@ -164,12 +164,12 @@ private static final class FakeDirectorySize implements DirectorySize {
         }
 
         @Override
-        public long getBytes(String location) {
+        public long getBytes(String location, String... filters) {
             return size;
         }
 
         @Override
-        public int getFiles(String location) {
+        public int getFiles(String location, String... filters) {
             return fileCount;
         }
     }