We Switched from Elasticsearch 8 to Typesense 0.25: 3x Faster Search Postmortem

import elasticsearch
from elasticsearch import Elasticsearch, helpers
import logging
import time
from typing import List, Dict, Optional

# Configure logging for audit trails
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Elasticsearch8Client:
    """Legacy Elasticsearch 8 client used pre-migration for product catalog search."""

    def __init__(self, nodes: List[str], index_name: str = "products"):
        self.index_name = index_name
        # Initialize client with retry logic and connection pooling
        self.client = Elasticsearch(
            nodes,
            retry_on_timeout=True,
            max_retries=3,
            request_timeout=30,
            # Disable sniffing for static node deployments
            sniff_on_start=False,
            sniff_on_node_failure=False
        )
        self._validate_connection()

    def _validate_connection(self) -> None:
        """Check cluster health and log node status."""
        try:
            health = self.client.cluster.health()
            logger.info(f"Elasticsearch cluster status: {health['status']}")
            logger.info(f"Connected nodes: {len(self.client.nodes.info()['nodes'])}")
        except elasticsearch.ConnectionError as e:
            logger.error(f"Failed to connect to Elasticsearch cluster: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected connection error: {e}")
            raise

    def index_documents(self, docs: List[Dict], batch_size: int = 1000) -> int:
        """Bulk index product documents with error handling."""
        success_count = 0
        failed_docs = []

        try:
            # Prepare actions for bulk indexing
            actions = [
                {
                    "_index": self.index_name,
                    "_id": doc["sku"],
                    "_source": doc
                }
                for doc in docs
            ]

            # Use streaming bulk helper for large datasets
            for ok, result in helpers.streaming_bulk(
                self.client,
                actions,
                chunk_size=batch_size,
                raise_on_error=False
            ):
                if ok:
                    success_count += 1
                else:
                    failed_docs.append(result)

            if failed_docs:
                logger.warning(f"Failed to index {len(failed_docs)} documents")
                with open("es_index_failures.log", "a") as f:
                    for fail in failed_docs[:10]:  # Log first 10 failures
                        f.write(f"{fail}\n")

            logger.info(f"Indexed {success_count}/{len(docs)} documents successfully")
            return success_count
        except elasticsearch.BulkIndexError as e:
            logger.error(f"Bulk index error: {e}")
            return success_count
        except Exception as e:
            logger.error(f"Unexpected indexing error: {e}")
            return success_count

    def search_products(self, query: str, filters: Optional[Dict] = None, size: int = 10) -> List[Dict]:
        """Execute a search query with optional filters using Elasticsearch DSL."""
        filters = filters or {}
        try:
            # Build Elasticsearch DSL query
            es_query = {
                "query": {
                    "bool": {
                        "must": [
                            {
                                "multi_match": {
                                    "query": query,
                                    "fields": ["name^3", "description^2", "category"],
                                    "fuzziness": "AUTO"
                                }
                            }
                        ],
                        "filter": []
                    }
                },
                "size": size,
                "sort": [{"_score": "desc"}, {"price": "asc"}]
            }

            # Add filter conditions
            if "price_min" in filters:
                es_query["query"]["bool"]["filter"].append({
                    "range": {"price": {"gte": filters["price_min"]}}
                })
            if "price_max" in filters:
                es_query["query"]["bool"]["filter"].append({
                    "range": {"price": {"lte": filters["price_max"]}}
                })
            if "category" in filters:
                es_query["query"]["bool"]["filter"].append({
                    "term": {"category.keyword": filters["category"]}
                })

            # Execute search
            response = self.client.search(index=self.index_name, body=es_query)
            return [hit["_source"] for hit in response["hits"]["hits"]]
        except elasticsearch.NotFoundError:
            logger.error(f"Index {self.index_name} not found")
            return []
        except elasticsearch.ConnectionError:
            logger.error("Search failed: connection to Elasticsearch lost")
            return []
        except Exception as e:
            logger.error(f"Search error: {e}")
            return []

