-
Notifications
You must be signed in to change notification settings - Fork 24.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add base framework for snapshot retention (#43605)
* Add base framework for snapshot retention This adds a basic `SnapshotRetentionService` and `SnapshotRetentionTask` to start as the basis for SLM's retention implementation. Relates to #38461 * Remove extraneous 'public' * Use a local var instead of reading class var repeatedly
- Loading branch information
Showing
6 changed files
with
301 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
...ilm/src/main/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotRetentionService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.snapshotlifecycle; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.elasticsearch.cluster.LocalNodeMasterListener; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.common.Strings; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings; | ||
| import org.elasticsearch.xpack.core.scheduler.CronSchedule; | ||
| import org.elasticsearch.xpack.core.scheduler.SchedulerEngine; | ||
| import org.elasticsearch.xpack.core.snapshotlifecycle.SnapshotLifecyclePolicy; | ||
|
|
||
| import java.io.Closeable; | ||
| import java.time.Clock; | ||
| import java.util.function.Supplier; | ||
|
|
||
| /** | ||
| * The {@code SnapshotRetentionService} is responsible for scheduling the period kickoff of SLM's | ||
| * snapshot retention. This means that when the retention schedule setting is configured, the | ||
| * scheduler schedules a job that, when triggered, will delete snapshots according to the retention | ||
| * policy configured in the {@link SnapshotLifecyclePolicy}. | ||
| */ | ||
| public class SnapshotRetentionService implements LocalNodeMasterListener, Closeable { | ||
|
|
||
| static final String SLM_RETENTION_JOB_ID = "slm-retention-job"; | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(SnapshotRetentionService.class); | ||
|
|
||
| private final SchedulerEngine scheduler; | ||
|
|
||
| private volatile String slmRetentionSchedule; | ||
|
|
||
| public SnapshotRetentionService(Settings settings, | ||
| Supplier<SnapshotRetentionTask> taskSupplier, | ||
| ClusterService clusterService, | ||
| Clock clock) { | ||
| this.scheduler = new SchedulerEngine(settings, clock); | ||
| this.scheduler.register(taskSupplier.get()); | ||
| this.slmRetentionSchedule = LifecycleSettings.SLM_RETENTION_SCHEDULE_SETTING.get(settings); | ||
| clusterService.addLocalNodeMasterListener(this); | ||
| clusterService.getClusterSettings().addSettingsUpdateConsumer(LifecycleSettings.SLM_RETENTION_SCHEDULE_SETTING, | ||
| this::setUpdateSchedule); | ||
| } | ||
|
|
||
| void setUpdateSchedule(String retentionSchedule) { | ||
| this.slmRetentionSchedule = retentionSchedule; | ||
| // The schedule has changed, so reschedule the retention job | ||
| rescheduleRetentionJob(); | ||
| } | ||
|
|
||
| // Only used for testing | ||
| SchedulerEngine getScheduler() { | ||
| return this.scheduler; | ||
| } | ||
|
|
||
| @Override | ||
| public void onMaster() { | ||
| rescheduleRetentionJob(); | ||
| } | ||
|
|
||
| @Override | ||
| public void offMaster() { | ||
| cancelRetentionJob(); | ||
| } | ||
|
|
||
| private void rescheduleRetentionJob() { | ||
| final String schedule = this.slmRetentionSchedule; | ||
| if (Strings.hasText(schedule)) { | ||
| final SchedulerEngine.Job retentionJob = new SchedulerEngine.Job(SLM_RETENTION_JOB_ID, | ||
| new CronSchedule(schedule)); | ||
| logger.debug("scheduling SLM retention job for [{}]", schedule); | ||
| this.scheduler.add(retentionJob); | ||
| } else { | ||
| // The schedule has been unset, so cancel the scheduled retention job | ||
| cancelRetentionJob(); | ||
| } | ||
| } | ||
|
|
||
| private void cancelRetentionJob() { | ||
| this.scheduler.scheduledJobIds().forEach(this.scheduler::remove); | ||
| } | ||
|
|
||
| @Override | ||
| public String executorName() { | ||
| return ThreadPool.Names.SNAPSHOT; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| this.scheduler.stop(); | ||
| } | ||
| } |
99 changes: 99 additions & 0 deletions
99
...in/ilm/src/main/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotRetentionTask.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.snapshotlifecycle; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.cluster.ClusterState; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.snapshots.SnapshotInfo; | ||
| import org.elasticsearch.xpack.core.scheduler.SchedulerEngine; | ||
| import org.elasticsearch.xpack.core.snapshotlifecycle.SnapshotLifecyclePolicy; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * The {@code SnapshotRetentionTask} is invoked by the scheduled job from the | ||
| * {@link SnapshotRetentionService}. It is responsible for retrieving the snapshots for repositories | ||
| * that have an SLM policy configured, and then deleting the snapshots that fall outside the | ||
| * retention policy. | ||
| */ | ||
| public class SnapshotRetentionTask implements SchedulerEngine.Listener { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(SnapshotRetentionTask.class); | ||
| private static final AtomicBoolean running = new AtomicBoolean(false); | ||
|
|
||
| private final Client client; | ||
| private final ClusterService clusterService; | ||
|
|
||
| public SnapshotRetentionTask(Client client, ClusterService clusterService) { | ||
| this.client = client; | ||
| this.clusterService = clusterService; | ||
| } | ||
|
|
||
| @Override | ||
| public void triggered(SchedulerEngine.Event event) { | ||
| assert event.getJobName().equals(SnapshotRetentionService.SLM_RETENTION_JOB_ID) : | ||
| "expected id to be " + SnapshotRetentionService.SLM_RETENTION_JOB_ID + " but it was " + event.getJobName(); | ||
| if (running.compareAndSet(false, true)) { | ||
| try { | ||
| logger.info("starting SLM retention snapshot cleanup task"); | ||
| final ClusterState state = clusterService.state(); | ||
|
|
||
| // Find all SLM policies that have retention enabled | ||
| final Map<String, SnapshotLifecyclePolicy> policiesWithRetention = getAllPoliciesWithRetentionEnabled(state); | ||
|
|
||
| // For those policies (there may be more than one for the same repo), | ||
| // return the repos that we need to get the snapshots for | ||
| final Set<String> repositioriesToFetch = policiesWithRetention.values().stream() | ||
| .map(SnapshotLifecyclePolicy::getRepository) | ||
| .collect(Collectors.toSet()); | ||
|
|
||
| // Find all the snapshots that are past their retention date | ||
| // TODO: include min/max snapshot count as a criteria for deletion also | ||
| final List<SnapshotInfo> snapshotsToBeDeleted = getAllSnapshots(repositioriesToFetch).stream() | ||
| .filter(snapshot -> snapshotEligibleForDeletion(snapshot, policiesWithRetention)) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| // Finally, delete the snapshots that need to be deleted | ||
| deleteSnapshots(snapshotsToBeDeleted); | ||
|
|
||
| } finally { | ||
| running.set(false); | ||
| } | ||
| } else { | ||
| logger.debug("snapshot lifecycle retention task started, but a task is already running, skipping"); | ||
| } | ||
| } | ||
|
|
||
| static Map<String, SnapshotLifecyclePolicy> getAllPoliciesWithRetentionEnabled(final ClusterState state) { | ||
| // TODO: fill me in | ||
| return Collections.emptyMap(); | ||
| } | ||
|
|
||
| static boolean snapshotEligibleForDeletion(SnapshotInfo snapshot, Map<String, SnapshotLifecyclePolicy> policies) { | ||
| // TODO: fill me in | ||
| return false; | ||
| } | ||
|
|
||
| List<SnapshotInfo> getAllSnapshots(Collection<String> repositories) { | ||
| // TODO: fill me in | ||
| return Collections.emptyList(); | ||
| } | ||
|
|
||
| void deleteSnapshots(List<SnapshotInfo> snapshotsToDelete) { | ||
| // TODO: fill me in | ||
| logger.info("deleting {}", snapshotsToDelete); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
...rc/test/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotRetentionServiceTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| package org.elasticsearch.xpack.snapshotlifecycle; | ||
|
|
||
| import org.elasticsearch.Version; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.cluster.node.DiscoveryNodeRole; | ||
| import org.elasticsearch.cluster.service.ClusterService; | ||
| import org.elasticsearch.common.settings.ClusterSettings; | ||
| import org.elasticsearch.common.settings.Setting; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.test.ClusterServiceUtils; | ||
| import org.elasticsearch.test.ESTestCase; | ||
| import org.elasticsearch.threadpool.TestThreadPool; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.xpack.core.indexlifecycle.LifecycleSettings; | ||
| import org.elasticsearch.xpack.core.scheduler.SchedulerEngine; | ||
| import org.elasticsearch.xpack.core.watcher.watch.ClockMock; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.HashSet; | ||
| import java.util.Set; | ||
|
|
||
| import static org.hamcrest.Matchers.containsInAnyOrder; | ||
| import static org.hamcrest.Matchers.equalTo; | ||
|
|
||
| public class SnapshotRetentionServiceTests extends ESTestCase { | ||
|
|
||
| private static final ClusterSettings clusterSettings; | ||
| static { | ||
| Set<Setting<?>> internalSettings = new HashSet<>(ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); | ||
| internalSettings.add(LifecycleSettings.SLM_RETENTION_SCHEDULE_SETTING); | ||
| clusterSettings = new ClusterSettings(Settings.EMPTY, internalSettings); | ||
| } | ||
|
|
||
| public void testJobsAreScheduled() { | ||
| final DiscoveryNode discoveryNode = new DiscoveryNode("node", ESTestCase.buildNewFakeTransportAddress(), | ||
| Collections.emptyMap(), DiscoveryNodeRole.BUILT_IN_ROLES, Version.CURRENT); | ||
| ClockMock clock = new ClockMock(); | ||
|
|
||
| try (ThreadPool threadPool = new TestThreadPool("test"); | ||
| ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool, discoveryNode, clusterSettings); | ||
| SnapshotRetentionService service = new SnapshotRetentionService(Settings.EMPTY, | ||
| FakeRetentionTask::new, clusterService, clock)) { | ||
| assertThat(service.getScheduler().jobCount(), equalTo(0)); | ||
|
|
||
| service.setUpdateSchedule(SnapshotLifecycleServiceTests.randomSchedule()); | ||
| assertThat(service.getScheduler().scheduledJobIds(), containsInAnyOrder(SnapshotRetentionService.SLM_RETENTION_JOB_ID)); | ||
|
|
||
| service.offMaster(); | ||
| assertThat(service.getScheduler().jobCount(), equalTo(0)); | ||
|
|
||
| service.onMaster(); | ||
| assertThat(service.getScheduler().scheduledJobIds(), containsInAnyOrder(SnapshotRetentionService.SLM_RETENTION_JOB_ID)); | ||
|
|
||
| service.setUpdateSchedule(""); | ||
| assertThat(service.getScheduler().jobCount(), equalTo(0)); | ||
| threadPool.shutdownNow(); | ||
| } | ||
| } | ||
|
|
||
| private static class FakeRetentionTask extends SnapshotRetentionTask { | ||
| FakeRetentionTask() { | ||
| super(null, null); | ||
| } | ||
|
|
||
| @Override | ||
| public void triggered(SchedulerEngine.Event event) { | ||
| super.triggered(event); | ||
| } | ||
| } | ||
| } |