Handle snapshot lifecycle policy updates and deletions

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/indexlifecycle/IndexLifecycle.java

@@ -81,6 +81,7 @@
 import org.elasticsearch.xpack.indexlifecycle.action.TransportStartILMAction;
 import org.elasticsearch.xpack.indexlifecycle.action.TransportStopILMAction;
 import org.elasticsearch.xpack.snapshotlifecycle.SnapshotLifecycleService;
+import org.elasticsearch.xpack.snapshotlifecycle.SnapshotLifecycleTask;
 import org.elasticsearch.xpack.snapshotlifecycle.action.DeleteSnapshotLifecycleAction;
 import org.elasticsearch.xpack.snapshotlifecycle.action.GetSnapshotLifecycleAction;
 import org.elasticsearch.xpack.snapshotlifecycle.action.PutSnapshotLifecycleAction;
@@ -151,7 +152,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
         }
         indexLifecycleInitialisationService.set(new IndexLifecycleService(settings, client, clusterService, threadPool,
                 getClock(), System::currentTimeMillis, xContentRegistry));
-        snapshotLifecycleService.set(new SnapshotLifecycleService(settings, client, clusterService, getClock()));
+        snapshotLifecycleService.set(new SnapshotLifecycleService(settings,
+            () -> new SnapshotLifecycleTask(client), clusterService, getClock()));
         return Arrays.asList(indexLifecycleInitialisationService.get(), snapshotLifecycleService.get());
     }
 

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotLifecycleMetadata.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -35,7 +36,7 @@ public class SnapshotLifecycleMetadata implements XPackMetaDataCustom {
     private final Map<String, SnapshotLifecyclePolicyMetadata> snapshotConfigurations;
 
     public SnapshotLifecycleMetadata(Map<String, SnapshotLifecyclePolicyMetadata> snapshotConfigurations) {
-        this.snapshotConfigurations = Collections.unmodifiableMap(snapshotConfigurations);
+        this.snapshotConfigurations = new HashMap<>(snapshotConfigurations);
         // TODO: maybe operation mode here so it can be disabled/re-enabled separately like ILM is
     }
 
@@ -44,7 +45,7 @@ public SnapshotLifecycleMetadata(StreamInput in) throws IOException {
     }
 
     public Map<String, SnapshotLifecyclePolicyMetadata> getSnapshotConfigurations() {
-        return this.snapshotConfigurations;
+        return Collections.unmodifiableMap(this.snapshotConfigurations);
     }
 
     @Override

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotLifecycleService.java

@@ -8,7 +8,6 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
@@ -23,6 +22,10 @@
 import java.io.Closeable;
 import java.time.Clock;
 import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 /**
  * {@code SnapshotLifecycleService} manages snapshot policy scheduling and triggering of the
@@ -32,28 +35,31 @@
 public class SnapshotLifecycleService implements LocalNodeMasterListener, Closeable, ClusterStateListener {
 
     private static final Logger logger = LogManager.getLogger(SnapshotLifecycleMetadata.class);
+    private static final String JOB_PATTERN_SUFFIX = "-\\d+$";
 
     private final SchedulerEngine scheduler;
     private final ClusterService clusterService;
     private final SnapshotLifecycleTask snapshotTask;
     private final Map<String, SchedulerEngine.Job> scheduledTasks = ConcurrentCollections.newConcurrentMap();
     private volatile boolean isMaster = false;
 
-    public SnapshotLifecycleService(Settings settings, Client client, ClusterService clusterService,
+    public SnapshotLifecycleService(Settings settings,
+                                    Supplier<SnapshotLifecycleTask> taskSupplier,
+                                    ClusterService clusterService,
                                     Clock clock) {
         this.scheduler = new SchedulerEngine(settings, clock);
         this.clusterService = clusterService;
-        this.snapshotTask = new SnapshotLifecycleTask(client);
+        this.snapshotTask = taskSupplier.get();
         clusterService.addLocalNodeMasterListener(this); // TODO: change this not to use 'this'
         clusterService.addListener(this);
     }
 
     @Override
-    public void clusterChanged(ClusterChangedEvent event) {
+    public void clusterChanged(final ClusterChangedEvent event) {
         if (this.isMaster) {
-            // TODO: handle modified policies (currently they are ignored)
-            // TODO: handle deleted policies
-            scheduleSnapshotJobs(event.state());
+            final ClusterState state = event.state();
+            scheduleSnapshotJobs(state);
+            cleanupDeletedPolicies(state);
         }
     }
 
@@ -71,6 +77,11 @@ public void offMaster() {
         cancelSnapshotJobs();
     }
 
+    // Only used for testing
+    SchedulerEngine getScheduler() {
+        return this.scheduler;
+    }
+
     /**
      * Schedule all non-scheduled snapshot jobs contained in the cluster state
      */
@@ -81,35 +92,85 @@ public void scheduleSnapshotJobs(final ClusterState state) {
         }
     }
 
