[ML][HLRC] adds new running_state field to datafeed stats (#73926)

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/datafeed/DatafeedStats.java

@@ -32,10 +32,13 @@ public class DatafeedStats implements ToXContentObject {
     private final String assignmentExplanation;
     @Nullable
     private final DatafeedTimingStats timingStats;
+    @Nullable
+    private final RunningState runningState;
 
     public static final ParseField ASSIGNMENT_EXPLANATION = new ParseField("assignment_explanation");
     public static final ParseField NODE = new ParseField("node");
     public static final ParseField TIMING_STATS = new ParseField("timing_stats");
+    public static final ParseField RUNNING_STATE = new ParseField("running_state");
 
     public static final ConstructingObjectParser<DatafeedStats, Void> PARSER = new ConstructingObjectParser<>("datafeed_stats",
     true,
@@ -45,7 +48,8 @@ public class DatafeedStats implements ToXContentObject {
         NodeAttributes nodeAttributes = (NodeAttributes)a[2];
         String assignmentExplanation = (String)a[3];
         DatafeedTimingStats timingStats = (DatafeedTimingStats)a[4];
-        return new DatafeedStats(datafeedId, datafeedState, nodeAttributes, assignmentExplanation, timingStats);
+        RunningState runningState = (RunningState) a[5];
+        return new DatafeedStats(datafeedId, datafeedState, nodeAttributes, assignmentExplanation, timingStats, runningState);
     } );
 
     static {
@@ -54,15 +58,21 @@ public class DatafeedStats implements ToXContentObject {
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), NodeAttributes.PARSER, NODE);
         PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), ASSIGNMENT_EXPLANATION);
         PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), DatafeedTimingStats.PARSER, TIMING_STATS);
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), RunningState.PARSER, RUNNING_STATE);
     }
 
-    public DatafeedStats(String datafeedId, DatafeedState datafeedState, @Nullable NodeAttributes node,
-                         @Nullable String assignmentExplanation, @Nullable DatafeedTimingStats timingStats) {
+    public DatafeedStats(String datafeedId,
+                         DatafeedState datafeedState,
+                         @Nullable NodeAttributes node,
+                         @Nullable String assignmentExplanation,
+                         @Nullable DatafeedTimingStats timingStats,
+                         @Nullable RunningState runningState) {
         this.datafeedId = Objects.requireNonNull(datafeedId);
         this.datafeedState = Objects.requireNonNull(datafeedState);
         this.node = node;
         this.assignmentExplanation = assignmentExplanation;
         this.timingStats = timingStats;
+        this.runningState = runningState;
     }
 
     public String getDatafeedId() {
@@ -85,6 +95,10 @@ public DatafeedTimingStats getDatafeedTimingStats() {
         return timingStats;
     }
 
+    public RunningState getRunningState() {
+        return runningState;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
         builder.startObject();
@@ -112,13 +126,16 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params par
         if (timingStats != null) {
             builder.field(TIMING_STATS.getPreferredName(), timingStats);
         }
+        if (runningState != null) {
+            builder.field(RUNNING_STATE.getPreferredName(), runningState);
+        }
         builder.endObject();
         return builder;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(datafeedId, datafeedState.toString(), node, assignmentExplanation, timingStats);
+        return Objects.hash(datafeedId, datafeedState.toString(), node, assignmentExplanation, timingStats, runningState);
     }
 
     @Override
@@ -134,6 +151,7 @@ public boolean equals(Object obj) {
             Objects.equals(this.datafeedState, other.datafeedState) &&
             Objects.equals(this.node, other.node) &&
             Objects.equals(this.assignmentExplanation, other.assignmentExplanation) &&
+            Objects.equals(this.runningState, other.runningState) &&
             Objects.equals(this.timingStats, other.timingStats);
     }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/datafeed/RunningState.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.client.ml.datafeed;
+
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ParseField;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class RunningState implements ToXContentObject {
+
+    private static final ParseField REAL_TIME_CONFIGURED = new ParseField("real_time_configured");
+    private static final ParseField REAL_TIME_RUNNING = new ParseField("real_time_running");
+
+    public static final ConstructingObjectParser<RunningState, Void> PARSER = new ConstructingObjectParser<>(
+        "datafeed_running_state",
+        true,
+        a -> new RunningState((Boolean)a[0], (Boolean)a[1])
+        );
+
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), REAL_TIME_CONFIGURED);
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), REAL_TIME_RUNNING);
+    }
+
+    // Is the datafeed a "realtime" datafeed, meaning it was started without an end_time
+    private final boolean realTimeConfigured;
+    // Has the reading historical data has finished and are we now running on "real-time" data
+    private final boolean realTimeRunning;
+
+    public RunningState(boolean realTimeConfigured, boolean realTimeRunning) {
+        this.realTimeConfigured = realTimeConfigured;
+        this.realTimeRunning = realTimeRunning;
+    }
+
+    /**
+     * Indicates if the datafeed is configured to run in real time
+     *
+     * @return true if the datafeed is configured to run in real time.
+     */
+    public boolean isRealTimeConfigured() {
+        return realTimeConfigured;
+    }
+
+    /**
+     * Indicates if the datafeed has processed all historical data available at the start time and is now processing "real-time" data.
+     * @return true if the datafeed is now running in real-time
+     */
+    public boolean isRealTimeRunning() {
+        return realTimeRunning;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        RunningState that = (RunningState) o;
+        return realTimeConfigured == that.realTimeConfigured && realTimeRunning == that.realTimeRunning;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(realTimeConfigured, realTimeRunning);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(REAL_TIME_CONFIGURED.getPreferredName(), realTimeConfigured);
+        builder.field(REAL_TIME_RUNNING.getPreferredName(), realTimeRunning);
+        builder.endObject();
+        return builder;
+    }
+
+
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java

