Merge branch '2.x' into 2.x

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

@@ -16,7 +16,8 @@ public enum FunctionName {
     ANOMALY_LOCALIZATION,
     RCF_SUMMARIZE,
     LOGISTIC_REGRESSION,
-    TEXT_EMBEDDING;
+    TEXT_EMBEDDING,
+    METRICS_CORRELATION;
 
     public static FunctionName from(String value) {
         try {

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

@@ -19,6 +19,7 @@
 import org.opensearch.ml.common.model.MLModelFormat;
 import org.opensearch.ml.common.model.MLModelState;
 import org.opensearch.ml.common.model.TextEmbeddingModelConfig;
+import org.opensearch.ml.common.model.MetricsCorrelationModelConfig;
 
 import java.io.IOException;
 import java.time.Instant;
@@ -302,7 +303,7 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params par
         return builder;
     }
 
-    public static MLModel parse(XContentParser parser) throws IOException {
+    public static MLModel parse(XContentParser parser, String algorithmName) throws IOException {
         String name = null;
         FunctionName algorithm = null;
         String version = null;
@@ -385,7 +386,11 @@ public static MLModel parse(XContentParser parser) throws IOException {
                     modelContentHash = parser.text();
                     break;
                 case MODEL_CONFIG_FIELD:
-                    modelConfig = TextEmbeddingModelConfig.parse(parser);
+                    if (FunctionName.METRICS_CORRELATION.name().equals(algorithmName)) {
+                        modelConfig = MetricsCorrelationModelConfig.parse(parser);
+                    } else {
+                        modelConfig = TextEmbeddingModelConfig.parse(parser);
+                    }
                     break;
                 case PLANNING_WORKER_NODE_COUNT_FIELD:
                     planningWorkerNodeCount = parser.intValue();

common/src/main/java/org/opensearch/ml/common/exception/ExecuteException.java

@@ -0,0 +1,7 @@
+package org.opensearch.ml.common.exception;
+
+public class ExecuteException extends MLException{
+    public ExecuteException(String msg) { super(msg); }
+    public ExecuteException(Throwable cause) { super(cause); }
+    public ExecuteException(String msg, Throwable cause) { super(msg, cause); }
+}

common/src/main/java/org/opensearch/ml/common/input/execute/metricscorrelation/MetricsCorrelationInput.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ml.common.input.execute.metricscorrelation;
+
+import lombok.Builder;
+import lombok.Data;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.core.ParseField;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.ml.common.FunctionName;
+import org.opensearch.ml.common.annotation.ExecuteInput;
+import org.opensearch.ml.common.input.Input;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+
+@ExecuteInput(algorithms={FunctionName.METRICS_CORRELATION})
+@Data
+public class MetricsCorrelationInput implements Input {
+    public static final String PARSE_FIELD_NAME = FunctionName.METRICS_CORRELATION.name();
+    public static final NamedXContentRegistry.Entry XCONTENT_REGISTRY = new NamedXContentRegistry.Entry(
+            Input.class,
+            new ParseField(PARSE_FIELD_NAME),
+            it -> parse(it)
+    );
+
+    public static final String METRICS_FIELD = "metrics";
+
+    List<float[]> inputData;
+
+    @Builder
+    public MetricsCorrelationInput(List<float[]> inputData) {
+        if (inputData == null || inputData.size() == 0) {
+            throw new IllegalArgumentException("empty input data");
+        }
+        int expectedLength = inputData.get(0).length;
+        for (int i = 1; i < inputData.size(); i++) {
+            float[] array = inputData.get(i);
+            if (array.length != expectedLength) {
+                // found an array with different length
+                throw new IllegalArgumentException("All the input metrics sizes should be same");
+            }
+        }
+        if (inputData.size() >= expectedLength) {
+            throw new IllegalArgumentException("The number of metrics to correlate must be smaller than the length of each time series.");
+        }
+        this.inputData = inputData;
+    }
+
+    public MetricsCorrelationInput(StreamInput in) throws IOException {
+        this.inputData = in.readList(StreamInput::readFloatArray);
+    }
+
+    public static MetricsCorrelationInput parse(XContentParser parser) throws IOException {
+        List<float[]> inputData = new ArrayList<>();
+
+        ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser);
+        while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+            String fieldName = parser.currentName();
+            parser.nextToken();
+
+            switch (fieldName) {
+                case METRICS_FIELD:
+                    ensureExpectedToken(XContentParser.Token.START_ARRAY, parser.currentToken(), parser);
+                    while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                        ensureExpectedToken(XContentParser.Token.START_ARRAY, parser.currentToken(), parser);
+                        List<Float> inputItem = new ArrayList<>();
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                            inputItem.add(parser.floatValue());
+                        }
+                        float[] floatArray = new float[inputItem.size()];
+                        int i = 0;
+
+                        for (Float f : inputItem) {
+                            floatArray[i++] = (f != null ? f : Float.NaN);
+                        }
+                        inputData.add(floatArray);
+                    }
+                    break;
+                default:
+                    parser.skipChildren();
+                    break;
+            }
+        }
+        return new MetricsCorrelationInput(inputData);
+    }
+
+    @Override
+    public FunctionName getFunctionName() {
+        return FunctionName.METRICS_CORRELATION;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeCollection(this.inputData, StreamOutput::writeFloatArray);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(METRICS_FIELD, inputData);
+        builder.endObject();
+        return builder;
+    }
+}

common/src/main/java/org/opensearch/ml/common/model/MetricsCorrelationModelConfig.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ml.common.model;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.ml.common.FunctionName;
+
+import java.io.IOException;
+import java.util.Locale;
+
+import static org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+
+@Setter
+@Getter
+public class MetricsCorrelationModelConfig extends MLModelConfig {
+
+    public static final String PARSE_FIELD_NAME = FunctionName.METRICS_CORRELATION.name();
+
+    @Builder(toBuilder = true)
+    public MetricsCorrelationModelConfig(String modelType, String allConfig) {
+        super(modelType, allConfig);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return PARSE_FIELD_NAME;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (modelType != null) {
+            builder.field(MODEL_TYPE_FIELD, modelType);
+        }
+        if (allConfig != null) {
+            builder.field(ALL_CONFIG_FIELD, allConfig);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public static MetricsCorrelationModelConfig parse(XContentParser parser) throws IOException {
+        String modelType = null;
+        String allConfig = null;
+
+        ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser);
+        while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+            String fieldName = parser.currentName();
+            parser.nextToken();
+
+            switch (fieldName) {
+                case MODEL_TYPE_FIELD:
+                    modelType = parser.text();
+                    break;
+                case ALL_CONFIG_FIELD:
+                    allConfig = parser.text();
+                    break;
+                default:
+                    parser.skipChildren();
+                    break;
+            }
+        }
+        return new MetricsCorrelationModelConfig(modelType, allConfig);
+    }
+}

common/src/main/java/org/opensearch/ml/common/output/MLOutputType.java

@@ -9,5 +9,6 @@ public enum MLOutputType {
     TRAINING,
     PREDICTION,
     SAMPLE_ALGO,
-    MODEL_TENSOR
+    MODEL_TENSOR,
+    MCORR_TENSOR
 }

common/src/main/java/org/opensearch/ml/common/output/execute/metrics_correlation/MCorrModelTensor.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ml.common.output.execute.metrics_correlation;
+
+import lombok.Builder;
+import lombok.Data;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.io.stream.Writeable;
+import org.opensearch.core.xcontent.ToXContentObject;
+import org.opensearch.core.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+@Data
+public class MCorrModelTensor implements Writeable, ToXContentObject {
+
+    public static final String EVENT_WINDOW = "event_window";
+    public static final String EVENT_PATTERN = "event_pattern";
+    public static final String SUSPECTED_METRICS = "suspected_metrics";
+
+    private float[] event_window;
+    private float[] event_pattern;
+    private long[] suspected_metrics;
+
+    @Builder
+    public MCorrModelTensor(float[] event_window, float[] event_pattern, long[] suspected_metrics) {
+        this.event_window = event_window;
+        this.event_pattern = event_pattern;
+        this.suspected_metrics = suspected_metrics;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (event_window != null) {
+            builder.field(EVENT_WINDOW, event_window);
+        }
+        if (event_pattern != null) {
+            builder.field(EVENT_PATTERN, event_pattern);
+        }
+        if (suspected_metrics != null) {
+            builder.field(SUSPECTED_METRICS, suspected_metrics);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public MCorrModelTensor(StreamInput in) throws IOException {
+        if (in.readBoolean()) {
+            this.event_window = in.readFloatArray();
+        }
+        if (in.readBoolean()) {
+            this.event_pattern = in.readFloatArray();
+        }
+        if (in.readBoolean()) {
+            this.suspected_metrics = in.readLongArray();
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        if (event_window != null) {
+            out.writeBoolean(true);
+            out.writeFloatArray(event_window);
+        } else {
+            out.writeBoolean(false);
+        }
+
+        if (event_pattern != null) {
+            out.writeBoolean(true);
+            out.writeFloatArray(event_pattern);
+        } else {
+            out.writeBoolean(false);
+        }
+        if (suspected_metrics != null) {
+            out.writeBoolean(true);
+            out.writeLongArray(suspected_metrics);
+        } else {
+            out.writeBoolean(false);
+        }
+    }
+}