[Transform][Rollup] add processing stats to record the time spent for processing results

client/rest-high-level/src/main/java/org/elasticsearch/client/core/IndexerJobStats.java

@@ -31,8 +31,10 @@ public abstract class IndexerJobStats {
     public static ParseField NUM_INVOCATIONS = new ParseField("trigger_count");
     public static ParseField INDEX_TIME_IN_MS = new ParseField("index_time_in_ms");
     public static ParseField SEARCH_TIME_IN_MS = new ParseField("search_time_in_ms");
+    public static ParseField PROCESSING_TIME_IN_MS = new ParseField("processing_time_in_ms");
     public static ParseField INDEX_TOTAL = new ParseField("index_total");
     public static ParseField SEARCH_TOTAL = new ParseField("search_total");
+    public static ParseField PROCESSING_TOTAL = new ParseField("processing_total");
     public static ParseField SEARCH_FAILURES = new ParseField("search_failures");
     public static ParseField INDEX_FAILURES = new ParseField("index_failures");
 
@@ -44,11 +46,14 @@ public abstract class IndexerJobStats {
     protected final long indexTotal;
     protected final long searchTime;
     protected final long searchTotal;
+    protected final long processingTime;
+    protected final long processingTotal;
     protected final long indexFailures;
     protected final long searchFailures;
 
     public IndexerJobStats(long numPages, long numInputDocuments, long numOutputDocuments, long numInvocations,
-                           long indexTime, long searchTime, long indexTotal, long searchTotal, long indexFailures, long searchFailures) {
+                           long indexTime, long searchTime, long processingTime, long indexTotal, long searchTotal, long processingTotal,
+                           long indexFailures, long searchFailures) {
         this.numPages = numPages;
         this.numInputDocuments = numInputDocuments;
         this.numOuputDocuments = numOutputDocuments;
@@ -57,6 +62,8 @@ public IndexerJobStats(long numPages, long numInputDocuments, long numOutputDocu
         this.indexTotal = indexTotal;
         this.searchTime = searchTime;
         this.searchTotal = searchTotal;
+        this.processingTime = processingTime;
+        this.processingTotal = processingTotal;
         this.indexFailures = indexFailures;
         this.searchFailures = searchFailures;
     }
@@ -117,6 +124,13 @@ public long getSearchTime() {
         return searchTime;
     }
 
+    /**
+     * Returns the time spent processing (cumulative) in milliseconds
+     */
+    public long getProcessingTime() {
+        return processingTime;
+    }
+
     /**
      * Returns the total number of indexing requests that have been processed
      * (Note: this is not the number of _documents_ that have been indexed)
@@ -132,6 +146,14 @@ public long getSearchTotal() {
         return searchTotal;
     }
 
+    /**
+     * Returns the total number of processing runs that have been made
+     */
+    public long getProcessingTotal() {
+        return processingTotal;
+    }
+
+
     @Override
     public boolean equals(Object other) {
         if (this == other) {
@@ -149,16 +171,19 @@ public boolean equals(Object other) {
                 && Objects.equals(this.numInvocations, that.numInvocations)
                 && Objects.equals(this.indexTime, that.indexTime)
                 && Objects.equals(this.searchTime, that.searchTime)
+                && Objects.equals(this.processingTime, that.processingTime)
                 && Objects.equals(this.indexFailures, that.indexFailures)
                 && Objects.equals(this.searchFailures, that.searchFailures)
                 && Objects.equals(this.searchTotal, that.searchTotal)
+                && Objects.equals(this.processingTotal, that.processingTotal)
                 && Objects.equals(this.indexTotal, that.indexTotal);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(numPages, numInputDocuments, numOuputDocuments, numInvocations,
-                indexTime, searchTime, indexFailures, searchFailures, searchTotal, indexTotal);
+                indexTime, searchTime, processingTime, indexFailures, searchFailures, searchTotal,
+                indexTotal, processingTotal);
     }
 
     @Override
@@ -172,6 +197,8 @@ public final String toString() {
                 + ", index_time_in_ms=" + indexTime
                 + ", index_total=" + indexTotal
                 + ", search_time_in_ms=" + searchTime
-                + ", search_total=" + searchTotal+ "}";
+                + ", search_total=" + searchTotal
+                + ", processing_time_in_ms=" + processingTime
+                + ", processing_total=" + processingTotal + "}";
     }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/rollup/GetRollupJobResponse.java

@@ -177,16 +177,18 @@ public final String toString() {
     public static class RollupIndexerJobStats extends IndexerJobStats {
 
         RollupIndexerJobStats(long numPages, long numInputDocuments, long numOuputDocuments, long numInvocations,
-                              long indexTime, long indexTotal, long searchTime, long searchTotal, long indexFailures, long searchFailures) {
+                              long indexTime, long indexTotal, long searchTime, long searchTotal, long processingTime,
+                              long processingTotal, long indexFailures, long searchFailures) {
             super(numPages, numInputDocuments, numOuputDocuments, numInvocations,
-                    indexTime, searchTime, indexTotal, searchTotal, indexFailures, searchFailures);
+                    indexTime, searchTime, processingTime, indexTotal, searchTotal, processingTotal, indexFailures, searchFailures);
         }
 
         private static final ConstructingObjectParser<RollupIndexerJobStats, Void> PARSER = new ConstructingObjectParser<>(
                 STATS.getPreferredName(),
                 true,
                 args -> new RollupIndexerJobStats((long) args[0], (long) args[1], (long) args[2], (long) args[3],
-                    (long) args[4], (long) args[5], (long) args[6], (long) args[7], (long) args[8], (long) args[9]));
+                    (long) args[4], (long) args[5], (long) args[6], (long) args[7], (long) args[8], (long) args[9],
+                    (long) args[10], (long) args[11]));
         static {
             PARSER.declareLong(constructorArg(), NUM_PAGES);
             PARSER.declareLong(constructorArg(), NUM_INPUT_DOCUMENTS);
@@ -196,6 +198,8 @@ public static class RollupIndexerJobStats extends IndexerJobStats {
             PARSER.declareLong(constructorArg(), INDEX_TOTAL);
             PARSER.declareLong(constructorArg(), SEARCH_TIME_IN_MS);
             PARSER.declareLong(constructorArg(), SEARCH_TOTAL);
+            PARSER.declareLong(constructorArg(), PROCESSING_TIME_IN_MS);
+            PARSER.declareLong(constructorArg(), PROCESSING_TOTAL);
             PARSER.declareLong(constructorArg(), INDEX_FAILURES);
             PARSER.declareLong(constructorArg(), SEARCH_FAILURES);
         }

client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/TransformIndexerStats.java

@@ -27,7 +27,6 @@
 import java.io.IOException;
 import java.util.Objects;
 
-import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
 
 public class TransformIndexerStats extends IndexerJobStats {
@@ -39,21 +38,38 @@ public class TransformIndexerStats extends IndexerJobStats {
     public static final ConstructingObjectParser<TransformIndexerStats, Void> LENIENT_PARSER = new ConstructingObjectParser<>(
         NAME,
         true,
-        args -> new TransformIndexerStats((long) args[0], (long) args[1], (long) args[2],
-            (long) args[3], (long) args[4], (long) args[5], (long) args[6], (long) args[7], (long) args[8], (long) args[9],
-            (Double) args[10], (Double) args[11], (Double) args[12]));
+        args -> new TransformIndexerStats(
+            unboxSafe(args[0], 0L),
+            unboxSafe(args[1], 0L),
+            unboxSafe(args[2], 0L),
+            unboxSafe(args[3], 0L),
+            unboxSafe(args[4], 0L),
+            unboxSafe(args[5], 0L),
+            unboxSafe(args[6], 0L),
+            unboxSafe(args[7], 0L),
+            unboxSafe(args[8], 0L),
+            unboxSafe(args[9], 0L),
+            unboxSafe(args[10], 0L),
+            unboxSafe(args[11], 0L),
+            unboxSafe(args[12], 0.0),
+            unboxSafe(args[13], 0.0),
+            unboxSafe(args[14], 0.0)
+        )
+    );
 
     static {
-        LENIENT_PARSER.declareLong(constructorArg(), NUM_PAGES);
-        LENIENT_PARSER.declareLong(constructorArg(), NUM_INPUT_DOCUMENTS);
-        LENIENT_PARSER.declareLong(constructorArg(), NUM_OUTPUT_DOCUMENTS);
-        LENIENT_PARSER.declareLong(constructorArg(), NUM_INVOCATIONS);
-        LENIENT_PARSER.declareLong(constructorArg(), INDEX_TIME_IN_MS);
-        LENIENT_PARSER.declareLong(constructorArg(), SEARCH_TIME_IN_MS);
-        LENIENT_PARSER.declareLong(constructorArg(), INDEX_TOTAL);
-        LENIENT_PARSER.declareLong(constructorArg(), SEARCH_TOTAL);
-        LENIENT_PARSER.declareLong(constructorArg(), INDEX_FAILURES);
-        LENIENT_PARSER.declareLong(constructorArg(), SEARCH_FAILURES);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), NUM_PAGES);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), NUM_INPUT_DOCUMENTS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), NUM_OUTPUT_DOCUMENTS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), NUM_INVOCATIONS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), INDEX_TIME_IN_MS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), SEARCH_TIME_IN_MS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), PROCESSING_TIME_IN_MS);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), INDEX_TOTAL);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), SEARCH_TOTAL);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), PROCESSING_TOTAL);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), INDEX_FAILURES);
+        LENIENT_PARSER.declareLong(optionalConstructorArg(), SEARCH_FAILURES);
         LENIENT_PARSER.declareDouble(optionalConstructorArg(), EXPONENTIAL_AVG_CHECKPOINT_DURATION_MS);
         LENIENT_PARSER.declareDouble(optionalConstructorArg(), EXPONENTIAL_AVG_DOCUMENTS_INDEXED);
         LENIENT_PARSER.declareDouble(optionalConstructorArg(), EXPONENTIAL_AVG_DOCUMENTS_PROCESSED);
