History Status API

sample-extension-plugin/src/test/java/org/opensearch/jobscheduler/sampleextension/SampleJobRunnerRestIT.java

@@ -36,6 +36,7 @@ public class SampleJobRunnerRestIT extends SampleExtensionIntegTestCase {
     public static final String LOCK_INFO_URI = "/_plugins/_job_scheduler/api/locks";
     public static final String SCHEDULER_INFO_URI = "/_plugins/_job_scheduler/api/jobs?by_node";
     public static final String SCHEDULER_INFO_URI_CLUSTER = "/_plugins/_job_scheduler/api/jobs";
+    public static final String HISTORY_INFO_URI = "/_plugins/_job_scheduler/api/history";
 
     public void testJobCreateWithCorrectParams() throws IOException {
         SampleJobParameter jobParameter = new SampleJobParameter();
@@ -211,6 +212,94 @@ public void testRunThenListJobs() throws Exception {
         deleteWatcherJob(jobId);
     }
 
+    public void testJobHistoryService() throws Exception {
+        String index = createTestIndex();
+        SampleJobParameter jobParameter = new SampleJobParameter();
+        jobParameter.setJobName("sample-job-lock-test-it");
+        jobParameter.setIndexToWatch(index);
+        // ensures that the next job tries to run even before the previous job finished & released its lock. Also look at
+        // SampleJobRunner.runTaskForLockIntegrationTests
+        jobParameter.setSchedule(new IntervalSchedule(Instant.now(), 5, ChronoUnit.SECONDS));
+        jobParameter.setLockDurationSeconds(10L);
+
+        // Creates a new watcher job.
+        String jobId = OpenSearchRestTestCase.randomAlphaOfLength(10);
+        createWatcherJob(jobId, jobParameter);
+
+        waitUntilLockIsAcquiredAndReleased(jobId);
+
+        Response response = makeRequest(client(), "GET", HISTORY_INFO_URI, Map.of(), null);
+        Map<String, Object> responseJson = parseResponse(response);
+
+        Assert.assertTrue("Response should contain total_history", responseJson.containsKey("total_history"));
+        Assert.assertTrue("Response should contain history", responseJson.containsKey("history"));
+
+        Integer totalHistory = (Integer) responseJson.get("total_history");
+        Assert.assertTrue("Total history should be greater than 0", totalHistory > 0);
+
+        Map<String, Object> history = (Map<String, Object>) responseJson.get("history");
+        Assert.assertFalse("History should not be empty", history.isEmpty());
+
+        for (Map.Entry<String, Object> entry : history.entrySet()) {
+            Map<String, Object> historyRecord = (Map<String, Object>) entry.getValue();
+            Assert.assertTrue("History record should contain job_index_name", historyRecord.containsKey("job_index_name"));
+            Assert.assertTrue("History record should contain job_id", historyRecord.containsKey("job_id"));
+            Assert.assertTrue("History record should contain start_time", historyRecord.containsKey("start_time"));
+            Assert.assertTrue("History record should contain completion_status", historyRecord.containsKey("completion_status"));
+            Assert.assertTrue("History record should contain end_time", historyRecord.containsKey("end_time"));
+
+            Assert.assertEquals("Job ID should match", jobId, historyRecord.get("job_id"));
+            Assert.assertEquals("Completion status should be 0", 0, historyRecord.get("completion_status"));
+        }
+
+        deleteWatcherJob(jobId);
+    }
+
+    public void testJobHistoryServiceById() throws Exception {
+        String index = createTestIndex();
+        SampleJobParameter jobParameter = new SampleJobParameter();
+        jobParameter.setJobName("sample-job-lock-test-it");
+        jobParameter.setIndexToWatch(index);
+        // ensures that the next job tries to run even before the previous job finished & released its lock. Also look at
+        // SampleJobRunner.runTaskForLockIntegrationTests
+        jobParameter.setSchedule(new IntervalSchedule(Instant.now(), 5, ChronoUnit.SECONDS));
+        jobParameter.setLockDurationSeconds(10L);
+
+        // Creates a new watcher job.
+        String jobId = OpenSearchRestTestCase.randomAlphaOfLength(10);
+        createWatcherJob(jobId, jobParameter);
+
+        waitUntilLockIsAcquiredAndReleased(jobId);
+
+        String historyIndexById = HISTORY_INFO_URI + "/.scheduler_sample_extension-" + jobId;
+
+        Response response = makeRequest(client(), "GET", historyIndexById, Map.of(), null);
+        Map<String, Object> responseJson = parseResponse(response);
+
+        Assert.assertTrue("Response should contain total_history", responseJson.containsKey("total_history"));
+        Assert.assertTrue("Response should contain history", responseJson.containsKey("history"));
+
+        Integer totalHistory = (Integer) responseJson.get("total_history");
+        Assert.assertTrue("Total history should be greater than 0", totalHistory > 0);
+
+        Map<String, Object> history = (Map<String, Object>) responseJson.get("history");
+        Assert.assertFalse("History should not be empty", history.isEmpty());
+
+        for (Map.Entry<String, Object> entry : history.entrySet()) {
+            Map<String, Object> historyRecord = (Map<String, Object>) entry.getValue();
+            Assert.assertTrue("History record should contain job_index_name", historyRecord.containsKey("job_index_name"));
+            Assert.assertTrue("History record should contain job_id", historyRecord.containsKey("job_id"));
+            Assert.assertTrue("History record should contain start_time", historyRecord.containsKey("start_time"));
+            Assert.assertTrue("History record should contain completion_status", historyRecord.containsKey("completion_status"));
+            Assert.assertTrue("History record should contain end_time", historyRecord.containsKey("end_time"));
+
+            Assert.assertEquals("Job ID should match", jobId, historyRecord.get("job_id"));
+            Assert.assertEquals("Completion status should be 0", 0, historyRecord.get("completion_status"));
+        }
+
+        deleteWatcherJob(jobId);
+    }
+
     public void testAcquiredLockPreventExecOfTasks() throws Exception {
         String index = createTestIndex();
         SampleJobParameter jobParameter = new SampleJobParameter();

src/main/java/org/opensearch/jobscheduler/JobSchedulerPlugin.java

@@ -23,12 +23,15 @@
 import org.opensearch.jobscheduler.rest.action.RestGetLockAction;
 import org.opensearch.jobscheduler.rest.action.RestGetScheduledInfoAction;
 import org.opensearch.jobscheduler.rest.action.RestReleaseLockAction;
+import org.opensearch.jobscheduler.rest.action.RestGetHistoryAction;
 import org.opensearch.jobscheduler.spi.utils.LockService;
 import org.opensearch.jobscheduler.transport.PluginClient;
 import org.opensearch.jobscheduler.transport.action.GetAllLocksAction;
 import org.opensearch.jobscheduler.transport.action.GetScheduledInfoAction;
 import org.opensearch.jobscheduler.transport.action.TransportGetAllLocksAction;
 import org.opensearch.jobscheduler.transport.action.TransportGetScheduledInfoAction;
+import org.opensearch.jobscheduler.transport.action.TransportGetHistoryAction;
+import org.opensearch.jobscheduler.transport.action.GetHistoryAction;
 import org.opensearch.jobscheduler.scheduler.JobScheduler;
 import org.opensearch.jobscheduler.spi.JobSchedulerExtension;
 import org.opensearch.jobscheduler.spi.ScheduledJobParser;
@@ -89,6 +92,7 @@ public class JobSchedulerPlugin extends Plugin implements ActionPlugin, Extensib
     private PluginClient pluginClient;
 
     private JobDetailsService jobDetailsService;
+    private ClusterService clusterService;
 
     public JobSchedulerPlugin() {
         this.indicesToListen = new HashSet<>();
@@ -112,7 +116,7 @@ public Collection<Module> createGuiceModules() {
     public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings) {
         return List.of(
             new SystemIndexDescriptor(LockServiceImpl.LOCK_INDEX_NAME, "Stores lock documents used for plugin job execution"),
-            new SystemIndexDescriptor(JobHistoryService.JOB_HISTORY_INDEX_NAME, "Stores history documents used for plugin job execution")
+            new SystemIndexDescriptor(JobHistoryService.JOB_HISTORY_INDEX_NAME, "Stores job execution history")
         );
     }
 
@@ -130,7 +134,7 @@ public Collection<Object> createComponents(
         IndexNameExpressionResolver indexNameExpressionResolver,
         Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
-        Supplier<Boolean> statusHistoryEnabled = () -> JobSchedulerSettings.STATUS_HISTORY.get(environment.settings());
+        Supplier<Boolean> statusHistoryEnabled = () -> clusterService.getClusterSettings().get(JobSchedulerSettings.STATUS_HISTORY);
         this.pluginClient = new PluginClient(client);
         this.historyService = new JobHistoryService(pluginClient, clusterService);
         this.lockService = new LockServiceImpl(pluginClient, clusterService, historyService, statusHistoryEnabled);
@@ -272,20 +276,24 @@ public List getRestHandlers(
         RestReleaseLockAction restReleaseLockAction = new RestReleaseLockAction(lockService);
         RestGetScheduledInfoAction restGetScheduledInfoAction = new RestGetScheduledInfoAction();
         RestGetLocksAction restGetAllLocksAction = new RestGetLocksAction();
+        RestGetHistoryAction restGetHistoryAction = new RestGetHistoryAction();
         return List.of(
             restGetJobDetailsAction,
             restGetLockAction,
             restReleaseLockAction,
             restGetScheduledInfoAction,
-            restGetAllLocksAction
+            restGetAllLocksAction,
+            restGetHistoryAction
         );
     }
 
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
-        List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>(2);
+        List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> actions = new ArrayList<>(3);
         actions.add(new ActionHandler<>(GetScheduledInfoAction.INSTANCE, TransportGetScheduledInfoAction.class));
         actions.add(new ActionHandler<>(GetAllLocksAction.INSTANCE, TransportGetAllLocksAction.class));
+        actions.add(new ActionHandler<>(GetHistoryAction.INSTANCE, TransportGetHistoryAction.class));
+
         return actions;
     }
 

src/main/java/org/opensearch/jobscheduler/rest/action/RestGetHistoryAction.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.jobscheduler.rest.action;
+
+import org.opensearch.transport.client.node.NodeClient;
+import org.opensearch.jobscheduler.JobSchedulerPlugin;
+import org.opensearch.jobscheduler.transport.action.GetHistoryAction;
+import org.opensearch.jobscheduler.transport.request.GetHistoryRequest;
+import org.opensearch.rest.BaseRestHandler;
+import org.opensearch.rest.RestRequest;
+import org.opensearch.rest.action.RestToXContentListener;
+
+import java.util.List;
+
+import static org.opensearch.rest.RestRequest.Method.GET;
+
+/**
+ * REST handler for getting job history
+ */
+public class RestGetHistoryAction extends BaseRestHandler {
+
+    @Override
+    public String getName() {
+        return "get_history_action";
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(
+            new Route(GET, JobSchedulerPlugin.JS_BASE_URI + "/api/history"),
+            new Route(GET, JobSchedulerPlugin.JS_BASE_URI + "/api/history/{history_id}")
+        );
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) {
+        String history_id = request.param("history_id");
+        GetHistoryRequest getHistoryRequest = new GetHistoryRequest(history_id);
+        return channel -> client.execute(GetHistoryAction.INSTANCE, getHistoryRequest, new RestToXContentListener<>(channel));
+    }
+}

src/main/java/org/opensearch/jobscheduler/transport/action/GetHistoryAction.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.jobscheduler.transport.action;
+
+import org.opensearch.action.ActionType;
+import org.opensearch.jobscheduler.transport.response.GetHistoryResponse;
+
+public class GetHistoryAction extends ActionType<GetHistoryResponse> {
+    public static final String NAME = "cluster:admin/opensearch/jobscheduler/history";
+    public static final GetHistoryAction INSTANCE = new GetHistoryAction();
+
+    private GetHistoryAction() {
+        super(NAME, GetHistoryResponse::new);
+    }
+}