Add scheduler status filter to jobs endpoint (elastic/elasticsearch#350)

* Add scheduler status filter to jobs endpoint

* For scheduled jobs set the initial scheduler state

* Add status filter to job endpoint


Original commit: elastic/x-pack-elasticsearch@c7ed1627e2

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/action/GetJobAction.java

@@ -38,7 +38,9 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.prelert.job.DataCounts;
 import org.elasticsearch.xpack.prelert.job.Job;
+import org.elasticsearch.xpack.prelert.job.JobStatus;
 import org.elasticsearch.xpack.prelert.job.ModelSizeStats;
+import org.elasticsearch.xpack.prelert.job.SchedulerState;
 import org.elasticsearch.xpack.prelert.job.manager.AutodetectProcessManager;
 import org.elasticsearch.xpack.prelert.job.manager.JobManager;
 import org.elasticsearch.xpack.prelert.job.persistence.ElasticsearchJobProvider;
@@ -92,10 +94,11 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
         private boolean config;
         private boolean dataCounts;
         private boolean modelSizeStats;
+        private boolean schedulerStatus;
+        private boolean status;
         private PageParams pageParams = null;
 
         public Request() {
-
         }
 
         public void setJobId(String jobId) {
@@ -115,9 +118,11 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
         }
 
         public Request all() {
-            this.config = true;
+            config = true;
-            this.dataCounts = true;
+            dataCounts = true;
-            this.modelSizeStats = true;
+            modelSizeStats = true;
+            schedulerStatus = true;
+            status = true;
             return this;
         }
 
@@ -148,17 +153,36 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
             return this;
         }
 
+        public boolean schedulerStatus() {
+            return schedulerStatus;
+        }
+
+        public Request schedulerStatus(boolean schedulerStatus) {
+            this.schedulerStatus = schedulerStatus;
+            return this;
+        }
+
         public void setStats(Set<String> stats) {
             if (stats.contains("_all")) {
                 all();
-            }
+            } else {
-            else {
                 config(stats.contains("config"));
                 dataCounts(stats.contains("data_counts"));
                 modelSizeStats(stats.contains("model_size_stats"));
+                schedulerStatus(stats.contains("scheduler_state"));
+                status(stats.contains("status"));
             }
         }
 
+        public boolean status() {
+            return status;
+        }
+
+        public Request status(boolean status) {
+            this.status = status;
+            return this;
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             return null;
@@ -171,6 +195,8 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
             config = in.readBoolean();
             dataCounts = in.readBoolean();
             modelSizeStats = in.readBoolean();
+            schedulerStatus = in.readBoolean();
+            status = in.readBoolean();
             pageParams = in.readOptionalWriteable(PageParams::new);
         }
 
@@ -181,12 +207,14 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
             out.writeBoolean(config);
             out.writeBoolean(dataCounts);
             out.writeBoolean(modelSizeStats);
+            out.writeBoolean(schedulerStatus);
+            out.writeBoolean(status);
             out.writeOptionalWriteable(pageParams);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(jobId, config, dataCounts, modelSizeStats, pageParams);
+            return Objects.hash(jobId, config, dataCounts, modelSizeStats, schedulerStatus, status, pageParams);
         }
 
         @Override
@@ -202,6 +230,8 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                     && this.config == other.config
                     && this.dataCounts == other.dataCounts
                     && this.modelSizeStats == other.modelSizeStats
+                    && this.schedulerStatus == other.schedulerStatus
+                    && this.status == other.status
                     && Objects.equals(this.pageParams, other.pageParams);
         }
     }
@@ -222,17 +252,28 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
             private DataCounts dataCounts;
             @Nullable
             private ModelSizeStats modelSizeStats;
+            @Nullable
+            private SchedulerState schedulerState;
+            @Nullable
+            private JobStatus status;
 
