feature: job extension and static job

Author: pyek-bot

common/src/main/java/org/opensearch/ml/common/CommonValue.java

@@ -44,8 +44,11 @@ public class CommonValue {
     public static final String ML_MEMORY_META_INDEX = ".plugins-ml-memory-meta";
     public static final String ML_MEMORY_MESSAGE_INDEX = ".plugins-ml-memory-message";
     public static final String ML_STOP_WORDS_INDEX = ".plugins-ml-stop-words";
+    // index used in 2.19 to track MlTaskBatchUpdate task
     public static final String TASK_POLLING_JOB_INDEX = ".ml_commons_task_polling_job";
     public static final String MCP_SESSION_MANAGEMENT_INDEX = ".plugins-ml-mcp-session-management";
+    // index created in 3.0 to track all ml jobs created via job scheduler
+    public static final String ML_JOBS_INDEX = ".plugins-ml-jobs";
     public static final Set<String> stopWordsIndices = ImmutableSet.of(".plugins-ml-stop-words");
     public static final String TOOL_PARAMETERS_PREFIX = "tools.parameters.";
 

common/src/main/java/org/opensearch/ml/common/MLModel.java

@@ -40,6 +40,7 @@
 import org.opensearch.ml.common.model.MetricsCorrelationModelConfig;
 import org.opensearch.ml.common.model.QuestionAnsweringModelConfig;
 import org.opensearch.ml.common.model.TextEmbeddingModelConfig;
+import org.opensearch.telemetry.metrics.tags.Tags;
 
 import lombok.Builder;
 import lombok.Getter;
@@ -745,4 +746,15 @@ public static MLModel fromStream(StreamInput in) throws IOException {
         return new MLModel(in);
     }
 
+    public Tags getModelTags() {
+        return Tags
+            .create()
+            .addTag("type", algorithm == FunctionName.REMOTE ? "remote" : "local")
+            .addTag("provider", algorithm == FunctionName.REMOTE ? getRemoteModelType() : algorithm.name());
+    }
+
+    private String getRemoteModelType() {
+        return "remote_sub_tye";
+    }
+
 }

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

@@ -8,18 +8,31 @@
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_SYNC_UP_JOB_INTERVAL_IN_SECONDS;
 import static org.opensearch.ml.plugin.MachineLearningPlugin.GENERAL_THREAD_POOL;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 
+import org.opensearch.action.index.IndexRequest;
+import org.opensearch.action.support.WriteRequest;
 import org.opensearch.cluster.LocalNodeClusterManagerListener;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.lifecycle.LifecycleListener;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.xcontent.json.JsonXContent;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule;
 import org.opensearch.ml.autoredeploy.MLModelAutoReDeployer;
 import org.opensearch.ml.common.settings.MLFeatureEnabledSetting;
 import org.opensearch.ml.engine.encryptor.Encryptor;
 import org.opensearch.ml.engine.indices.MLIndicesHandler;
+import org.opensearch.ml.common.CommonValue;
+import org.opensearch.ml.engine.encryptor.Encryptor;
+import org.opensearch.ml.engine.indices.MLIndicesHandler;
+import org.opensearch.ml.jobs.MLJobParameter;
+import org.opensearch.ml.jobs.MLJobType;
+import org.opensearch.ml.settings.MLFeatureEnabledSetting;
 import org.opensearch.remote.metadata.client.SdkClient;
 import org.opensearch.threadpool.Scheduler;
 import org.opensearch.threadpool.ThreadPool;
@@ -95,6 +108,40 @@ public void onClusterManager() {
                 TimeValue.timeValueSeconds(jobInterval),
                 GENERAL_THREAD_POOL
             );
+//        startStatsCollectorJob();
+    }
+
+    public void startStatsCollectorJob() {
+        try {
+            int intervalInMinutes = 5;
+            Long lockDurationSeconds = 20L;
+
+            MLJobParameter jobParameter = new MLJobParameter(
+                MLJobType.STATS_COLLECTOR.name(),
+                new IntervalSchedule(Instant.now(), intervalInMinutes, ChronoUnit.MINUTES),
+                lockDurationSeconds,
+                null,
+                MLJobType.STATS_COLLECTOR
+            );
+
+            IndexRequest indexRequest = new IndexRequest()
+                .index(CommonValue.ML_JOBS_INDEX)
+                .id(MLJobType.STATS_COLLECTOR.name())
+                .source(jobParameter.toXContent(JsonXContent.contentBuilder(), null))
+                .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+
+            client
+                .index(
+                    indexRequest,
+                    ActionListener
+                        .wrap(
+                            r -> log.info("Indexed ml stats collection job successfully"),
+                            e -> log.error("Failed to index stats collection job", e)
+                        )
+                );
+        } catch (IOException e) {
+            log.error("Failed to index stats collection job", e);
+        }
     }
 
     private void startSyncModelRoutingCron() {