if __name__ == "__main__":
    # Example usage of legacy ES client
    es_client = Elasticsearch8Client(nodes=["https://es-node-1:9200", "https://es-node-2:9200"])
    # Sample product doc
    sample_doc = [{
        "sku": "PROD-12345",
        "name": "Wireless Bluetooth Headphones",
        "description": "Noise cancelling over-ear headphones with 30h battery",
        "category": "Electronics",
        "price": 129.99,
        "in_stock": True
    }]
    es_client.index_documents(sample_doc)
    results = es_client.search_products("bluetooth headphones", filters={"price_max": 150})
    logger.info(f"Found {len(results)} results for sample query")

import typesense
import logging
from typing import List, Dict, Optional
import time

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Typesense025Client:
    """Typesense 0.25 client used post-migration for product catalog search."""

    def __init__(self, nodes: List[Dict], api_key: str, collection_name: str = "products"):
        self.collection_name = collection_name
        # Initialize Typesense client with connection pooling
        self.client = typesense.Client({
            "nodes": nodes,
            "api_key": api_key,
            "connection_timeout_seconds": 30,
            "retry_interval_seconds": 2,
            "num_retries": 3
        })
        self._ensure_collection()

    def _ensure_collection(self) -> None:
        """Create collection if it doesn't exist with schema matching our product catalog."""
        try:
            # Define collection schema
            schema = {
                "name": self.collection_name,
                "fields": [
                    {"name": "sku", "type": "string", "index": True},
                    {"name": "name", "type": "string", "index": True, "infix": True},
                    {"name": "description", "type": "string", "index": True, "infix": True},
                    {"name": "category", "type": "string", "index": True, "facet": True},
                    {"name": "price", "type": "float", "index": True, "sort": True},
                    {"name": "in_stock", "type": "bool", "index": True, "facet": True},
                    {"name": "created_at", "type": "int64", "sort": True}
                ],
                "default_sorting_field": "created_at"
            }

            # Check if collection exists, create if not
            try:
                self.client.collections[self.collection_name].retrieve()
                logger.info(f"Collection {self.collection_name} already exists")
            except typesense.exceptions.ObjectNotFound:
                self.client.collections.create(schema)
                logger.info(f"Created collection {self.collection_name}")
        except typesense.exceptions.TypesenseError as e:
            logger.error(f"Failed to ensure collection: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected collection error: {e}")
            raise

    def index_documents(self, docs: List[Dict], batch_size: int = 1000) -> int:
        """Bulk import documents into Typesense with error handling."""
        success_count = 0
        try:
            # Typesense expects a list of dicts; convert sku to string if needed
            processed_docs = []
            for doc in docs:
                processed = doc.copy()
                processed["sku"] = str(processed["sku"])
                processed["price"] = float(processed["price"])
                processed["in_stock"] = bool(processed["in_stock"])
                processed["created_at"] = int(time.time())
                processed_docs.append(processed)

            # Import in batches
            for i in range(0, len(processed_docs), batch_size):
                batch = processed_docs[i:i+batch_size]
                try:
                    import_results = self.client.collections[self.collection_name].documents.import_(batch)
                    # Check for failures in import results
                    for result in import_results:
                        if result["success"]:
                            success_count += 1
                        else:
                            logger.warning(f"Failed to import doc {result.get('document', {}).get('sku')}: {result.get('error')}")
                except typesense.exceptions.TypesenseError as e:
                    logger.error(f"Batch import error: {e}")

            logger.info(f"Imported {success_count}/{len(docs)} documents successfully")
            return success_count
        except Exception as e:
            logger.error(f"Unexpected indexing error: {e}")
            return success_count

    def search_products(self, query: str, filters: Optional[Dict] = None, size: int = 10) -> List[Dict]:
        """Execute search query with optional filters using Typesense's simple API."""
        filters = filters or {}
        try:
            # Build search parameters
            search_params = {
                "q": query,
                "query_by": "name,description,category",
                "prefix": True,
                "infix": "always",
                "per_page": size,
                "sort_by": "_text_match:desc,price:asc"
            }

            # Add filter conditions
            filter_by = []
            if "price_min" in filters:
                filter_by.append(f"price:>={filters['price_min']}")
            if "price_max" in filters:
                filter_by.append(f"price:<={filters['price_max']}")
            if "category" in filters:
                filter_by.append(f"category:{filters['category']}")
            if "in_stock" in filters:
                filter_by.append(f"in_stock:{str(filters['in_stock']).lower()}")

            if filter_by:
                search_params["filter_by"] = " && ".join(filter_by)

            # Execute search
            results = self.client.collections[self.collection_name].documents.search(search_params)
            return [hit["document"] for hit in results["hits"]]
        except typesense.exceptions.TypesenseError as e:
            logger.error(f"Search error: {e}")
            return []
        except Exception as e:
            logger.error(f"Unexpected search error: {e}")
            return []

    def health_check(self) -> bool:
        """Check Typesense node health."""
        try:
            health = self.client.health()
            return all(node["is_ready"] for node in health["nodes"])
        except Exception as e:
            logger.error(f"Health check failed: {e}")
            return False

