<NME> DatabaseCleanupTask.java
<BEF> /*
 *
 *  Copyright 2016 Netflix, Inc.
 *
 *     Licensed under the Apache License, Version 2.0 (the "License");
 *     you may not use this file except in compliance with the License.
 *     You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *     Unless required by applicable law or agreed to in writing, software
 *     distributed under the License is distributed on an "AS IS" BASIS,
 *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *     See the License for the specific language governing permissions and
 *     limitations under the License.
 *
 */
package com.netflix.genie.web.tasks.leader;

import com.google.common.collect.Sets;
import com.netflix.genie.common.internal.dtos.ClusterStatus;
import com.netflix.genie.common.internal.dtos.CommandStatus;
import com.netflix.genie.web.tasks.GenieTaskScheduleType;
import com.netflix.genie.web.tasks.TaskUtils;
import com.netflix.spectator.api.Registry;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import com.netflix.genie.web.util.MetricsUtils;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.Trigger;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.atomic.AtomicLong;

/**
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * A {@link LeaderTask} which will clean up the database of old records if desired.
 *
 * @author tgianos
 * @since 3.0.0
 */
// TODO: The intention of this class is clear, it is to have the leader trigger a database cleanup action periodically
//       at system administrators discretion. The issue here is that this current implementation bleeds a lot of
//       details about the underlying implementation into this class. If someone were to re-implement the persistence
    private final JobPersistenceService jobPersistenceService;

    private final AtomicLong numDeletedJobs;

    /**
     * Constructor.
//       forget and hopefully come back to it once there is some time. - TJG 4/21/2020
@Slf4j
public class DatabaseCleanupTask extends LeaderTask {

    private static final String DATABASE_CLEANUP_DURATION_TIMER_NAME = "genie.tasks.databaseCleanup.duration.timer";
    private static final String APPLICATION_DELETION_TIMER = "genie.tasks.databaseCleanup.applicationDeletion.timer";
    private static final String CLUSTER_DELETION_TIMER = "genie.tasks.databaseCleanup.clusterDeletion.timer";
    private static final String COMMAND_DEACTIVATION_TIMER = "genie.tasks.databaseCleanup.commandDeactivation.timer";
    private static final String COMMAND_DELETION_TIMER = "genie.tasks.databaseCleanup.commandDeletion.timer";
    private static final String FILE_DELETION_TIMER = "genie.tasks.databaseCleanup.fileDeletion.timer";
    private static final String TAG_DELETION_TIMER = "genie.tasks.databaseCleanup.tagDeletion.timer";

        this.jobPersistenceService = jobPersistenceService;

        this.numDeletedJobs = registry.gauge("genie.tasks.databaseCleanup.numDeletedJobs.gauge", new AtomicLong());
    }

    /**
    // TODO: May want to make this a property. Currently this maintains consistent behavior with before but it would
    //       be nice to add OUT_OF_SERVICE
    private static final Set<ClusterStatus> TO_DELETE_CLUSTER_STATUSES = EnumSet.of(ClusterStatus.TERMINATED);

    private final DatabaseCleanupProperties cleanupProperties;
    private final Environment environment;
    private final PersistenceService persistenceService;

    private final MeterRegistry registry;
    private final AtomicLong numDeletedJobs;
    private final AtomicLong numDeletedClusters;
    private final AtomicLong numDeactivatedCommands;
    private final AtomicLong numDeletedCommands;
    private final AtomicLong numDeletedApplications;
    private final AtomicLong numDeletedTags;
    private final AtomicLong numDeletedFiles;

     */
    @Override
    public void run() {
        final Calendar cal = TaskUtils.getMidnightUTC();
        // Move the date back the number of days retention is set for
        TaskUtils.subtractDaysFromDate(cal, this.cleanupProperties.getRetention());
        final Date retentionLimit = cal.getTime();

        final long numberDeletedJobs = this.jobPersistenceService.deleteAllJobsCreatedBeforeDate(retentionLimit);
        log.info("Deleted {} jobs from before {}", numberDeletedJobs, this.dateFormat.format(retentionLimit));
        this.numDeletedJobs.set(numberDeletedJobs);
    }

    /**
        this.numDeactivatedCommands = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeactivatedCommands.gauge",
            new AtomicLong()
        );
        this.numDeletedCommands = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedCommands.gauge",
            new AtomicLong()
        );
        this.numDeletedApplications = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedApplications.gauge",
            new AtomicLong()
        );
        this.numDeletedTags = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedTags.gauge",
            new AtomicLong()
        );
        this.numDeletedFiles = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedFiles.gauge",
            new AtomicLong()
        );
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public GenieTaskScheduleType getScheduleType() {
        return GenieTaskScheduleType.TRIGGER;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Trigger getTrigger() {
        final String expression = this.environment.getProperty(
            DatabaseCleanupProperties.EXPRESSION_PROPERTY,
            String.class,
            this.cleanupProperties.getExpression()
        );
        return new CronTrigger(expression, JobConstants.UTC);
    }

    /**
     * Clean out database based on date.
     */
    @Override
    public void run() {
        final long start = System.nanoTime();
        final Instant runtime = Instant.now();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            this.deleteJobs();

            // Get now - 1 hour to avoid deleting references that were created as part of new resources recently
            final Instant creationThreshold = runtime.minus(1L, ChronoUnit.HOURS);

            this.deleteClusters(creationThreshold);
            this.deleteCommands(creationThreshold);
            this.deactivateCommands(runtime);
            this.deleteApplications(creationThreshold);
            this.deleteFiles(creationThreshold);
            this.deleteTags(creationThreshold);

            MetricsUtils.addSuccessTags(tags);
        } catch (final Throwable t) {
            MetricsUtils.addFailureTagsWithException(tags, t);
            throw t;
        } finally {
            this.registry
                .timer(DATABASE_CLEANUP_DURATION_TIMER_NAME, tags)
                .record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void cleanup() {
        this.numDeletedJobs.set(0L);
        this.numDeletedClusters.set(0L);
        this.numDeactivatedCommands.set(0L);
        this.numDeletedCommands.set(0L);
        this.numDeletedApplications.set(0L);
        this.numDeletedTags.set(0L);
        this.numDeletedFiles.set(0L);
    }

    /*
     * Delete jobs that are older than the retention threshold and are complete
     */
    private void deleteJobs() {
        final boolean skipJobs = this.environment.getProperty(
            DatabaseCleanupProperties.JobDatabaseCleanupProperties.SKIP_PROPERTY,
            Boolean.class,
            this.cleanupProperties.getJobCleanup().isSkip()
        );
        if (skipJobs) {
            log.info("Skipping job cleanup");
            this.numDeletedJobs.set(0);
        } else {
            // TODO: Maybe we shouldn't reset it to midnight no matter what... just go with runtime minus something
            final Instant midnightUTC = TaskUtils.getMidnightUTC();
            final Instant retentionLimit = midnightUTC.minus(
                this.environment.getProperty(
                    DatabaseCleanupProperties.JobDatabaseCleanupProperties.JOB_RETENTION_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getJobCleanup().getRetention()
                ),
                ChronoUnit.DAYS
            );
            final int batchSize = this.environment.getProperty(
                DatabaseCleanupProperties.JobDatabaseCleanupProperties.PAGE_SIZE_PROPERTY,
                Integer.class,
                this.cleanupProperties.getJobCleanup().getPageSize()
            );

            log.info(
                "Attempting to delete jobs from before {} in batches of {} jobs per iteration",
                retentionLimit,
                batchSize
            );
            long numDeletedJobsInBatch;
            long totalDeletedJobs = 0L;
            do {
                numDeletedJobsInBatch = this.persistenceService.deleteJobsCreatedBefore(
                    retentionLimit,
                    JobStatus.getActiveStatuses(),
                    batchSize
                );
                totalDeletedJobs += numDeletedJobsInBatch;
            } while (numDeletedJobsInBatch != 0);
            log.info(
                "Deleted {} jobs",
                totalDeletedJobs
            );
            this.numDeletedJobs.set(totalDeletedJobs);
        }
    }

    /*
     * Delete all clusters that are marked terminated and aren't attached to any jobs after jobs were deleted.
     */
    private void deleteClusters(final Instant creationThreshold) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipClusters = this.environment.getProperty(
                DatabaseCleanupProperties.ClusterDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getClusterCleanup().isSkip()
            );
            if (skipClusters) {
                log.info("Skipping clusters cleanup");
                this.numDeletedClusters.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );

                log.info(
                    "Attempting to delete unused clusters from before {} in batches of {}",
                    creationThreshold,
                    batchSize
                );

                long deleted;
                long totalDeleted = 0L;
                do {
                    deleted = this.persistenceService.deleteUnusedClusters(
                        TO_DELETE_CLUSTER_STATUSES,
                        creationThreshold,
                        batchSize
                    );
                    totalDeleted += deleted;
                } while (deleted > 0);

                log.info(
                    "Deleted {} clusters that were in one of {} states, were created before {} and weren't "
                        + " attached to any jobs",
                    totalDeleted,
                    TO_DELETE_CLUSTER_STATUSES,
                    creationThreshold
                );
                this.numDeletedClusters.set(totalDeleted);
            }
        } catch (final Exception e) {
            log.error("Unable to delete clusters from database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(CLUSTER_DELETION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }

    private void deleteFiles(final Instant creationThreshold) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipFiles = this.environment.getProperty(
                DatabaseCleanupProperties.FileDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getFileCleanup().isSkip()
            );
            if (skipFiles) {
                log.info("Skipping files cleanup");
                this.numDeletedFiles.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );
                log.info(
                    "Attempting to delete unused files from before {} in batches of {}",
                    creationThreshold,
                    batchSize
                );

                long deleted;
                long totalDeleted = 0L;
                do {
                    deleted = this.persistenceService.deleteUnusedFiles(creationThreshold, batchSize);
                    totalDeleted += deleted;
                } while (deleted > 0);
                log.info(
                    "Deleted {} files that were unused by any resource and created before {}",
                    totalDeleted,
                    creationThreshold
                );
                this.numDeletedFiles.set(totalDeleted);
            }
        } catch (final Exception e) {
            log.error("Unable to delete files from database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(FILE_DELETION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }

    private void deleteTags(final Instant creationThreshold) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipTags = this.environment.getProperty(
                DatabaseCleanupProperties.TagDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getTagCleanup().isSkip()
            );
            if (skipTags) {
                log.info("Skipping tags cleanup");
                this.numDeletedTags.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );

                log.info(
                    "Attempting to delete unused tags from before {} in batches of {}",
                    creationThreshold,
                    batchSize
                );

                long deleted;
                long totalDeleted = 0L;
                do {
                    deleted = this.persistenceService.deleteUnusedTags(creationThreshold, batchSize);
                    totalDeleted += deleted;
                } while (deleted > 0);
                log.info(
                    "Deleted {} tags that were unused by any resource and created before {}",
                    totalDeleted,
                    creationThreshold
                );
                this.numDeletedTags.set(totalDeleted);
            }
        } catch (final Exception e) {
            log.error("Unable to delete tags from database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(TAG_DELETION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }

    private void deactivateCommands(final Instant runtime) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipDeactivation = this.environment.getProperty(
                DatabaseCleanupProperties.CommandDeactivationDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getCommandDeactivation().isSkip()
            );
            if (skipDeactivation) {
                log.info("Skipping command deactivation");
                this.numDeactivatedCommands.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );

                final Instant commandCreationThreshold = runtime.minus(
                    this.environment.getProperty(
                        DatabaseCleanupProperties
                            .CommandDeactivationDatabaseCleanupProperties
                            .COMMAND_CREATION_THRESHOLD_PROPERTY,
                        Integer.class,
                        this.cleanupProperties.getCommandDeactivation().getCommandCreationThreshold()
                    ),
                    ChronoUnit.DAYS
                );
                log.info(
                    "Attempting to set commands to status {} that were previously in one of {} in batches of {}",
                    CommandStatus.INACTIVE,
                    TO_DEACTIVATE_COMMAND_STATUSES,
                    batchSize
                );
                long totalDeactivatedCommands = 0;
                long batchedDeactivated;
                do {
                    batchedDeactivated = this.persistenceService.updateStatusForUnusedCommands(
                        CommandStatus.INACTIVE,
                        commandCreationThreshold,
                        TO_DEACTIVATE_COMMAND_STATUSES,
                        batchSize
                    );
                    totalDeactivatedCommands += batchedDeactivated;
                } while (batchedDeactivated > 0);
                log.info(
                    "Set {} commands to status {} that were previously in one of {}",
                    totalDeactivatedCommands,
                    CommandStatus.INACTIVE,
                    TO_DEACTIVATE_COMMAND_STATUSES
                );
                this.numDeactivatedCommands.set(totalDeactivatedCommands);
            }
        } catch (final Exception e) {
            log.error("Unable to disable commands in database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(COMMAND_DEACTIVATION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }

    private void deleteCommands(final Instant creationThreshold) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipCommands = this.environment.getProperty(
                DatabaseCleanupProperties.CommandDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getCommandCleanup().isSkip()
            );
            if (skipCommands) {
                log.info("Skipping command cleanup");
                this.numDeletedCommands.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );
                log.info(
                    "Attempting to delete unused commands from before {} in batches of {}",
                    creationThreshold,
                    batchSize
                );

                long deleted;
                long totalDeleted = 0L;
                do {
                    deleted = this.persistenceService.deleteUnusedCommands(
                        TO_DELETE_COMMAND_STATUSES,
                        creationThreshold,
                        batchSize
                    );
                    totalDeleted += deleted;
                } while (deleted > 0);
                log.info(
                    "Deleted {} commands that were unused by any resource and created before {}",
                    totalDeleted,
                    creationThreshold
                );
                this.numDeletedCommands.set(totalDeleted);
            }
        } catch (final Exception e) {
            log.error("Unable to delete commands in database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(COMMAND_DELETION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }

    private void deleteApplications(final Instant creationThreshold) {
        final long startTime = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            final boolean skipApplications = this.environment.getProperty(
                DatabaseCleanupProperties.ApplicationDatabaseCleanupProperties.SKIP_PROPERTY,
                Boolean.class,
                this.cleanupProperties.getApplicationCleanup().isSkip()
            );
            if (skipApplications) {
                log.info("Skipping application cleanup");
                this.numDeletedCommands.set(0);
            } else {
                final int batchSize = this.environment.getProperty(
                    DatabaseCleanupProperties.BATCH_SIZE_PROPERTY,
                    Integer.class,
                    this.cleanupProperties.getBatchSize()
                );
                log.info(
                    "Attempting to delete unused applications from before {} in batches of {}",
                    creationThreshold,
                    batchSize
                );

                long deleted;
                long totalDeleted = 0L;
                do {
                    deleted = this.persistenceService.deleteUnusedApplications(
                        creationThreshold,
                        batchSize
                    );
                    totalDeleted += deleted;
                } while (deleted > 0);
                log.info(
                    "Deleted {} applications that were unused by any resource and created before {}",
                    totalDeleted,
                    creationThreshold
                );
                this.numDeletedApplications.set(totalDeleted);
            }
        } catch (final Exception e) {
            log.error("Unable to delete applications in database", e);
            MetricsUtils.addFailureTagsWithException(tags, e);
        } finally {
            this.registry
                .timer(APPLICATION_DELETION_TIMER, tags)
                .record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
        }
    }
}

