You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "arjunashok (via GitHub)" <gi...@apache.org> on 2024/10/23 21:51:41 UTC
[PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
arjunashok opened a new pull request, #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139
These changes are part of the effort to introduce bespoke Sidecar APIs to support key operational functionality currently managed through nodetool commands. The APIs for some of the commands can trigger long-running tasks on Cassandra.
The job management framework allows management of the async jobs submitted via the command APIs while coordinating downstream with the Cassandra node.
## Changes
- Job management framework
- Introduces APIs for `job status` and `list jobs`, with corresponding Sidecar client classes and tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852989244
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
Review Comment:
I had this renamed to `isRunningDownstream` in my earlier draft. Will update to that.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
Review Comment:
Ack
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
Review Comment:
Much better :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852989461
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
+ }
+ catch (final InterruptedException e)
+ {
+ return false;
Review Comment:
Ack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1840646927
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
+ {
+ super(initialCapacity);
+ this.capacity = initialCapacity;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job put(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.put(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job putIfAbsent(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.putIfAbsent(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void putAll(final Map<? extends UUID, ? extends Job> m)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.putAll(m);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job remove(final Object key)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.remove(key);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear()
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.clear();
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job get(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.get(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean containsKey(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.containsKey(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.size();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.isEmpty();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object clone()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.clone();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean removeEldestEntry(final Map.Entry<UUID, Job> eldest)
+ {
+ if (size() <= capacity)
Review Comment:
I am still not convinced as this doesn't feel like smooth user experience. Returning 404 after Sidecar restart sounds fine. But without any process restart, kicking out a ongoing job from the map and user seeing 404 doesn't feel like great user experience. It will confuse the user who just submitted the job, polling for status and seeing 404.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1881198709
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/data/OperationalJobsEntry.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.common.response.data;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Structure of the operational job instance within the list operational jobs API response
+ */
+public class OperationalJobsEntry
Review Comment:
Not needed. WIll remove.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ListOperationalJobsHandler.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.routes;
+
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.ListOperationalJobsResponse;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+
+/**
+ * Handler for retrieving the all the jobs running on the sidecar
+ */
+public class ListOperationalJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+ @Inject
+ public ListOperationalJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ protected void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ ListOperationalJobsResponse listResponse = new ListOperationalJobsResponse();
+ jobManager.allInflightJobs()
+ .stream()
+ .map(job -> new OperationalJobResponse(job.jobId, RUNNING, job.name(), ""))
Review Comment:
Yes. Will update.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationalJobHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operational jobs running on the sidecar
+ */
+public class OperationalJobHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+
+ @Inject
+ public OperationalJobHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobId = validateJobIdParam(context);
+ executorPools.service()
+ .executeBlocking(() -> {
+ OperationalJob job = jobManager.getJobIfExists(jobId);
+ if (job == null)
+ {
+ logger.info("No operational job found with the jobId. jobId={}", jobId);
+ throw wrapHttpException(HttpResponseStatus.NOT_FOUND,
+ String.format("Unknown job with ID: %s. Please retry the operation.", jobId));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobId, job));
+ }
+
+ private UUID validateJobIdParam(RoutingContext context)
+ {
+ String requestJobId = context.pathParam(OPERATIONAL_JOB_ID_PATH_PARAM.substring(1));
+ if (requestJobId == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ OPERATIONAL_JOB_ID_PATH_PARAM + " is required but not supplied");
+ }
+
+ UUID jobId;
+ try
+ {
+ jobId = UUID.fromString(requestJobId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ logger.info("Invalid jobId. jobId={}", requestJobId);
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, String.format("Invalid job ID provided :%s.", requestJobId));
+ }
+ return jobId;
+ }
+
+ public void sendStatusBasedResponse(RoutingContext context, UUID jobId, OperationalJob job)
+ {
+ OperationalJobStatus status = job.status();
+ String reason = null;
+ switch(status)
+ {
+ case SUCCEEDED:
+ context.response().setStatusCode(HttpResponseStatus.OK.code()); break;
+ case CREATED:
+ case RUNNING:
+ context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code()); break;
+ case FAILED:
+ reason = job.asyncResult().cause().getMessage();
Review Comment:
Added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1882773509
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationalJobHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operational jobs running on the sidecar
+ */
+public class OperationalJobHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+
+ @Inject
+ public OperationalJobHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobId = validateJobIdParam(context);
+ executorPools.service()
+ .executeBlocking(() -> {
+ OperationalJob job = jobManager.getJobIfExists(jobId);
+ if (job == null)
+ {
+ logger.info("No operational job found with the jobId. jobId={}", jobId);
+ throw wrapHttpException(HttpResponseStatus.NOT_FOUND,
+ String.format("Unknown job with ID: %s. Please retry the operation.", jobId));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobId, job));
+ }
+
+ private UUID validateJobIdParam(RoutingContext context)
+ {
+ String requestJobId = context.pathParam(OPERATIONAL_JOB_ID_PATH_PARAM.substring(1));
+ if (requestJobId == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ OPERATIONAL_JOB_ID_PATH_PARAM + " is required but not supplied");
+ }
+
+ UUID jobId;
+ try
+ {
+ jobId = UUID.fromString(requestJobId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ logger.info("Invalid jobId. jobId={}", requestJobId);
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, String.format("Invalid job ID provided :%s.", requestJobId));
+ }
+ return jobId;
+ }
+
+ public void sendStatusBasedResponse(RoutingContext context, UUID jobId, OperationalJob job)
+ {
+ OperationalJobStatus status = job.status();
+ String reason = null;
+ switch(status)
+ {
+ case SUCCEEDED:
+ context.response().setStatusCode(HttpResponseStatus.OK.code()); break;
+ case CREATED:
+ case RUNNING:
+ context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code()); break;
+ case FAILED:
+ reason = job.asyncResult().cause().getMessage();
Review Comment:
yep. 200 to indicate the request is successful. The status of the job is expressed in the response body.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1868225922
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
Review Comment:
`isRunning` also sounds good to me. But without the `Downstream` part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831813747
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobManager.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.JobResult.JobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class JobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobManager.class);
+
+ private final JobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public JobManager(ExecutorPools executorPools)
+ {
+ this.executorPools = executorPools;
+ jobTracker = new JobTracker(64);
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<Job> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == JobStatus.Pending || j.status() == JobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public Job getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
+ }
+
+ /**
+ * Asynchronously submit (and lazily create, via the supplier) the job, if it is not currently being
+ * tracked and is not running downstream. The job is triggered on a separate internal thread-pool.
+ * The job execution failure behavior is tracked within the {@link Job}.
+ * @param jobId job identifier
+ * @param jobSupplier supplier used to create an instance of the job
+ * @return the instance of the job that is either being tracked or was just submitted
+ */
+ public Job trySubmitJob(UUID jobId, Supplier<Job> jobSupplier)
+ {
+
+ return jobTracker.computeIfAbsent(jobId, id -> {
+ Job job = jobSupplier.get();
+ LOGGER.info("Created job with ID: {}, operation: {}", job.jobId(), job.operation());
+ jobTracker.put(job.jobId(), job);
Review Comment:
yep redundant, removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh merged PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852990474
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobManager.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class OperationsJobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJobManager.class);
+
+ private final OperationsJobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public OperationsJobManager(ExecutorPools executorPools, OperationsJobTracker jobTracker)
+ {
+ this.executorPools = executorPools;
+ this.jobTracker = jobTracker;
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<OperationsJob> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == OperationsJobStatus.Pending || j.status() == OperationsJobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public OperationsJob getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
+ }
+
+ /**
+ * Asynchronously submit (and lazily create, via the supplier) the job, if it is not currently being
+ * tracked and is not running downstream. The job is triggered on a separate internal thread-pool.
+ * The job execution failure behavior is tracked within the {@link OperationsJob}.
+ * @param jobId job identifier
+ * @param jobSupplier supplier used to create an instance of the job
+ * @return the instance of the job that is either being tracked or was just submitted
+ */
+ public OperationsJob trySubmitJob(UUID jobId, Supplier<OperationsJob> jobSupplier)
Review Comment:
I changed this to use a creator function (`Function<UUID, OperationsJob> creator`) that takes UUID argument, to improve clarity on which ID is used for the job. This should address the correctness issue as the returned job always corresponds to the headerId passed in (or a new UUID when absent).
As for the return-type, we need to job status and id to send the appropriate response.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = validateJobIdParam(context);
Review Comment:
Ack
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
Review Comment:
Ack
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = validateJobIdParam(context);
+
+ executorPools.service().executeBlocking(() -> {
+ OperationsJob job = jobManager.getJobIfExists(jobUUID);
+ if (job == null)
+ {
+ String response = String.format("Unknown job with ID:%s. Please retry the operation.", jobUUID);
+ logger.info(response);
+ context.fail(wrapHttpException(HttpResponseStatus.NOT_FOUND, response));
Review Comment:
Ack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1844358267
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarHttpHeaderNames.java:
##########
@@ -31,4 +31,9 @@ public final class SidecarHttpHeaderNames
* {@code "cassandra-content-xxhash32-seed"}
*/
public static final String CONTENT_XXHASH32_SEED = "cassandra-content-xxhash32-seed";
+
+ /**
+ * {@code "cassandra-operations-job-uuid"}
+ */
+ public static final String OPERATIONS_JOB_HEADER_NAME = "cassandra-operations-jobid";
Review Comment:
the header seems to be duplication since there is `OperationsJobsResponse` that contains jobId too.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/OperationsJobsRequest.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the status of a async operations job
+ */
+public class OperationsJobsRequest extends JsonRequest<OperationsJobsResponse>
+{
+
+ /**
+ * Constructs a request to retrieve status for a specified operations job
+ */
+ public OperationsJobsRequest(String jobId)
Review Comment:
if a jobId is a uuid, then the input type here should be uuid.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
+{
+ private final UUID jobId;
+ private final OperationsJobResult.OperationsJobStatus status;
+ private final String operation;
+ private final String reason;
Review Comment:
`OperationsJobResult` consists of status and reason. Why flatten here?
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
+
+ private OperationsJobStatus status;
+ private String reason;
Review Comment:
`public final` is usually preferred over `private`
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
Review Comment:
Add `@JsonIgnoreProperties(ignoreUnknown = true)` for compatibility.
Add `@JsonInclude(JsonInclude.Include.NON_NULL)` if any field is nullable.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/ListOperationsJobsRequest.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.ListOperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the list of async operations jobs
+ */
+public class ListOperationsJobsRequest extends JsonRequest<ListOperationsJobsResponse>
Review Comment:
"OperationsJobs" reads unnatural. Should it be "OperationalJobs"?
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/ListOperationsJobsResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.response;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Response structure of the list operations jobs API
+ */
+public class ListOperationsJobsResponse
+{
+ private final List<OperationsJobsResponse> jobs;
+
+ /**
+ * Constructs a {@link ListOperationsJobsResponse} object.
+ */
+ public ListOperationsJobsResponse()
+ {
+ this.jobs = new ArrayList<>();
+ }
+
+ public void addJob(OperationsJobsResponse job)
+ {
+ jobs.add(job);
+ }
+
+ @JsonProperty("jobs")
+ public List<OperationsJobsResponse> jobs()
+ {
+ return jobs;
+ }
+
+ /**
+ * Structure of the operations job instance within the list operations jobs API response
+ */
+ public static class OperationsJobsResponse
+ {
+ public final UUID jobId;
Review Comment:
I think the jobId does not necessarily be a `UUID`, since it is produced by the local instance and queried against the instance only. It can be just an integer.
But not necessarily to be addressed neither.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
Review Comment:
nit: one line per variant, and `Completed --> Succeeded`
Do we really need to distinguish between `pending` and `running`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831788281
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/JobStatusHandler.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.JobStatusResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.ASYNC_JOB_UUID;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async jobs running on the sidecar
+ */
+public class JobStatusHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public JobStatusHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = verifyJobIdPresent(context);
+
+ executorPools.service().executeBlocking(() -> {
+ Job job = jobManager.getJobIfExists(jobUUID);
+ if (job == null)
+ {
+ String response = String.format("Unknown job with ID:%s. Please retry the operation.", jobUUID);
+ logger.info(response);
+ context.fail(wrapHttpException(HttpResponseStatus.NOT_FOUND, response));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobUUID, job));
+ }
+
+ private UUID verifyJobIdPresent(RoutingContext context)
Review Comment:
Fair point, will update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1868153729
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/ListOperationsJobsRequest.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.ListOperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the list of async operations jobs
+ */
+public class ListOperationsJobsRequest extends JsonRequest<ListOperationsJobsResponse>
Review Comment:
"Operational" also describes something that is related to operations.
> of, relating to, or based on operations
See https://www.merriam-webster.com/dictionary/operational
It is confusing to see two nouns putting together. Looks like Hari has the same confusion.
Please update the name for clarity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1823489255
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -109,6 +109,9 @@ public final class ApiEndpointsV1
public static final String ABORT_RESTORE_JOB_ROUTE = RESTORE_JOB_ROUTE + ABORT;
public static final String RESTORE_JOB_PROGRESS_ROUTE = RESTORE_JOB_ROUTE + PROGRESS;
+ public static final String LIST_JOBS_ROUTE = API_V1 + CASSANDRA + "/jobs";
+ public static final String JOB_STATUS_ROUTE = API_V1 + CASSANDRA + "/jobs/" + JOB_ID_PATH_PARAM + "/status";
Review Comment:
> The job details API is not there. Is it reserved for later?
This is the only job API since it would return the job status, operation and failure reason when applicable. I do not see a use-case for a separate details API, although I do agree we could do away with the `status` suffix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1829892416
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarHttpHeaderNames.java:
##########
@@ -31,4 +31,9 @@ public final class SidecarHttpHeaderNames
* {@code "cassandra-content-xxhash32-seed"}
*/
public static final String CONTENT_XXHASH32_SEED = "cassandra-content-xxhash32-seed";
+
+ /**
+ * {@code "cassandra-sidecar-job-uuid"}
+ */
+ public static final String ASYNC_JOB_UUID = "cassandra-sidecar-job-uuid";
Review Comment:
I don't think this should be a header name, instead we should return a the UUID as part of the JSON payload
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobManager.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.JobResult.JobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class JobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobManager.class);
+
+ private final JobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public JobManager(ExecutorPools executorPools)
+ {
+ this.executorPools = executorPools;
+ jobTracker = new JobTracker(64);
Review Comment:
shouldn't job tracker be injected and we should use the singleton instance?
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
Review Comment:
let's remove unnecessary `final` usage in this class
##########
server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java:
##########
@@ -588,6 +599,13 @@ public LocalTokenRangesProvider localTokenRangesProvider(InstancesConfig instanc
return new CachedLocalTokenRanges(instancesConfig, dnsResolver);
}
+ @Provides
+ @Singleton
+ public JobTracker jobTracker()
+ {
+ return new JobTracker(64);
Review Comment:
should we make this configurable?
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/JobStatusHandler.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.JobStatusResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.ASYNC_JOB_UUID;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async jobs running on the sidecar
+ */
+public class JobStatusHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public JobStatusHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = verifyJobIdPresent(context);
+
+ executorPools.service().executeBlocking(() -> {
Review Comment:
this is getting data from memory, no need to execute it in the service executor pool
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ListJobsHandler.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.routes;
+
+import java.util.List;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.ListJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+/**
+ * Handler for retrieving the all the jobs running on the sidecar
+ */
+public class ListJobsHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public ListJobsHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
Review Comment:
let's add the `@Override` annotation.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ListJobsHandler.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.routes;
+
+import java.util.List;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.ListJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+/**
+ * Handler for retrieving the all the jobs running on the sidecar
+ */
+public class ListJobsHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public ListJobsHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ protected void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ executorPools.service()
Review Comment:
There's no need to run this on the service executor pool. These jobs are in memory.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/JobStatusHandler.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.JobStatusResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.ASYNC_JOB_UUID;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async jobs running on the sidecar
+ */
+public class JobStatusHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public JobStatusHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
Review Comment:
this should return the jobUUID
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobManager.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.JobResult.JobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class JobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobManager.class);
+
+ private final JobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public JobManager(ExecutorPools executorPools)
+ {
+ this.executorPools = executorPools;
+ jobTracker = new JobTracker(64);
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<Job> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == JobStatus.Pending || j.status() == JobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public Job getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
+ }
+
+ /**
+ * Asynchronously submit (and lazily create, via the supplier) the job, if it is not currently being
+ * tracked and is not running downstream. The job is triggered on a separate internal thread-pool.
+ * The job execution failure behavior is tracked within the {@link Job}.
+ * @param jobId job identifier
+ * @param jobSupplier supplier used to create an instance of the job
+ * @return the instance of the job that is either being tracked or was just submitted
+ */
+ public Job trySubmitJob(UUID jobId, Supplier<Job> jobSupplier)
+ {
+
+ return jobTracker.computeIfAbsent(jobId, id -> {
+ Job job = jobSupplier.get();
+ LOGGER.info("Created job with ID: {}, operation: {}", job.jobId(), job.operation());
+ jobTracker.put(job.jobId(), job);
Review Comment:
this seems wrong, you shouldn't call put here, that's the job of computeIfAbsent, no?
##########
server/src/main/java/org/apache/cassandra/sidecar/job/Job.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.JobResult;
+import org.apache.cassandra.sidecar.common.utils.JobResult.JobStatus;
+
+/**
+ * An abstract class representing a Job managed by the sidecar.
+ *
+ */
+public abstract class Job
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(Job.class);
+
+ protected UUID jobId;
+ protected JobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected Job()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected Job(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = JobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected Job(UUID jobId, JobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(JobStatus status)
+ {
+ this.status = status;
+ }
+ public JobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<JobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ JobResult result = jobOperationSupplier().get();
Review Comment:
it feels we should use Future primitives here instead, execute should return a future as well, instead of doing latch countdown
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -109,6 +109,9 @@ public final class ApiEndpointsV1
public static final String ABORT_RESTORE_JOB_ROUTE = RESTORE_JOB_ROUTE + ABORT;
public static final String RESTORE_JOB_PROGRESS_ROUTE = RESTORE_JOB_ROUTE + PROGRESS;
+ public static final String LIST_JOBS_ROUTE = API_V1 + CASSANDRA + "/jobs";
+ public static final String JOB_STATUS_ROUTE = API_V1 + CASSANDRA + "/jobs/" + JOB_ID_PATH_PARAM + "/status";
Review Comment:
I think we should qualify these endpoints as admin. Also, I think we should not have the `status` path part . I propose we use the following instead:
```
/api/v1/admin/cassandra/jobs
/api/v1/admin/cassandra/jobs/<job_id>
```
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
do we really need to expose all the methods from Map here? I only see usages of `getJobsView` , `get`, `computeIfAbsent` (which is not even guarded by the lock), and `put` which seems wrong to have . I think I'd prefer to have a smaller API, that exposes only the things we need. Maybe we can simplify by doing this:
```
Map<UUID, Job> internalTracker = Collections.synchronizedMap(new LinkedHashMap<UUID, Job>()
{
@Override
protected boolean removeEldestEntry(Map.Entry<UUID, Job> eldest)
{
if (size() <= capacity)
{
return false;
}
LOGGER.warn("Job tracker reached max size, so expiring job uuid={}", eldest.getKey());
return true;
}
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1827774693
##########
client/src/main/java/org/apache/cassandra/sidecar/client/RequestContext.java:
##########
@@ -486,6 +489,29 @@ public Builder connectedClientStatsRequest()
return request(new ConnectedClientStatsRequest());
}
+ /**
+ * Sets the {@code request} to be a {@link JobStatusRequest} and returns a reference to this Builder
+ * enabling method chaining.
+ *
+ * @return a reference to this Builder
+ */
+ public Builder jobStatusRequest(String jobId)
Review Comment:
Minor, to be in.line with listJobsRequest() naming style below, can we have this named statusJobRequest?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1868270482
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/ListOperationsJobsRequest.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.ListOperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the list of async operations jobs
+ */
+public class ListOperationsJobsRequest extends JsonRequest<ListOperationsJobsResponse>
Review Comment:
sounds reasonable. "Operational" it is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852993115
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
+
+ private OperationsJobStatus status;
+ private String reason;
Review Comment:
coding style conventions for Cassandra
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852988622
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/ListOperationsJobsRequest.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.ListOperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the list of async operations jobs
+ */
+public class ListOperationsJobsRequest extends JsonRequest<ListOperationsJobsResponse>
Review Comment:
I don't have a strong opinion here. The reason I did not choose "Operational" is because it seems to suggest something is "working" or "in-use" and not specifically jobs that are used in Operations context (which seemed more appropriate). Lmk what you think.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/request/OperationsJobsRequest.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.common.request;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+
+/**
+ * Represents a request to retrieve the status of a async operations job
+ */
+public class OperationsJobsRequest extends JsonRequest<OperationsJobsResponse>
+{
+
+ /**
+ * Constructs a request to retrieve status for a specified operations job
+ */
+ public OperationsJobsRequest(String jobId)
Review Comment:
Done
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/ListOperationsJobsResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.response;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Response structure of the list operations jobs API
+ */
+public class ListOperationsJobsResponse
+{
+ private final List<OperationsJobsResponse> jobs;
Review Comment:
Will do
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1853002379
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
+
+ private OperationsJobStatus status;
+ private String reason;
Review Comment:
https://cassandra.apache.org/_/development/code_style.html
```
Boilerplate
Prefer public final fields to private fields with getters (but prefer encapsulating behavior in "real" methods to either).
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1882741426
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationalJobHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operational jobs running on the sidecar
+ */
+public class OperationalJobHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+
+ @Inject
+ public OperationalJobHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobId = validateJobIdParam(context);
+ executorPools.service()
+ .executeBlocking(() -> {
+ OperationalJob job = jobManager.getJobIfExists(jobId);
+ if (job == null)
+ {
+ logger.info("No operational job found with the jobId. jobId={}", jobId);
+ throw wrapHttpException(HttpResponseStatus.NOT_FOUND,
+ String.format("Unknown job with ID: %s. Please retry the operation.", jobId));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobId, job));
+ }
+
+ private UUID validateJobIdParam(RoutingContext context)
+ {
+ String requestJobId = context.pathParam(OPERATIONAL_JOB_ID_PATH_PARAM.substring(1));
+ if (requestJobId == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ OPERATIONAL_JOB_ID_PATH_PARAM + " is required but not supplied");
+ }
+
+ UUID jobId;
+ try
+ {
+ jobId = UUID.fromString(requestJobId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ logger.info("Invalid jobId. jobId={}", requestJobId);
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, String.format("Invalid job ID provided :%s.", requestJobId));
+ }
+ return jobId;
+ }
+
+ public void sendStatusBasedResponse(RoutingContext context, UUID jobId, OperationalJob job)
+ {
+ OperationalJobStatus status = job.status();
+ String reason = null;
+ switch(status)
+ {
+ case SUCCEEDED:
+ context.response().setStatusCode(HttpResponseStatus.OK.code()); break;
+ case CREATED:
+ case RUNNING:
+ context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code()); break;
+ case FAILED:
+ reason = job.asyncResult().cause().getMessage();
Review Comment:
does 200 make sense here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831844315
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
I think @yifan-c and I are suggesting the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1827785462
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
+ {
+ super(initialCapacity);
+ this.capacity = initialCapacity;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job put(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.put(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job putIfAbsent(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.putIfAbsent(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void putAll(final Map<? extends UUID, ? extends Job> m)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.putAll(m);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job remove(final Object key)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.remove(key);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear()
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.clear();
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job get(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.get(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean containsKey(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.containsKey(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.size();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.isEmpty();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object clone()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.clone();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean removeEldestEntry(final Map.Entry<UUID, Job> eldest)
+ {
+ if (size() <= capacity)
Review Comment:
Let's say capacity is 5 (just assuming for the discussion), all 5 jobs are ongoing, if a new job comes up do we force delete the eldest ongoing job from tracker? If so, how can a user track that deleted eldest job?
Instead of having capacity, if we are just worried about too many ongoing jobs and jobtracker taking up too much memory, can we have a cap on ongoing jobs and reject incoming requests if jobtracker reached the maximum limit and ask user to wait to finish an ongoing job to submit a new one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1876481811
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
+{
+ private final UUID jobId;
+ private final OperationsJobResult.OperationsJobStatus status;
+ private final String operation;
+ private final String reason;
Review Comment:
Latest revision removes the need for status and reason as they are inferred from the Future state which is cached.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831829316
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
How about _composing_ a concurrent hash map, instead of _inheriting_ a map. Composition provides a better flexibility to expose only the necessary APIs.
I am also not convinced with removing the old jobs. If bounded maps is required, it is clearer to me to reject the creation of new jobs. Otherwise, it (i.e. removing jobs underneath) loses the purpose of tracking.
```java
public class JobTracker
{
private ConcurrentHashMap<UUID, Job> jobsById;
```
(somehow my last comment got swallowed. Writing up the same comment again...)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831784749
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
+ {
+ super(initialCapacity);
+ this.capacity = initialCapacity;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job put(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.put(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job putIfAbsent(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.putIfAbsent(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void putAll(final Map<? extends UUID, ? extends Job> m)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.putAll(m);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job remove(final Object key)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.remove(key);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear()
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.clear();
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job get(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.get(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean containsKey(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.containsKey(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.size();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.isEmpty();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object clone()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.clone();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean removeEldestEntry(final Map.Entry<UUID, Job> eldest)
+ {
+ if (size() <= capacity)
Review Comment:
So, the intention behind the tracker is only to maintain the state for ongoing jobs without growing unbounded, but not to throttle/limit the no. of jobs that can be posted.
Since jobs are typically short-lived, the entry's age based eviction seems reasonable as a capping mechanism.
> how can a user track that deleted eldest job
Ideally, we should size this optimally to minimize evictions of ongoing jobs.
If a job that is inflight on the C* node gets evicted from the cache, the subsequent polling will return a 404.
The expected behavior for the client at this point is to retry the command (eg. decommission), here the downstream check will determine that there is an ongoing operation and return a 409 Conflict, and start tracking it with a new UUID (which is part of the response).
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
+ {
+ super(initialCapacity);
+ this.capacity = initialCapacity;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job put(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.put(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job putIfAbsent(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.putIfAbsent(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void putAll(final Map<? extends UUID, ? extends Job> m)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.putAll(m);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job remove(final Object key)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.remove(key);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear()
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.clear();
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job get(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.get(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean containsKey(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.containsKey(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.size();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.isEmpty();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object clone()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.clone();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean removeEldestEntry(final Map.Entry<UUID, Job> eldest)
+ {
+ if (size() <= capacity)
Review Comment:
So, the intention behind the tracker is only to maintain the state for ongoing jobs without growing unbounded, but not to throttle/limit the no. of jobs that can be posted.
Since jobs are typically short-lived, the entry's age based eviction seems reasonable as a capping mechanism.
> how can a user track that deleted eldest job
Ideally, we should size this optimally to minimize evictions of ongoing jobs.
If a job that is inflight on the C* node gets evicted from the cache, the subsequent polling will return a 404.
The expected behavior for the client at this point is to retry the command (eg. decommission), here the downstream check will determine that there is an ongoing operation and return a 409 Conflict, and start tracking it with a new UUID (which is part of the response).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831815069
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
In addition to overriding the LHM, we also intend to make it thread-safe with read/write locks around the operations, so we will need all the methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1815795344
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -109,6 +109,9 @@ public final class ApiEndpointsV1
public static final String ABORT_RESTORE_JOB_ROUTE = RESTORE_JOB_ROUTE + ABORT;
public static final String RESTORE_JOB_PROGRESS_ROUTE = RESTORE_JOB_ROUTE + PROGRESS;
+ public static final String LIST_JOBS_ROUTE = API_V1 + CASSANDRA + "/jobs";
+ public static final String JOB_STATUS_ROUTE = API_V1 + CASSANDRA + "/jobs/" + JOB_ID_PATH_PARAM + "/status";
Review Comment:
`/status` feels redundant.
`.../jobs/:jobId` can be used to retrieve the job details, including status. The job details API is not there. Is it reserved for later?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "nvharikrishna (via GitHub)" <gi...@apache.org>.
nvharikrishna commented on PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#issuecomment-2512232843
I see operations and jobs coming together often in this PR. Small question on terminology being used. Is a job seen as a set of operations? Or are they the same?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1868224767
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
+{
+ private final UUID jobId;
+ private final OperationsJobResult.OperationsJobStatus status;
+ private final String operation;
+ private final String reason;
Review Comment:
`OperationsJobsResponse.status.operation == OperationsJobsResponse.operation` and `OperationsJobsResponse.status.reason == OperationsJobsResponse.reason`.
They are duplicated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1882746280
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationalJobHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operational jobs running on the sidecar
+ */
+public class OperationalJobHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+
+ @Inject
+ public OperationalJobHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobId = validateJobIdParam(context);
+ executorPools.service()
+ .executeBlocking(() -> {
+ OperationalJob job = jobManager.getJobIfExists(jobId);
+ if (job == null)
+ {
+ logger.info("No operational job found with the jobId. jobId={}", jobId);
+ throw wrapHttpException(HttpResponseStatus.NOT_FOUND,
+ String.format("Unknown job with ID: %s. Please retry the operation.", jobId));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobId, job));
+ }
+
+ private UUID validateJobIdParam(RoutingContext context)
+ {
+ String requestJobId = context.pathParam(OPERATIONAL_JOB_ID_PATH_PARAM.substring(1));
+ if (requestJobId == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ OPERATIONAL_JOB_ID_PATH_PARAM + " is required but not supplied");
+ }
+
+ UUID jobId;
+ try
+ {
+ jobId = UUID.fromString(requestJobId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ logger.info("Invalid jobId. jobId={}", requestJobId);
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, String.format("Invalid job ID provided :%s.", requestJobId));
+ }
+ return jobId;
+ }
+
+ public void sendStatusBasedResponse(RoutingContext context, UUID jobId, OperationalJob job)
+ {
+ OperationalJobStatus status = job.status();
+ String reason = null;
+ switch(status)
+ {
+ case SUCCEEDED:
+ context.response().setStatusCode(HttpResponseStatus.OK.code()); break;
+ case CREATED:
+ case RUNNING:
+ context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code()); break;
+ case FAILED:
+ reason = job.asyncResult().cause().getMessage();
Review Comment:
Yes, since 200 is the status of the (current) job API request, which succeeded. The response body has the status of the job being queried which is `FAILED`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#issuecomment-2460973932
There are empty lines in many files. Those empty lines should be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831855064
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarHttpHeaderNames.java:
##########
@@ -31,4 +31,9 @@ public final class SidecarHttpHeaderNames
* {@code "cassandra-content-xxhash32-seed"}
*/
public static final String CONTENT_XXHASH32_SEED = "cassandra-content-xxhash32-seed";
+
+ /**
+ * {@code "cassandra-sidecar-job-uuid"}
+ */
+ public static final String ASYNC_JOB_UUID = "cassandra-sidecar-job-uuid";
Review Comment:
As discussed offline, the responses for operation APIs will have a payload.
The UUID header, however is meant to provide a consistent interface for all long running jobs, independent of whether it has a response or not (eg. long-running GET operations).
So, for a client triggering a operation, the expectation is to use the header value to poll for completion (using the job status API) without having to parse the response when the status is 202.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#issuecomment-2454826207
Good work 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#issuecomment-2512335607
> I see operations and jobs coming together often in this PR. Small question on terminology being used. Is a job seen as a set of operations? Or are they the same?
So, the term "OperationsJob" or "OperationalJob" in the context of this PR refers to a long-running sidecar functionality to perform operational tasks for the node, such as `decommission`, `repair`, `rebuild` etc. Such jobs, when initiated, would return an identifier that can be used by the client to poll for completion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#issuecomment-2486887529
Submitted the comments of the full review. Thanks for the patch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1827794030
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/JobStatusHandler.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.JobStatusResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.Job;
+import org.apache.cassandra.sidecar.job.JobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.ASYNC_JOB_UUID;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async jobs running on the sidecar
+ */
+public class JobStatusHandler extends AbstractHandler<Void>
+{
+ private final JobManager jobManager;
+ @Inject
+ public JobStatusHandler(InstanceMetadataFetcher metadataFetcher, ExecutorPools executorPools, CassandraInputValidator validator, JobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = verifyJobIdPresent(context);
+
+ executorPools.service().executeBlocking(() -> {
+ Job job = jobManager.getJobIfExists(jobUUID);
+ if (job == null)
+ {
+ String response = String.format("Unknown job with ID:%s. Please retry the operation.", jobUUID);
+ logger.info(response);
+ context.fail(wrapHttpException(HttpResponseStatus.NOT_FOUND, response));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobUUID, job));
+ }
+
+ private UUID verifyJobIdPresent(RoutingContext context)
Review Comment:
From the name verifyJobIdPresent, I was expecting this function to lookup jobId in the ongoing jobIds list and error if not found. Whereas this function only ensuring given jobid is in UUID format. Not sure am I missing something.
If we just want to ensure jobId string is a valid UUID, can we rename this function as validateJobId (or similar)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831845873
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
The tracker does not need to be a map. It just uses map internally, and the map should be encapsulated within the tracker, not exposing the map APIs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "skoppu22 (via GitHub)" <gi...@apache.org>.
skoppu22 commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1840646927
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
+{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobTracker.class);
+ int capacity;
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public JobTracker(final int initialCapacity)
+ {
+ super(initialCapacity);
+ this.capacity = initialCapacity;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job put(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.put(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job putIfAbsent(final UUID key, final Job value)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.putIfAbsent(key, value);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void putAll(final Map<? extends UUID, ? extends Job> m)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.putAll(m);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job remove(final Object key)
+ {
+ lock.writeLock().lock();
+ try
+ {
+ return super.remove(key);
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void clear()
+ {
+ lock.writeLock().lock();
+ try
+ {
+ super.clear();
+ }
+ finally
+ {
+ lock.writeLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Job get(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.get(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean containsKey(final Object key)
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.containsKey(key);
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public int size()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.size();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public boolean isEmpty()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.isEmpty();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Object clone()
+ {
+ lock.readLock().lock();
+ try
+ {
+ return super.clone();
+ }
+ finally
+ {
+ lock.readLock().unlock();
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected boolean removeEldestEntry(final Map.Entry<UUID, Job> eldest)
+ {
+ if (size() <= capacity)
Review Comment:
I am still not convinced, as this doesn't feel like smooth user experience. Returning 404 after Sidecar restart sounds fine. But without any process restart, kicking out an ongoing job from the map and user seeing 404 doesn't feel like great user experience. It will confuse the user who just submitted the job, polling for status and seeing 404.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852988965
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
Review Comment:
Ack
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
+
+ private OperationsJobStatus status;
+ private String reason;
Review Comment:
Why have public attributes, when we have the accessors?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1876480284
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
Discussed offline. Simplified tracker API in latest revision
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1864705374
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class OperationsJobTracker extends LinkedHashMap<UUID, OperationsJob>
Review Comment:
@yifan-c Can you elaborate on simplifications?
Regarding the need for map, we already had a discussion in this thread about this - the LRU eviction policy and thread-safety is driving this need. https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1830081175.
> I think internally we should keep track of Futures
Not sure I understand, there are no Futures referenced in this class. Can you elaborate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1830257859
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -109,6 +109,9 @@ public final class ApiEndpointsV1
public static final String ABORT_RESTORE_JOB_ROUTE = RESTORE_JOB_ROUTE + ABORT;
public static final String RESTORE_JOB_PROGRESS_ROUTE = RESTORE_JOB_ROUTE + PROGRESS;
+ public static final String LIST_JOBS_ROUTE = API_V1 + CASSANDRA + "/jobs";
+ public static final String JOB_STATUS_ROUTE = API_V1 + CASSANDRA + "/jobs/" + JOB_ID_PATH_PARAM + "/status";
Review Comment:
Yep. I mean exactly this, replacing `jobs/:jobId/status` with `jobs/:jobId`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1849095200
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
Review Comment:
nit:
```suggestion
public abstract boolean isInflight();
```
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
+ }
+ catch (final InterruptedException e)
+ {
+ return false;
Review Comment:
preserve the interrupted flag.
```java
Thread.currentThread().interrupt();
```
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
Review Comment:
nit: consider adding `final` to forbidden overriding this method, since the concrete logic is implemented in `jobOperationSupplier`
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobManager.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class OperationsJobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJobManager.class);
+
+ private final OperationsJobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public OperationsJobManager(ExecutorPools executorPools, OperationsJobTracker jobTracker)
+ {
+ this.executorPools = executorPools;
+ this.jobTracker = jobTracker;
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<OperationsJob> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == OperationsJobStatus.Pending || j.status() == OperationsJobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public OperationsJob getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
+ }
+
+ /**
+ * Asynchronously submit (and lazily create, via the supplier) the job, if it is not currently being
+ * tracked and is not running downstream. The job is triggered on a separate internal thread-pool.
+ * The job execution failure behavior is tracked within the {@link OperationsJob}.
+ * @param jobId job identifier
+ * @param jobSupplier supplier used to create an instance of the job
+ * @return the instance of the job that is either being tracked or was just submitted
+ */
+ public OperationsJob trySubmitJob(UUID jobId, Supplier<OperationsJob> jobSupplier)
Review Comment:
Passing a `Supplier` would have correctness issue. (Explained at the end)
My suggestion is to have this signature instead.
```suggestion
public boolean trySubmitJob(OperationsJob job);
```
1. Use boolean to indicate the submission result.
2. `OperationsJob` already contains jobId. The parameter `UUID jobId` can be omitted.
Now, why there is potential correctness issue.
Say there are 3 threads submitting the same jobId with different suppliers at the same time. Let's say thread A wins, thread B and C lose. Since the signature returns a `OperationsJob`, there is no way to tell which thread has succeeded. All A, B and C will consider they have created the job successfully. The ambiguity can break the subsequent execution.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobManager.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class OperationsJobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJobManager.class);
+
+ private final OperationsJobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public OperationsJobManager(ExecutorPools executorPools, OperationsJobTracker jobTracker)
+ {
+ this.executorPools = executorPools;
+ this.jobTracker = jobTracker;
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<OperationsJob> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == OperationsJobStatus.Pending || j.status() == OperationsJobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public OperationsJob getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
Review Comment:
It is odd that the `allInflightJobs` is backed by jobView, but `getJobIfExists` reads from jobTracker directly. It makes me wondering that `getJobsView` is unnecessary to have.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
Review Comment:
```suggestion
return latch.await(waitSeconds, TimeUnit.SECONDS);
```
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
+ }
+ catch (final InterruptedException e)
+ {
+ return false;
+ }
+ }
Review Comment:
Not at a fan of this blocking API.
Consider this option. Schedule a future task / timer, and check the result after timer times out.
```
public boolean isComplete()
{
// of course, not necessary to use CountDownLatch in this case
return latch.getCount() == 0;
}
// at call-site
Vertx vertx;
OperationsJob operationsJob;
vertx.timer(waitSeconds, TimeUnit.SECONDS).map(v -> operationsJob.isComplete())...
```
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/ListOperationsJobsResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.response;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Response structure of the list operations jobs API
+ */
+public class ListOperationsJobsResponse
+{
+ private final List<OperationsJobsResponse> jobs;
Review Comment:
It read confusing to me that the array in the json response body is also named "Response". Can you place the current `OperationsJobsResponse` class in package `org.apache.cassandra.sidecar.common.response.data` and rename the class to `OperationalJobEntry`? In other words, the `ListOperationalJobsResponse` contains an array of `OperationalJobEntry`
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class OperationsJobTracker extends LinkedHashMap<UUID, OperationsJob>
Review Comment:
The class can use some simplifications.
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
Review Comment:
Returning a function seems to complex.
The equivalent is probably easier to understand.
```java
protected abstract OperationsJobResult executeInternal() throws OperationalJobException;
```
Note that the access modifier can be `protected` and prefer a domain specific exception type.
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = validateJobIdParam(context);
+
+ executorPools.service().executeBlocking(() -> {
+ OperationsJob job = jobManager.getJobIfExists(jobUUID);
+ if (job == null)
+ {
+ String response = String.format("Unknown job with ID:%s. Please retry the operation.", jobUUID);
+ logger.info(response);
+ context.fail(wrapHttpException(HttpResponseStatus.NOT_FOUND, response));
Review Comment:
```suggestion
throw wrapHttpException(HttpResponseStatus.NOT_FOUND, response);
```
Otherwise, it is going to return job and hit the `onSucesss` handler (unnecessarily).
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
Review Comment:
`@Override`
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationsJobsHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.OperationsJobsResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationsJob;
+import org.apache.cassandra.sidecar.job.OperationsJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONS_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.OPERATIONS_JOB_HEADER_NAME;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operations jobs running on the sidecar
+ */
+public class OperationsJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationsJobManager jobManager;
+ @Inject
+ public OperationsJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationsJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobUUID = validateJobIdParam(context);
Review Comment:
nit: rename to `jobId`
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/utils/OperationsJobResult.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.common.utils;
+
+/**
+ * Holder class for the results of a job execution. Captures the job status and reason
+ * as a result of performing the downstream operation.
+ */
+public class OperationsJobResult
+{
+ /**
+ * Encapsulates the states of the job lifecycle. All new jobs are in Pending state.
+ */
+ public enum OperationsJobStatus
+ { Pending, Running, Completed, Failed }
Review Comment:
Enum variants name should be in upper case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1853278755
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class OperationsJobTracker extends LinkedHashMap<UUID, OperationsJob>
Review Comment:
+1 we don't need a full blown map here. Also, I think internally we should keep track of Futures
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1853087360
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ OperationsJobResult result = jobOperationSupplier().get();
+ status = result.status();
+ failureReason = result.reason();
+ LOGGER.debug("Job with ID: {} returned with status: {}", jobId, status);
+ }
+ catch (Exception e)
+ {
+ String reason = (e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+ LOGGER.error("Failed to execute job {} with reason: {}", jobId, reason);
+ status = OperationsJobStatus.Failed;
+ failureReason = reason;
+ }
+ finally
+ {
+ latch.countDown();
+ }
+ }
+
+ /**
+ * Returns if the job execution completes within the provided wait time (in seconds)
+ * @param waitSeconds no. of seconds to wait for the job execution to complete
+ * @return true if the job completed within the specified time
+ */
+ public boolean isResultAvailable(long waitSeconds)
+ {
+ try
+ {
+ return (!latch.await(waitSeconds, TimeUnit.SECONDS)) ? false : true;
+ }
+ catch (final InterruptedException e)
+ {
+ return false;
+ }
+ }
Review Comment:
I see the concern, but in this case, we will have -
1) handler of long-running job invoking `trySubmitJob` within `executorPools.service().executeBlocking(()`
2) Job execution/submission done within `executorPools.internal().runBlocking()`
So we have a multi-level indirection. For vertx to wait for the execution to complete, we would need some form of Future chaining to detect completion of the inner most execution thread.
We will have to weigh the complexity of that to the existing implementation. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852988804
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/ListOperationsJobsResponse.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.common.response;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Response structure of the list operations jobs API
+ */
+public class ListOperationsJobsResponse
+{
+ private final List<OperationsJobsResponse> jobs;
+
+ /**
+ * Constructs a {@link ListOperationsJobsResponse} object.
+ */
+ public ListOperationsJobsResponse()
+ {
+ this.jobs = new ArrayList<>();
+ }
+
+ public void addJob(OperationsJobsResponse job)
+ {
+ jobs.add(job);
+ }
+
+ @JsonProperty("jobs")
+ public List<OperationsJobsResponse> jobs()
+ {
+ return jobs;
+ }
+
+ /**
+ * Structure of the operations job instance within the list operations jobs API response
+ */
+ public static class OperationsJobsResponse
+ {
+ public final UUID jobId;
Review Comment:
I see your point, but it does help for it to be guaranteed to be globally unique, when we are debugging these jobs across multiple nodes (even though they are currently node-local) to avoid confusion.
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
Review Comment:
Ack
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/OperationsJobsResponse.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.response;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+
+/**
+ * Response structure of the operations jobs API
+ */
+public class OperationsJobsResponse
+{
+ private final UUID jobId;
+ private final OperationsJobResult.OperationsJobStatus status;
+ private final String operation;
+ private final String reason;
Review Comment:
Not sure I understand. Are you asking why we're not using `OperationsJobResult` as a member here?
If so, I did not see a reason to have the hierarchy since all these are attributes of the jobs API response.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1852989670
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJobManager.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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.List;
+import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import javax.inject.Inject;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+
+/**
+ * An abstraction of the management and tracking of long-running jobs running on the sidecar.
+ */
+@Singleton
+public class OperationsJobManager
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJobManager.class);
+
+ private final OperationsJobTracker jobTracker;
+ private final ExecutorPools executorPools;
+
+ /**
+ * Creates a manager instance with a default sized job-tracker.
+ * @param executorPools
+ */
+ @Inject
+ public OperationsJobManager(ExecutorPools executorPools, OperationsJobTracker jobTracker)
+ {
+ this.executorPools = executorPools;
+ this.jobTracker = jobTracker;
+ }
+
+ /**
+ * Fetches the inflight jobs being tracked on the sidecar
+ * @return instances of the jobs that are in pending or running states
+ */
+ public List<OperationsJob> allInflightJobs()
+ {
+ return jobTracker.getJobsView().values()
+ .stream()
+ .filter(j -> j.status() == OperationsJobStatus.Pending || j.status() == OperationsJobStatus.Running)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Fetch the job using its UUID
+ * @param jobId identifier of the job
+ * @return instance of the job or null
+ */
+ public OperationsJob getJobIfExists(UUID jobId)
+ {
+ return jobTracker.get(jobId);
Review Comment:
`allInflightJobs` iterates over all entries, so doing this without holding locks is beneficial to not affect other actions.
`getJobIfExists` on the other hand is a point lookup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1853037276
##########
server/src/main/java/org/apache/cassandra/sidecar/job/OperationsJob.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult;
+import org.apache.cassandra.sidecar.common.utils.OperationsJobResult.OperationsJobStatus;
+
+/**
+ * An abstract class representing a Operations job managed by the sidecar.
+ *
+ */
+public abstract class OperationsJob
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(OperationsJob.class);
+
+ protected UUID jobId;
+ protected OperationsJobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected OperationsJob()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected OperationsJob(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = OperationsJobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected OperationsJob(UUID jobId, OperationsJobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(OperationsJobStatus status)
+ {
+ this.status = status;
+ }
+ public OperationsJobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<OperationsJobResult> jobOperationSupplier() throws Exception;
Review Comment:
I like this idea. Will give it a shot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1881178656
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ListOperationalJobsHandler.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.routes;
+
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.response.ListOperationalJobsResponse;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.RUNNING;
+
+/**
+ * Handler for retrieving the all the jobs running on the sidecar
+ */
+public class ListOperationalJobsHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+ @Inject
+ public ListOperationalJobsHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ protected void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ ListOperationalJobsResponse listResponse = new ListOperationalJobsResponse();
+ jobManager.allInflightJobs()
+ .stream()
+ .map(job -> new OperationalJobResponse(job.jobId, RUNNING, job.name(), ""))
Review Comment:
can the reason be null here? or do we have some constraint that it Neds to be an empty string?
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/response/data/OperationalJobsEntry.java:
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.common.response.data;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * Structure of the operational job instance within the list operational jobs API response
+ */
+public class OperationalJobsEntry
Review Comment:
is this class needed?
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/OperationalJobHandler.java:
##########
@@ -0,0 +1,120 @@
+/*
+ * 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.routes;
+
+import java.util.UUID;
+import javax.inject.Inject;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.job.OperationalJob;
+import org.apache.cassandra.sidecar.job.OperationalJobManager;
+import org.apache.cassandra.sidecar.utils.CassandraInputValidator;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+
+import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
+import static org.apache.cassandra.sidecar.common.data.OperationalJobStatus.FAILED;
+import static org.apache.cassandra.sidecar.utils.HttpExceptions.wrapHttpException;
+
+/**
+ * Handler for retrieving the status of async operational jobs running on the sidecar
+ */
+public class OperationalJobHandler extends AbstractHandler<Void>
+{
+ private final OperationalJobManager jobManager;
+
+ @Inject
+ public OperationalJobHandler(InstanceMetadataFetcher metadataFetcher,
+ ExecutorPools executorPools,
+ CassandraInputValidator validator,
+ OperationalJobManager jobManager)
+ {
+ super(metadataFetcher, executorPools, validator);
+ this.jobManager = jobManager;
+ }
+
+ @Override
+ protected Void extractParamsOrThrow(RoutingContext context)
+ {
+ return null;
+ }
+
+ @Override
+ public void handleInternal(RoutingContext context, HttpServerRequest httpRequest, String host, SocketAddress remoteAddress, Void request)
+ {
+ UUID jobId = validateJobIdParam(context);
+ executorPools.service()
+ .executeBlocking(() -> {
+ OperationalJob job = jobManager.getJobIfExists(jobId);
+ if (job == null)
+ {
+ logger.info("No operational job found with the jobId. jobId={}", jobId);
+ throw wrapHttpException(HttpResponseStatus.NOT_FOUND,
+ String.format("Unknown job with ID: %s. Please retry the operation.", jobId));
+ }
+ return job;
+ })
+ .onFailure(cause -> processFailure(cause, context, host, remoteAddress, request))
+ .onSuccess(job -> sendStatusBasedResponse(context, jobId, job));
+ }
+
+ private UUID validateJobIdParam(RoutingContext context)
+ {
+ String requestJobId = context.pathParam(OPERATIONAL_JOB_ID_PATH_PARAM.substring(1));
+ if (requestJobId == null)
+ {
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST,
+ OPERATIONAL_JOB_ID_PATH_PARAM + " is required but not supplied");
+ }
+
+ UUID jobId;
+ try
+ {
+ jobId = UUID.fromString(requestJobId);
+ }
+ catch (IllegalArgumentException e)
+ {
+ logger.info("Invalid jobId. jobId={}", requestJobId);
+ throw wrapHttpException(HttpResponseStatus.BAD_REQUEST, String.format("Invalid job ID provided :%s.", requestJobId));
+ }
+ return jobId;
+ }
+
+ public void sendStatusBasedResponse(RoutingContext context, UUID jobId, OperationalJob job)
+ {
+ OperationalJobStatus status = job.status();
+ String reason = null;
+ switch(status)
+ {
+ case SUCCEEDED:
+ context.response().setStatusCode(HttpResponseStatus.OK.code()); break;
+ case CREATED:
+ case RUNNING:
+ context.response().setStatusCode(HttpResponseStatus.ACCEPTED.code()); break;
+ case FAILED:
+ reason = job.asyncResult().cause().getMessage();
Review Comment:
do we need to send a status code here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-150: Adds job management framework and APIs [cassandra-sidecar]
Posted by "arjunashok (via GitHub)" <gi...@apache.org>.
arjunashok commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1876479704
##########
server/src/main/java/org/apache/cassandra/sidecar/job/Job.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.sidecar.common.utils.JobResult;
+import org.apache.cassandra.sidecar.common.utils.JobResult.JobStatus;
+
+/**
+ * An abstract class representing a Job managed by the sidecar.
+ *
+ */
+public abstract class Job
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(Job.class);
+
+ protected UUID jobId;
+ protected JobStatus status;
+ protected String failureReason;
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ protected Job()
+ {
+
+ }
+ /**
+ * Constructs a job with a unique UUID, in Pending state
+ * @param jobId UUID representing the Job to be created
+ */
+ protected Job(UUID jobId)
+ {
+ this.jobId = jobId;
+ this.status = JobStatus.Pending;
+ this.failureReason = "";
+ }
+
+ @VisibleForTesting
+ protected Job(UUID jobId, JobStatus status)
+ {
+ this.jobId = jobId;
+ this.status = status;
+ this.failureReason = "";
+ }
+
+ public void setStatus(JobStatus status)
+ {
+ this.status = status;
+ }
+ public JobStatus status()
+ {
+ return status;
+ }
+ public String failureReason()
+ {
+ return failureReason;
+ }
+
+ public UUID jobId()
+ {
+ return jobId;
+ }
+
+ /**
+ * Supplier specifying the functionality of the job to be triggered when the job is executed. Subclasses to
+ * provide operation-specific implementations
+ * @return a function with the operation implementation that returns a {@code JobResult}
+ * @throws Exception
+ */
+ public abstract Supplier<JobResult> jobOperationSupplier() throws Exception;
+
+ /**
+ * Provide a meaningful name of the operation executed by the concrete subclass.
+ * @return the name of the operation. eg. nodetool command name
+ */
+ public abstract String operation();
+
+ /**
+ * Specifies the downstream operation to be performed to check if the job is running on the Cassandra node/cluster.
+ * This functionality is provided by the Job when it is asynchronously triggering the job via the {@code JobManager}
+ * For synchronous jobs this should always return false.
+ * @return true if the job is running downstream
+ */
+ public abstract boolean checkInflightJob();
+
+ /**
+ * Execute the job behavior as specified in the operation supplier {@link #jobOperationSupplier()},
+ * while tracking the status of the job's lifecycle.
+ */
+ public void execute()
+ {
+ try
+ {
+ LOGGER.info("Executing job with ID: {}", jobId);
+ JobResult result = jobOperationSupplier().get();
Review Comment:
Addressed in latest revision
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRASC-150 Adds job management framework and APIs [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #139:
URL: https://github.com/apache/cassandra-sidecar/pull/139#discussion_r1831843899
##########
server/src/main/java/org/apache/cassandra/sidecar/job/JobTracker.java:
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import com.google.common.collect.ImmutableMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Tracks and stores the results of long-running jobs running on the sidecar
+ */
+public class JobTracker extends LinkedHashMap<UUID, Job>
Review Comment:
Wrapping the map in `Collections.synchronizedMap` makes it thread safe. and we reduce the exposure of the internals
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org