+    public void cleanupDeletedPolicies(final ClusterState state) {
+        SnapshotLifecycleMetadata snapMeta = state.metaData().custom(SnapshotLifecycleMetadata.TYPE);
+        if (snapMeta != null) {
+            // Retrieve all of the expected policy job ids from the policies in the metadata
+            final Set<String> policyJobIds = snapMeta.getSnapshotConfigurations().values().stream()
+                .map(SnapshotLifecycleService::getJobId)
+                .collect(Collectors.toSet());
+
+            // Cancel all jobs that are *NOT* in the scheduled tasks map
+            scheduledTasks.keySet().stream()
+                .filter(jobId -> policyJobIds.contains(jobId) == false)
+                .forEach(this::cancelScheduledSnapshot);
+        }
+    }
+
     /**
-     * Schedule the {@link SnapshotLifecyclePolicy} job if it does not already exist. If the job already
-     * exists it is not interfered with.
+     * Schedule the {@link SnapshotLifecyclePolicy} job if it does not already exist. First checks
+     * to see if any previous versions of the policy were scheduled, and if so, cancels those. If
+     * the same version of a policy has already been scheduled it does not overwrite the job.
      */
     public void maybeScheduleSnapshot(final SnapshotLifecyclePolicyMetadata snapshotLifecyclePolicy) {
-        final String jobId = snapshotLifecyclePolicy.getPolicy().getId();
+        final String jobId = getJobId(snapshotLifecyclePolicy);
+        final Pattern existingJobPattern = Pattern.compile(snapshotLifecyclePolicy.getPolicy().getId() + JOB_PATTERN_SUFFIX);
+
+        // Find and cancel any existing jobs for this policy
+        final boolean existingJobsFoundAndCancelled = scheduledTasks.keySet().stream()
+            // Find all jobs matching the `jobid-\d+` pattern
+            .filter(jId -> existingJobPattern.matcher(jId).matches())
+            // Filter out a job that has not been changed (matches the id exactly meaning the version is the same)
+            .filter(jId -> jId.equals(jobId) == false)
+            .map(existingJobId -> {
+                // Cancel existing job so the new one can be scheduled
+                logger.debug("removing existing snapshot lifecycle job [{}] as it has been updated", existingJobId);
+                scheduledTasks.remove(existingJobId);
+                boolean existed = scheduler.remove(existingJobId);
+                assert existed : "expected job for " + existingJobId + " to exist in scheduler";
+                return existed;
+            })
+            .reduce(false, (a, b) -> a || b);
+
+        // Now atomically schedule the new job and add it to the scheduled tasks map. If the jobId
+        // is identical to an existing job (meaning the version has not changed) then this does
+        // not reschedule it.
         scheduledTasks.computeIfAbsent(jobId, id -> {
             final SchedulerEngine.Job job = new SchedulerEngine.Job(jobId,
                 new CronSchedule(snapshotLifecyclePolicy.getPolicy().getSchedule()));
-            logger.info("scheduling snapshot lifecycle job [{}]", jobId);
+            if (existingJobsFoundAndCancelled) {
+                logger.info("rescheduling updated snapshot lifecycle job [{}]", jobId);
+            } else {
+                logger.info("scheduling snapshot lifecycle job [{}]", jobId);
+            }
             scheduler.add(job);
             return job;
         });
     }
 
+    /**
+     * Generate the job id for a given policy metadata. The job id is {@code <policyid>-<version>}
+     */
+    static String getJobId(SnapshotLifecyclePolicyMetadata policyMeta) {
+        return policyMeta.getPolicy().getId() + "-" + policyMeta.getVersion();
+    }
+
     /**
      * Cancel all scheduled snapshot jobs
      */
     public void cancelSnapshotJobs() {
+        logger.trace("cancelling all snapshot lifecycle jobs");
         scheduler.scheduledJobIds().forEach(scheduler::remove);
         scheduledTasks.clear();
     }
 
     /**
-     * Cancel the given snapshot lifecycle id
+     * Cancel the given policy job id (from {@link #getJobId(SnapshotLifecyclePolicyMetadata)}
      */
-    public void cancelScheduledSnapshot(final String snapshotLifecycleId) {
-        scheduledTasks.remove(snapshotLifecycleId);
-        scheduler.remove(snapshotLifecycleId);
+    public void cancelScheduledSnapshot(final String lifecycleJobId) {
+        logger.debug("cancelling snapshot lifecycle job [{}] as it no longer exists", lifecycleJobId);
+        scheduledTasks.remove(lifecycleJobId);
+        scheduler.remove(lifecycleJobId);
     }
 
     @Override

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/snapshotlifecycle/SnapshotLifecycleTask.java

@@ -17,7 +17,7 @@ public class SnapshotLifecycleTask implements SchedulerEngine.Listener {
 
     private final Client client;
 
-    SnapshotLifecycleTask(final Client client) {
+    public SnapshotLifecycleTask(final Client client) {
         this.client = client;
     }