fix: add additional checks for initializing the stats job collector to minimize jvm usage

Signed-off-by: Pavan Yekbote <pybot@amazon.com>

Author: pyek-bot

plugin/src/main/java/org/opensearch/ml/cluster/MLCommonsClusterEventListener.java

@@ -5,7 +5,7 @@
 
 package org.opensearch.ml.cluster;
 
-import static org.opensearch.ml.common.CommonValue.TASK_POLLING_JOB_INDEX;
+import static org.opensearch.ml.common.CommonValue.ML_JOBS_INDEX;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_MONITORING_REQUEST_COUNT;
 
 import java.util.List;
@@ -39,6 +39,7 @@ public class MLCommonsClusterEventListener implements ClusterStateListener {
     private final MLModelAutoReDeployer mlModelAutoReDeployer;
     private final Client client;
     private final MLFeatureEnabledSetting mlFeatureEnabledSetting;
+    private boolean startedStatsJob;
 
     public MLCommonsClusterEventListener(
         ClusterService clusterService,
@@ -90,12 +91,11 @@ public void clusterChanged(ClusterChangedEvent event) {
          */
         for (DiscoveryNode node : state.nodes()) {
             if (node.isDataNode() && node.getVersion().onOrAfter(Version.V_3_1_0)) {
-                if (mlFeatureEnabledSetting.isMetricCollectionEnabled() && mlFeatureEnabledSetting.isStaticMetricCollectionEnabled()) {
-                    mlTaskManager.indexStatsCollectorJob(true);
-                }
-
-                if (clusterService.state().getMetadata().hasIndex(TASK_POLLING_JOB_INDEX)) {
-                    mlTaskManager.startTaskPollingJob();
+                if (mlFeatureEnabledSetting.isMetricCollectionEnabled() && mlFeatureEnabledSetting.isStaticMetricCollectionEnabled()
+                        && !clusterService.state().getMetadata().hasIndex(ML_JOBS_INDEX) && !this.startedStatsJob) {
+                        mlTaskManager.indexStatsCollectorJob(true);
+                        // using this variable in case if same node has a cluster state change event and the state is not updated yet
+                        this.startedStatsJob = true;
                 }
 
                 break;

plugin/src/main/java/org/opensearch/ml/task/MLTaskManager.java

@@ -577,6 +577,10 @@ public void onStaticMetricCollectionEnabledChanged(boolean isEnabled) {
     }
 
     public void indexStatsCollectorJob(boolean enabled) {
+        if (this.statsCollectorJobStarted && enabled) {
+            return;
+        }
+
         try {
             MLJobParameter jobParameter = new MLJobParameter(
                 MLJobType.STATS_COLLECTOR.name(),
@@ -593,7 +597,7 @@ public void indexStatsCollectorJob(boolean enabled) {
                 .source(jobParameter.toXContent(JsonXContent.contentBuilder(), null))
                 .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
 
-            indexJob(indexRequest, MLJobType.STATS_COLLECTOR, () -> {});
+            indexJob(indexRequest, MLJobType.STATS_COLLECTOR, () -> this.statsCollectorJobStarted = enabled);
         } catch (IOException e) {
             log.error("Failed to index stats collection job", e);
         }