[Backport 2.x] Backports Job Scheduler Communication mechanism to 2.x (…

…#382) * Communication mechanism for js (#289) * Job Details from Extension for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Commnunication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> * Communication Mechanism for JS Signed-off-by: Varun Jain <[email protected]> Signed-off-by: Varun Jain <[email protected]> * [Extensions] Synchronize opensearch-plugin-job-details with JobDetailsService map (#299) * Added JobDetailsService as an indexOperationListener to synchronize metadata index with internal job details m and indicesToListen set. Signed-off-by: Joshua Palis <[email protected]> * Changes indexToJobDetails to a ConcurrentMap, adds getter method for indexToJobDetails Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments, fixing log message Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments Signed-off-by: Joshua Palis <[email protected]> * Added test to updateIndexToJobDetails Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments, changing extensionId to extensionUniqueId Signed-off-by: Joshua Palis <[email protected]> * Addressing PR Comments : Updating Job Details index Name and mapping file to opensearch-job-scheduler-job-details Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments, enabling extensions to submit more than 1 job details entry to support extensions registering multiple types of jobs, updated all integration, multinode tests now that the rest response value is the document Id Signed-off-by: Joshua Palis <[email protected]> * Renaming TestHelper base URI name Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments, made multinode test request strings constant Signed-off-by: Joshua Palis <[email protected]> Signed-off-by: Joshua Palis <[email protected]> * [Extensions] Create a proxy ScheduledJobRunner, ScheduledJobParser to invoke corresponding registered Job Detail actions (#306) * Draft : Added JobParameterRequest to translate ScheduledJobParser parameters nto compatible inputs for the ExtensionActionRequest. Completed initial proxy scheduled job parser implementation. Added ExtensionJobParameter class to deserialize ExtensionActionResponsebyte array into an object of type ScheduledJobParameter Signed-off-by: Joshua Palis <[email protected]> * Added JobRunnerRequest, modified JobExecutionContext to implement writeable, created initial proxy ScheduledJobRunner, fixed failing tests Signed-off-by: Joshua Palis <[email protected]> * Added generic ExtensionJobActionRequest that extends ExtensionActionRequest, modified JobParameter/Runner request to implement writeable, refactored proxy object creation so that the requests are added to the ExtensionJobActionRequest, which in turn upcasts the request into an ExtensionActionRequest Signed-off-by: Joshua Palis <[email protected]> * Added byte array constructors for the JobRunner/Parameter requests, added javadocs Signed-off-by: Joshua Palis <[email protected]> * Fixing javadocs Signed-off-by: Joshua Palis <[email protected]> * Fixing javadocs Signed-off-by: Joshua Palis <[email protected]> * Added placeholder string for an eventual access token to be sent to extensions to validate prior to invoking an action Signed-off-by: Joshua Palis <[email protected]> * Added ExtensionJobActionResponse, JobParameterResponse, JobRunnerResponse to facilitate the response of extension actions Signed-off-by: Joshua Palis <[email protected]> * Fixing javadocs Signed-off-by: Joshua Palis <[email protected]> * Added JobRunnerResponse handling Signed-off-by: Joshua Palis <[email protected]> * Separating updateIndexToJobProviders into separate methods Signed-off-by: Joshua Palis <[email protected]> * Fixing javadocs Signed-off-by: Joshua Palis <[email protected]> * SpotlessApply Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments Signed-off-by: Joshua Palis <[email protected]> * Added tests for serialization/deserialization of JobExecutionContext, JobDocVersion, ExtensionJobParameter, ExtensionJobActionRequest, ExtensionJobActionResponse, JobRunner/JobParameter/Request/Response Signed-off-by: Joshua Palis <[email protected]> * Writing ExtensionActionRequest/Response to bytestream output to test deserialization Signed-off-by: Joshua Palis <[email protected]> * Fixing imports Signed-off-by: Joshua Palis <[email protected]> * Changing to extensionActionResponse constructor Signed-off-by: Joshua Palis <[email protected]> * Adding tests for updateIndexToJobProviders Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments, added getters for lock duration seconds and jitter values, added javadocs to ScheduleType enum and made this public Signed-off-by: Joshua Palis <[email protected]> * Removing Strings dependency from lock model to fix BWC tests Signed-off-by: Joshua Palis <[email protected]> * Fixes BWC fullRestartClusterTask, rollingUpgradeClusterTask, oldVersionCluster task. mixedClusterTask is still failing Signed-off-by: Joshua Palis <[email protected]> * Modified createProxyScheduledJobRunner to return the document Id of the extension job parameter entry within the registered job index, modifed the ExtensionJobParameter to null check the jitter value and setting this to 0.0 if null Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Joshua Palis <[email protected]> * [Extensions] Exposes a GetLock REST API to enable extensions to acquire a lock model for their job execution (#311) * Added RestGetLockAction API, added AcquireLockRequest, enables extensions to retrieve a lock model object to run their Job Signed-off-by: Joshua Palis <[email protected]> * Adding Lock ID field to RestGetLockAction response Signed-off-by: Joshua Palis <[email protected]> * Added multi node integration tests for GetLockApi Signed-off-by: Joshua Palis <[email protected]> * Making lock service in RestGetLockAction private Signed-off-by: Joshua Palis <[email protected]> * Added Rest integration tests for RestGetLockAction Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments Signed-off-by: Joshua Palis <[email protected]> * Updating get lock rest path in tests Signed-off-by: Joshua Palis <[email protected]> * Addressing PR comments Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Joshua Palis <[email protected]> * Communication Mechanism Work Item 6 Release lock api (#312) * Release lock API Signed-off-by: Varun Jain <[email protected]> * Release Lock API Signed-off-by: Varun Jain <[email protected]> * Release Lock API Signed-off-by: Varun Jain <[email protected]> * Release Lock API Signed-off-by: Varun Jain <[email protected]> * Reformatting Signed-off-by: Varun Jain <[email protected]> * Reformatting Signed-off-by: Varun Jain <[email protected]> * Reformatting Signed-off-by: Varun Jain <[email protected]> * Addressing Dan's Comments Signed-off-by: Varun Jain <[email protected]> * Addressing Dan's Comments Signed-off-by: Varun Jain <[email protected]> * Addressing Sarat's Comments Signed-off-by: Varun Jain <[email protected]> * Addressing Sarat's Comments Signed-off-by: Varun Jain <[email protected]> * Fixing test cases Signed-off-by: Varun Jain <[email protected]> * Fixing test cases Signed-off-by: Varun Jain <[email protected]> * Fixing ReleaseLocktestcase Signed-off-by: Varun Jain <[email protected]> * Fixing ReleaseLocktestcase Signed-off-by: Varun Jain <[email protected]> --------- Signed-off-by: Varun Jain <[email protected]> * Bumping BWC version to 2.7 and Fixing xcontent imports (#322) * [Extensions] Add all fields of LockModel to RestGetLockAction response (#342) * Modifies the RestGetLockAction response to include all fields of the lock model object, rather than the lock model object itself Signed-off-by: Joshua Palis <[email protected]> * Removing unused fields Signed-off-by: Joshua Palis <[email protected]> * reverting field removal Signed-off-by: Joshua Palis <[email protected]> * Adding lockID to back to response Signed-off-by: Joshua Palis <[email protected]> * Adding checks to multinode get lock rest integration tests to ensure that all response fields are populated Signed-off-by: Joshua Palis <[email protected]> * fixing lock time parsing Signed-off-by: Joshua Palis <[email protected]> * Adds a new Response class to house serde logic for RestGetLockAction response Signed-off-by: Joshua Palis <[email protected]> * Moving parser.nextToken() within AcquireLockResponse.parse() Signed-off-by: Joshua Palis <[email protected]> * Implementing toXContentObject for AcquireLockRequest Signed-off-by: Joshua Palis <[email protected]> * Moving AcquireLockResponse to transpor package Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Joshua Palis <[email protected]> * Increasing JobDetailsService request timeoufrom 10 to 15, since the initial job detail registration request will initialize the job details metadata index, which takes some time (#346) Signed-off-by: Joshua Palis <[email protected]> * Fixes serde logic for proxy scheduled jobrunner/parser requests/responses (#349) Signed-off-by: Joshua Palis <[email protected]> * Fixes serde logic for JobParameterRequest/JobRunnerRequest. Removes extra trimming of request bytes for the streaminput constructor for the JobParameter/RunnerRequest, as this is already trimmed by the SDK prior to forwarding the actionrequest to the transport action (#351) Signed-off-by: Joshua Palis <[email protected]> * [Extensions] Makes JobRunner/Parameter/Request/Response classes extend from ActionRequest/Response (#352) * Replaces ExtensionActionRequest class name with the JobParameter/RunnerRequest fully qualified class names, modifes JobParameter/Runner/Request/Response classes to extend from ActionRequest/Response Signed-off-by: Joshua Palis <[email protected]> * Removes ExtensionJobActionResponse and fixes affected test classes Signed-off-by: Joshua Palis <[email protected]> * Fixing javadocs Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Joshua Palis <[email protected]> * Consuming breaking changes from moving ExtensionActionRequest (#381) Signed-off-by: Sarat Vemulapalli <[email protected]> * spotless Signed-off-by: Joshua Palis <[email protected]> * Fixing apache imports for TestHelpers class Signed-off-by: Joshua Palis <[email protected]> * Modofies ODFERestTestCase to work with 2.x Signed-off-by: Joshua Palis <[email protected]> * Fixing wipeAllODFEIndices Signed-off-by: Joshua Palis <[email protected]> --------- Signed-off-by: Varun Jain <[email protected]> Signed-off-by: Joshua Palis <[email protected]> Signed-off-by: Sarat Vemulapalli <[email protected]> Co-authored-by: Varun Jain <[email protected]> Co-authored-by: Sarat Vemulapalli <[email protected]>
opensearch-project · May 10, 2023 · 0201df6 · 0201df6
1 parent 7dcf7f2
commit 0201df6
Show file tree

Hide file tree

Showing 33 changed files with 3,621 additions and 19 deletions.
diff --git a/build.gradle b/build.gradle
@@ -152,6 +152,8 @@ repositories {
 
 dependencies {
     implementation project(path: ":${rootProject.name}-spi", configuration: 'shadow')
+    implementation group: 'com.google.guava', name: 'guava', version:'31.0.1-jre'
+    implementation group: 'com.google.guava', name: 'failureaccess', version:'1.0.1'
     javaRestTestImplementation project.sourceSets.main.runtimeClasspath
 }
 
@@ -252,7 +254,8 @@ task integTestRemote(type: RestIntegTestTask) {
     systemProperty "security", System.getProperty("security")
     systemProperty "user", System.getProperty("user")
     systemProperty "password", System.getProperty("password")
-
+    systemProperty 'tests.rest.cluster', 'localhost:9200'
+    systemProperty 'tests.clustername', 'opensearch-job-scheduler-cluster'
     if (System.getProperty("tests.rest.cluster") != null) {
         filter {
             includeTestsMatching "org.opensearch.jobscheduler.*RestIT"

diff --git a/spi/src/main/java/org/opensearch/jobscheduler/spi/JobDocVersion.java b/spi/src/main/java/org/opensearch/jobscheduler/spi/JobDocVersion.java
@@ -8,12 +8,17 @@
  */
 package org.opensearch.jobscheduler.spi;
 
+import java.io.IOException;
 import java.util.Locale;
 
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.io.stream.Writeable;
+
 /**
  * Structure to represent scheduled job document version. JobScheduler use this to determine this job
  */
-public class JobDocVersion implements Comparable<JobDocVersion> {
+public class JobDocVersion implements Comparable<JobDocVersion>, Writeable {
     private final long primaryTerm;
     private final long seqNo;
     private final long version;
@@ -24,6 +29,19 @@ public JobDocVersion(long primaryTerm, long seqNo, long version) {
         this.version = version;
     }
 
+    public JobDocVersion(StreamInput in) throws IOException {
+        this.primaryTerm = in.readLong();
+        this.seqNo = in.readLong();
+        this.version = in.readLong();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeLong(this.primaryTerm);
+        out.writeLong(this.seqNo);
+        out.writeLong(this.version);
+    }
+
     public long getPrimaryTerm() {
         return primaryTerm;
     }

diff --git a/spi/src/main/java/org/opensearch/jobscheduler/spi/JobExecutionContext.java b/spi/src/main/java/org/opensearch/jobscheduler/spi/JobExecutionContext.java
@@ -8,11 +8,15 @@
  */
 package org.opensearch.jobscheduler.spi;
 
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.io.stream.Writeable;
 import org.opensearch.jobscheduler.spi.utils.LockService;
 
+import java.io.IOException;
 import java.time.Instant;
 
-public class JobExecutionContext {
+public class JobExecutionContext implements Writeable {
     private final Instant expectedExecutionTime;
     private final JobDocVersion jobVersion;
     private final LockService lockService;
@@ -33,6 +37,22 @@ public JobExecutionContext(
         this.jobId = jobId;
     }
 
+    public JobExecutionContext(StreamInput in) throws IOException {
+        this.expectedExecutionTime = in.readInstant();
+        this.jobVersion = new JobDocVersion(in);
+        this.lockService = null;
+        this.jobIndexName = in.readString();
+        this.jobId = in.readString();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeInstant(this.expectedExecutionTime);
+        this.jobVersion.writeTo(out);
+        out.writeString(this.jobIndexName);
+        out.writeString(this.jobId);
+    }
+
     public Instant getExpectedExecutionTime() {
         return this.expectedExecutionTime;
     }
@@ -52,4 +72,5 @@ public String getJobIndexName() {
     public String getJobId() {
         return this.jobId;
     }
+
 }
diff --git a/spi/src/main/java/org/opensearch/jobscheduler/spi/LockModel.java b/spi/src/main/java/org/opensearch/jobscheduler/spi/LockModel.java
@@ -8,12 +8,14 @@
  */
 package org.opensearch.jobscheduler.spi;
 
-import org.opensearch.common.Strings;
+import org.opensearch.ExceptionsHelper;
+import org.opensearch.OpenSearchException;
+import org.opensearch.common.bytes.BytesReference;
 import org.opensearch.core.xcontent.ToXContentObject;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.common.xcontent.XContentParserUtils;
-import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.common.xcontent.json.JsonXContent;
 import org.opensearch.index.seqno.SequenceNumbers;
 
 import java.io.IOException;
@@ -30,6 +32,13 @@ public final class LockModel implements ToXContentObject {
     public static final String LOCK_DURATION = "lock_duration_seconds";
     public static final String RELEASED = "released";
 
+    // Rest Fields
+    public static final String GET_LOCK_ACTION = "get_lock_action";
+    public static final String SEQUENCE_NUMBER = "seq_no";
+    public static final String PRIMARY_TERM = "primary_term";
+    public static final String LOCK_ID = "lock_id";
+    public static final String LOCK_MODEL = "lock_model";
+
     private final String lockId;
     private final String jobIndexName;
     private final String jobId;
@@ -174,7 +183,23 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
 
     @Override
     public String toString() {
-        return Strings.toString(XContentType.JSON, this, false, true);
+        try {
+            XContentBuilder builder = JsonXContent.contentBuilder();
+            builder.humanReadable(true);
+            this.toXContent(builder, EMPTY_PARAMS);
+            return BytesReference.bytes(builder).utf8ToString();
+        } catch (IOException e) {
+            try {
+                XContentBuilder builder = JsonXContent.contentBuilder();
+                builder.startObject();
+                builder.field("error", "error building toString out of XContent: " + e.getMessage());
+                builder.field("stack_trace", ExceptionsHelper.stackTrace(e));
+                builder.endObject();
+                return BytesReference.bytes(builder).utf8ToString();
+            } catch (IOException e2) {
+                throw new OpenSearchException("cannot generate error message for deserialization", e);
+            }
+        }
     }
 
     public String getLockId() {

diff --git a/spi/src/main/java/org/opensearch/jobscheduler/spi/utils/LockService.java b/spi/src/main/java/org/opensearch/jobscheduler/spi/utils/LockService.java
@@ -243,7 +243,7 @@ private void createLock(final LockModel tempLock, ActionListener<LockModel> list
         }
     }
 
-    private void findLock(final String lockId, ActionListener<LockModel> listener) {
+    public void findLock(final String lockId, ActionListener<LockModel> listener) {
         GetRequest getRequest = new GetRequest(LOCK_INDEX_NAME).id(lockId);
         client.get(getRequest, ActionListener.wrap(response -> {
             if (!response.isExists()) {

diff --git a/src/main/java/org/opensearch/jobscheduler/JobSchedulerPlugin.java b/src/main/java/org/opensearch/jobscheduler/JobSchedulerPlugin.java
@@ -8,6 +8,15 @@
  */
 package org.opensearch.jobscheduler;
 
+import org.opensearch.cluster.node.DiscoveryNodes;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.IndexScopedSettings;
+import org.opensearch.common.settings.SettingsFilter;
+import org.opensearch.jobscheduler.rest.action.RestGetJobDetailsAction;
+import org.opensearch.jobscheduler.rest.action.RestGetLockAction;
+import org.opensearch.jobscheduler.rest.action.RestReleaseLockAction;
 import org.opensearch.jobscheduler.scheduler.JobScheduler;
 import org.opensearch.jobscheduler.spi.JobSchedulerExtension;
 import org.opensearch.jobscheduler.spi.ScheduledJobParser;
@@ -23,35 +32,39 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.core.ParseField;
 import org.opensearch.common.io.stream.NamedWriteableRegistry;
-import org.opensearch.common.settings.Setting;
-import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.env.Environment;
 import org.opensearch.env.NodeEnvironment;
 import org.opensearch.index.IndexModule;
+import org.opensearch.jobscheduler.utils.JobDetailsService;
+import org.opensearch.plugins.ActionPlugin;
 import org.opensearch.plugins.ExtensiblePlugin;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.rest.RestController;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ExecutorBuilder;
 import org.opensearch.threadpool.FixedExecutorBuilder;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.watcher.ResourceWatcherService;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.ArrayList;
 import java.util.function.Supplier;
 
-public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
+import com.google.common.collect.ImmutableList;
+
+public class JobSchedulerPlugin extends Plugin implements ActionPlugin, ExtensiblePlugin {
 
     public static final String OPEN_DISTRO_JOB_SCHEDULER_THREAD_POOL_NAME = "open_distro_job_scheduler";
+    public static final String JS_BASE_URI = "/_plugins/_job_scheduler";
 
     private static final Logger log = LogManager.getLogger(JobSchedulerPlugin.class);
 
@@ -61,6 +74,8 @@ public class JobSchedulerPlugin extends Plugin implements ExtensiblePlugin {
     private Map<String, ScheduledJobProvider> indexToJobProviders;
     private Set<String> indicesToListen;
 
+    private JobDetailsService jobDetailsService;
+
     public JobSchedulerPlugin() {
         this.indicesToListen = new HashSet<>();
         this.indexToJobProviders = new HashMap<>();
@@ -81,6 +96,7 @@ public Collection<Object> createComponents(
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
         this.lockService = new LockService(client, clusterService);
+        this.jobDetailsService = new JobDetailsService(client, clusterService, this.indicesToListen, this.indexToJobProviders);
         this.scheduler = new JobScheduler(threadPool, this.lockService);
         this.sweeper = initSweeper(
             environment.settings(),
@@ -89,7 +105,8 @@ public Collection<Object> createComponents(
             threadPool,
             xContentRegistry,
             this.scheduler,
-            this.lockService
+            this.lockService,
+            this.jobDetailsService
         );
         clusterService.addListener(this.sweeper);
         clusterService.addLifecycleListener(this.sweeper);
@@ -135,6 +152,10 @@ public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
 
     @Override
     public void onIndexModule(IndexModule indexModule) {
+        if (indexModule.getIndex().getName().equals(JobDetailsService.JOB_DETAILS_INDEX_NAME)) {
+            indexModule.addIndexOperationListener(this.jobDetailsService);
+            log.info("JobDetailsService started listening to operations on index {}", JobDetailsService.JOB_DETAILS_INDEX_NAME);
+        }
         if (this.indicesToListen.contains(indexModule.getIndex().getName())) {
             indexModule.addIndexOperationListener(this.sweeper);
             log.info("JobSweeper started listening to operations on index {}", indexModule.getIndex().getName());
@@ -181,8 +202,36 @@ private JobSweeper initSweeper(
         ThreadPool threadPool,
         NamedXContentRegistry registry,
         JobScheduler scheduler,
-        LockService lockService
+        LockService lockService,
+        JobDetailsService jobDetailsService
     ) {
-        return new JobSweeper(settings, client, clusterService, threadPool, registry, this.indexToJobProviders, scheduler, lockService);
+        return new JobSweeper(
+            settings,
+            client,
+            clusterService,
+            threadPool,
+            registry,
+            this.indexToJobProviders,
+            scheduler,
+            lockService,
+            jobDetailsService
+        );
     }
+
+    @Override
+    public List getRestHandlers(
+        Settings settings,
+        RestController restController,
+        ClusterSettings clusterSettings,
+        IndexScopedSettings indexScopedSettings,
+        SettingsFilter settingsFilter,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<DiscoveryNodes> nodesInCluster
+    ) {
+        RestGetJobDetailsAction restGetJobDetailsAction = new RestGetJobDetailsAction(jobDetailsService);
+        RestGetLockAction restGetLockAction = new RestGetLockAction(lockService);
+        RestReleaseLockAction restReleaseLockAction = new RestReleaseLockAction(lockService);
+        return ImmutableList.of(restGetJobDetailsAction, restGetLockAction, restReleaseLockAction);
+    }
+
 }