-            JobInfo(@Nullable Job job, @Nullable DataCounts dataCounts, @Nullable ModelSizeStats modelSizeStats) {
+
+
+            JobInfo(@Nullable Job job, @Nullable DataCounts dataCounts, @Nullable ModelSizeStats modelSizeStats,
+                    @Nullable SchedulerState schedulerStatus, @Nullable JobStatus status) {
                 this.jobConfig = job;
                 this.dataCounts = dataCounts;
                 this.modelSizeStats = modelSizeStats;
+                this.schedulerState = schedulerStatus;
+                this.status = status;
             }
 
             JobInfo(StreamInput in) throws IOException {
                 jobConfig = in.readOptionalWriteable(Job::new);
                 dataCounts = in.readOptionalWriteable(DataCounts::new);
                 modelSizeStats = in.readOptionalWriteable(ModelSizeStats::new);
+                schedulerState = in.readOptionalWriteable(SchedulerState::new);
+                status = in.readOptionalWriteable(JobStatus::fromStream);
             }
 
             @Override
@@ -247,6 +288,12 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                 if (modelSizeStats != null) {
                     builder.field("model_size_stats", modelSizeStats);
                 }
+                if (schedulerState != null) {
+                    builder.field("scheduler_state", schedulerState);
+                }
+                if (status != null) {
+                    builder.field("status", status);
+                }
                 builder.endObject();
 
                 return builder;
@@ -257,11 +304,13 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                 out.writeOptionalWriteable(jobConfig);
                 out.writeOptionalWriteable(dataCounts);
                 out.writeOptionalWriteable(modelSizeStats);
+                out.writeOptionalWriteable(schedulerState);
+                out.writeOptionalWriteable(status);
             }
 
             @Override
             public int hashCode() {
-                return Objects.hash(jobConfig, dataCounts, modelSizeStats);
+                return Objects.hash(jobConfig, dataCounts, modelSizeStats, schedulerState, status);
             }
 
             @Override
@@ -275,7 +324,9 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                 JobInfo other = (JobInfo) obj;
                 return Objects.equals(jobConfig, other.jobConfig)
                         && Objects.equals(this.dataCounts, other.dataCounts)
-                        && Objects.equals(this.modelSizeStats, other.modelSizeStats);
+                        && Objects.equals(this.modelSizeStats, other.modelSizeStats)
+                        && Objects.equals(this.schedulerState, other.schedulerState)
+                        && Objects.equals(this.status, other.status);
             }
         }
 
@@ -400,8 +451,10 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                 Job jobConfig = request.config() ? jobs.hits().get(0) : null;
                 DataCounts dataCounts = readDataCounts(request.dataCounts(), request.getJobId());
                 ModelSizeStats modelSizeStats = readModelSizeStats(request.modelSizeStats(), request.getJobId());
+                SchedulerState schedulerStatus = readSchedulerState(request.schedulerStatus(), request.getJobId());
+                JobStatus jobStatus = readJobStatus(request.status(), request.getJobId());
 
-                Response.JobInfo jobInfo = new Response.JobInfo(jobConfig, dataCounts, modelSizeStats);
+                Response.JobInfo jobInfo = new Response.JobInfo(jobConfig, dataCounts, modelSizeStats, schedulerStatus, jobStatus);
                 response = new QueryPage<>(Collections.singletonList(jobInfo), 1);
 
             } else {
@@ -412,7 +465,9 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
                     Job jobConfig = request.config() ? job : null;
                     DataCounts dataCounts = readDataCounts(request.dataCounts(), job.getJobId());
                     ModelSizeStats modelSizeStats = readModelSizeStats(request.modelSizeStats(), job.getJobId());
-                    Response.JobInfo jobInfo = new Response.JobInfo(jobConfig, dataCounts, modelSizeStats);
+                    SchedulerState schedulerStatus = readSchedulerState(request.schedulerStatus(), job.getJobId());
+                    JobStatus jobStatus = readJobStatus(request.status(), job.getJobId());
+                    Response.JobInfo jobInfo = new Response.JobInfo(jobConfig, dataCounts, modelSizeStats, schedulerStatus, jobStatus);
                     jobInfoList.add(jobInfo);
                 }
                 response = new QueryPage<>(jobInfoList, jobsPage.hitCount());
