Make SnapshotLifecycleService project aware (#131735)

Author: samxbr

x-pack/plugin/slm/src/main/java/org/elasticsearch/xpack/slm/SnapshotLifecycle.java

@@ -134,7 +134,7 @@ public Collection<?> createComponents(PluginServices services) {
         snapshotLifecycleService.set(
             new SnapshotLifecycleService(
                 settings,
-                () -> new SnapshotLifecycleTask(client, clusterService, snapshotHistoryStore.get()),
+                projectId -> new SnapshotLifecycleTask(projectId, client, clusterService, snapshotHistoryStore.get()),
                 clusterService,
                 getClock()
             )

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

@@ -13,11 +13,14 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.scheduler.SchedulerEngine;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
@@ -31,8 +34,9 @@
 import java.time.Clock;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -47,22 +51,23 @@ public class SnapshotLifecycleService implements Closeable, ClusterStateListener
     private static final Logger logger = LogManager.getLogger(SnapshotLifecycleService.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 final AtomicBoolean running = new AtomicBoolean(true);
+    private final Function<ProjectId, SnapshotLifecycleTask> taskProvider;
+    private final Settings settings;
+    private final Clock clock;
     private volatile boolean isMaster = false;
+    private final ConcurrentMap<ProjectId, SnapshotLifecycleProjectState> projects = ConcurrentCollections.newConcurrentMap();
 
     public SnapshotLifecycleService(
         Settings settings,
-        Supplier<SnapshotLifecycleTask> taskSupplier,
+        Function<ProjectId, SnapshotLifecycleTask> taskProvider,
         ClusterService clusterService,
         Clock clock
     ) {
-        this.scheduler = new SchedulerEngine(settings, clock);
+        this.settings = settings;
+        this.taskProvider = taskProvider;
         this.clusterService = clusterService;
-        this.snapshotTask = taskSupplier.get();
+        this.clock = clock;
     }
 
     /**
@@ -73,36 +78,40 @@ public void init() {
     }
 
     @Override
+    @FixForMultiProject(description = "Project deletion should cancel the scheduled SLM tasks")
     public void clusterChanged(final ClusterChangedEvent event) {
         // Instead of using a LocalNodeMasterListener to track master changes, this service will
         // track them here to avoid conditions where master listener events run after other
         // listeners that depend on what happened in the master listener
         final boolean prevIsMaster = this.isMaster;
-        if (prevIsMaster != event.localNodeMaster()) {
-            this.isMaster = event.localNodeMaster();
-            if (this.isMaster) {
-                scheduler.register(snapshotTask);
-            } else {
-                scheduler.unregister(snapshotTask);
-                cancelSnapshotJobs();
-            }
-        }
+        final boolean masterChanged = prevIsMaster != event.localNodeMaster();
+        this.isMaster = event.localNodeMaster();
 
-        if (this.isMaster) {
-            final ClusterState state = event.state();
-
-            if (slmStoppedOrStopping(state)) {
-                if (scheduler.scheduledJobIds().size() > 0) {
-                    cancelSnapshotJobs();
-                }
-                if (slmStopping(state)) {
-                    submitUnbatchedTask("slm_operation_mode_update[stopped]", OperationModeUpdateTask.slmMode(OperationMode.STOPPED));
+        for (ProjectMetadata metadata : event.state().metadata().projects().values()) {
+            SnapshotLifecycleProjectState project = getOrCreateProjectState(metadata.id());
+            if (masterChanged) {
+                if (this.isMaster) {
+                    project.scheduler.register(project.snapshotTask);
+                } else {
+                    project.scheduler.unregister(project.snapshotTask);
+                    cancelSnapshotJobs(project);
                 }
-                return;
             }
 
-            scheduleSnapshotJobs(state);
-            cleanupDeletedPolicies(state);
+            if (this.isMaster) {
+                if (slmStoppedOrStopping(metadata)) {
+                    if (project.scheduler.scheduledJobIds().isEmpty() == false) {
+                        cancelSnapshotJobs(project);
+                    }
+                    if (slmStopping(metadata)) {
+                        submitUnbatchedTask("slm_operation_mode_update[stopped]", OperationModeUpdateTask.slmMode(OperationMode.STOPPED));
+                    }
+                    continue;
+                }
+
+                scheduleSnapshotJobs(project, metadata);
+                cleanupDeletedPolicies(project, metadata);
+            }
         }
     }
 
@@ -112,48 +121,61 @@ private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String
     }
 
     // Only used for testing
-    SchedulerEngine getScheduler() {
-        return this.scheduler;
+    SchedulerEngine getScheduler(ProjectId projectId) {
+        SnapshotLifecycleProjectState project = projects.get(projectId);
+        return project == null ? null : project.scheduler;
     }
 
     /**
      * Returns true if SLM is in the stopping or stopped state
      */
+    @Deprecated(forRemoval = true)
     static boolean slmStoppedOrStopping(ClusterState state) {
         OperationMode mode = currentSLMMode(state);
         return OperationMode.STOPPING == mode || OperationMode.STOPPED == mode;
     }
 
+    /**
+     * Returns true if SLM is in the stopping or stopped state
+     */
+    static boolean slmStoppedOrStopping(ProjectMetadata metadata) {
+        OperationMode mode = currentSLMMode(metadata);
+        return OperationMode.STOPPING == mode || OperationMode.STOPPED == mode;
+    }
+
     /**
      * Returns true if SLM is in the stopping state
      */
-    static boolean slmStopping(ClusterState state) {
-        OperationMode mode = currentSLMMode(state);
+    static boolean slmStopping(ProjectMetadata metadata) {
+        OperationMode mode = currentSLMMode(metadata);
         return OperationMode.STOPPING == mode;
     }
 
     /**
      * Schedule all non-scheduled snapshot jobs contained in the cluster state
      */
-    public void scheduleSnapshotJobs(final ClusterState state) {
-        SnapshotLifecycleMetadata snapMeta = state.metadata().getProject().custom(SnapshotLifecycleMetadata.TYPE);
+    private void scheduleSnapshotJobs(final SnapshotLifecycleProjectState project, final ProjectMetadata metadata) {
+        SnapshotLifecycleMetadata snapMeta = metadata.custom(SnapshotLifecycleMetadata.TYPE);
         if (snapMeta != null) {
-            snapMeta.getSnapshotConfigurations().values().forEach(this::maybeScheduleSnapshot);
+            snapMeta.getSnapshotConfigurations().values().forEach((config) -> maybeScheduleSnapshot(project, config));
         }
     }
 
-    public void cleanupDeletedPolicies(final ClusterState state) {
-        SnapshotLifecycleMetadata snapMeta = state.metadata().getProject().custom(SnapshotLifecycleMetadata.TYPE);
+    private void cleanupDeletedPolicies(final SnapshotLifecycleProjectState project, final ProjectMetadata metadata) {
+        SnapshotLifecycleMetadata snapMeta = metadata.custom(SnapshotLifecycleMetadata.TYPE);
         if (snapMeta != null) {
-            // Retrieve all of the expected policy job ids from the policies in the metadata
+            // Retrieve all 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);
+            project.scheduledTasks.keySet()
+                .stream()
+                .filter(jobId -> policyJobIds.contains(jobId) == false)
+                .forEach(jobId -> cancelScheduledSnapshot(project, jobId));
         }
     }
 
@@ -162,16 +184,16 @@ public void cleanupDeletedPolicies(final ClusterState state) {
      * 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) {
-        if (this.running.get() == false) {
+    void maybeScheduleSnapshot(final SnapshotLifecycleProjectState project, final SnapshotLifecyclePolicyMetadata snapshotLifecyclePolicy) {
+        if (project.running.get() == false) {
             return;
         }
 
         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()
+        final boolean existingJobsFoundAndCancelled = project.scheduledTasks.keySet()
             .stream()
             // Find all jobs matching the `jobid-\d+` pattern
             .filter(jId -> existingJobPattern.matcher(jId).matches())
@@ -180,8 +202,8 @@ public void maybeScheduleSnapshot(final SnapshotLifecyclePolicyMetadata snapshot
             .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);
+                project.scheduledTasks.remove(existingJobId);
+                boolean existed = project.scheduler.remove(existingJobId);
                 assert existed : "expected job for " + existingJobId + " to exist in scheduler";
                 return existed;
             })
@@ -190,15 +212,15 @@ public void maybeScheduleSnapshot(final SnapshotLifecyclePolicyMetadata snapshot
         // 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 -> {
+        project.scheduledTasks.computeIfAbsent(jobId, id -> {
             if (existingJobsFoundAndCancelled) {
                 logger.info("rescheduling updated snapshot lifecycle job [{}]", jobId);
             } else {
                 logger.info("scheduling snapshot lifecycle job [{}]", jobId);
             }
 
             final SchedulerEngine.Job job = snapshotLifecyclePolicy.buildSchedulerJob(jobId);
-            scheduler.add(job);
+            project.scheduler.add(job);
             return job;
         });
     }
@@ -213,25 +235,34 @@ public static String getJobId(SnapshotLifecyclePolicyMetadata policyMeta) {
     /**
      * Cancel all scheduled snapshot jobs
      */
-    public void cancelSnapshotJobs() {
+    private void cancelSnapshotJobs(SnapshotLifecycleProjectState project) {
         logger.trace("cancelling all snapshot lifecycle jobs");
-        scheduler.scheduledJobIds().forEach(scheduler::remove);
-        scheduledTasks.clear();
+        project.scheduler.scheduledJobIds().forEach(project.scheduler::remove);
+        project.scheduledTasks.clear();
     }
 
     /**
      * Cancel the given policy job id (from {@link #getJobId(SnapshotLifecyclePolicyMetadata)}
      */
-    public void cancelScheduledSnapshot(final String lifecycleJobId) {
-        logger.debug("cancelling snapshot lifecycle job [{}] as it no longer exists", lifecycleJobId);
-        scheduledTasks.remove(lifecycleJobId);
-        scheduler.remove(lifecycleJobId);
+    private void cancelScheduledSnapshot(final SnapshotLifecycleProjectState project, final String lifecycleJobId) {
+        logger.debug("cancelling project [{}] snapshot lifecycle job [{}] as it no longer exists", project.projectId, lifecycleJobId);
+        project.scheduledTasks.remove(lifecycleJobId);
+        project.scheduler.remove(lifecycleJobId);
+    }
+
+    private SnapshotLifecycleProjectState getOrCreateProjectState(ProjectId projectId) {
+        // initialize the project state if it does not exist (i.e. new project)
+        return projects.computeIfAbsent(
+            projectId,
+            id -> new SnapshotLifecycleProjectState(projectId, new SchedulerEngine(settings, clock), taskProvider.apply(projectId))
+        );
     }
 
     /**
      * Validates that the {@code repository} exists as a registered snapshot repository
      * @throws IllegalArgumentException if the repository does not exist
      */
+    @Deprecated(forRemoval = true)
     public static void validateRepositoryExists(final String repository, final ClusterState state) {
         if (RepositoriesMetadata.get(state).repository(repository) == null) {
             throw new IllegalArgumentException("no such repository [" + repository + "]");
@@ -260,8 +291,27 @@ public static void validateMinimumInterval(final SnapshotLifecyclePolicy lifecyc
 
     @Override
     public void close() {
-        if (this.running.compareAndSet(true, false)) {
-            this.scheduler.stop();
+        projects.values().forEach(project -> {
+            if (project.running.compareAndSet(true, false)) {
+                project.scheduler.stop();
+            }
+        });
+    }
+
+    /**
+     * Internal wrapper class to hold the state of a project's snapshot lifecycle
+     */
+    static class SnapshotLifecycleProjectState {
+        final ProjectId projectId;
+        final SchedulerEngine scheduler;
+        final SnapshotLifecycleTask snapshotTask;
+        final Map<String, SchedulerEngine.Job> scheduledTasks = ConcurrentCollections.newConcurrentMap();
+        final AtomicBoolean running = new AtomicBoolean(true);
+
+        SnapshotLifecycleProjectState(ProjectId projectId, SchedulerEngine scheduler, SnapshotLifecycleTask snapshotTask) {
+            this.projectId = projectId;
+            this.scheduler = scheduler;
+            this.snapshotTask = snapshotTask;
         }
     }
 }