Custom Anomaly Detection Model with ML Commons – Real-time Inference in Ingest Pipelines?

[Yash-Patil-2004]

Summary

I am developing multiple machine learning models as a Flask-based API that detect various log anomalies, such as:

• Unusual status codes,
• Spikes in error rates,
• Error categorization related to databases.

The Flask app exposes a /predict endpoint, which accepts log data and returns predictions. I want to integrate this setup into OpenSearch to enable real-time anomaly detection.

---

Intended Architecture

I aim to:

1. Create an OpenSearch ingest pipeline that:

   • Sends incoming log data to the /predict endpoint of my external model.
   • Receives predictions (anomalies), and
   • Routes anomaly logs to a separate index for visualization and dashboarding.

2. Reduce infrastructure costs by enabling real-time anomaly detection during ingestion, rather than batch processing.

---

Current Issue

While exploring ML Commons and the ml_inference ingest processor, I noticed:

• It supports specific types of models: text embedding models, sparse encoding, cross-encoders, and question-answering.
• It appears to only work with registered ML Commons models, not arbitrary external APIs like my Flask /predict.
• There is no clear support for invoking a custom anomaly detection model hosted externally in real-time from within an ingest pipeline.
• Even if I allow private IP's into the cluster setting there are errors in input/output mapping int the ingestion pipeline and also with accessing that endpoint.

---

Questions

1. Is it currently possible to:

   • Register and invoke a custom anomaly detection model via ML Commons (hosted externally),
   • And use it within an ingest pipeline to enrich incoming documents in real time?

2. Which architecture is more suitable and cost-effective for this use case:

   • Option A: Deploying the model on OpenSearch and using ML Commons / ingest pipeline,
   • Option B: Hosting the model externally and using a scheduled job to:
     • Fetch logs from the past 5 minutes,
     • Run inference,
     • Index anomalies into a different index.

---

My Goal

To build a robust, low-latency anomaly detection pipeline integrated with OpenSearch for logs analysis and dashboarding—while keeping infra cost and maintenance complexity low.

Any guidance on the supported approach or roadmap plans for ML Commons and ingest pipelines would be highly appreciated.

Thanks in advance!

[ylwu-amzn]

I think you can do these

1. create a remote model with connector, refer to https://docs.opensearch.org/docs/latest/ml-commons-plugin/remote-models/connectors/ . If your external hosted model URL using private IP, you should enable this setting "plugins.ml_commons.connector.private_ip_enabled": true
2. create ingest pipeline with ML inference processor
3. We don't have scheduled job support in ml-commons now. This sounds a new requirement.

@mingshl , can you help take a look ?

[mingshl]

you can try host the model remotely in SageMaker, here is a sample tutorial that I am working on to host a model in SageMaker from Huggingface #3966

Feel free to contribute to the tutorial folder of using ml inference processor https://github.com/mingshl/ml-commons/blob/main-multi-language-tutorials/docs/tutorials/ml_inference/semantic_search/semantic_search_for_long_document.md

[Yash-Patil-2004]

Connector Creation and Model Registration
Create Connector Request

POST /_plugins/_ml/connectors/_create
{
  "name": "status-anomaly-detector-connector-final",
  "description": "Connector for the external Flask Status Anomaly Detector model (final payload)",
  "version": 1,
  "protocol": "http",
  "parameters": {
    "endpoint": "172.19.0.3:5000/predict"
  },
  "credential": {
    "security_type": "NO_AUTH",
    "value": "{}"
  },
  "actions": [
    {
      "action_type": "predict",
      "method": "POST",
      "url": "http://${parameters.endpoint}",
      "headers": {
        "Content-Type": "application/json"
      },
      "request_body": """{
        "input": [
          {
            "features": {
              "log_text_input": "${parameters.input.log_text_input}",
              "http_status_input": ${parameters.input.http_status_input}
            }
          }
        ]
      }"""
    }
  ]
}

Connector Creation Response
JSON

{
"connector_id": "8y_N8pcBwfV9kd1470ai"
}
Register Model Request
JSON

POST /_plugins/_ml/models/_register
{
  "name": "status-anomaly-model",
  "description": "Remote Status Anomaly Detection model hosted by Flask",
  "function_name": "remote",
  "model_group_id": "XFI41JcB07ze-zorSuir",
  "connector_id": "8y_N8pcBwfV9kd1470ai"
}
Model Registration Response
JSON

{
  "task_id": "-S_O8pcBwfV9kd14cUYk",
  "status": "CREATED",
  "model_id": "-i_O8pcBwfV9kd14cUY5"
}
Deploy Model Request
JSON

POST /_plugins/_ml/models/-i_O8pcBwfV9kd14cUY5/_deploy
{
  "task_id": "AS_O8pcBwfV9kd143Ufo",
  "task_type": "DEPLOY_MODEL",
  "status": "COMPLETED"
}
Ingest Pipeline Configuration
Pipeline Definition
JSON

