CASSSIDECAR-377: Implement job coordination for cluster-wide operations

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Implement job coordination for cluster-wide operations (CASSSIDECAR-377)
  * Implement durable operational job tracker (CASSSIDECAR-374)
  * Remove filesystem path from Http response (CASSSIDECAR-477)
  * Add basic configuration retrieval logic to ConfigurationManager (CASSSIDECAR-427)

server/src/main/java/org/apache/cassandra/sidecar/job/DisabledOperationalJobCoordinator.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.sidecar.job;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An {@link OperationalJobCoordinator} used when a Sidecar instance is not configured to support
+ * coordinated cluster-wide operations.
+ * <p>
+ * Jobs that do not require coordination ({@link OperationalJob#requiresCoordination()} returns
+ * {@code false}) never reach this coordinator, so uncoordinated operations (e.g. decommission) are
+ * unaffected. 
+ */
+public class DisabledOperationalJobCoordinator implements OperationalJobCoordinator
+{
+    private static final String NOT_SUPPORTED_MESSAGE =
+    "Operational job coordination is not supported by this Sidecar instance. "
+    + "Configure a coordinator to enable coordinated cluster-wide operations.";
+
+    @Override
+    public boolean trySetActive(OperationType operationType, UUID operationId)
+    {
+        throw new UnsupportedOperationException(NOT_SUPPORTED_MESSAGE);
+    }
+
+    @Override
+    public boolean clearActive(OperationType operationType, UUID operationId)
+    {
+        throw new UnsupportedOperationException(NOT_SUPPORTED_MESSAGE);
+    }
+
+    @Override
+    @Nullable
+    public UUID getActiveOperation(OperationType operationType)
+    {
+        return null;
+    }
+
+    @Override
+    @NotNull
+    public Map<OperationType, UUID> getActiveOperations()
+    {
+        return Collections.emptyMap();
+    }
+}

server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJob.java

@@ -183,6 +183,24 @@ public boolean requiresCoordination()
         return false;
     }
 
+    /**
+     * Whether the manager should release the active operation lock (via
+     * {@link OperationalJobCoordinator#clearActive}) when this job completes locally. Only consulted for jobs that
+     * {@link #requiresCoordination() require coordination}.
+     * <p>
+     * Single-node coordinated jobs return {@code true} (the default): the Sidecar that acquires the lock also
+     * finishes the work, so releasing on local completion is correct. Distributed cluster-wide jobs whose work
+     * finishes on other nodes return {@code false} and rely on the orchestration layer to call
+     * {@link OperationalJobCoordinator#clearActive} once all nodes reach a terminal state.
+     *
+     * @return {@code true} if the manager should release the lock on local completion; {@code false} to defer
+     *         release to the orchestration layer
+     */
+    public boolean releasesOnCompletion()
+    {
+        return true;
+    }
+
     @Override
     public final Void result()
     {
@@ -343,6 +361,19 @@ public Future<Void> asyncResult(TaskExecutorPool executorPool, DurationSpec wait
                      });
     }
 
+    /**
+     * Marks the job as failed before it begins executing, for example when it cannot be started due to a
+     * coordination conflict. The execution result is completed exceptionally so the job reports
+     * {@link OperationalJobStatus#FAILED} with the supplied reason, and {@link #executeInternal()} is never invoked.
+     *
+     * @param cause the reason the job could not be started
+     */
+    public void failToStart(Throwable cause)
+    {
+        lastUpdate = Instant.now();
+        executionPromise.tryFail(cause);
+    }
+
     /**
      * OperationalJob body. The implementation returns a Future representing the job execution.
      */

