2410

pending completion

Build # 2410

Build Type

push

travis-ci

Committed by

tgianos

Commit Message

<a href="https://github.com/Netflix/genie/commit/<a class=hub.com/Netflix/genie/commit/e7ae9e1fff09e05df52907b69deaecda4982e06c">e7ae9e1ff<a href="https://github.com/Netflix/genie/commit/e7ae9e1fff09e05df52907b69deaecda4982e06c">&quot;&gt;Revert &amp;quot;Modify command arg parsing for jobs&quot;

This reverts commit </a><a class="double-link" href="https://github.com/Netflix/genie/commit/<a class="double-link" href="https://github.com/Netflix/genie/commit/b01a44f522c0ef03428376d869262b4bafcf54cb">b01a44f52</a>">b01a44f52</a><a href="https://github.com/Netflix/genie/commit/e7ae9e1fff09e05df52907b69deaecda4982e06c">.

Run Details

26 of 26 new or added lines in 7 files covered. (100.0%)

12263 of 13369 relevant lines covered (91.73%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.06

/genie-web/src/main/java/com/netflix/genie/web/tasks/leader/DatabaseCleanupTask.java

/*
 *
 *  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.jobs.JobConstants;
import com.netflix.genie.web.data.services.ClusterPersistenceService;
import com.netflix.genie.web.data.services.FilePersistenceService;
import com.netflix.genie.web.data.services.JobPersistenceService;
import com.netflix.genie.web.data.services.TagPersistenceService;
import com.netflix.genie.web.properties.DatabaseCleanupProperties;
import com.netflix.genie.web.tasks.GenieTaskScheduleType;
import com.netflix.genie.web.tasks.TaskUtils;
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.scheduling.Trigger;
import org.springframework.scheduling.support.CronTrigger;

import javax.validation.constraints.NotNull;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * A task which will clean up the database of old jobs if desired.
 *
 * @author tgianos
 * @since 3.0.0
 */
@Slf4j
public class DatabaseCleanupTask extends LeadershipTask {

    private static final String DATABASE_CLEANUP_DURATION_TIMER_NAME = "genie.tasks.databaseCleanup.duration.timer";
    private final DatabaseCleanupProperties cleanupProperties;
    private final JobPersistenceService jobPersistenceService;
    private final ClusterPersistenceService clusterPersistenceService;
    private final FilePersistenceService filePersistenceService;
    private final TagPersistenceService tagPersistenceService;

    private final MeterRegistry registry;
    private final AtomicLong numDeletedJobs;
    private final AtomicLong numDeletedClusters;
    private final AtomicLong numDeletedTags;
    private final AtomicLong numDeletedFiles;

    /**
     * Constructor.
     *
     * @param cleanupProperties         The properties to use to configure this task
     * @param jobPersistenceService     The persistence service to use to cleanup the data store
     * @param clusterPersistenceService The cluster service to use to delete terminated clusters
     * @param filePersistenceService    The file service to use to delete unused file references
     * @param tagPersistenceService     The tag service to use to delete unused tag references
     * @param registry                  The metrics registry
     */
    public DatabaseCleanupTask(
        @NotNull final DatabaseCleanupProperties cleanupProperties,
        @NotNull final JobPersistenceService jobPersistenceService,
        @NotNull final ClusterPersistenceService clusterPersistenceService,
        @NotNull final FilePersistenceService filePersistenceService,
        @NotNull final TagPersistenceService tagPersistenceService,
        @NotNull final MeterRegistry registry
    ) {
        this.registry = registry;
        this.cleanupProperties = cleanupProperties;
        this.jobPersistenceService = jobPersistenceService;
        this.clusterPersistenceService = clusterPersistenceService;
        this.filePersistenceService = filePersistenceService;
        this.tagPersistenceService = tagPersistenceService;

        this.numDeletedJobs = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedJobs.gauge",
            new AtomicLong()
        );
        this.numDeletedClusters = this.registry.gauge(
            "genie.tasks.databaseCleanup.numDeletedClusters.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() {
        return new CronTrigger(this.cleanupProperties.getExpression(), JobConstants.UTC);
    }

    /**
     * Clean out database based on date.
     */
    @Override
    public void run() {
        final long start = System.nanoTime();
        final Set<Tag> tags = Sets.newHashSet();
        try {
            // Delete jobs that are older than the retention threshold and are complete
            if (this.cleanupProperties.isSkipJobsCleanup()) {
                log.debug("Skipping job cleanup");
                this.numDeletedJobs.set(0);
            } else {
                final long countDeletedJobs = this.deleteJobs();
                log.info(
                    "Deleted {} jobs",
                    countDeletedJobs
                );
                this.numDeletedJobs.set(countDeletedJobs);
            }

            // Delete all clusters that are marked terminated and aren't attached to any jobs after jobs were deleted
            if (this.cleanupProperties.isSkipClustersCleanup()) {
                log.debug("Skipping clusters cleanup");
                this.numDeletedClusters.set(0);
            } else {
                final long countDeletedClusters = this.clusterPersistenceService.deleteTerminatedClusters();
                log.info(
                    "Deleted {} clusters that were in TERMINATED state and weren't attached to any jobs",
                    countDeletedClusters
                );
                this.numDeletedClusters.set(countDeletedClusters);
            }

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

            if (this.cleanupProperties.isSkipFilesCleanup()) {
                log.debug("Skipping files cleanup");
                this.numDeletedFiles.set(0);
            } else {
                final long countDeletedFiles = this.filePersistenceService.deleteUnusedFiles(creationThreshold);
                log.info(
                    "Deleted {} files that were unused by any resource and created over an hour ago",
                    countDeletedFiles
                );
                this.numDeletedFiles.set(countDeletedFiles);
            }

            if (this.cleanupProperties.isSkipTagsCleanup()) {
                log.debug("Skipping tags cleanup");
                this.numDeletedTags.set(0);
            } else {
                final long countDeletedTags = this.tagPersistenceService.deleteUnusedTags(creationThreshold);
                log.info(
                    "Deleted {} tags that were unused by any resource and created over an hour ago",
                    countDeletedTags
                );
                this.numDeletedTags.set(countDeletedTags);
            }

            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.numDeletedTags.set(0L);
        this.numDeletedFiles.set(0L);
    }

    private long deleteJobs() {
        final Instant midnightUTC = TaskUtils.getMidnightUTC();
        final Instant retentionLimit = midnightUTC.minus(this.cleanupProperties.getRetention(), ChronoUnit.DAYS);
        final int batchSize = this.cleanupProperties.getMaxDeletedPerTransaction();
        final int pageSize = this.cleanupProperties.getPageSize();

        log.info(
            "Attempting to delete jobs from before {} in batches of {} jobs per iteration",
            retentionLimit,
            batchSize
        );
        long totalDeletedJobs = 0;
        while (true) {
            final long numberDeletedJobs = this.jobPersistenceService.deleteBatchOfJobsCreatedBeforeDate(
                retentionLimit,
                batchSize,
                pageSize
            );
            totalDeletedJobs += numberDeletedJobs;
            if (numberDeletedJobs == 0) {
                break;
            }
        }
        return totalDeletedJobs;
    }
}

1	/*
2	*
3	* Copyright 2016 Netflix, Inc.
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License");
6	* you may not use this file except in compliance with the License.
7	* You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*
17	*/
18	package com.netflix.genie.web.tasks.leader;
19
20	import com.google.common.collect.Sets;
21	import com.netflix.genie.common.internal.jobs.JobConstants;
22	import com.netflix.genie.web.data.services.ClusterPersistenceService;
23	import com.netflix.genie.web.data.services.FilePersistenceService;
24	import com.netflix.genie.web.data.services.JobPersistenceService;
25	import com.netflix.genie.web.data.services.TagPersistenceService;
26	import com.netflix.genie.web.properties.DatabaseCleanupProperties;
27	import com.netflix.genie.web.tasks.GenieTaskScheduleType;
28	import com.netflix.genie.web.tasks.TaskUtils;
29	import com.netflix.genie.web.util.MetricsUtils;
30	import io.micrometer.core.instrument.MeterRegistry;
31	import io.micrometer.core.instrument.Tag;
32	import lombok.extern.slf4j.Slf4j;
33	import org.springframework.scheduling.Trigger;
34	import org.springframework.scheduling.support.CronTrigger;
35
36	import javax.validation.constraints.NotNull;
37	import java.time.Instant;
38	import java.time.temporal.ChronoUnit;
39	import java.util.Set;
40	import java.util.concurrent.TimeUnit;
41	import java.util.concurrent.atomic.AtomicLong;
42
43	/**
44	* A task which will clean up the database of old jobs if desired.
45	*
46	* @author tgianos
47	* @since 3.0.0
48	*/
49	@Slf4j	1✔
50	public class DatabaseCleanupTask extends LeadershipTask {
51
52	private static final String DATABASE_CLEANUP_DURATION_TIMER_NAME = "genie.tasks.databaseCleanup.duration.timer";
53	private final DatabaseCleanupProperties cleanupProperties;
54	private final JobPersistenceService jobPersistenceService;
55	private final ClusterPersistenceService clusterPersistenceService;
56	private final FilePersistenceService filePersistenceService;
57	private final TagPersistenceService tagPersistenceService;
58
59	private final MeterRegistry registry;
60	private final AtomicLong numDeletedJobs;
61	private final AtomicLong numDeletedClusters;
62	private final AtomicLong numDeletedTags;
63	private final AtomicLong numDeletedFiles;
64
65	/**
66	* Constructor.
67	*
68	* @param cleanupProperties The properties to use to configure this task
69	* @param jobPersistenceService The persistence service to use to cleanup the data store
70	* @param clusterPersistenceService The cluster service to use to delete terminated clusters
71	* @param filePersistenceService The file service to use to delete unused file references
72	* @param tagPersistenceService The tag service to use to delete unused tag references
73	* @param registry The metrics registry
74	*/
75	public DatabaseCleanupTask(
76	@NotNull final DatabaseCleanupProperties cleanupProperties,
77	@NotNull final JobPersistenceService jobPersistenceService,
78	@NotNull final ClusterPersistenceService clusterPersistenceService,
79	@NotNull final FilePersistenceService filePersistenceService,
80	@NotNull final TagPersistenceService tagPersistenceService,
81	@NotNull final MeterRegistry registry
82	) {	1✔
83	this.registry = registry;	1✔
84	this.cleanupProperties = cleanupProperties;	1✔
85	this.jobPersistenceService = jobPersistenceService;	1✔
86	this.clusterPersistenceService = clusterPersistenceService;	1✔
87	this.filePersistenceService = filePersistenceService;	1✔
88	this.tagPersistenceService = tagPersistenceService;	1✔
89
90	this.numDeletedJobs = this.registry.gauge(	1✔
91	"genie.tasks.databaseCleanup.numDeletedJobs.gauge",
92	new AtomicLong()
93	);
94	this.numDeletedClusters = this.registry.gauge(	1✔
95	"genie.tasks.databaseCleanup.numDeletedClusters.gauge",
96	new AtomicLong()
97	);
98	this.numDeletedTags = this.registry.gauge(	1✔
99	"genie.tasks.databaseCleanup.numDeletedTags.gauge",
100	new AtomicLong()
101	);
102	this.numDeletedFiles = this.registry.gauge(	1✔
103	"genie.tasks.databaseCleanup.numDeletedFiles.gauge",
104	new AtomicLong()
105	);
106	}	1✔
107
108	/**
109	* {@inheritDoc}
110	*/
111	@Override
112	public GenieTaskScheduleType getScheduleType() {
113	return GenieTaskScheduleType.TRIGGER;	1✔
114	}
115
116	/**
117	* {@inheritDoc}
118	*/
119	@Override
120	public Trigger getTrigger() {
121	return new CronTrigger(this.cleanupProperties.getExpression(), JobConstants.UTC);	1✔
122	}
123
124	/**
125	* Clean out database based on date.
126	*/
127	@Override
128	public void run() {
129	final long start = System.nanoTime();	1✔
130	final Set<Tag> tags = Sets.newHashSet();	1✔
131	try {
132	// Delete jobs that are older than the retention threshold and are complete
133	if (this.cleanupProperties.isSkipJobsCleanup()) {	1✔
134	log.debug("Skipping job cleanup");	1✔
135	this.numDeletedJobs.set(0);	1✔
136	} else {
137	final long countDeletedJobs = this.deleteJobs();	1✔
138	log.info(	1✔
139	"Deleted {} jobs",
140	countDeletedJobs	1✔
141	);
142	this.numDeletedJobs.set(countDeletedJobs);	1✔
143	}
144
145	// Delete all clusters that are marked terminated and aren't attached to any jobs after jobs were deleted
146	if (this.cleanupProperties.isSkipClustersCleanup()) {	1✔
147	log.debug("Skipping clusters cleanup");	1✔
148	this.numDeletedClusters.set(0);	1✔
149	} else {
150	final long countDeletedClusters = this.clusterPersistenceService.deleteTerminatedClusters();	1✔
151	log.info(	1✔
152	"Deleted {} clusters that were in TERMINATED state and weren't attached to any jobs",
153	countDeletedClusters	1✔
154	);
155	this.numDeletedClusters.set(countDeletedClusters);	1✔
156	}
157
158	// Get now - 1 hour to avoid deleting references that were created as part of new resources recently
159	final Instant creationThreshold = Instant.now().minus(1L, ChronoUnit.HOURS);	1✔
160
161	if (this.cleanupProperties.isSkipFilesCleanup()) {	1✔
162	log.debug("Skipping files cleanup");	1✔
163	this.numDeletedFiles.set(0);	1✔
164	} else {
165	final long countDeletedFiles = this.filePersistenceService.deleteUnusedFiles(creationThreshold);	1✔
166	log.info(	1✔
167	"Deleted {} files that were unused by any resource and created over an hour ago",
168	countDeletedFiles	1✔
169	);
170	this.numDeletedFiles.set(countDeletedFiles);	1✔
171	}
172
173	if (this.cleanupProperties.isSkipTagsCleanup()) {	1✔
174	log.debug("Skipping tags cleanup");	1✔
175	this.numDeletedTags.set(0);	1✔
176	} else {
177	final long countDeletedTags = this.tagPersistenceService.deleteUnusedTags(creationThreshold);	1✔
178	log.info(	1✔
179	"Deleted {} tags that were unused by any resource and created over an hour ago",
180	countDeletedTags	1✔
181	);
182	this.numDeletedTags.set(countDeletedTags);	1✔
183	}
184
185	MetricsUtils.addSuccessTags(tags);	1✔
186	} catch (final Throwable t) {	1✔
187	MetricsUtils.addFailureTagsWithException(tags, t);	1✔
188	throw t;	1✔
189	} finally {
190	this.registry	1✔
191	.timer(DATABASE_CLEANUP_DURATION_TIMER_NAME, tags)	1✔
192	.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);	1✔
193	}
194	}	1✔
195
196	/**
197	* {@inheritDoc}
198	*/
199	@Override
200	public void cleanup() {
201	this.numDeletedJobs.set(0L);	×
202	this.numDeletedClusters.set(0L);	×
203	this.numDeletedTags.set(0L);	×
204	this.numDeletedFiles.set(0L);	×
205	}	×
206
207	private long deleteJobs() {
208	final Instant midnightUTC = TaskUtils.getMidnightUTC();	1✔
209	final Instant retentionLimit = midnightUTC.minus(this.cleanupProperties.getRetention(), ChronoUnit.DAYS);	1✔
210	final int batchSize = this.cleanupProperties.getMaxDeletedPerTransaction();	1✔
211	final int pageSize = this.cleanupProperties.getPageSize();	1✔
212
213	log.info(	1✔
214	"Attempting to delete jobs from before {} in batches of {} jobs per iteration",
215	retentionLimit,
216	batchSize	1✔
217	);
218	long totalDeletedJobs = 0;	1✔
219	while (true) {
220	final long numberDeletedJobs = this.jobPersistenceService.deleteBatchOfJobsCreatedBeforeDate(	1✔
221	retentionLimit,
222	batchSize,
223	pageSize
224	);
225	totalDeletedJobs += numberDeletedJobs;	1✔
226	if (numberDeletedJobs == 0) {	1✔
227	break;	1✔
228	}
229	}	1✔
230	return totalDeletedJobs;	1✔
231	}
232	}

Netflix / genie / 2410

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous