Warn when checkpoint duration is longer than frequency

Author: prwhelan

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformContext.java

@@ -48,6 +48,9 @@ public interface Listener {
     private volatile boolean shouldRecreateDestinationIndex = false;
     private volatile AuthorizationState authState;
     private volatile int pageSize = 0;
+    private AtomicInteger skippedCheckpoints = new AtomicInteger(0);
+    private AtomicInteger totalTimeWasGreaterThanFrequency = new AtomicInteger(0);
+    private volatile Instant lastTotalTimeRecorded;
 
     /**
      * If the destination index is blocked (e.g. during a reindex), the Transform will fail to write to it.
@@ -273,6 +276,36 @@ boolean doesNotHaveFailures() {
         return getFailureCount() == 0 && getStatePersistenceFailureCount() == 0 && getStartUpFailureCount() == 0;
     }
 
+    int skippedCheckpoints() {
+        return skippedCheckpoints.get();
+    }
+
+    void checkpointSkipped() {
+        skippedCheckpoints.incrementAndGet();
+    }
+
+    void resetSkippedCheckpoint() {
+        skippedCheckpoints.set(0);
+    }
+
+    int numberOfTimesTotalTimeWasGreaterThanFrequency() {
+        return totalTimeWasGreaterThanFrequency.get();
+    }
+
+    void totalTimeWasGreaterThanFrequency() {
+        totalTimeWasGreaterThanFrequency.incrementAndGet();
+        lastTotalTimeRecorded = Instant.now();
+    }
+
+    void totalTimeWasLessThanFrequency() {
+        totalTimeWasGreaterThanFrequency.decrementAndGet();
+        lastTotalTimeRecorded = Instant.now();
+    }
+
+    Instant lastTotalTimeRecorded() {
+        return lastTotalTimeRecorded;
+    }
+
     void shutdown() {
         taskListener.shutdown();
     }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformHealthChecker.java

@@ -39,7 +39,9 @@ public enum IssueType {
         TRANSFORM_TASK_FAILED("Transform task state is [failed]"),
         TRANSFORM_INDEXER_FAILED("Transform indexer failed"),
         TRANSFORM_INTERNAL_STATE_UPDATE_FAILED("Task encountered failures updating internal state"),
-        TRANSFORM_STARTUP_FAILED("Transform task is automatically retrying its startup process");
+        TRANSFORM_STARTUP_FAILED("Transform task is automatically retrying its startup process"),
+        TRANSFORM_IS_SLOW("Transform task is taking longer than its configured frequency. Data processing will be delayed."),
+        TRANSFORM_WAS_SLOW("Transform task was recently taking longer than its configured frequency, but it is recovering.");
 
         private final String issue;
 
@@ -163,6 +165,30 @@ public static TransformHealth checkTransform(TransformTask transformTask, @Nulla
             );
         }
 
+        if (transformContext.numberOfTimesTotalTimeWasGreaterThanFrequency() > 0) {
+            if (HealthStatus.RED.equals(maxStatus) == false) {
+                maxStatus = HealthStatus.YELLOW;
+            }
+            var checkpoints = transformContext.skippedCheckpoints();
+            /*if (checkpoints > 0) {
+                issues.add(
+                    IssueType.TRANSFORM_IS_SLOW.newIssue(
+                        "Transform has skipped " + checkpoints + " checkpoints.",
+                        transformContext.numberOfTimesTotalTimeWasGreaterThanFrequency(),
+                        transformContext.lastTotalTimeRecorded()
+                    )
+                );
+            } else {
+                issues.add(
+                    IssueType.TRANSFORM_WAS_SLOW.newIssue(
+                        null,
+                        transformContext.numberOfTimesTotalTimeWasGreaterThanFrequency(),
+                        transformContext.lastTotalTimeRecorded()
+                    )
+                );
+            }*/
+        }
+
         return new TransformHealth(maxStatus, issues);
     }
 

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformIndexer.java

@@ -605,6 +605,14 @@ private void finalizeCheckpoint(ActionListener<Void> listener) {
 
                 long durationMs = System.currentTimeMillis() - lastCheckpoint.getTimestamp();
                 getStats().incrementCheckpointExponentialAverages(durationMs < 0 ? 0 : durationMs, docsIndexed, docsProcessed);
+
+                if (getConfig().getFrequency() != null) {
+                    if (durationMs > getConfig().getFrequency().getMillis()) {
+                        context.totalTimeWasGreaterThanFrequency();
+                    } else {
+                        context.totalTimeWasLessThanFrequency();
+                    }
+                }
             }
             if (shouldAuditOnFinish(checkpoint)) {
                 auditor.info(getJobId(), "Finished indexing for transform checkpoint [" + checkpoint + "].");
@@ -659,10 +667,19 @@ public boolean maybeTriggerAsyncJob(long now) {
             IndexerState indexerState = getState();
             if (IndexerState.INDEXING.equals(indexerState) || IndexerState.STOPPING.equals(indexerState)) {
                 logger.debug("[{}] indexer for transform has state [{}]. Ignoring trigger.", getJobId(), indexerState);
+                if (IndexerState.INDEXING.equals(indexerState)) {
+                    context.checkpointSkipped();
+                }
                 return false;
             }
 
-            return super.maybeTriggerAsyncJob(now);
+            var triggered = super.maybeTriggerAsyncJob(now);
+            if (triggered) {
+                context.resetSkippedCheckpoint();
+            } else {
+                context.checkpointSkipped();
+            }
+            return triggered;
         }
     }