server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobCoordinator.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.cassandra.sidecar.job;
+
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.cassandra.sidecar.common.data.OperationType;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Coordinates cluster-wide operational jobs by managing active operation state.
+ * Ensures mutual exclusion so that only one operation of a given type is active
+ * at a time within a datacenter.
+ */
+public interface OperationalJobCoordinator
+{
+    /**
+     * Attempt to set an operation as active. Implementations must ensure mutual exclusion
+     * so that only one operation of the given type is active at a time.
+     *
+     * @param operationType the type of operation
+     * @param operationId   the unique identifier for this operation
+     * @return {@code true} if the operation was successfully set as active,
+     *         {@code false} if an operation of the same type is already active
+     */
+    boolean trySetActive(OperationType operationType, UUID operationId);
+
+    /**
+     * Clear the active operation lock, but only if the provided operation ID matches
+     * the currently active one.
+     * <p>
+     * This method is not called by {@code OperationalJobManager} during job submission.
+     * For cluster-wide operations, the Sidecar that creates the job is not necessarily
+     * the one that finishes it. Clearing is the responsibility of the orchestration layer
+     * once all participating nodes have completed their work.
+     *
+     * @param operationType the type of operation
+     * @param operationId   the operation ID to clear
+     * @return {@code true} if the active operation was cleared,
+     *         {@code false} if the provided operation ID did not match the active one
+     */
+    boolean clearActive(OperationType operationType, UUID operationId);
+
+    /**
+     * Get the active operation ID for a given operation type.
+     *
+     * @param operationType the type of operation
+     * @return the active operation ID, or {@code null} if no operation of this type is active
+     */
+    @Nullable
+    UUID getActiveOperation(OperationType operationType);
+
+    /**
+     * Get all active operations.
+     *
+     * @return a map of operation type to operation ID for all currently active operations,
+     *         or an empty map if no operations are active
+     */
+    @NotNull
+    Map<OperationType, UUID> getActiveOperations();
+}

server/src/main/java/org/apache/cassandra/sidecar/job/OperationalJobManager.java

@@ -28,10 +28,12 @@
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.vertx.core.Future;
 import org.apache.cassandra.sidecar.common.server.utils.DurationSpec;
 import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
 import org.apache.cassandra.sidecar.concurrent.TaskExecutorPool;
 import org.apache.cassandra.sidecar.exceptions.OperationalJobConflictException;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * An abstraction of the management and tracking of long-running jobs running on the sidecar.
@@ -41,18 +43,24 @@ public class OperationalJobManager
 {
     protected final Logger logger = LoggerFactory.getLogger(this.getClass());
     private final OperationalJobTracker jobTracker;
-
+    private final OperationalJobCoordinator coordinator;
     private final TaskExecutorPool internalExecutorPool;
 
     /**
-     * Creates a manager instance with a default sized job-tracker.
+     * Creates a manager instance with a coordinator for cluster-wide operation mutual exclusion. Instances that
+     * do not support coordination bind a {@link DisabledOperationalJobCoordinator}, which fails coordination
+     * requests.
      *
-     * @param jobTracker the tracker for the operational jobs
+     * @param jobTracker  the tracker for the operational jobs
+     * @param coordinator the coordinator for cluster-wide operations
      */
     @Inject
-    public OperationalJobManager(OperationalJobTracker jobTracker, ExecutorPools executorPools)
+    public OperationalJobManager(OperationalJobTracker jobTracker,
+                                 OperationalJobCoordinator coordinator,
+                                 ExecutorPools executorPools)
     {
         this.jobTracker = jobTracker;
+        this.coordinator = coordinator;
         this.internalExecutorPool = executorPools.internal();
     }
 
@@ -99,20 +107,65 @@ public void trySubmitJob(OperationalJob job,
         try
         {
             checkConflict(job);
-
-            // Track the job first, then start execution separately
-            OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
-            if (tracked == job)
-            {
-                internalExecutorPool.executeBlocking(job::execute);
-            }
         }
         catch (OperationalJobConflictException oje)
         {
             onComplete.accept(job, oje);
             return;
         }
 
+        if (!job.requiresCoordination())
+        {
+            trackAndExecute(job, onComplete, serviceExecutorPool, waitTime);
+            return;
+        }
+
+        // Acquiring the active operation lock might perform blocking storage I/O so it must
+        // run off the event loop
+        acquireActiveOperationLock(job)
+        .onComplete(ar ->
+        {
+            if (ar.succeeded() && Boolean.TRUE.equals(ar.result()))
+            {
+                // Distributed jobs that finish on other nodes opt out of auto-release; the orchestration
+                // layer clears the lock once all nodes reach a terminal state.
+                if (job.releasesOnCompletion())
+                {
+                    job.asyncResult().onComplete(result -> releaseActiveOperationLock(job));
+                }
+                trackAndExecute(job, onComplete, serviceExecutorPool, waitTime);
+            }
+            else
+            {
+                OperationalJobConflictException conflict = coordinationConflict(job, ar.cause());
+                job.failToStart(conflict);
+                jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
+                onComplete.accept(job, conflict);
+            }
+        });
+    }
+
+    /**
+     * Tracks the job and submits it for asynchronous execution on the internal executor pool, then arranges for
+     * {@code onComplete} to be invoked with the result (or after {@code waitTime} elapses).
+     *
+     * @param job                 the job to track and execute
+     * @param onComplete          callback to invoke when the job completes
+     * @param serviceExecutorPool the executor pool to use for waiting on job completion
+     * @param waitTime            the maximum time to wait for job completion before returning
+     */
+    private void trackAndExecute(OperationalJob job,
+                                 BiConsumer<OperationalJob, OperationalJobConflictException> onComplete,
+                                 TaskExecutorPool serviceExecutorPool,
+                                 DurationSpec waitTime)
+    {
+        // New job is submitted for all cases when we do not have a corresponding downstream job
+        OperationalJob tracked = jobTracker.computeIfAbsent(job.jobId(), jobId -> job);
+        if (tracked == job)
+        {
+            internalExecutorPool.executeBlocking(job::execute);
+        }
+
         // Get the result, waiting for the specified wait time for result
         job.asyncResult(serviceExecutorPool, waitTime)
            .onComplete(v -> onComplete.accept(job, null));
@@ -132,4 +185,51 @@ private void checkConflict(OperationalJob job) throws OperationalJobConflictExce
             throw new OperationalJobConflictException("The same operational job is already running on Cassandra. operationName='" + job.name() + '\'');
         }
     }
