Creating an embeddings-first OpenSearch flow using Terraform

If your search still relies on exact keyword matches, you have probably seen the same issues: good results get missed because users phrase things differently. An embeddings-first flow fixes that by generating vectors at ingest time and making semantic matching part of your default path.

What this setup does

1. Creates a secure OpenSearch domain with Terraform.
2. Generates a setup script from opensearch_setup.py.tpl with environment-aware values.
3. Configures ML Commons and model lifecycle calls.
4. Builds an ingest pipeline with the text_embedding processor.
5. Creates a k-NN index that uses that pipeline by default.

Provisioning the domain with Terraform

The Terraform side in opensearch.tf focuses on practical defaults: HTTPS enforcement, encryption at rest, node-to-node encryption, and fine-grained access control. That gives you a safer baseline before you start loading vectors.

resource "aws_opensearch_domain" "search" {
  domain_name    = var.opensearch_domain_name
  engine_version = "OpenSearch_2.13"

  cluster_config {
    instance_type             = var.opensearch_instance_type
    instance_count            = var.opensearch_instance_count
    dedicated_master_enabled  = false
  }

  ebs_options {
    ebs_enabled = true
    volume_type = "gp3"
    volume_size = var.opensearch_ebs_volume_size
  }

  encrypt_at_rest { enabled = true }
  node_to_node_encryption { enabled = true }

  domain_endpoint_options {
    enforce_https       = true
    tls_security_policy = "Policy-Min-TLS-1-2-2019-07"
  }

  advanced_security_options {
    enabled                        = true
    internal_user_database_enabled = true
    master_user_options {
      master_user_name     = var.opensearch_username
      master_user_password = local.opensearch_password
    }
  }
}

variable "ml_model_name" {
  description = "OpenSearch ML Commons pre-trained model name (from the OpenSearch model repository)"
  type        = string
  default     = "huggingface/sentence-transformers/all-MiniLM-L6-v2"
}

variable "ml_model_version" {
  description = "Version of the ML model to register"
  type        = string
  default     = "1.0.2"
}

variable "ml_model_format" {
  description = "Format of the ML model (TORCH_SCRIPT or ONNX)"
  type        = string
  default     = "TORCH_SCRIPT"
}

Terraform also writes a concrete setup script from template and runs it with environment variables for endpoint, credentials, and index name. That keeps one-off manual setup out of the console and in versioned code.

Where opensearch_setup.py is generated from the template

This is the Terraform block that renders opensearch_setup.py from opensearch_setup.py.tpl using templatefile():

resource "local_file" "opensearch_setup_script" {
  filename = "${path.module}/opensearch_setup.py"
  content  = templatefile("${path.module}/opensearch_setup.py.tpl", {
    endpoint         = aws_opensearch_domain.search.endpoint
    username         = var.opensearch_username
    password         = local.opensearch_password
    index_name       = var.opensearch_index
    ml_model_name    = var.ml_model_name
    ml_model_version = var.ml_model_version
    ml_model_format  = var.ml_model_format
  })
}

After that file is written, the setup step executes it with local-exec, which is what applies the ML Commons and index configuration against the domain.

Password handling for local.opensearch_password

The password flow is designed so you can pass a value explicitly, or let Terraform generate and persist one. In practice, that means:

• random_password.opensearch generates a strong password when no input variable is set.
• aws_secretsmanager_secret_version.opensearch_password stores either the provided value or the generated one.
• data.aws_secretsmanager_secret_version.opensearch_password reads the stored value on later runs.
• local.opensearch_password can then safely resolve from input var, stored secret, or generated fallback.

Caveat: You can also skip passing the password through Terraform variables/environment and have opensearch_setup.py read the secret directly from AWS Secrets Manager (for example via boto3). That can reduce secret exposure in local execution contexts, but it means the script runtime needs IAM permission to read that secret.

How the setup script wires in embeddings

The Python template flow:

• Wait until the domain API responds.
• Apply ML Commons cluster settings.
• Check for an existing model or register one.
• Deploy the model and poll task status.
• Create an ingest pipeline that maps embedding_input to a vector field.
• Create an index with knn enabled and set default_pipeline.

#!/usr/bin/env python3
"""
OpenSearch setup script — registers the ML embedding model, creates the ingest pipeline,
and creates the k-NN index for data.

Environment variables (set by Terraform provisioner):
  OPENSEARCH_ENDPOINT - OpenSearch domain endpoint (hostname only)
  OPENSEARCH_USERNAME - Master username
  OPENSEARCH_PASSWORD - Master password
  OPENSEARCH_INDEX    - Index name
  ML_MODEL_NAME       - OpenSearch ML Commons model name
  ML_MODEL_VERSION    - Model version
  ML_MODEL_FORMAT     - Model format (TORCH_SCRIPT or ONNX)
"""

