Add external scheduler metrics (#827)

* Add external scheduler metrics Signed-off-by: Tomoyuki Morita <[email protected]> * Format code Signed-off-by: Tomoyuki Morita <[email protected]> --------- Signed-off-by: Tomoyuki Morita <[email protected]> Signed-off-by: Tomoyuki MORITA <[email protected]>
opensearch-project · Oct 31, 2024 · c590d29 · c590d29
1 parent a2a9838
commit c590d29
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/flint-core/src/main/java/org/opensearch/flint/core/metrics/MetricConstants.java b/flint-core/src/main/java/org/opensearch/flint/core/metrics/MetricConstants.java
@@ -160,6 +160,11 @@ public final class MetricConstants {
      */
     public static final String CHECKPOINT_DELETE_TIME_METRIC = "checkpoint.delete.processingTime";
 
+    /**
+     * Prefix for externalScheduler metrics
+     */
+    public static final String EXTERNAL_SCHEDULER_METRIC_PREFIX = "externalScheduler.";
+
     /**
      * Metric prefix for tracking the index state transitions
      */

diff --git a/...n/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobExternalSchedulingService.scala b/...n/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobExternalSchedulingService.scala
@@ -10,6 +10,7 @@ import java.time.Instant
 import org.opensearch.flint.common.metadata.log.FlintMetadataLogEntry.IndexState
 import org.opensearch.flint.common.scheduler.AsyncQueryScheduler
 import org.opensearch.flint.common.scheduler.model.{AsyncQuerySchedulerRequest, LangType}
+import org.opensearch.flint.core.metrics.{MetricConstants, MetricsUtil}
 import org.opensearch.flint.core.storage.OpenSearchClientUtils
 import org.opensearch.flint.spark.FlintSparkIndex
 import org.opensearch.flint.spark.refresh.util.RefreshMetricsAspect
@@ -83,8 +84,16 @@ class FlintSparkJobExternalSchedulingService(
         case AsyncQuerySchedulerAction.REMOVE => flintAsyncQueryScheduler.removeJob(request)
         case _ => throw new IllegalArgumentException(s"Unsupported action: $action")
       }
+      addExternalSchedulerMetrics(action)
 
       None // Return None for all cases
     }
   }
+
+  private def addExternalSchedulerMetrics(action: AsyncQuerySchedulerAction): Unit = {
+    val actionName = action.name().toLowerCase()
+    MetricsUtil.addHistoricGauge(
+      MetricConstants.EXTERNAL_SCHEDULER_METRIC_PREFIX + actionName + ".count",
+      1)
+  }
 }