+
+    /**
+     * For jobs that require cluster-wide coordination, attempts to acquire the active operation lock via the
+     * coordinator. The acquisition runs on the internal executor pool because it might performs blocking storage I/O
+     *
+     * @param job the job requiring coordination
+     * @return a future resolving to {@code true} if the lock was acquired, {@code false} if another operation
+     *         already holds it, or a failed future if coordination could not be attempted (e.g. coordination is
+     *         disabled on this instance or the storage call failed)
+     */
+    private Future<Boolean> acquireActiveOperationLock(OperationalJob job)
+    {
+        return internalExecutorPool.executeBlocking(() -> coordinator.trySetActive(job.operationType(), job.jobId()), false);
+    }
+
+    /**
+     * Releases the active operation lock previously acquired for the given job. The release performs
+     * blocking storage I/O, so it runs on the internal executor pool off the event loop. A failure to clear is logged
+     * rather than surfaced, since the job has already completed by this point.
+     *
+     * @param job the job whose active operation lock should be released
+     */
+    private void releaseActiveOperationLock(OperationalJob job)
+    {
+        internalExecutorPool.executeBlocking(() -> coordinator.clearActive(job.operationType(), job.jobId()), false)
+                            .onFailure(e -> logger.error("Failed to clear active operation lock. jobId={} operationType={}",
+                                                         job.jobId(), job.operationType(), e));
+    }
+
+    /**
+     * Builds the conflict exception describing why the active operation lock could not be acquired.
+     *
+     * @param job   the job that could not be coordinated
+     * @param cause the failure cause when coordination could not be attempted, or {@code null} when the lock is
+     *              simply held by another active operation
+     * @return the conflict exception to report to the caller
+     */
+    private OperationalJobConflictException coordinationConflict(OperationalJob job, @Nullable Throwable cause)
+    {
+        if (cause != null)
+        {
+            return new OperationalJobConflictException("Unable to coordinate operation. operationType='"
+                                                       + job.operationType() + "', reason='" + cause.getMessage() + '\'');
+        }
+        return new OperationalJobConflictException("An active operation already exists. operationType='"
+                                                   + job.operationType() + '\'');
+    }
 }