import os
import sys
import time
import json
import requests
from urllib.parse import urljoin
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

endpoint = os.environ.get('OPENSEARCH_ENDPOINT', '${endpoint}')
username = os.environ.get('OPENSEARCH_USERNAME', '${username}')
password = os.environ.get('OPENSEARCH_PASSWORD', '${password}')
index_name = os.environ.get('OPENSEARCH_INDEX', '${index_name}')
ml_model_name    = os.environ.get('ML_MODEL_NAME',    '${ml_model_name}')
ml_model_version = os.environ.get('ML_MODEL_VERSION', '${ml_model_version}')
ml_model_format  = os.environ.get('ML_MODEL_FORMAT',  '${ml_model_format}')

base_url = f"https://{endpoint}"
auth = (username, password)
headers = {'Content-Type': 'application/json'}

def api_call(method, endpoint_path, data=None, timeout=10):
  """Make a REST call to OpenSearch."""
  url = urljoin(base_url, endpoint_path)
  try:
    if method == 'GET':
      r = requests.get(url, auth=auth, headers=headers, verify=False, timeout=timeout)
    elif method == 'POST':
      r = requests.post(url, auth=auth, headers=headers, json=data, verify=False, timeout=timeout)
    elif method == 'PUT':
      r = requests.put(url, auth=auth, headers=headers, json=data, verify=False, timeout=timeout)
    else:
      raise ValueError(f"Unsupported method: {method}")
    r.raise_for_status()
    return r.json() if r.text else {}
  except Exception as e:
    print(f"[ERROR] {method} {endpoint_path}: {e}")
    raise

def wait_for_domain(max_retries=180, delay=10):
  """Wait for the domain to be accessible (up to 30 minutes)."""
  for i in range(max_retries):
    try:
      resp = api_call('GET', '/')
      if resp.get('version', {}).get('number'):
        print(f"[OK] Domain is accessible (version {resp['version']['number']})")
        return True
    except Exception as e:
      print(f"[WAITING] Domain not yet ready ({i+1}/{max_retries}): {e}")
      time.sleep(delay)
  raise RuntimeError("Domain did not become ready within timeout")

def enable_ml_commons():
  """Enable ML Commons on data nodes."""
  print("\n[STEP 1] Enabling ML Commons...")
  data = {
    "persistent": {
      "plugins.ml_commons.only_run_on_ml_node": "false",
      "plugins.ml_commons.model_access_control_enabled": "true",
      "plugins.ml_commons.native_memory_threshold": "99",
      "cluster.max_shards_per_node": "100" # needed for k-NN indices, adjust as needed for larger deployments

      # Amazon OpenSearch Service doesn't support use of the following ML Commons settings:
      # https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ml.html
      # "plugins.ml_commons.allow_registering_model_via_url": "true"
    }
  }
  api_call('PUT', '/_cluster/settings', data)
  print("[OK] ML Commons enabled")

def check_existing_model():
  """Check if a model with the given name already exists."""
  print(f"\n[STEP 2a] Checking for existing model '{ml_model_name}'...")
  try:
    resp = api_call('GET', '/_plugins/_ml/models?search_term=' + ml_model_name)
    models = resp.get('data', [])
    for model in models:
      if model.get('name') == ml_model_name:
        model_id = model.get('id')
        print(f"[OK] Found existing model: {model_id}")
        return model_id
    print(f"[OK] No existing model found")
    return None
  except Exception as e:
    print(f"[WARNING] Could not check for existing models: {e}")
    return None

def register_model():
  """Register the embedding model."""
  print(f"\n[STEP 2b] Registering embedding model '{ml_model_name}' v{ml_model_version}...")
  data = {
    "name": ml_model_name,
    "version": ml_model_version,
    "model_format": ml_model_format
  }
  resp = api_call('POST', '/_plugins/_ml/models/_register', data)
  task_id = resp.get('task_id')
  if not task_id:
    raise RuntimeError(f"No task_id returned: {resp}")
  print(f"[OK] Model registration task: {task_id}")
  return task_id

def wait_for_task(task_id, max_retries=120, delay=2):
  """Poll a task until it completes."""
  for i in range(max_retries):
    resp = api_call('GET', f"/_plugins/_ml/tasks/{task_id}")
    state = resp.get('state')
    if state == 'COMPLETED':
      print(f"[OK] Task {task_id} completed")
      return resp
    elif state == 'FAILED':
      #raise RuntimeError(f"Task {task_id} failed: {resp}")
      # Soft fail for now...
      print(f"[FAILED] Task {task_id} failed: {resp}")
      return resp
    else:
      print(f"[WAITING] Task {task_id}: state={state} ({i+1}/{max_retries})")
      time.sleep(delay)
  raise RuntimeError(f"Task {task_id} did not complete within timeout")