<MSG> Modify database cleanup task to delete jobs in configurable batches

<DFF> @@ -23,6 +23,7 @@ import com.netflix.genie.web.properties.DatabaseCleanupProperties;
 import com.netflix.genie.web.tasks.GenieTaskScheduleType;
 import com.netflix.genie.web.tasks.TaskUtils;
 import com.netflix.spectator.api.Registry;
+import com.netflix.spectator.api.Timer;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -35,6 +36,7 @@ import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.Calendar;
 import java.util.Date;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -53,6 +55,7 @@ public class DatabaseCleanupTask extends LeadershipTask {
     private final JobPersistenceService jobPersistenceService;
 
     private final AtomicLong numDeletedJobs;
+    private final Timer deletionTimer;
 
     /**
      * Constructor.
@@ -71,6 +74,7 @@ public class DatabaseCleanupTask extends LeadershipTask {
         this.jobPersistenceService = jobPersistenceService;
 
         this.numDeletedJobs = registry.gauge("genie.tasks.databaseCleanup.numDeletedJobs.gauge", new AtomicLong());
+        this.deletionTimer = registry.timer("genie.tasks.databaseCleanup.duration.timer");
     }
 
     /**
@@ -94,14 +98,27 @@ public class DatabaseCleanupTask extends LeadershipTask {
      */
     @Override
     public void run() {
-        final Calendar cal = TaskUtils.getMidnightUTC();
-        // Move the date back the number of days retention is set for
-        TaskUtils.subtractDaysFromDate(cal, this.cleanupProperties.getRetention());
-        final Date retentionLimit = cal.getTime();
+        final long start = System.nanoTime();
+        try {
+            final Calendar cal = TaskUtils.getMidnightUTC();
+            // Move the date back the number of days retention is set for
+            TaskUtils.subtractDaysFromDate(cal, this.cleanupProperties.getRetention());
+            final Date retentionLimit = cal.getTime();
+            final String retentionLimitString = this.dateFormat.format(retentionLimit);
+            final int batchSize = this.cleanupProperties.getBatchSize();
 
-        final long numberDeletedJobs = this.jobPersistenceService.deleteAllJobsCreatedBeforeDate(retentionLimit);
-        log.info("Deleted {} jobs from before {}", numberDeletedJobs, this.dateFormat.format(retentionLimit));
-        this.numDeletedJobs.set(numberDeletedJobs);
+            log.info(
+                "Attempting to delete jobs from before {} in batches of {} jobs per iteration",
+                retentionLimitString,
+                batchSize
+            );
+            final long numberDeletedJobs
+                = this.jobPersistenceService.deleteAllJobsCreatedBeforeDate(retentionLimit, batchSize);
+            log.info("Deleted {} jobs from before {}", numberDeletedJobs, retentionLimitString);
+            this.numDeletedJobs.set(numberDeletedJobs);
+        } finally {
+            this.deletionTimer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+        }
     }
 
     /**

Modify database cleanup task to delete jobs in configurable batches