if __name__ == "__main__":
    # Example usage of Typesense client
    ts_nodes = [{"host": "ts-node-1", "port": "8108", "protocol": "http"}]
    ts_client = Typesense025Client(
        nodes=ts_nodes,
        api_key="super-secret-api-key",
        collection_name="products"
    )
    # Sample product doc (same as ES example)
    sample_doc = [{
        "sku": "PROD-12345",
        "name": "Wireless Bluetooth Headphones",
        "description": "Noise cancelling over-ear headphones with 30h battery",
        "category": "Electronics",
        "price": 129.99,
        "in_stock": True
    }]
    ts_client.index_documents(sample_doc)
    results = ts_client.search_products("bluetooth headphones", filters={"price_max": 150})
    logger.info(f"Found {len(results)} results for sample query")

import elasticsearch
import typesense
import logging
from typing import List, Dict
import time
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class SearchMigrator:
    """Handles migration of product catalog data from Elasticsearch 8 to Typesense 0.25."""

    def __init__(self, es_client: elasticsearch.Elasticsearch, ts_client: typesense.Client, batch_size: int = 5000):
        self.es_client = es_client
        self.ts_client = ts_client
        self.batch_size = batch_size
        self.total_migrated = 0
        self.total_failed = 0

    def _fetch_es_documents(self, index_name: str, last_sku: Optional[str] = None) -> List[Dict]:
        """Fetch documents from Elasticsearch in batches using search_after for efficiency."""
        docs = []
        try:
            # Build search query with search_after for pagination
            query = {
                "query": {"match_all": {}},
                "sort": [{"sku.keyword": "asc"}],
                "size": self.batch_size,
                "_source": ["sku", "name", "description", "category", "price", "in_stock", "created_at"]
            }
            if last_sku:
                query["search_after"] = [last_sku]

            response = self.es_client.search(index=index_name, body=query)
            hits = response["hits"]["hits"]

            for hit in hits:
                source = hit["_source"]
                # Ensure sku is string for Typesense
                source["sku"] = str(source.get("sku", hit["_id"]))
                docs.append(source)

            return docs
        except elasticsearch.NotFoundError:
            logger.error(f"Elasticsearch index {index_name} not found")
            return []
        except elasticsearch.ConnectionError:
            logger.error("Failed to fetch docs: Elasticsearch connection lost")
            return []
        except Exception as e:
            logger.error(f"Error fetching ES documents: {e}")
            return []

    def _validate_migration(self, es_doc: Dict, ts_doc: Dict) -> bool:
        """Validate that a migrated document matches the source."""
        # Check critical fields
        checks = [
            str(es_doc.get("sku")) == str(ts_doc.get("sku")),
            es_doc.get("name") == ts_doc.get("name"),
            float(es_doc.get("price", 0)) == float(ts_doc.get("price", 0)),
            es_doc.get("category") == ts_doc.get("category")
        ]
        return all(checks)

    def run_migration(self, es_index: str, ts_collection: str, validate: bool = True) -> None:
        """Execute full migration with progress tracking and validation."""
        logger.info(f"Starting migration from ES index {es_index} to TS collection {ts_collection}")
        start_time = time.time()
        last_sku = None

        while True:
            # Fetch batch from ES
            es_docs = self._fetch_es_documents(es_index, last_sku)
            if not es_docs:
                logger.info("No more documents to migrate")
                break

            # Prepare docs for Typesense
            ts_docs = []
            for doc in es_docs:
                ts_doc = {
                    "sku": str(doc["sku"]),
                    "name": doc["name"],
                    "description": doc.get("description", ""),
                    "category": doc["category"],
                    "price": float(doc["price"]),
                    "in_stock": bool(doc.get("in_stock", True)),
                    "created_at": int(doc.get("created_at", time.time()))
                }
                ts_docs.append(ts_doc)

            # Import into Typesense
            try:
                import_results = self.ts_client.collections[ts_collection].documents.import_(ts_docs)
                batch_success = sum(1 for r in import_results if r["success"])
                batch_failed = len(import_results) - batch_success
                self.total_migrated += batch_success
                self.total_failed += batch_failed

                # Log batch results
                logger.info(f"Migrated batch: {batch_success} success, {batch_failed} failed. Total: {self.total_migrated}")

                # Optional validation (slows migration, disable for large datasets)
                if validate and batch_success > 0:
                    for i, es_doc in enumerate(es_docs[:10]):  # Validate first 10 per batch
                        ts_doc = ts_docs[i]
                        if not self._validate_migration(es_doc, ts_doc):
                            logger.warning(f"Validation failed for SKU {es_doc.get('sku')}")
            except typesense.exceptions.TypesenseError as e:
                logger.error(f"Typesense import error: {e}")
                self.total_failed += len(ts_docs)

            # Update last_sku for next batch
            last_sku = es_docs[-1].get("sku")

            # Log progress every 100k docs
            if self.total_migrated % 100000 == 0:
                elapsed = time.time() - start_time
                logger.info(f"Progress: {self.total_migrated} docs migrated in {elapsed:.2f}s ({self.total_migrated/elapsed:.0f} docs/s)")

        # Final report
        elapsed = time.time() - start_time
        logger.info(f"Migration complete. Total migrated: {self.total_migrated}, Failed: {self.total_failed}")
        logger.info(f"Total time: {elapsed:.2f}s ({self.total_migrated/elapsed:.0f} docs/s)")