PUT _ingest/pipeline/status-anomaly-detection-pipeline-final
{
  "description": "Pipeline to detect unusual HTTP status codes using external ML model (Final Fix)",
  "processors": [
    {
      "set": {
        "field": "ml_temp_message",
        "value": "{{Event.Message}",
        "if": "ctx.Event?.Message != null",
        "override": true
      }
    },
    {
      "set": {
        "field": "ml_temp_log_type",
        "value": "{{Event.Type}}",
        "if": "ctx.Event?.Type != null",
        "override": true
      }
    },
    {
      "grok": {
        "field": "Event.Message",
        "patterns": [
          ".status=%{INT:status:long} ."
        ],
        "ignore_missing": true,
        "on_failure": [
          {
            "set": {
              "field": "error.message",
              "value": "Grok failed on Event.Message"
            }
          }
        ]
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": """
          String combinedText = "";
          if (ctx.ml_temp_log_type != null) {
            combinedText += ctx.ml_temp_log_type;
          }
          if (ctx.ml_temp_message != null) {
            if (combinedText.length() > 0) {
              combinedText += " - "; // Add a separator if both exist
            }
            combinedText += ctx.ml_temp_message;
          }
          if (combinedText.length() > 0) {
            ctx.ml_combined_text_input = combinedText;
          }
          // Create the single input object for ml_inference
          ctx.ml_inference_input_features = new HashMap();
          if (ctx.ml_combined_text_input != null) {
            ctx.ml_inference_input_features.put("log_text_input", ctx.ml_combined_text_input);
          }
          if (ctx.status != null) {
            ctx.ml_inference_input_features.put("http_status_input", ctx.status);
          }
        """,
        "if": "ctx.ml_temp_message != null || ctx.ml_temp_log_type != null || ctx.status != null"
      }
    },
    {
      "ml_inference": {
        "model_id": "-i_O8pcBwfV9kd14cUY5",
        "input_map": [
          {
            "input": "ml_inference_input_features"
          }
        ],
        "output_map": [
          {
            "prediction_output": "ml_inference_raw_response"
          }
        ],
        "if": "ctx.ml_inference_input_features != null && ctx.ml_inference_input_features.size() > 0",
        "ignore_failure": false
      }
    },
    {
      "script": {
        "lang": "painless",
        "source": """
          if (ctx.ml_inference_raw_response != null) {
            ctx.ml.anomaly_detection_results = new HashMap();
            ctx.ml.anomaly_detection_results.is_anomaly = ctx.ml_inference_raw_response.is_anomaly;
            ctx.ml.anomaly_detection_results.reason = ctx.ml_inference_raw_response.reason;
            ctx.ml.anomaly_detection_results.detected_status_code = ctx.ml_inference_raw_response.detected_status_code;
            ctx.ml.anomaly_detection_results.rrcf_score = ctx.ml_inference_raw_response.rrcf_score;
          }
        """,
        "if": "ctx.ml_inference_raw_response != null"
      }
    },
    {
      "remove": {
        "field": "ml_temp_message",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": "ml_temp_log_type",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": "ml_combined_text_input",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": "ml_inference_input_features",
        "ignore_failure": true
      }
    },
    {
      "remove": {
        "field": "ml_inference_raw_response",
        "ignore_failure": true
      }
    }
  ]
}
Simulate Pipeline Request
JSON

POST _ingest/pipeline/status-anomaly-detection-pipeline-final/_simulate?pretty&verbose=true
{
  "docs": [
    {
      "_index": "test_logs",
      "_id": "1",
      "_source": {
        "Event": {
          "Message": "INFO Request: Method=GET, Path=/api/data, status=200, duration=150ms",
          "Type": "webserver_access"
        },
        "timestamp": "2025-07-07T10:00:00Z",
        "service": "web-app"
      }
    },
    {
      "_index": "test_logs",
      "_id": "2",
      "_source": {
        "Event": {
          "Message": "ERROR Request: Method=POST, Path=/admin, status=503, duration=5000ms",
          "Type": "webserver_error"
        },
        "timestamp": "2025-07-07T10:01:00Z",
        "service": "web-app"
      }
    }
  ]
}

I followed these steps with exact input and output mappings and allowed the private IP, still I am getting an issue.

Is the ML inference too complex to map the input and output?

My model expects the input in the format:

JSON

{
  "input": [
    {
      "features": {
        "log_text_input": "...",
        "http_status_input": ...
      }
    }
  ]
}

And my logs which are coming to the index are like:

JSON

{
  "Caller": "database_driver/sql.go:123",
  "Event": {
    "Date": "2025-07-09T08:25:59.110+00:00Z",
    "Type": "WARN",
    "Message": "ERROR: Could not establish DB endpoint connection timeout. Ensure DB instance is running. (status=503) (duration=5.053s)"
  },
  "Host": {
    "HostName": "dev-db-server"
  },
  "Service": {
    "Name": "user-auth-service",
    "Version": "0.1.0",
    "Owner": "Database Team",
    "BuildNumber": "0.1.0-52",
    "CommitId": "fb0569225e4361ca685094768ab715354cfbd0db",
    "Environment": "qa"
  },
  "Resource": {
    "CorrelationId": "8019f724-cc7b-4c78-8268-",
    "UserId": "767a2bb8e7f043b65929f4b4",
    "PartnerId": "618813b7331ceb7a92fb1172"
  },
  "event_type": "error",
  "status": 503,
  "duration": 5.052952558749363
}
JSON

{
  "Caller": "database_driver/sql.go:123",
  "Event": {
    "Date": "2025-07-09T08:25:59.211+00:00Z",
    "Type": "WARN",
    "Message": "ERROR: Could not establish Database server is unreachable. Ensure DB instance is running. (status=503) (duration=6.473s)"
  },
  "Host": {
    "HostName": "reporting-db-eu"
  },
  "Service": {
    "Name": "analytics-microservice",
    "Version": "0.1.0",
    "Owner": "Database Team",
    "BuildNumber": "0.1.0-86",
    "CommitId": "d32d5ba7cf779651294a73d2fc4284f5f59e1a62",
    "Environment": "int"
  },
  "Resource": {
    "CorrelationId": "227529a7-f5d5-4e73-b566-",
    "UserId": "75e37099307d52fccc5f824a",
    "PartnerId": "a7e9e920c6773a483a7bab4c"
  },
  "event_type": "error",
  "status": 503,
  "duration": 6.472930680763097
}