Merge pull request #815 from arunagrawal84/3.11

Disable backups

priam/src/main/java/com/netflix/priam/PriamServer.java

@@ -31,7 +31,7 @@
 import com.netflix.priam.identity.InstanceIdentity;
 import com.netflix.priam.restore.RestoreContext;
 import com.netflix.priam.scheduler.PriamScheduler;
-import com.netflix.priam.tuner.TuneCassandra;
+import com.netflix.priam.tuner.CassandraTunerService;
 import com.netflix.priam.utils.Sleeper;
 import com.netflix.priam.utils.SystemUtils;
 import java.io.IOException;
@@ -49,6 +49,7 @@ public class PriamServer implements IService {
     private final RestoreContext restoreContext;
     private final IService backupV2Service;
     private final IService backupService;
+    private final IService cassandraTunerService;
     private static final int CASSANDRA_MONITORING_INITIAL_DELAY = 10;
     private static final Logger logger = LoggerFactory.getLogger(PriamServer.class);
 
@@ -61,7 +62,8 @@ public PriamServer(
             ICassandraProcess cassProcess,
             RestoreContext restoreContext,
             BackupService backupService,
-            BackupV2Service backupV2Service) {
+            BackupV2Service backupV2Service,
+            CassandraTunerService cassandraTunerService) {
         this.config = config;
         this.scheduler = scheduler;
         this.instanceIdentity = id;
@@ -70,6 +72,7 @@ public PriamServer(
         this.restoreContext = restoreContext;
         this.backupService = backupService;
         this.backupV2Service = backupV2Service;
+        this.cassandraTunerService = cassandraTunerService;
     }
 
     private void createDirectories() throws IOException {
@@ -107,8 +110,8 @@ public void scheduleService() throws Exception {
                     UpdateSecuritySettings.getTimer(instanceIdentity));
         }
 
-        // Run the task to tune Cassandra
-        scheduler.runTaskNow(TuneCassandra.class);
+        // Set up cassandra tuning.
+        cassandraTunerService.scheduleService();
 
         // Determine if we need to restore from backup else start cassandra.
         if (restoreContext.isRestoreEnabled()) {
@@ -151,6 +154,12 @@ public void scheduleService() throws Exception {
         backupV2Service.scheduleService();
     }
 
+    @Override
+    public void updateServicePre() throws Exception {}
+
+    @Override
+    public void updateServicePost() throws Exception {}
+
     public InstanceIdentity getInstanceIdentity() {
         return instanceIdentity;
     }

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

@@ -24,10 +24,14 @@
 import com.netflix.priam.scheduler.Task;
 import com.netflix.priam.utils.SystemUtils;
 import java.io.File;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.text.ParseException;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.Future;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -169,6 +173,40 @@ protected final void initiateBackup(
     protected abstract void processColumnFamily(
             String keyspace, String columnFamily, File backupDir) throws Exception;
 
+    /**
+     * Get all the backup directories for Cassandra.
+     *
+     * @param config to get the location of the data folder.
+     * @param monitoringFolder folder where cassandra backup's are configured.
+     * @return Set of the path(s) containing the backup folder for each columnfamily.
+     * @throws Exception incase of IOException.
+     */
+    public static Set<Path> getBackupDirectories(IConfiguration config, String monitoringFolder)
+            throws Exception {
+        HashSet<Path> backupPaths = new HashSet<>();
+        if (config.getDataFileLocation() == null) return backupPaths;
+        Path dataPath = Paths.get(config.getDataFileLocation());
+        if (Files.exists(dataPath) && Files.isDirectory(dataPath))
+            try (DirectoryStream<Path> directoryStream =
+                    Files.newDirectoryStream(dataPath, path -> Files.isDirectory(path))) {
+                for (Path keyspaceDirPath : directoryStream) {
+                    try (DirectoryStream<Path> keyspaceStream =
+                            Files.newDirectoryStream(
+                                    keyspaceDirPath, path -> Files.isDirectory(path))) {
+                        for (Path columnfamilyDirPath : keyspaceStream) {
+                            Path backupDirPath =
+                                    Paths.get(columnfamilyDirPath.toString(), monitoringFolder);
+                            if (Files.exists(backupDirPath) && Files.isDirectory(backupDirPath)) {
+                                logger.debug("Backup folder: {}", backupDirPath);
+                                backupPaths.add(backupDirPath);
+                            }
+                        }
+                    }
+                }
+            }
+        return backupPaths;
+    }
+
     /** Filters unwanted keyspaces */
     private boolean isValidBackupDir(File keyspaceDir, File backupDir) {
         if (backupDir == null || !backupDir.isDirectory() || !backupDir.exists()) return false;

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

@@ -19,49 +19,68 @@
 
 import com.google.inject.Inject;
 import com.netflix.priam.aws.UpdateCleanupPolicy;
+import com.netflix.priam.config.IBackupRestoreConfig;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.defaultimpl.IService;
-import com.netflix.priam.identity.InstanceIdentity;
 import com.netflix.priam.scheduler.PriamScheduler;
-import org.apache.commons.collections4.CollectionUtils;
+import com.netflix.priam.scheduler.TaskTimer;
+import com.netflix.priam.tuner.CassandraTunerService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Created by aagrawal on 3/9/19. */
+/**
+ * Encapsulate the backup service 1.0 - Execute all the tasks required to run backup service.
+ *
+ * <p>Created by aagrawal on 3/9/19.
+ */
 public class BackupService implements IService {
     private final PriamScheduler scheduler;
     private final IConfiguration config;
-    private final InstanceIdentity instanceIdentity;
+    private final IBackupRestoreConfig backupRestoreConfig;
+    private final CassandraTunerService cassandraTunerService;
     private static final Logger logger = LoggerFactory.getLogger(BackupService.class);
 
     @Inject
     public BackupService(
             IConfiguration config,
+            IBackupRestoreConfig backupRestoreConfig,
             PriamScheduler priamScheduler,
-            InstanceIdentity instanceIdentity) {
+            CassandraTunerService cassandraTunerService) {
         this.config = config;
+        this.backupRestoreConfig = backupRestoreConfig;
         this.scheduler = priamScheduler;
-        this.instanceIdentity = instanceIdentity;
+        this.cassandraTunerService = cassandraTunerService;
     }
 
     @Override
     public void scheduleService() throws Exception {
         // Start the snapshot backup schedule - Always run this. (If you want to
         // set it off, set backup hour to -1) or set backup cron to "-1"
-        if (SnapshotBackup.getTimer(config) != null
-                && (CollectionUtils.isEmpty(config.getBackupRacs())
-                        || config.getBackupRacs()
-                                .contains(instanceIdentity.getInstanceInfo().getRac()))) {
-            scheduleTask(scheduler, SnapshotBackup.class, SnapshotBackup.getTimer(config));
+        TaskTimer snapshotTimer = SnapshotBackup.getTimer(config);
+        scheduleTask(scheduler, SnapshotBackup.class, snapshotTimer);
 
-            // Start the Incremental backup schedule
-            scheduleTask(scheduler, IncrementalBackup.class, IncrementalBackup.getTimer(config));
+        if (snapshotTimer != null) {
+            // Schedule commit log task
+            scheduleTask(
+                    scheduler, CommitLogBackupTask.class, CommitLogBackupTask.getTimer(config));
         }
 
-        // Schedule commit log task
-        scheduleTask(scheduler, CommitLogBackupTask.class, CommitLogBackupTask.getTimer(config));
+        // Start the Incremental backup schedule if enabled
+        scheduleTask(
+                scheduler,
+                IncrementalBackup.class,
+                IncrementalBackup.getTimer(config, backupRestoreConfig));
 
         // Set cleanup
         scheduleTask(scheduler, UpdateCleanupPolicy.class, UpdateCleanupPolicy.getTimer());
     }
+
+    @Override
+    public void updateServicePre() throws Exception {
+        // Run the task to tune Cassandra
+        cassandraTunerService.onChangeUpdateService();
+    }
+
+    @Override
+    public void updateServicePost() throws Exception {}
 }

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

@@ -20,13 +20,17 @@
 import com.google.inject.Provider;
 import com.google.inject.Singleton;
 import com.netflix.priam.backup.AbstractBackupPath.BackupFileType;
+import com.netflix.priam.backupv2.SnapshotMetaTask;
 import com.netflix.priam.config.IBackupRestoreConfig;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.scheduler.SimpleTimer;
 import com.netflix.priam.scheduler.TaskTimer;
 import com.netflix.priam.utils.DateUtil;
 import java.io.File;
+import java.nio.file.Path;
 import java.util.List;
+import java.util.Set;
+import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -65,11 +69,46 @@ public void execute() throws Exception {
     }
 
     /** Run every 10 Sec */
-    public static TaskTimer getTimer(IConfiguration config) {
-        if (config.isIncrementalBackupEnabled()) return new SimpleTimer(JOBNAME, 10L * 1000);
+    public static TaskTimer getTimer(
+            IConfiguration config, IBackupRestoreConfig backupRestoreConfig) {
+        if (IncrementalBackup.isEnabled(config, backupRestoreConfig))
+            return new SimpleTimer(JOBNAME, 10L * 1000);
         return null;
     }
 
+    private static void cleanOldBackups(IConfiguration configuration) throws Exception {
+        Set<Path> backupPaths =
+                AbstractBackup.getBackupDirectories(configuration, INCREMENTAL_BACKUP_FOLDER);
+        for (Path backupDirPath : backupPaths) {
+            FileUtils.cleanDirectory(backupDirPath.toFile());
+        }
+    }
+
+    public static boolean isEnabled(
+            IConfiguration configuration, IBackupRestoreConfig backupRestoreConfig) {
+        boolean enabled = false;
+        try {
+            // Once backup 1.0 is gone, we should not check for enableV2Backups.
+            enabled =
+                    (configuration.isIncrementalBackupEnabled()
+                            && (SnapshotBackup.isBackupEnabled(configuration)
+                                    || (backupRestoreConfig.enableV2Backups()
+                                            && SnapshotMetaTask.isBackupEnabled(
+                                                    configuration, backupRestoreConfig))));
+            logger.info("Incremental backups are enabled: {}", enabled);
+
+            if (!enabled) {
+                // Clean up the incremental backup folder.
+                cleanOldBackups(configuration);
+            }
+        } catch (Exception e) {
+            logger.error(
+                    "Error while trying to find if incremental backup is enabled: "
+                            + e.getMessage());
+        }
+        return enabled;
+    }
+
     @Override
     public String getName() {
         return JOBNAME;

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

@@ -31,10 +31,14 @@
 import com.netflix.priam.utils.DateUtil;
 import com.netflix.priam.utils.ThreadSleeper;
 import java.io.File;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -96,6 +100,8 @@ public void execute() throws Exception {
         }
 
         try {
+            // Clean up all the backup directories, if any.
+            cleanOldBackups(config);
             executeSnapshot();
         } finally {
             lock.unlock();
@@ -169,7 +175,25 @@ public static boolean isBackupEnabled(IConfiguration config) throws Exception {
     }
 
     public static TaskTimer getTimer(IConfiguration config) throws Exception {
-        return CronTimer.getCronTimer(JOBNAME, config.getBackupCronExpression());
+        TaskTimer timer = CronTimer.getCronTimer(JOBNAME, config.getBackupCronExpression());
+        if (timer == null) {
+            // Clean up all the backup directories, if any.
+            cleanOldBackups(config);
+        }
+        return timer;
+    }
+
+    private static void cleanOldBackups(IConfiguration configuration) throws Exception {
+        Set<Path> backupPaths = AbstractBackup.getBackupDirectories(configuration, SNAPSHOT_FOLDER);
+        for (Path backupDirPath : backupPaths)
+            try (DirectoryStream<Path> directoryStream =
+                    Files.newDirectoryStream(backupDirPath, path -> Files.isDirectory(path))) {
+                for (Path backupDir : directoryStream) {
+                    if (isValidBackupDir(backupDir)) {
+                        FileUtils.deleteDirectory(backupDir.toFile());
+                    }
+                }
+            }
     }
 
     @Override
@@ -188,4 +212,10 @@ protected void processColumnFamily(String keyspace, String columnFamily, File ba
         abstractBackupPaths.addAll(
                 upload(snapshotDir, BackupFileType.SNAP, config.enableAsyncSnapshot(), true));
     }
+
+    private static boolean isValidBackupDir(Path backupDir) {
+        String backupDirName = backupDir.toFile().getName();
+        // Check if it of format yyyyMMddHHmm
+        return (DateUtil.getDate(backupDirName) != null);
+    }
 }