if __name__ == "__main__":
    # Initialize clients (reuse earlier client classes for brevity)
    es_client = elasticsearch.Elasticsearch(["https://es-node-1:9200"])
    ts_client = typesense.Client({
        "nodes": [{"host": "ts-node-1", "port": "8108", "protocol": "http"}],
        "api_key": "super-secret-api-key",
        "connection_timeout_seconds": 30
    })

    migrator = SearchMigrator(es_client, ts_client, batch_size=5000)
    migrator.run_migration(es_index="products", ts_collection="products", validate=False)

def test_query_parity(es_client, ts_client):
    test_queries = ["bluetooth headphones", "gaming laptop", "running shoes"]
    for query in test_queries:
        es_results = es_client.search_products(query)
        ts_results = ts_client.search_products(query)
        # Compare top 5 SKUs
        es_skus = [doc["sku"] for doc in es_results[:5]]
        ts_skus = [doc["sku"] for doc in ts_results[:5]]
        assert set(es_skus) == set(ts_skus), f"Parity mismatch for query: {query}"

schema = {
    "name": "products",
    "fields": [
        {"name": "price", "type": "float", "index": True, "sort": True, "facet": False},
        {"name": "category", "type": "string", "index": True, "facet": True},
        {"name": "name", "type": "string", "index": True, "infix": True}
    ],
    "default_sorting_field": "created_at"
}

# Short NGINX traffic mirroring snippet
location /search {
    proxy_pass http://elasticsearch-cluster;
    # Mirror 10% of traffic to Typesense canary
    mirror /typesense-mirror;
    mirror_request_body on;
}

location /typesense-mirror {
    internal;
    proxy_pass http://typesense-canary;
}