@@ -814,6 +814,8 @@ public void testGetDatafeedStats() throws Exception {
         assertThat(response.datafeedStats(), hasSize(1));
         assertThat(response.datafeedStats().get(0).getDatafeedId(), equalTo(datafeedId1));
         assertThat(response.datafeedStats().get(0).getDatafeedState().toString(), equalTo(DatafeedState.STARTED.toString()));
+        assertThat(response.datafeedStats().get(0).getRunningState(), is(notNullValue()));
+        assertThat(response.datafeedStats().get(0).getRunningState().isRealTimeConfigured(), is(true));
 
         // Test getting all explicitly
         request = GetDatafeedStatsRequest.getAllDatafeedStatsRequest();

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/datafeed/DatafeedStatsTests.java

@@ -40,7 +40,14 @@ public static DatafeedStats createRandomInstance() {
         }
         String assignmentReason = randomBoolean() ? randomAlphaOfLength(10) : null;
         DatafeedTimingStats timingStats = DatafeedTimingStatsTests.createRandomInstance();
-        return new DatafeedStats(datafeedId, datafeedState, nodeAttributes, assignmentReason, timingStats);
+        return new DatafeedStats(
+            datafeedId,
+            datafeedState,
+            nodeAttributes,
+            assignmentReason,
+            timingStats,
+            randomBoolean() ? null : RunningStateTests.createRandomInstance()
+        );
     }
 
     @Override

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/datafeed/RunningStateTests.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.client.ml.datafeed;
+
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+
+
+public class RunningStateTests extends AbstractXContentTestCase<RunningState> {
+
+    public static RunningState createRandomInstance() {
+        return new RunningState(randomBoolean(), randomBoolean());
+    }
+
+    @Override
+    protected RunningState createTestInstance() {
+        return createRandomInstance();
+    }
+
+    @Override
+    protected RunningState doParseInstance(XContentParser parser) throws IOException {
+        return RunningState.PARSER.apply(parser, null);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+}