Amazon OpenSearch Service

Master real-time analytics, log analytics, SIEM, and full-text search with OpenSearch on AWS.

20 min readAdvanced

What is Amazon OpenSearch Service?

Amazon OpenSearch Service (successor to Amazon Elasticsearch Service) makes it easy to deploy, operate, and scale OpenSearch clusters in the AWS Cloud. OpenSearch is a fork of Elasticsearch that supports full-text search, analytics, and observability use cases.

Core Concepts

OpenSearch is a distributed search and analytics engine built on Apache Lucene. It provides a RESTful API for indexing, searching, and analyzing large volumes of data in near real-time.

Concept	Description
Domain	An OpenSearch cluster with its configuration, endpoints, and data
Index	A collection of documents with a defined schema (analogous to a database table)
Shard	A partition of an index that can be hosted on any node in the cluster
Replica	A copy of a shard for high availability and read scaling
Cluster	A collection of nodes that hold data and provide indexing and search capabilities
Document	A JSON object stored in an index (analogous to a database row)

OpenSearch Architecture

How OpenSearch Works

When a document is indexed, the coordinating node determines which shard should hold the document using the document ID hash. The request is routed to the appropriate primary shard, which then replicates the document to replica shards. During search, the coordinating node sends the query to all relevant shards, merges results, and returns the final response.

Real-World Project Structure

Architecture Diagram

opensearch-production/
├── domains/
│   ├── log-analytics/
│   │   ├── domain-config.json
│   │   ├── index-templates/
│   │   │   ├── cloudwatch-logs.json
│   │   │   ├── application-logs.json
│   │   │   └── vpc-flow-logs.json
│   │   └── ilm-policies/
│   │       └── log-lifecycle.json
│   ├── metrics-analytics/
│   │   ├── domain-config.json
│   │   └── index-templates/
│   │       └── custom-metrics.json
│   └── security-analytics/
│       ├── domain-config.json
│       └── detector-rules/
│           └── threat-intel.json
├── kibana/
│   ├── dashboards/
│   │   ├── operational-overview.json
│   │   ├── security-dashboard.json
│   │   └── business-metrics.json
│   └── saved-objects/
│       └── index-patterns.json
├── data-pipeline/
│   ├── kinesis-firehose/
│   │   ├── log-delivery-stream.yaml
│   │   └── metrics-delivery-stream.yaml
│   └── lambda-transform/
│       └── transform-function.py
├── monitoring/
│   ├── cloudwatch-alarms.yaml
│   └── dashboards/
│       └── domain-health.json
└── security/
    ├── iam-roles/
    │   ├── opensearch-service-role.json
    │   └── kinesis-firehose-role.json
    └── resource-policies/
        └── domain-access.json

Production Python Code

import boto3
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth

logger = logging.getLogger(__name__)