def deploy_model(model_id):
  """Deploy the registered model."""
  print(f"\n[STEP 3] Deploying model {model_id}...")
  resp = api_call('POST', f"/_plugins/_ml/models/{model_id}/_deploy")
  deploy_task_id = resp.get('task_id')
  if not deploy_task_id:
    raise RuntimeError(f"No task_id returned: {resp}")
  print(f"[OK] Model deployment task: {deploy_task_id}")
  return deploy_task_id

def create_pipeline(model_id):
  """Create the text-embedding ingest pipeline."""
  print(f"\n[STEP 4] Creating ingest pipeline with model {model_id}...")
  data = {
    "description": "Generate text embeddings for our documents",
    "processors": [
      {
        "text_embedding": {
          "model_id": model_id,
          "field_map": {"embedding_input": "embedding"}
        }
      }
    ]
  }
  api_call('PUT', '/_ingest/pipeline/embedding-pipeline', data)
  print("[OK] Pipeline created: embedding-pipeline")

def create_index(index_name):
  """Create the k-NN index with default pipeline."""
  print(f"\n[STEP 5] Creating k-NN index '{index_name}'...")
  data = {
    "settings": {
      "index": {
        "knn": True,
        "default_pipeline": "embedding-pipeline"
      }
    },
    "mappings": {
      "properties": {
        "partNumber": {"type": "keyword"},
        "name": {"type": "text"},
        "status": {"type": "keyword"},
        "user": {"type": "keyword"},
        "embedding_input": {"type": "text"},
        "embedding": {
          "type": "knn_vector",
          "dimension": 384,
          "method": {
            "name": "hnsw",
            "space_type": "l2",
            "engine": "lucene"
          }
        }
      }
    }
  }
  api_call('PUT', f"/{index_name}", data)
  print(f"[OK] Index created: {index_name}")

def main():
  """Run the full setup sequence."""
  print(f"OpenSearch Setup Script")
  print(f"Endpoint: {base_url}")
  print(f"Index: {index_name}")
  print(f"Pipeline: embedding-pipeline")
  print("=" * 80)

  try:
    # Wait for domain to be healthy
    print("\n[STEP 0] Waiting for OpenSearch domain to be accessible...")
    wait_for_domain()

    # Enable ML Commons
    enable_ml_commons()

    # Check for existing model or register new one
    model_id = check_existing_model()
    if not model_id:
      reg_task_id = register_model()
      reg_resp = wait_for_task(reg_task_id)
      model_id = reg_resp.get('model_id')
      if not model_id:
        raise RuntimeError(f"No model_id in registration response: {reg_resp}")
      print(f"[OK] New model ID: {model_id}")
    else:
      print(f"[OK] Using existing model ID: {model_id}")

    # Deploy model
    deploy_task_id = deploy_model(model_id)
    wait_for_task(deploy_task_id, max_retries=300)  # Deployment takes longer
    print(f"[OK] Model deployed and ready")

    # Create pipeline
    create_pipeline(model_id)

    # Create index
    create_index(index_name)

    print("\n" + "=" * 80)
    print("[SUCCESS] OpenSearch setup complete!")
    print(f"  - ML Commons enabled")
    print(f"  - Model {model_id} deployed")
    print(f"  - Pipeline embedding-pipeline created")
    print(f"  - Index {index_name} created with k-NN + embeddings")
    print("\nYou can now ingest our data. Set embedding_input on each doc before indexing.")
    return 0

  except Exception as e:
    print(f"\n[FATAL] Setup failed: {e}", file=sys.stderr)
    return 1

if __name__ == "__main__":
  sys.exit(main())

Why this pattern works well

• It is repeatable. Infra and setup steps live together in source control.
• It is easier to operate. You can move the same flow into CI/CD later.
• It is ready for semantic retrieval from day one instead of as a follow-up project.

References

• Terraform AWS provider: aws_opensearch_domain
• Terraform Random provider: random_password
• Terraform AWS provider: aws_secretsmanager_secret_version (resource)
• Terraform AWS provider: aws_secretsmanager_secret_version (data source)
• Amazon OpenSearch Service: Fine-grained access control
• Amazon OpenSearch Service: ML Commons support on AWS
• OpenSearch docs: ML Commons
• OpenSearch docs: text_embedding processor
• OpenSearch docs: approximate k-NN search