@@ -441,6 +496,14 @@ public class GetJobAction extends Action<GetJobAction.Request, GetJobAction.Resp
             }
             return null;
         }
+
+        private SchedulerState readSchedulerState(boolean schedulerState, String jobId) {
+            return schedulerState ? jobManager.getSchedulerState(jobId).orElse(null) : null;
+        }
+
+        private JobStatus readJobStatus(boolean status, String jobId) {
+            return status ? jobManager.getJobStatus(jobId) : null;
+        }
     }
 
 }

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/SchedulerState.java

@@ -35,16 +35,17 @@ public class SchedulerState extends ToXContentToBytes implements Writeable {
     static {
         PARSER.declareField(ConstructingObjectParser.constructorArg(), p -> JobSchedulerStatus.fromString(p.text()), STATUS,
                 ValueType.STRING);
-        PARSER.declareLong(ConstructingObjectParser.constructorArg(), START_TIME_MILLIS);
+        PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), START_TIME_MILLIS);
         PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), END_TIME_MILLIS);
     }
 
     private JobSchedulerStatus status;
-    private long startTimeMillis;
+    @Nullable
+    private Long startTimeMillis;
     @Nullable
     private Long endTimeMillis;
 
-    public SchedulerState(JobSchedulerStatus status, long startTimeMillis, Long endTimeMillis) {
+    public SchedulerState(JobSchedulerStatus status, Long startTimeMillis, Long endTimeMillis) {
         this.status = status;
         this.startTimeMillis = startTimeMillis;
         this.endTimeMillis = endTimeMillis;
@@ -52,7 +53,7 @@ public class SchedulerState extends ToXContentToBytes implements Writeable {
 
     public SchedulerState(StreamInput in) throws IOException {
         status = JobSchedulerStatus.fromStream(in);
-        startTimeMillis = in.readLong();
+        startTimeMillis = in.readOptionalLong();
         endTimeMillis = in.readOptionalLong();
     }
 
@@ -60,7 +61,7 @@ public class SchedulerState extends ToXContentToBytes implements Writeable {
         return status;
     }
 
-    public long getStartTimeMillis() {
+    public Long getStartTimeMillis() {
         return startTimeMillis;
     }
 
@@ -98,7 +99,7 @@ public class SchedulerState extends ToXContentToBytes implements Writeable {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         status.writeTo(out);
-        out.writeLong(startTimeMillis);
+        out.writeOptionalLong(startTimeMillis);
         out.writeOptionalLong(endTimeMillis);
     }
 
@@ -106,7 +107,9 @@ public class SchedulerState extends ToXContentToBytes implements Writeable {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(STATUS.getPreferredName(), status.name().toUpperCase(Locale.ROOT));
-        builder.field(START_TIME_MILLIS.getPreferredName(), startTimeMillis);
+        if (startTimeMillis != null) {
+            builder.field(START_TIME_MILLIS.getPreferredName(), startTimeMillis);
+        }
         if (endTimeMillis != null) {
             builder.field(END_TIME_MILLIS.getPreferredName(), endTimeMillis);
         }

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/manager/JobManager.java

@@ -340,6 +340,16 @@ public class JobManager {
         }
     }
 
+    public Optional<SchedulerState> getSchedulerState(String jobId) {
+        Job job = getJobOrThrowIfUnknown(clusterService.state(), jobId);
+        if (job.getSchedulerConfig() == null) {
+            return Optional.empty();
+        }
+
+        Allocation allocation = getAllocation(clusterService.state(), jobId);
+        return Optional.ofNullable(allocation.getSchedulerState());
+    }
+
     public void updateSchedulerStatus(String jobId, JobSchedulerStatus newStatus) {
         clusterService.submitStateUpdateTask("update-scheduler-status-job-" + jobId, new ClusterStateUpdateTask() {
 
@@ -506,6 +516,10 @@ public class JobManager {
         });
     }
 
+    public JobStatus getJobStatus(String jobId) {
+        return getJobAllocation(jobId).getStatus();
+    }
+
     public void setJobStatus(String jobId, JobStatus newStatus) {
         clusterService.submitStateUpdateTask("set-paused-status-job-" + jobId, new ClusterStateUpdateTask() {
 

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/metadata/Allocation.java

@@ -186,7 +186,8 @@ public class Allocation extends AbstractDiffable<Allocation> implements ToXConte
                 }
                 break;
             case STOPPED:
-                if (currentSchedulerStatus != JobSchedulerStatus.STOPPING) {
+                if ((currentSchedulerStatus != JobSchedulerStatus.STOPPED ||
+                     currentSchedulerStatus != JobSchedulerStatus.STOPPING) == false) {
                     String msg = Messages.getMessage(Messages.JOB_SCHEDULER_CANNOT_STOP_IN_CURRENT_STATE, jobId, newSchedulerStatus);
                     throw ExceptionsHelper.conflictStatusException(msg);
                 }

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/metadata/JobAllocator.java

@@ -50,7 +50,13 @@ public class JobAllocator extends AbstractComponent implements ClusterStateListe
 
         for (String jobId : prelertMetadata.getJobs().keySet()) {
             if (prelertMetadata.getAllocations().containsKey(jobId) == false) {
-                builder.putAllocation(prelertNode.getId(), jobId);
+                boolean addSchedulderState = prelertMetadata.getJobs().get(jobId).getSchedulerConfig() != null;
+                if (addSchedulderState) {
+                    builder.putAllocationWithScheduler(prelertNode.getId(), jobId);
+                }
+                else {
+                    builder.putAllocation(prelertNode.getId(), jobId);
+                }
             }
         }
 

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/job/metadata/PrelertMetadata.java

@@ -18,6 +18,8 @@ import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.xpack.prelert.job.Job;
+import org.elasticsearch.xpack.prelert.job.JobSchedulerStatus;
+import org.elasticsearch.xpack.prelert.job.SchedulerState;
 import org.elasticsearch.xpack.prelert.utils.ExceptionsHelper;
 
 import java.io.IOException;
@@ -219,6 +221,15 @@ public class PrelertMetadata implements MetaData.Custom {
             return this;
         }
 
+        public Builder putAllocationWithScheduler(String nodeId, String jobId) {
+            Allocation.Builder builder = new Allocation.Builder();
+            builder.setJobId(jobId);
+            builder.setNodeId(nodeId);
+            builder.setSchedulerState(new SchedulerState(JobSchedulerStatus.STOPPED, null, null));
+            this.allocations.put(jobId, builder.build());
+            return this;
+        }
+
         public Builder updateAllocation(String jobId, Allocation updated) {
             Allocation previous = this.allocations.put(jobId, updated);
             if (previous == null) {

elasticsearch/src/main/java/org/elasticsearch/xpack/prelert/rest/job/RestGetJobAction.java

@@ -51,7 +51,6 @@ public class RestGetJobAction extends BaseRestHandler {
 
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
-
         final GetJobAction.Request request;
         if (RestActions.hasBodyContent(restRequest)) {
             BytesReference bodyBytes = RestActions.getRestContent(restRequest);
@@ -66,8 +65,7 @@ public class RestGetJobAction extends BaseRestHandler {
             request.setPageParams(new PageParams(restRequest.paramAsInt(PageParams.FROM.getPreferredName(), DEFAULT_FROM),
                     restRequest.paramAsInt(PageParams.SIZE.getPreferredName(), DEFAULT_SIZE)));
         }
+
         return channel -> transportGetJobAction.execute(request, new RestStatusToXContentListener<>(channel));
-
-
     }
 }

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/action/GetJobActionRequestTests.java

@@ -17,6 +17,8 @@ public class GetJobActionRequestTests extends AbstractStreamableTestCase<GetJobA
         instance.config(randomBoolean());
         instance.dataCounts(randomBoolean());
         instance.modelSizeStats(randomBoolean());
+        instance.schedulerStatus(randomBoolean());
+        instance.status(randomBoolean());
         if (randomBoolean()) {
             int from = randomInt(PageParams.MAX_FROM_SIZE_SUM);
             int maxSize = PageParams.MAX_FROM_SIZE_SUM - from;

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/action/GetJobActionResponseTests.java

@@ -13,10 +13,13 @@ import org.elasticsearch.xpack.prelert.job.DataDescription;
 import org.elasticsearch.xpack.prelert.job.Detector;
 import org.elasticsearch.xpack.prelert.job.IgnoreDowntime;
 import org.elasticsearch.xpack.prelert.job.Job;
+import org.elasticsearch.xpack.prelert.job.JobSchedulerStatus;
+import org.elasticsearch.xpack.prelert.job.JobStatus;
 import org.elasticsearch.xpack.prelert.job.ModelDebugConfig;
 import org.elasticsearch.xpack.prelert.job.ModelSizeStats;
 import org.elasticsearch.xpack.prelert.job.SchedulerConfig;
 import org.elasticsearch.xpack.prelert.job.persistence.QueryPage;
+import org.elasticsearch.xpack.prelert.job.SchedulerState;
 import org.elasticsearch.xpack.prelert.job.transform.TransformConfig;
 import org.elasticsearch.xpack.prelert.job.transform.TransformType;
 import org.elasticsearch.xpack.prelert.support.AbstractStreamableTestCase;
@@ -25,6 +28,7 @@ import org.joda.time.DateTime;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
 
@@ -70,18 +74,30 @@ public class GetJobActionResponseTests extends AbstractStreamableTestCase<GetJob
 
 
             DataCounts dataCounts = null;
-            ModelSizeStats sizeStats = null;
-
             if (randomBoolean()) {
                 dataCounts = new DataCounts(randomAsciiOfLength(10), randomIntBetween(1, 1_000_000),
                         randomIntBetween(1, 1_000_000), randomIntBetween(1, 1_000_000), randomIntBetween(1, 1_000_000),
                         randomIntBetween(1, 1_000_000), randomIntBetween(1, 1_000_000), randomIntBetween(1, 1_000_000),
                         new DateTime(randomDateTimeZone()).toDate(), new DateTime(randomDateTimeZone()).toDate());
             }
+
+            ModelSizeStats sizeStats = null;
             if (randomBoolean()) {
                 sizeStats = new ModelSizeStats.Builder("foo").build();
             }
-            Response.JobInfo jobInfo = new Response.JobInfo(job, dataCounts, sizeStats);
+
+            SchedulerState schedulerState = null;
+            if (randomBoolean()) {
+                schedulerState = new SchedulerState(randomFrom(EnumSet.allOf(JobSchedulerStatus.class)), randomPositiveLong(),
+                        randomPositiveLong());
+            }
+
+            JobStatus jobStatus = null;
+            if (randomBoolean()) {
+                jobStatus = randomFrom(EnumSet.allOf(JobStatus.class));
+            }
+
+            Response.JobInfo jobInfo = new Response.JobInfo(job, dataCounts, sizeStats, schedulerState, jobStatus);
             jobInfoList.add(jobInfo);
         }
 

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/action/ScheduledJobsIT.java

@@ -93,7 +93,7 @@ public class ScheduledJobsIT extends ESIntegTestCase {
         PutJobAction.Response putJobResponse = client().execute(PutJobAction.INSTANCE, putJobRequest).get();
         assertTrue(putJobResponse.isAcknowledged());
 
-        SchedulerState schedulerState = new SchedulerState(JobSchedulerStatus.STARTING, 0, now);
+        SchedulerState schedulerState = new SchedulerState(JobSchedulerStatus.STARTING, 0L, now);
         StartJobSchedulerAction.Request startSchedulerRequest = new StartJobSchedulerAction.Request("_job_id", schedulerState);
         StartJobSchedulerAction.Response startJobResponse = client().execute(StartJobSchedulerAction.INSTANCE, startSchedulerRequest)
                 .get();
@@ -123,7 +123,7 @@ public class ScheduledJobsIT extends ESIntegTestCase {
         PutJobAction.Response putJobResponse = client().execute(PutJobAction.INSTANCE, putJobRequest).get();
         assertTrue(putJobResponse.isAcknowledged());
 
-        SchedulerState schedulerState = new SchedulerState(JobSchedulerStatus.STARTING, 0, null);
+        SchedulerState schedulerState = new SchedulerState(JobSchedulerStatus.STARTING, 0L, null);
         StartJobSchedulerAction.Request startSchedulerRequest = new StartJobSchedulerAction.Request("_job_id", schedulerState);
         StartJobSchedulerAction.Response startJobResponse = client().execute(StartJobSchedulerAction.INSTANCE, startSchedulerRequest)
                 .get();

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/SchedulerStateTests.java

@@ -30,7 +30,7 @@ public class SchedulerStateTests extends AbstractSerializingTestCase<SchedulerSt
 
     public void testEquals_GivenDifferentClass() {
 
-        assertFalse(new SchedulerState(JobSchedulerStatus.STARTED, 0, null).equals("a string"));
+        assertFalse(new SchedulerState(JobSchedulerStatus.STARTED, 0L, null).equals("a string"));
     }
 
     public void testEquals_GivenSameReference() {
@@ -58,7 +58,7 @@ public class SchedulerStateTests extends AbstractSerializingTestCase<SchedulerSt
     }
 
     public void testEquals_GivenDifferentStartTimeMillis() {
-        SchedulerState schedulerState1 = new SchedulerState(JobSchedulerStatus.STARTED, 18L, 42L);
+        SchedulerState schedulerState1 = new SchedulerState(JobSchedulerStatus.STARTED, null, 42L);
         SchedulerState schedulerState2 = new SchedulerState(JobSchedulerStatus.STOPPED, 19L, schedulerState1.getEndTimeMillis());
 
         assertFalse(schedulerState1.equals(schedulerState2));

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/metadata/JobAllocatorTests.java

@@ -17,8 +17,13 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.LocalTransportAddress;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xpack.prelert.job.DataDescription;
+import org.elasticsearch.xpack.prelert.job.Job;
+import org.elasticsearch.xpack.prelert.job.JobSchedulerStatus;
+import org.elasticsearch.xpack.prelert.job.SchedulerConfig;
 import org.junit.Before;
 
+import java.util.Collections;
 import java.util.concurrent.ExecutorService;
 
 import static org.mockito.Mockito.doAnswer;
@@ -176,4 +181,34 @@ public class JobAllocatorTests extends ESTestCase {
         verify(clusterService, times(1)).submitStateUpdateTask(any(), any());
     }
 
+    public void testScheduledJobHasDefaultSchedulerState() {
+        PrelertMetadata.Builder pmBuilder = new PrelertMetadata.Builder();
+
+        SchedulerConfig.Builder schedulerConfigBuilder = new SchedulerConfig.Builder(SchedulerConfig.DataSource.ELASTICSEARCH);
+        schedulerConfigBuilder.setBaseUrl("http://server");
+        schedulerConfigBuilder.setIndexes(Collections.singletonList("foo"));
+        schedulerConfigBuilder.setTypes(Collections.singletonList("bar"));
+
+        Job.Builder jobBuilder = buildJobBuilder("_job_id");
+        jobBuilder.setSchedulerConfig(schedulerConfigBuilder);
+        DataDescription.Builder dataDescriptionBuilder = new DataDescription.Builder();
+        dataDescriptionBuilder.setFormat(DataDescription.DataFormat.ELASTICSEARCH);
+        jobBuilder.setDataDescription(dataDescriptionBuilder);
+
+        pmBuilder.putJob(jobBuilder.build(), false);
+
+        ClusterState cs = ClusterState.builder(new ClusterName("_cluster_name")).metaData(MetaData.builder()
+                .putCustom(PrelertMetadata.TYPE, pmBuilder.build()))
+                .nodes(DiscoveryNodes.builder()
+                        .add(new DiscoveryNode("_id", new LocalTransportAddress("_id"), Version.CURRENT))
+                        .masterNodeId("_id")
+                        .localNodeId("_id"))
+                .build();
+
+
+        ClusterState clusterStateWithAllocation = jobAllocator.allocateJobs(cs);
+        PrelertMetadata metadata = clusterStateWithAllocation.metaData().custom(PrelertMetadata.TYPE);
+        assertEquals(JobSchedulerStatus.STOPPED, metadata.getAllocations().get("_job_id").getSchedulerState().getStatus());
+    }
+
 }

elasticsearch/src/test/java/org/elasticsearch/xpack/prelert/job/scheduler/ScheduledJobServiceTests.java

@@ -147,7 +147,7 @@ public class ScheduledJobServiceTests extends ESTestCase {
     public void testStop_GivenStartedScheduledJob() throws IOException {
         Job.Builder builder = createScheduledJob();
         Allocation allocation1 =
-                new Allocation("_nodeId", "foo", JobStatus.RUNNING, new SchedulerState(JobSchedulerStatus.STARTED, 0, null));
+                new Allocation("_nodeId", "foo", JobStatus.RUNNING, new SchedulerState(JobSchedulerStatus.STARTED, 0L, null));
         when(jobManager.getJobAllocation("foo")).thenReturn(allocation1);
 
         DataExtractor dataExtractor = mock(DataExtractor.class);
@@ -161,7 +161,7 @@ public class ScheduledJobServiceTests extends ESTestCase {
 
         // Properly stop it to avoid leaking threads in the test
         Allocation allocation2 =
-                new Allocation("_nodeId", "foo", JobStatus.RUNNING, new SchedulerState(JobSchedulerStatus.STOPPING, 0, null));
+                new Allocation("_nodeId", "foo", JobStatus.RUNNING, new SchedulerState(JobSchedulerStatus.STOPPING, 0L, null));
         scheduledJobService.registry.put("foo", scheduledJobService.createJobScheduler(builder.build()));
         scheduledJobService.stop(allocation2);
 

elasticsearch/src/test/resources/rest-api-spec/api/xpack.prelert.get_job.json

@@ -2,20 +2,20 @@
   "xpack.prelert.get_job": {
     "methods": [ "GET", "POST" ],
     "url": {
-      "path": "/_xpack/prelert/jobs/{jobId}",
+      "path": "/_xpack/prelert/jobs/{job_id}",
       "paths": [
         "/_xpack/prelert/jobs",
-        "/_xpack/prelert/jobs/{jobId}",
+        "/_xpack/prelert/jobs/{job_id}",
-        "/_xpack/prelert/jobs/{jobId}/{metric}"
+        "/_xpack/prelert/jobs/{job_id}/{metric}"
       ],
       "parts": {
-        "jobId": {
+        "job_id": {
           "type": "string",
           "description": "The ID of the job to fetch"
         },
         "metric" : {
           "type" : "list",
-          "options" : ["_all", "config", "data_counts", "model_size_stats"],
+          "options" : ["_all", "config", "data_counts", "model_size_stats", "scheduler_state", "status"],
           "description" : "Limit the information returned to the specified statistics"
         }
       },

elasticsearch/src/test/resources/rest-api-spec/test/job_get_stats.yaml

@@ -16,15 +16,41 @@ setup:
             }
           }
 
+  - do:
+      xpack.prelert.put_job:
+        body:  >
+          {
+            "jobId":"scheduled-job",
+            "description":"A job with a scheduler",
+            "analysisConfig" : {
+                "bucketSpan":3600,
+                "detectors" :[{"function":"metric","fieldName":"responsetime","byFieldName":"airline"}]
+            },
+            "dataDescription" : {
+                "format" : "ELASTICSEARCH",
+                "timeField":"time",
+                "timeFormat":"yyyy-MM-dd'T'HH:mm:ssX"
+            },
+            "schedulerConfig": {
+                "dataSource":"ELASTICSEARCH",
+                "baseUrl":"http://marple:9202",
+                "indexes":["farequote"],
+                "types":["response"],
+                "retrieveWholeSource":true
+            }
+          }
+
 ---
 "Test get job stats after uploading data prompting the creation of some stats":
   - do:
       xpack.prelert.get_job:
-        jobId: job-stats-test
+        job_id: job-stats-test
 
   - is_true: hits.0.config
   - is_false: hits.0.data_counts
   - is_false: hits.0.model_size_stats
+  - is_false: hits.0.scheduler_state
+
 
   - do:
       xpack.prelert.post_data:
@@ -41,7 +67,7 @@ setup:
 
   - do:
       xpack.prelert.get_job:
-        jobId: job-stats-test
+        job_id: job-stats-test
         metric: data_counts
 
   - match: { hits.0.data_counts.processed_record_count: 2 }
@@ -53,23 +79,53 @@ setup:
 # won't be created with such a small data sample
   - do:
       xpack.prelert.get_job:
-        jobId: "job-stats-test"
+        job_id: "job-stats-test"
         metric: "data_counts"
   - is_false: hits.0.config
   - is_true: hits.0.data_counts
   - is_false: hits.0.model_size_stats
+  - is_false: hits.0.scheduler_state
 
   - do:
       xpack.prelert.get_job:
-        jobId: "job-stats-test"
+        job_id: "job-stats-test"
         metric: "model_size_stats"
   - is_false: hits.0.config
   - is_false: hits.0.data_counts
+  - is_false: hits.0.scheduler_state
+
+  - do:
+      xpack.prelert.get_job:
+        job_id: "job-stats-test"
+        metric: "scheduler_state"
+  - is_false: hits.0.config
+  - is_false: hits.0.data_counts
 
   - do:
       xpack.prelert.get_job:
-        jobId: "job-stats-test"
+        job_id: "job-stats-test"
+        metric: "status"
+  - is_false: hits.0.config
+  - is_false: hits.0.data_counts
+  - is_false: hits.0.model_size_stats
+  - is_false: hits.0.scheduler_state
+  - match: { hits.0.status: CLOSED }
+
+  - do:
+      xpack.prelert.get_job:
+        job_id: "job-stats-test"
         metric: "_all"
   - is_true: hits.0.config
   - is_true: hits.0.data_counts
+  - is_false: hits.0.scheduler_state
+  - match: { hits.0.status: CLOSED }
+
+  - do:
+      xpack.prelert.get_job:
+        job_id: "scheduled-job"
+        metric: "scheduler_state"
+  - is_false: hits.0.config
+  - is_false: hits.0.data_counts
+  - is_false: hits.0.model_size_stats
+  - match: { hits.0.scheduler_state.status: STOPPED }
 

elasticsearch/src/test/resources/rest-api-spec/test/jobs_crud.yaml

@@ -32,7 +32,7 @@
 
   - do:
       xpack.prelert.get_job:
-        jobId: "farequote"
+        job_id: "farequote"
   - match: { hitCount: 1 }
   - match: { hits.0.config.jobId: "farequote" }
 
@@ -44,7 +44,7 @@
   - do:
       catch: missing
       xpack.prelert.get_job:
-        jobId: "farequote"
+        job_id: "farequote"
 
   - do:
       indices.exists:
@@ -56,7 +56,7 @@
   - do:
       catch: missing
       xpack.prelert.get_job:
-        jobId: "non-existing"
+        job_id: "non-existing"
 
 ---
 "Test put job with id that is already taken":

elasticsearch/src/test/resources/rest-api-spec/test/revert_model_snapshot.yaml

@@ -219,7 +219,7 @@ setup:
 
   - do:
       xpack.prelert.get_job:
-        jobId: foo
+        job_id: foo
         metric: data_counts
 
   - match: { hits.0.data_counts.latest_record_timestamp: 1464739200000 }