class OpenSearchManager:
    """Production-grade Amazon OpenSearch Service manager."""

    def __init__(self, domain_endpoint: str, region: str = 'us-east-1'):
        self.region = region
        self.domain_endpoint = domain_endpoint
        self.credentials = boto3.Session().get_credentials()
        self.awsauth = AWSV4SignerAuth(
            self.credentials, region, 'es'
        )
        self.client = OpenSearch(
            hosts=[{'host': domain_endpoint, 'port': 443}],
            http_auth=self.awsauth,
            use_ssl=True,
            verify_certs=True,
            connection_class=RequestsHttpConnection
        )

    def create_index_with_template(
        self,
        index_name: str,
        template_body: Dict
    ) -> Dict:
        """Create an index with a component template."""
        try:
            response = self.client.indices.create_index(
                index=index_name,
                body=template_body
            )
            logger.info(f"Created index: {index_name}")
            return response
        except Exception as e:
            logger.error(f"Failed to create index {index_name}: {e}")
            raise

    def bulk_index_documents(
        self,
        index_name: str,
        documents: List[Dict],
        batch_size: int = 500
    ) -> Dict:
        """Index documents in batches with error handling."""
        results = {'success': 0, 'failed': 0, 'errors': []}

        for i in range(0, len(documents), batch_size):
            batch = documents[i:i + batch_size]
            bulk_body = []
            for doc in batch:
                bulk_body.append(json.dumps({
                    'index': {'_index': index_name}
                }))
                bulk_body.append(json.dumps(doc))

            try:
                response = self.client.bulk(
                    body='\n'.join(bulk_body) + '\n'
                )
                if response.get('errors'):
                    for item in response['items']:
                        if 'error' in item.get('index', {}):
                            results['failed'] += 1
                            results['errors'].append(item['index']['error'])
                        else:
                            results['success'] += 1
                else:
                    results['success'] += len(batch)
            except Exception as e:
                logger.error(f"Bulk index batch {i} failed: {e}")
                results['failed'] += len(batch)
                results['errors'].append(str(e))

        logger.info(
            f"Indexed {results['success']} docs, "
            f"{results['failed']} failed"
        )
        return results

    def search_with_aggregations(
        self,
        index_name: str,
        query: Dict,
        aggregations: Optional[Dict] = None,
        size: int = 0
    ) -> Dict:
        """Execute search with optional aggregations."""
        body = {'query': query, 'size': size}
        if aggregations:
            body['aggs'] = aggregations

        try:
            response = self.client.search(
                index=index_name,
                body=body
            )
            return response
        except Exception as e:
            logger.error(f"Search failed: {e}")
            raise

    def create_ism_policy(self, policy_name: str, policy_body: Dict) -> Dict:
        """Create an Index State Management policy."""
        try:
            response = self.client.indices.put_ism_policy(
                policy=policy_name,
                body=policy_body
            )
            logger.info(f"Created ISM policy: {policy_name}")
            return response
        except Exception as e:
            logger.error(f"Failed to create ISM policy: {e}")
            raise


class OpenSearchProvisioner:
    """Provision and manage OpenSearch domains using boto3."""

    def __init__(self, region: str = 'us-east-1'):
        self.client = boto3.client('opensearch', region_name=region)

    def create_domain(
        self,
        domain_name: str,
        instance_type: str = 'r5.large.search',
        instance_count: int = 3,
        dedicated_master_enabled: bool = True,
        zone_awareness_enabled: bool = True,
        ebs_enabled: bool = True,
        ebs_volume_size: int = 100,
        encryption_at_rest: bool = True,
        node_to_node_encryption: bool = True,
        enforce_https: bool = True
    ) -> Dict:
        """Create a production OpenSearch domain."""
        try:
            response = self.client.create_domain(
                DomainName=domain_name,
                EngineVersion='OpenSearch_2.11',
                ClusterConfig={
                    'InstanceType': instance_type,
                    'InstanceCount': instance_count,
                    'DedicatedMasterEnabled': dedicated_master_enabled,
                    'ZoneAwarenessEnabled': zone_awareness_enabled,
                    'DedicatedMasterType': 'm5.large.search',
                    'DedicatedMasterCount': 3
                },
                EBSOptions={
                    'EBSEnabled': ebs_enabled,
                    'VolumeType': 'gp3',
                    'VolumeSize': ebs_volume_size
                },
                EncryptionAtRestOptions={
                    'Enabled': encryption_at_rest
                },
                NodeToNodeEncryptionOptions={
                    'Enabled': node_to_node_encryption
                },
                DomainEndpointOptions={
                    'EnforceHTTPS': enforce_https
                },
                AccessPolicies=json.dumps({
                    'Version': '2012-10-17',
                    'Statement': [{
                        'Effect': 'Allow',
                        'Principal': {'AWS': '*'},
                        'Action': 'es:*',
                        'Resource': f'arn:aws:es:us-east-1:*:domain/{domain_name}/*'
                    }]
                })
            )
            logger.info(f"Created domain: {response['DomainStatus']['ARN']}")
            return response
        except Exception as e:
            logger.error(f"Failed to create domain: {e}")
            raise

    def create_kinesis_firehose_to_opensearch(
        self,
        stream_name: str,
        domain_arn: str,
        s3_backup_bucket: str
    ) -> Dict:
        """Create a Kinesis Firehose delivery stream to OpenSearch."""
        firehose = boto3.client('firehose')
        try:
            response = firehose.create_delivery_stream(
                DeliveryStreamName=stream_name,
                DeliveryStreamType='DirectPut',
                AmazonopensearchDestinationConfiguration={
                    'DomainARN': domain_arn,
                    'RoleARN': f'arn:aws:iam::*:role/firehose-opensearch-role',
                    'IndexName': 'logs',
                    'TypeName': 'INLINE',
                    'S3Configuration': {
                        'RoleARN': f'arn:aws:iam::*:role/firehose-s3-role',
                        'BucketARN': f'arn:aws:s3:::{s3_backup_bucket}',
                        'BufferingHints': {
                            'SizeInMBs': 64,
                            'IntervalInSeconds': 300
                        }
                    },
                    'BufferingHints': {
                        'SizeInMBs': 5,
                        'IntervalInSeconds': 60
                    },
                    'RetryOptions': {
                        'DurationInSeconds': 300
                    }
                }
            )
            logger.info(f"Created Firehose stream: {stream_name}")
            return response
        except Exception as e:
            logger.error(f"Failed to create Firehose: {e}")
            raise


