[ML-DataFrame] Combine task_state and indexer_state in _stats (#45324)

This commit replaces task_state and indexer_state in the
data frame _stats output with a single top level state
that combines the two. It is defined as:

- failed if what's currently reported as task_state is failed
- stopped if there is no persistent task
- Otherwise what's currently reported as indexer_state

Backport of #45276

client/rest-high-level/src/main/java/org/elasticsearch/client/dataframe/transforms/DataFrameTransformCheckpointStats.java

@@ -19,10 +19,8 @@
 
 package org.elasticsearch.client.dataframe.transforms;
 
-import org.elasticsearch.client.core.IndexerState;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
-import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
@@ -33,16 +31,14 @@
 public class DataFrameTransformCheckpointStats {
 
     public static final ParseField CHECKPOINT = new ParseField("checkpoint");
-    public static final ParseField INDEXER_STATE = new ParseField("indexer_state");
     public static final ParseField POSITION = new ParseField("position");
     public static final ParseField CHECKPOINT_PROGRESS = new ParseField("checkpoint_progress");
     public static final ParseField TIMESTAMP_MILLIS = new ParseField("timestamp_millis");
     public static final ParseField TIME_UPPER_BOUND_MILLIS = new ParseField("time_upper_bound_millis");
 
-    public static final DataFrameTransformCheckpointStats EMPTY = new DataFrameTransformCheckpointStats(0L, null, null, null, 0L, 0L);
+    public static final DataFrameTransformCheckpointStats EMPTY = new DataFrameTransformCheckpointStats(0L, null, null, 0L, 0L);
 
     private final long checkpoint;
-    private final IndexerState indexerState;
     private final DataFrameIndexerPosition position;
     private final DataFrameTransformProgress checkpointProgress;
     private final long timestampMillis;
@@ -51,19 +47,16 @@ public class DataFrameTransformCheckpointStats {
     public static final ConstructingObjectParser<DataFrameTransformCheckpointStats, Void> LENIENT_PARSER = new ConstructingObjectParser<>(
             "data_frame_transform_checkpoint_stats", true, args -> {
         long checkpoint = args[0] == null ? 0L : (Long) args[0];
-        IndexerState indexerState = (IndexerState) args[1];
-        DataFrameIndexerPosition position = (DataFrameIndexerPosition) args[2];
-        DataFrameTransformProgress checkpointProgress = (DataFrameTransformProgress) args[3];
-        long timestamp = args[4] == null ? 0L : (Long) args[4];
-        long timeUpperBound = args[5] == null ? 0L : (Long) args[5];
+        DataFrameIndexerPosition position = (DataFrameIndexerPosition) args[1];
+        DataFrameTransformProgress checkpointProgress = (DataFrameTransformProgress) args[2];
+        long timestamp = args[3] == null ? 0L : (Long) args[3];
+        long timeUpperBound = args[4] == null ? 0L : (Long) args[4];
 
-        return new DataFrameTransformCheckpointStats(checkpoint, indexerState, position, checkpointProgress, timestamp, timeUpperBound);
+        return new DataFrameTransformCheckpointStats(checkpoint, position, checkpointProgress, timestamp, timeUpperBound);
     });
 
     static {
         LENIENT_PARSER.declareLong(optionalConstructorArg(), CHECKPOINT);
-        LENIENT_PARSER.declareField(optionalConstructorArg(), p -> IndexerState.fromString(p.text()), INDEXER_STATE,
-            ObjectParser.ValueType.STRING);
         LENIENT_PARSER.declareObject(optionalConstructorArg(), DataFrameIndexerPosition.PARSER, POSITION);
         LENIENT_PARSER.declareObject(optionalConstructorArg(), DataFrameTransformProgress.PARSER, CHECKPOINT_PROGRESS);
         LENIENT_PARSER.declareLong(optionalConstructorArg(), TIMESTAMP_MILLIS);
@@ -74,11 +67,10 @@ public static DataFrameTransformCheckpointStats fromXContent(XContentParser pars
         return LENIENT_PARSER.parse(parser, null);
     }
 
-    public DataFrameTransformCheckpointStats(final long checkpoint, final IndexerState indexerState,
-                                             final DataFrameIndexerPosition position, final DataFrameTransformProgress checkpointProgress,
-                                             final long timestampMillis, final long timeUpperBoundMillis) {
+    public DataFrameTransformCheckpointStats(final long checkpoint, final DataFrameIndexerPosition position,
+                                             final DataFrameTransformProgress checkpointProgress, final long timestampMillis,
+                                             final long timeUpperBoundMillis) {
         this.checkpoint = checkpoint;
-        this.indexerState = indexerState;
         this.position = position;
         this.checkpointProgress = checkpointProgress;
         this.timestampMillis = timestampMillis;
@@ -89,10 +81,6 @@ public long getCheckpoint() {
         return checkpoint;
     }
 
-    public IndexerState getIndexerState() {
-        return indexerState;
-    }
-
     public DataFrameIndexerPosition getPosition() {
         return position;
     }
@@ -111,7 +99,7 @@ public long getTimeUpperBoundMillis() {
 
     @Override
     public int hashCode() {
-        return Objects.hash(checkpoint, indexerState, position, checkpointProgress, timestampMillis, timeUpperBoundMillis);
+        return Objects.hash(checkpoint, position, checkpointProgress, timestampMillis, timeUpperBoundMillis);
     }
 
     @Override
@@ -127,7 +115,6 @@ public boolean equals(Object other) {
         DataFrameTransformCheckpointStats that = (DataFrameTransformCheckpointStats) other;
 
         return this.checkpoint == that.checkpoint
-            && Objects.equals(this.indexerState, that.indexerState)
             && Objects.equals(this.position, that.position)
             && Objects.equals(this.checkpointProgress, that.checkpointProgress)
             && this.timestampMillis == that.timestampMillis

client/rest-high-level/src/main/java/org/elasticsearch/client/dataframe/transforms/DataFrameTransformStats.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Locale;
 import java.util.Objects;
 
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
@@ -33,20 +34,20 @@
 public class DataFrameTransformStats {
 
     public static final ParseField ID = new ParseField("id");
-    public static final ParseField TASK_STATE_FIELD = new ParseField("task_state");
+    public static final ParseField STATE_FIELD = new ParseField("state");
     public static final ParseField REASON_FIELD = new ParseField("reason");
     public static final ParseField NODE_FIELD = new ParseField("node");
     public static final ParseField STATS_FIELD = new ParseField("stats");
     public static final ParseField CHECKPOINTING_INFO_FIELD = new ParseField("checkpointing");
 
     public static final ConstructingObjectParser<DataFrameTransformStats, Void> PARSER = new ConstructingObjectParser<>(
-            "data_frame_transform_state_and_stats_info", true,
-            a -> new DataFrameTransformStats((String) a[0], (DataFrameTransformTaskState) a[1], (String) a[2],
-                (NodeAttributes) a[3], (DataFrameIndexerTransformStats) a[4], (DataFrameTransformCheckpointingInfo) a[5]));
+        "data_frame_transform_state_and_stats_info", true,
+        a -> new DataFrameTransformStats((String) a[0], (State) a[1], (String) a[2],
+            (NodeAttributes) a[3], (DataFrameIndexerTransformStats) a[4], (DataFrameTransformCheckpointingInfo) a[5]));
 
     static {
         PARSER.declareString(constructorArg(), ID);
-        PARSER.declareField(optionalConstructorArg(), p -> DataFrameTransformTaskState.fromString(p.text()), TASK_STATE_FIELD,
+        PARSER.declareField(optionalConstructorArg(), p -> State.fromString(p.text()), STATE_FIELD,
             ObjectParser.ValueType.STRING);
         PARSER.declareString(optionalConstructorArg(), REASON_FIELD);
         PARSER.declareField(optionalConstructorArg(), NodeAttributes.PARSER::apply, NODE_FIELD, ObjectParser.ValueType.OBJECT);
@@ -61,16 +62,15 @@ public static DataFrameTransformStats fromXContent(XContentParser parser) throws
 
     private final String id;
     private final String reason;
-    private final DataFrameTransformTaskState taskState;
+    private final State state;
     private final NodeAttributes node;
     private final DataFrameIndexerTransformStats indexerStats;
     private final DataFrameTransformCheckpointingInfo checkpointingInfo;
 
-    public DataFrameTransformStats(String id, DataFrameTransformTaskState taskState, String reason, NodeAttributes node,
-                                   DataFrameIndexerTransformStats stats,
+    public DataFrameTransformStats(String id, State state, String reason, NodeAttributes node, DataFrameIndexerTransformStats stats,
                                    DataFrameTransformCheckpointingInfo checkpointingInfo) {
         this.id = id;
-        this.taskState = taskState;
+        this.state = state;
         this.reason = reason;
         this.node = node;
         this.indexerStats = stats;
@@ -81,8 +81,8 @@ public String getId() {
         return id;
     }
 
-    public DataFrameTransformTaskState getTaskState() {
-        return taskState;
+    public State getState() {
+        return state;
     }
 
     public String getReason() {
@@ -103,7 +103,7 @@ public DataFrameTransformCheckpointingInfo getCheckpointingInfo() {
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, taskState, reason, node, indexerStats, checkpointingInfo);
+        return Objects.hash(id, state, reason, node, indexerStats, checkpointingInfo);
     }
 
     @Override
@@ -119,10 +119,23 @@ public boolean equals(Object other) {
         DataFrameTransformStats that = (DataFrameTransformStats) other;
 
         return Objects.equals(this.id, that.id)
-            && Objects.equals(this.taskState, that.taskState)
+            && Objects.equals(this.state, that.state)
             && Objects.equals(this.reason, that.reason)
             && Objects.equals(this.node, that.node)
             && Objects.equals(this.indexerStats, that.indexerStats)
             && Objects.equals(this.checkpointingInfo, that.checkpointingInfo);
     }
+
+    public enum State {
+
+        STARTED, INDEXING, ABORTING, STOPPING, STOPPED, FAILED;
+
+        public static State fromString(String name) {
+            return valueOf(name.trim().toUpperCase(Locale.ROOT));
+        }
+
+        public String value() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+    }
 }

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

@@ -44,7 +44,6 @@
 import org.elasticsearch.client.dataframe.transforms.DataFrameTransformConfig;
 import org.elasticsearch.client.dataframe.transforms.DataFrameTransformConfigUpdate;
 import org.elasticsearch.client.dataframe.transforms.DataFrameTransformStats;
-import org.elasticsearch.client.dataframe.transforms.DataFrameTransformTaskState;
 import org.elasticsearch.client.dataframe.transforms.DestConfig;
 import org.elasticsearch.client.dataframe.transforms.SourceConfig;
 import org.elasticsearch.client.dataframe.transforms.TimeSyncConfig;
@@ -306,10 +305,11 @@ public void testStartStop() throws IOException {
         GetDataFrameTransformStatsResponse statsResponse = execute(new GetDataFrameTransformStatsRequest(id),
                 client::getDataFrameTransformStats, client::getDataFrameTransformStatsAsync);
         assertThat(statsResponse.getTransformsStats(), hasSize(1));
-        DataFrameTransformTaskState taskState = statsResponse.getTransformsStats().get(0).getTaskState();
+        DataFrameTransformStats.State taskState = statsResponse.getTransformsStats().get(0).getState();
 
         // Since we are non-continuous, the transform could auto-stop between being started earlier and us gathering the statistics
-        assertThat(taskState, is(oneOf(DataFrameTransformTaskState.STARTED, DataFrameTransformTaskState.STOPPED)));
+        assertThat(taskState, oneOf(DataFrameTransformStats.State.STARTED, DataFrameTransformStats.State.INDEXING,
+            DataFrameTransformStats.State.STOPPING, DataFrameTransformStats.State.STOPPED));
 
         StopDataFrameTransformRequest stopRequest = new StopDataFrameTransformRequest(id, Boolean.TRUE, null);
         StopDataFrameTransformResponse stopResponse =
@@ -321,8 +321,8 @@ public void testStartStop() throws IOException {
         // Calling stop with wait_for_completion assures that we will be in the `STOPPED` state for the transform task
         statsResponse = execute(new GetDataFrameTransformStatsRequest(id),
             client::getDataFrameTransformStats, client::getDataFrameTransformStatsAsync);
-        taskState = statsResponse.getTransformsStats().get(0).getTaskState();
-        assertThat(taskState, is(DataFrameTransformTaskState.STOPPED));
+        taskState = statsResponse.getTransformsStats().get(0).getState();
+        assertThat(taskState, is(DataFrameTransformStats.State.STOPPED));
     }
 
     @SuppressWarnings("unchecked")
@@ -405,7 +405,7 @@ public void testGetStats() throws Exception {
 
         assertEquals(1, statsResponse.getTransformsStats().size());
         DataFrameTransformStats stats = statsResponse.getTransformsStats().get(0);
-        assertEquals(DataFrameTransformTaskState.STOPPED, stats.getTaskState());
+        assertEquals(DataFrameTransformStats.State.STOPPED, stats.getState());
 
         DataFrameIndexerTransformStats zeroIndexerStats = new DataFrameIndexerTransformStats(0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L);
         assertEquals(zeroIndexerStats, stats.getIndexerStats());
@@ -420,8 +420,8 @@ public void testGetStats() throws Exception {
                     client::getDataFrameTransformStats, client::getDataFrameTransformStatsAsync);
             DataFrameTransformStats stateAndStats = response.getTransformsStats().get(0);
             assertNotEquals(zeroIndexerStats, stateAndStats.getIndexerStats());
-            assertThat(stateAndStats.getTaskState(),
-                is(oneOf(DataFrameTransformTaskState.STARTED, DataFrameTransformTaskState.STOPPED)));
+            assertThat(stateAndStats.getState(), oneOf(DataFrameTransformStats.State.STARTED, DataFrameTransformStats.State.INDEXING,
+                DataFrameTransformStats.State.STOPPING, DataFrameTransformStats.State.STOPPED));
             assertThat(stateAndStats.getReason(), is(nullValue()));
         });
     }

client/rest-high-level/src/test/java/org/elasticsearch/client/dataframe/transforms/DataFrameTransformCheckpointStatsTests.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.client.dataframe.transforms;
 
-import org.elasticsearch.client.core.IndexerState;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.test.ESTestCase;
 
@@ -41,7 +40,6 @@ public void testFromXContent() throws IOException {
 
     public static DataFrameTransformCheckpointStats randomDataFrameTransformCheckpointStats() {
         return new DataFrameTransformCheckpointStats(randomLongBetween(1, 1_000_000),
-            randomBoolean() ? null : randomFrom(IndexerState.values()),
             randomBoolean() ? null : DataFrameIndexerPositionTests.randomDataFrameIndexerPosition(),
             randomBoolean() ? null : DataFrameTransformProgressTests.randomInstance(),
             randomLongBetween(1, 1_000_000), randomLongBetween(0, 1_000_000));
@@ -50,9 +48,6 @@ public static DataFrameTransformCheckpointStats randomDataFrameTransformCheckpoi
     public static void toXContent(DataFrameTransformCheckpointStats stats, XContentBuilder builder) throws IOException {
         builder.startObject();
         builder.field(DataFrameTransformCheckpointStats.CHECKPOINT.getPreferredName(), stats.getCheckpoint());
-        if (stats.getIndexerState() != null) {
-            builder.field(DataFrameTransformCheckpointStats.INDEXER_STATE.getPreferredName(), stats.getIndexerState().value());
-        }
         if (stats.getPosition() != null) {
             builder.field(DataFrameTransformCheckpointStats.POSITION.getPreferredName());
             DataFrameIndexerPositionTests.toXContent(stats.getPosition(), builder);

client/rest-high-level/src/test/java/org/elasticsearch/client/dataframe/transforms/DataFrameTransformStatsTests.java

@@ -41,7 +41,7 @@ public void testFromXContent() throws IOException {
 
     public static DataFrameTransformStats randomInstance() {
         return new DataFrameTransformStats(randomAlphaOfLength(10),
-            randomBoolean() ? null : randomFrom(DataFrameTransformTaskState.values()),
+            randomBoolean() ? null : randomFrom(DataFrameTransformStats.State.values()),
             randomBoolean() ? null : randomAlphaOfLength(100),
             randomBoolean() ? null : NodeAttributesTests.createRandom(),
             DataFrameIndexerTransformStatsTests.randomStats(),
@@ -51,9 +51,9 @@ public static DataFrameTransformStats randomInstance() {
     public static void toXContent(DataFrameTransformStats stats, XContentBuilder builder) throws IOException {
         builder.startObject();
         builder.field(DataFrameTransformStats.ID.getPreferredName(), stats.getId());
-        if (stats.getTaskState() != null) {
-            builder.field(DataFrameTransformStats.TASK_STATE_FIELD.getPreferredName(),
-                stats.getTaskState().value());
+        if (stats.getState() != null) {
+            builder.field(DataFrameTransformStats.STATE_FIELD.getPreferredName(),
+                stats.getState().value());
         }
         if (stats.getReason() != null) {
             builder.field(DataFrameTransformStats.REASON_FIELD.getPreferredName(), stats.getReason());

client/rest-high-level/src/test/java/org/elasticsearch/client/dataframe/transforms/hlrc/DataFrameTransformCheckpointStatsTests.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.client.AbstractHlrcXContentTestCase;
 import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformCheckpointStats;
-import org.elasticsearch.xpack.core.indexing.IndexerState;
 
 import java.io.IOException;
 import java.util.function.Predicate;
@@ -34,7 +33,6 @@ public class DataFrameTransformCheckpointStatsTests extends AbstractHlrcXContent
     public static DataFrameTransformCheckpointStats fromHlrc(
             org.elasticsearch.client.dataframe.transforms.DataFrameTransformCheckpointStats instance) {
         return new DataFrameTransformCheckpointStats(instance.getCheckpoint(),
-            (instance.getIndexerState() != null) ? IndexerState.fromString(instance.getIndexerState().value()) : null,
             DataFrameIndexerPositionTests.fromHlrc(instance.getPosition()),
             DataFrameTransformProgressTests.fromHlrc(instance.getCheckpointProgress()),
             instance.getTimestampMillis(),
@@ -55,7 +53,6 @@ public DataFrameTransformCheckpointStats convertHlrcToInternal(
 
     public static DataFrameTransformCheckpointStats randomDataFrameTransformCheckpointStats() {
         return new DataFrameTransformCheckpointStats(randomLongBetween(1, 1_000_000),
-            randomBoolean() ? null : randomFrom(IndexerState.values()),
             DataFrameIndexerPositionTests.randomDataFrameIndexerPosition(),
             randomBoolean() ? null : DataFrameTransformProgressTests.randomDataFrameTransformProgress(),
             randomLongBetween(1, 1_000_000), randomLongBetween(0, 1_000_000));