@@ -67,16 +83,40 @@ public static TransformIndexerStats fromXContent(XContentParser parser) throws I
     private final double expAvgDocumentsIndexed;
     private final double expAvgDocumentsProcessed;
 
-    public TransformIndexerStats(long numPages, long numInputDocuments, long numOuputDocuments,
-                                 long numInvocations, long indexTime, long searchTime,
-                                 long indexTotal, long searchTotal, long indexFailures, long searchFailures,
-                                 Double expAvgCheckpointDurationMs, Double expAvgDocumentsIndexed,
-                                 Double expAvgDocumentsProcessed) {
-        super(numPages, numInputDocuments, numOuputDocuments, numInvocations, indexTime, searchTime,
-                indexTotal, searchTotal, indexFailures, searchFailures);
-        this.expAvgCheckpointDurationMs = expAvgCheckpointDurationMs == null ? 0.0 : expAvgCheckpointDurationMs;
-        this.expAvgDocumentsIndexed = expAvgDocumentsIndexed == null ? 0.0 : expAvgDocumentsIndexed;
-        this.expAvgDocumentsProcessed = expAvgDocumentsProcessed == null ? 0.0 : expAvgDocumentsProcessed;
+    public TransformIndexerStats(
+        long numPages,
+        long numInputDocuments,
+        long numOuputDocuments,
+        long numInvocations,
+        long indexTime,
+        long searchTime,
+        long processingTime,
+        long indexTotal,
+        long searchTotal,
+        long processingTotal,
+        long indexFailures,
+        long searchFailures,
+        double expAvgCheckpointDurationMs,
+        double expAvgDocumentsIndexed,
+        double expAvgDocumentsProcessed
+    ) {
+        super(
+            numPages,
+            numInputDocuments,
+            numOuputDocuments,
+            numInvocations,
+            indexTime,
+            searchTime,
+            processingTime,
+            indexTotal,
+            searchTotal,
+            processingTotal,
+            indexFailures,
+            searchFailures
+        );
+        this.expAvgCheckpointDurationMs = expAvgCheckpointDurationMs;
+        this.expAvgDocumentsIndexed = expAvgDocumentsIndexed;
+        this.expAvgDocumentsProcessed = expAvgDocumentsProcessed;
     }
 
     public double getExpAvgCheckpointDurationMs() {
@@ -109,19 +149,44 @@ public boolean equals(Object other) {
             && Objects.equals(this.numInvocations, that.numInvocations)
             && Objects.equals(this.indexTime, that.indexTime)
             && Objects.equals(this.searchTime, that.searchTime)
+            && Objects.equals(this.processingTime, that.processingTime)
             && Objects.equals(this.indexFailures, that.indexFailures)
             && Objects.equals(this.searchFailures, that.searchFailures)
             && Objects.equals(this.indexTotal, that.indexTotal)
             && Objects.equals(this.searchTotal, that.searchTotal)
+            && Objects.equals(this.processingTotal, that.processingTotal)
             && Objects.equals(this.expAvgCheckpointDurationMs, that.expAvgCheckpointDurationMs)
             && Objects.equals(this.expAvgDocumentsIndexed, that.expAvgDocumentsIndexed)
             && Objects.equals(this.expAvgDocumentsProcessed, that.expAvgDocumentsProcessed);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(numPages, numInputDocuments, numOuputDocuments, numInvocations,
-            indexTime, searchTime, indexFailures, searchFailures, indexTotal, searchTotal,
-            expAvgCheckpointDurationMs, expAvgDocumentsIndexed, expAvgDocumentsProcessed);
+        return Objects.hash(
+            numPages,
+            numInputDocuments,
+            numOuputDocuments,
+            numInvocations,
+            indexTime,
+            searchTime,
+            processingTime,
+            indexFailures,
+            searchFailures,
+            indexTotal,
+            searchTotal,
+            processingTotal,
+            expAvgCheckpointDurationMs,
+            expAvgDocumentsIndexed,
+            expAvgDocumentsProcessed
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T unboxSafe(Object l, T default_value) {
+        if (l == null) {
+            return default_value;
+        } else {
+            return (T) l;
+        }
     }
 }

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

@@ -441,6 +441,8 @@ public void testGetStats() throws Exception {
             0L,
             0L,
             0L,
+            0L,
+            0L,
             0.0,
             0.0,
             0.0);

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

@@ -64,8 +64,9 @@ private GetRollupJobResponse createTestInstance() {
 
     private RollupIndexerJobStats randomStats() {
         return new RollupIndexerJobStats(randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
-            randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
-            randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong());
+            randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
+            randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong(),
+            randomNonNegativeLong());
     }
 
     private RollupJobStatus randomStatus() {
@@ -120,6 +121,8 @@ public void toXContent(RollupIndexerJobStats stats, XContentBuilder builder, ToX
         builder.field(IndexerJobStats.SEARCH_TIME_IN_MS.getPreferredName(), stats.getSearchTime());
         builder.field(IndexerJobStats.SEARCH_TOTAL.getPreferredName(), stats.getSearchTotal());
         builder.field(IndexerJobStats.SEARCH_FAILURES.getPreferredName(), stats.getSearchFailures());
+        builder.field(IndexerJobStats.PROCESSING_TIME_IN_MS.getPreferredName(), stats.getProcessingTime());
+        builder.field(IndexerJobStats.PROCESSING_TOTAL.getPreferredName(), stats.getProcessingTotal());
         builder.endObject();
     }