# Production usage
if __name__ == '__main__':
    provisioner = OpenSearchProvisioner()
    domain = provisioner.create_domain(
        domain_name='log-analytics-prod',
        instance_type='r5.xlarge.search',
        instance_count=6,
        ebs_volume_size=500
    )
    print(f"Domain ARN: {domain['DomainStatus']['ARN']}")

Production Bash Commands

#!/bin/bash
# OpenSearch domain health check and monitoring script

set -euo pipefail

DOMAIN_NAME="${1:-log-analytics-prod}"
REGION="${AWS_REGION:-us-east-1}"
ALERT_EMAIL="${ALERT_EMAIL:-ops-team@company.com}"

echo "=== OpenSearch Domain Health Check ==="
echo "Domain: ${DOMAIN_NAME}"
echo "Region: ${REGION}"
echo "Timestamp: $(date -u +%Y-%m-%dT%H:%M:%SZ)"

# Get domain status
DOMAIN_STATUS=$(aws opensearch describe-domain \
  --domain-name "${DOMAIN_NAME}" \
  --region "${REGION}" \
  --query 'DomainStatus' \
  --output json)

PROCESSING=$(echo "$DOMAIN_STATUS" | jq -r '.Processing')
SEARCHABLE=$(echo "$DOMAIN_STATUS" | jq -r '.Searchable')
ENDPOINT=$(echo "$DOMAIN_STATUS" | jq -r '.Endpoint')

echo "Status: Processing=${PROCESSING}, Searchable=${SEARCHABLE}"
echo "Endpoint: https://${ENDPOINT}"

