Metrics framework integration with ml-commons (#3661)

Author: pyek-bot

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

@@ -44,9 +44,12 @@ 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";
     public static final String MCP_TOOLS_INDEX = ".plugins-ml-mcp-tools";
+    // index created in 3.1 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

@@ -342,4 +342,12 @@ private MLCommonsSettings() {}
     /** This setting sets the remote metadata service name */
     public static final Setting<String> REMOTE_METADATA_SERVICE_NAME = Setting
         .simpleString("plugins.ml_commons." + REMOTE_METADATA_SERVICE_NAME_KEY, Setting.Property.NodeScope, Setting.Property.Final);
+
+    // Feature flag for enabling telemetry metric collection via metrics framework
+    public static final Setting<Boolean> ML_COMMONS_METRIC_COLLECTION_ENABLED = Setting
+        .boolSetting("plugins.ml_commons.metrics_collection_enabled", false, Setting.Property.NodeScope, Setting.Property.Dynamic);
+
+    // Feature flag for enabling telemetry static metric collection job -- MLStatsJobProcessor
+    public static final Setting<Boolean> ML_COMMONS_STATIC_METRIC_COLLECTION_ENABLED = Setting
+        .boolSetting("plugins.ml_commons.metrics_static_collection_enabled", false, Setting.Property.NodeScope, Setting.Property.Dynamic);
 }

common/src/main/java/org/opensearch/ml/common/settings/MLCommonsSettings.java

@@ -12,10 +12,13 @@
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_CONTROLLER_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_LOCAL_MODEL_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_MCP_SERVER_ENABLED;
+import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_METRIC_COLLECTION_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_MULTI_TENANCY_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_OFFLINE_BATCH_INFERENCE_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_OFFLINE_BATCH_INGESTION_ENABLED;
+import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_RAG_PIPELINE_FEATURE_ENABLED;
 import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_REMOTE_INFERENCE_ENABLED;
+import static org.opensearch.ml.common.settings.MLCommonsSettings.ML_COMMONS_STATIC_METRIC_COLLECTION_ENABLED;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -43,6 +46,11 @@ public class MLFeatureEnabledSetting {
 
     private volatile Boolean isMcpServerEnabled;
 
+    private volatile Boolean isRagSearchPipelineEnabled;
+
+    private volatile Boolean isMetricCollectionEnabled;
+    private volatile Boolean isStaticMetricCollectionEnabled;
+
     private final List<SettingsChangeListener> listeners = new ArrayList<>();
 
     public MLFeatureEnabledSetting(ClusterService clusterService, Settings settings) {
@@ -55,6 +63,9 @@ public MLFeatureEnabledSetting(ClusterService clusterService, Settings settings)
         isBatchInferenceEnabled = ML_COMMONS_OFFLINE_BATCH_INFERENCE_ENABLED.get(settings);
         isMultiTenancyEnabled = ML_COMMONS_MULTI_TENANCY_ENABLED.get(settings);
         isMcpServerEnabled = ML_COMMONS_MCP_SERVER_ENABLED.get(settings);
+        isRagSearchPipelineEnabled = ML_COMMONS_RAG_PIPELINE_FEATURE_ENABLED.get(settings);
+        isMetricCollectionEnabled = ML_COMMONS_METRIC_COLLECTION_ENABLED.get(settings);
+        isStaticMetricCollectionEnabled = ML_COMMONS_STATIC_METRIC_COLLECTION_ENABLED.get(settings);
 
         clusterService
             .getClusterSettings()
@@ -74,6 +85,15 @@ public MLFeatureEnabledSetting(ClusterService clusterService, Settings settings)
             .getClusterSettings()
             .addSettingsUpdateConsumer(ML_COMMONS_OFFLINE_BATCH_INFERENCE_ENABLED, it -> isBatchInferenceEnabled = it);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(ML_COMMONS_MCP_SERVER_ENABLED, it -> isMcpServerEnabled = it);
+        clusterService
+            .getClusterSettings()
+            .addSettingsUpdateConsumer(MLCommonsSettings.ML_COMMONS_RAG_PIPELINE_FEATURE_ENABLED, it -> isRagSearchPipelineEnabled = it);
+        clusterService
+            .getClusterSettings()
+            .addSettingsUpdateConsumer(ML_COMMONS_METRIC_COLLECTION_ENABLED, it -> isMetricCollectionEnabled = it);
+        clusterService
+            .getClusterSettings()
+            .addSettingsUpdateConsumer(ML_COMMONS_STATIC_METRIC_COLLECTION_ENABLED, it -> isStaticMetricCollectionEnabled = it);
     }
 
     /**
@@ -148,6 +168,22 @@ public void addListener(SettingsChangeListener listener) {
         listeners.add(listener);
     }
 
+    /**
+     * Whether the rag search pipeline feature is enabled. If disabled, APIs in ml-commons will block rag search pipeline.
+     * @return whether the feature is enabled.
+     */
+    public boolean isRagSearchPipelineEnabled() {
+        return isRagSearchPipelineEnabled;
+    }
+
+    public boolean isMetricCollectionEnabled() {
+        return isMetricCollectionEnabled;
+    }
+
+    public boolean isStaticMetricCollectionEnabled() {
+        return isStaticMetricCollectionEnabled;
+    }
+
     @VisibleForTesting
     public void notifyMultiTenancyListeners(boolean isEnabled) {
         for (SettingsChangeListener listener : listeners) {