# Check cluster health via OpenSearch API
HEALTH_RESPONSE=$(curl -s -w "\n%{http_code}" \
  "https://${ENDPOINT}/_cluster/health" \
  --aws-sigv4 "aws:amz:${REGION}:es" \
  --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" 2>/dev/null)

HTTP_CODE=$(echo "$HEALTH_RESPONSE" | tail -1)
HEALTH_BODY=$(echo "$HEALTH_RESPONSE" | head -n -1)

if [ "$HTTP_CODE" -eq 200 ]; then
  CLUSTER_STATUS=$(echo "$HEALTH_BODY" | jq -r '.status')
  echo "Cluster Health: ${CLUSTER_STATUS}"

  if [ "$CLUSTER_STATUS" = "red" ]; then
    echo "ALERT: Cluster is in RED state!"
    aws sns publish \
      --topic-arn "arn:aws:sns:${REGION}:$(aws sts get-caller-identity --query Account --output text):opensearch-alerts" \
      --message "OpenSearch cluster ${DOMAIN_NAME} is in RED state" \
      --region "${REGION}"
  fi
else
  echo "WARNING: Could not fetch cluster health (HTTP ${HTTP_CODE})"
fi

# Check node count
NODES=$(curl -s "https://${ENDPOINT}/_cat/nodes?format=json" \
  --aws-sigv4 "aws:amz:${REGION}:es" \
  --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" 2>/dev/null | jq length)
echo "Active Nodes: ${NODES}"

# Check pending tasks
PENDING=$(curl -s "https://${ENDPOINT}/_cluster/pending_tasks" \
  --aws-sigv4 "aws:amz:${REGION}:es" \
  --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" 2>/dev/null | jq '.tasks | length')
echo "Pending Tasks: ${PENDING}"

# Check disk usage across nodes
DISK_USAGE=$(curl -s "https://${ENDPOINT}/_cat/allocation?v&format=json" \
  --aws-sigv4 "aws:amz:${REGION}:es" \
  --user "${AWS_ACCESS_KEY_ID}:${AWS_SECRET_ACCESS_KEY}" 2>/dev/null)
echo "Disk Usage per Node:"
echo "$DISK_USAGE" | jq -r '.[] | "  Node: \(.node), Disk Used: \(.disk.percent)"'

# CloudWatch metrics
echo ""
echo "=== CloudWatch Metrics ==="
aws cloudwatch get-metric-statistics \
  --namespace AWS/ES \
  --metric-name FreeStorageSpace \
  --dimensions Name=DomainName,Value="${DOMAIN_NAME}" Name=ClientId,Value="$(aws sts get-caller-identity --query Account --output text)" \
  --start-time "$(date -u -d '1 hour ago' +%Y-%m-%dT%H:%M:%SZ)" \
  --end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
  --period 300 \
  --statistics Average \
  --region "${REGION}" | jq '.Datapoints[] | {Time: .Timestamp, FreeGB: (.Average / 1073741824 | floor)}'

echo ""
echo "Health check complete."

Why This Matters

OpenSearch provides a managed, scalable solution for log analytics, observability, and security analytics. Unlike self-managed Elasticsearch, OpenSearch on AWS handles cluster management, patching, backups, and scaling automatically. The integration with CloudWatch, Kinesis Firehose, and S3 enables end-to-end data pipelines for operational intelligence.

Real-World Project Architecture

Mathematical Formulas

Performance Considerations

Factor	Recommendation	Impact
Shard Size	Target 10-50GB per shard	High - affects query and indexing performance
Replica Count	Use 1 replica minimum for production	High - read scaling and fault tolerance
Bulk Size	Batch 500-5000 documents per request	Medium - reduces API overhead
Refresh Interval	Set to 30s-60s for batch ingestion	Medium - improves indexing throughput
Index Templates	Define mappings and settings upfront	High - ensures consistent schema
Document Size	Keep under 100KB per document	Medium - affects memory and search
Field Count	Limit to 1000 fields per index	Medium - reduces mapping overhead
UltraWarm	Move indices older than 30 days	High - reduces storage costs by 90%

Security Considerations

VPC Deployment: Deploy OpenSearch domains within VPC for network isolation
Fine-Grained Access Control: Enable to control access at index, document, and field levels
Encryption at Rest: Use AWS KMS for encrypting index data on EBS volumes
Node-to-Node Encryption: Enable TLS for all inter-node communication
IAM Integration: Use IAM roles and resource-based policies for authentication
Cognito Integration: Use Amazon Cognito for Kibana user authentication
Audit Logging: Enable to track all user activity and API calls
CloudTrail Integration: Log all OpenSearch API calls for compliance

Common Pitfalls

Pitfall	Consequence	Solution
Too many shards	High memory usage, slow cluster	Target 10-50GB per shard
No index lifecycle policy	Unbounded storage costs	Implement ILM for tiered storage
Ignoring replica count	Single point of failure	Use minimum 1 replica in production
Skipping field mappings	Mapping explosion, poor performance	Define explicit mappings
Large bulk requests	Out of memory errors	Keep batches under 5000 documents
No UltraWarm tier	Overpaying for cold data	Migrate old indices to UltraWarm
Disabling HTTPS	Man-in-the-middle attacks	Always enforce HTTPS
No monitoring	Undetected performance issues	Set up CloudWatch alarms

Interview Questions & Answers

Q1: What is Amazon OpenSearch Service and when would you use it?

Answer: Amazon OpenSearch Service is a managed service that makes it easy to deploy, operate, and scale OpenSearch clusters in the AWS Cloud. Use it for log analytics, real-time application monitoring, full-text search, security analytics (SIEM), and observability. It integrates natively with CloudWatch, Kinesis Firehose, and S3 for end-to-end data pipelines.

Q2: How does OpenSearch handle data distribution across nodes?

Answer: OpenSearch distributes data using indices and shards. Each index is split into one or more primary shards distributed across data nodes. Each primary shard can have zero or more replica shards for fault tolerance. The consistent hashing algorithm determines shard placement, ensuring even distribution. Zone awareness ensures replicas are placed in different availability zones.

Q3: What is UltraWarm storage and when should you use it?

Answer: UltraWarm is a storage tier for OpenSearch that uses S3-backed storage at roughly one-tenth the cost of standard EBS-based storage. It is ideal for older, less frequently accessed data (typically indices older than 30-90 days). UltraWarm supports read operations and searches but has higher latency than hot storage. Use ILM policies to automatically migrate data between tiers.

Q4: How do you optimize OpenSearch query performance?

Answer: Key optimizations include: (1) Use index lifecycle management to separate hot and cold data. (2) Define explicit field mappings to avoid mapping explosion. (3) Use the _source filtering to return only needed fields. (4) Implement routing for related documents. (5) Use filter context instead of query context for non-scoring filters. (6) Set appropriate refresh intervals for batch ingestion.

Q5: What is the difference between OpenSearch and Elasticsearch?

Answer: OpenSearch is an open-source fork of Elasticsearch 7.10 created by AWS. Key differences include: OpenSearch is community-driven with no proprietary features, includes integrated security by default, has built-in alerting and anomaly detection, and supports S3-based UltraWarm and cold storage tiers. Both use similar query DSL and APIs.

Q6: How do you implement cross-cluster replication in OpenSearch?

Answer: Cross-cluster replication (CCR) enables replicating an index from a leader domain to a follower domain. Configure a replication rule on the follower domain pointing to the leader domain's endpoint and index. The follower pulls data from the leader in near real-time. Use cases include disaster recovery, read scaling across regions, and migration between clusters.

Q7: What security features are available in OpenSearch Service?

Answer: Security features include: VPC deployment for network isolation, fine-grained access control at index/document/field levels, encryption at rest with KMS, node-to-node TLS encryption, IAM-based authentication, Amazon Cognito for Kibana SSO, audit logging for compliance, and resource-based access policies. All features are enabled through domain configuration.

Q8: How does OpenSearch differ from CloudWatch Logs Insights?

Answer: OpenSearch provides full-text search, complex aggregations, and real-time analytics on log data, while CloudWatch Logs Insights is a simpler query tool for ad-hoc log analysis. OpenSearch is better for building dashboards, alerting rules, and complex analytics pipelines. CloudWatch Logs Insights is simpler for quick log exploration and is tightly integrated with CloudWatch metrics and alarms.

Amazon OpenSearch Service for Data Engineers