Vector Store Operational Management Architecture

mindmap
  root((Vector Store Ops))
    Performance Monitoring
      Query Latency p50/p95/p99
      Throughput QPS
      Concurrent Connections
      Index Scan Efficiency
      Memory Pressure
      Cache Hit Rate
    Index Optimization
      HNSW Parameters
        ef_construction
        m (graph connectivity)
        ef_search
      Segment Compaction
      Auto-Reindex Triggers
      Benchmark-Driven Tuning
      Recall@K Tracking
    Data Quality
      Embedding Freshness
      Stale Content Detection
      Embedding Model Version Tracking
      Clustering Coherence
      Nearest-Neighbor Accuracy
      Drift Detection
    Capacity Management
      Storage Utilization
      OCU Scaling
      Cost per Query
      Growth Projection
      Dimension Optimization
    Operational Reliability
      Health Checks
      Failover Procedures
      Backup and Restore
      Replication Lag
      Disaster Recovery

graph TB
    subgraph IngestionPipeline["Ingestion Pipeline"]
        style IngestionPipeline fill:#1a1a2e,stroke:#16213e,color:#fff
        DDB["DynamoDB<br/>Product Catalog"] -->|Change Streams| LAMBDA_ETL["Lambda ETL<br/>Extract & Transform"]
        LAMBDA_ETL -->|Text chunks| TITAN_EMB_I["Bedrock Titan<br/>Embeddings v2"]
        TITAN_EMB_I -->|1536-dim vectors| BULK_IDX["OpenSearch Serverless<br/>Bulk Indexing API"]
        BULK_IDX --> OSS_COLL["OpenSearch Serverless<br/>manga-products Collection"]
    end

    subgraph QueryPipeline["Query Pipeline"]
        style QueryPipeline fill:#0f3460,stroke:#16213e,color:#fff
        USER_Q["User Query<br/>via WebSocket"] --> ECS_ORCH["ECS Orchestrator<br/>Fargate Service"]
        ECS_ORCH --> TITAN_EMB_Q["Bedrock Titan<br/>Embedding"]
        TITAN_EMB_Q -->|Query vector| KNN_SEARCH["OpenSearch k-NN<br/>HNSW Search"]
        KNN_SEARCH --> RERANK["Results Reranking<br/>Score Normalization"]
        RERANK --> FM_CTX["Claude 3 Context<br/>Augmented Prompt"]
    end

    subgraph MonitoringLayer["Monitoring Layer"]
        style MonitoringLayer fill:#533483,stroke:#16213e,color:#fff
        CW_METRICS["CloudWatch Metrics<br/>Latency, Errors, OCU Usage"]
        CUSTOM_METRICS["Custom Metrics<br/>Embedding Freshness<br/>Quality Scores<br/>Recall@K"]
        HEALTH_CHK["Automated Health Checks<br/>Every 60 seconds"]
    end

    subgraph OptimizationLayer["Optimization Layer"]
        style OptimizationLayer fill:#e94560,stroke:#16213e,color:#fff
        IDX_OPT["Index Parameter<br/>Optimizer"]
        SEG_COMP["Segment Compaction<br/>Scheduler"]
        EMB_REFRESH["Embedding Refresh<br/>Pipeline"]
    end

    subgraph AlertingLayer["Alerting & Response"]
        style AlertingLayer fill:#0a1931,stroke:#16213e,color:#fff
        CW_ALARM["CloudWatch Alarms<br/>Threshold Breach"]
        SNS_TOPIC["SNS Topic<br/>Notifications"]
        AUTO_SCALE["Auto-Scale OCU<br/>Capacity Policy"]
        RUNBOOK["Runbook Automation<br/>Lambda Remediation"]
    end

    OSS_COLL --> CW_METRICS
    KNN_SEARCH --> CW_METRICS
    KNN_SEARCH --> CUSTOM_METRICS
    BULK_IDX --> CUSTOM_METRICS
    HEALTH_CHK --> CW_METRICS
    HEALTH_CHK --> CUSTOM_METRICS

    CW_METRICS --> IDX_OPT
    CUSTOM_METRICS --> IDX_OPT
    CUSTOM_METRICS --> SEG_COMP
    CUSTOM_METRICS --> EMB_REFRESH
    EMB_REFRESH --> LAMBDA_ETL

    CW_METRICS --> CW_ALARM
    CUSTOM_METRICS --> CW_ALARM
    CW_ALARM --> SNS_TOPIC
    CW_ALARM --> AUTO_SCALE
    CW_ALARM --> RUNBOOK

Memory per vector ≈ (dimensions × 4 bytes) + (m × 2 × 4 bytes) + overhead

flowchart LR
    A["User Query Arrives"] -->|T0: timestamp| B["Generate Query<br/>Embedding"]
    B -->|T1: embedding_latency_ms| C["Execute k-NN<br/>Search"]
    C -->|T2: knn_latency_ms| D["Score & Rerank<br/>Results"]
    D -->|T3: rerank_latency_ms| E["Return Top-K<br/>Results"]
    E -->|T4: total_latency_ms| F["Emit CloudWatch<br/>Metrics"]

    B -.->|Monitor| M1["Embedding API<br/>latency & errors"]
    C -.->|Monitor| M2["k-NN search latency<br/>segments scanned<br/>scores distribution"]
    D -.->|Monitor| M3["Rerank duration<br/>score delta<br/>results filtered"]
    F -.->|Monitor| M4["Total e2e latency<br/>cache hit/miss<br/>result count"]

    style A fill:#2d3436,stroke:#00b894,color:#fff
    style F fill:#2d3436,stroke:#00b894,color:#fff
    style M1 fill:#6c5ce7,stroke:#a29bfe,color:#fff
    style M2 fill:#6c5ce7,stroke:#a29bfe,color:#fff
    style M3 fill:#6c5ce7,stroke:#a29bfe,color:#fff
    style M4 fill:#6c5ce7,stroke:#a29bfe,color:#fff

flowchart TD
    A["Performance Degradation<br/>Detected"] --> B{"Recall@10<br/>below target?"}
    B -->|Yes| C{"Current ef_search<br/>< max configured?"}
    C -->|Yes| D["Increase ef_search<br/>(runtime, no downtime)"]
    C -->|No| E{"Segment count<br/>> 20?"}
    E -->|Yes| F["Run Segment<br/>Compaction"]
    E -->|No| G{"m parameter<br/>< 48?"}
    G -->|Yes| H["Schedule Full Re-index<br/>with higher m"]
    G -->|No| I["Scale up OCUs<br/>or optimize queries"]

    B -->|No| J{"Latency p95<br/>above target?"}
    J -->|Yes| K{"Search OCU<br/>> 80%?"}
    K -->|Yes| L["Increase max OCU<br/>capacity policy"]
    K -->|No| M{"Segment count<br/>> 15?"}
    M -->|Yes| F
    M -->|No| N["Decrease ef_search<br/>(trade recall for speed)"]

    J -->|No| O["No action needed<br/>Continue monitoring"]

    style A fill:#e74c3c,stroke:#c0392b,color:#fff
    style D fill:#27ae60,stroke:#2ecc71,color:#fff
    style F fill:#f39c12,stroke:#e67e22,color:#fff
    style H fill:#e74c3c,stroke:#c0392b,color:#fff
    style I fill:#8e44ad,stroke:#9b59b6,color:#fff
    style L fill:#8e44ad,stroke:#9b59b6,color:#fff
    style N fill:#f39c12,stroke:#e67e22,color:#fff
    style O fill:#27ae60,stroke:#2ecc71,color:#fff

{
  "document_id": "prod-manga-12345",
  "source_updated_at": "2026-03-29T10:15:00Z",
  "embedding_created_at": "2026-03-29T11:00:00Z",
  "embedding_model_version": "amazon.titan-embed-text-v2:0",
  "source_hash": "sha256:abc123...",
  "staleness_hours": 0.75
}

flowchart TD
    subgraph NightlyBatch["Nightly Batch Validation (2am JST)"]
        style NightlyBatch fill:#1a1a2e,stroke:#e94560,color:#fff
        NB1["Scan all documents<br/>in collection"] --> NB2["Compute freshness<br/>for each document"]
        NB2 --> NB3["Calculate clustering<br/>metrics (sample 10K)"]
        NB3 --> NB4["Run known-pair<br/>accuracy test"]
        NB4 --> NB5["Check model version<br/>consistency"]
        NB5 --> NB6{"All metrics<br/>within targets?"}
        NB6 -->|Yes| NB7["Publish healthy<br/>status to CloudWatch"]
        NB6 -->|No| NB8["Identify degraded<br/>metrics"]
        NB8 --> NB9["Trigger targeted<br/>re-embedding for<br/>stale documents"]
        NB9 --> NB10["Publish degraded<br/>status + alert"]
    end

    subgraph ContinuousCheck["Continuous Spot-Check"]
        style ContinuousCheck fill:#0f3460,stroke:#e94560,color:#fff
        CC1["DynamoDB Stream<br/>detects product update"] --> CC2["Compare source_updated_at<br/>vs embedding_created_at"]
        CC2 --> CC3{"Staleness ><br/>threshold?"}
        CC3 -->|Yes| CC4["Add to re-embedding<br/>priority queue"]
        CC3 -->|No| CC5["Log freshness metric<br/>continue"]
        CC4 --> CC6["Lambda processes<br/>queue every 5 min"]
        CC6 --> CC7["Re-embed document<br/>via Titan v2"]
        CC7 --> CC8["Update OpenSearch<br/>document in-place"]
    end

    NB10 -.->|Feed into| CC4

from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
from enum import Enum


class IndexHealthStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    REINDEXING = "reindexing"
    STALE = "stale"
    CORRUPTED = "corrupted"


class AlertSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"


@dataclass
class VectorQueryMetrics:
    """Metrics captured for each vector search query."""
    query_id: str
    collection_name: str
    timestamp: datetime
    latency_ms: float
    results_count: int
    top_score: float
    min_score: float
    query_vector_norm: float
    index_size: int
    ef_search_used: int
    segments_scanned: int = 0
    cache_hit: bool = False


@dataclass
class IndexHealthReport:
    """Periodic health assessment of a vector index."""
    collection_name: str
    timestamp: datetime
    status: IndexHealthStatus
    total_vectors: int
    dimension: int
    index_size_bytes: int
    segment_count: int
    hnsw_m: int
    hnsw_ef_construction: int
    avg_query_latency_ms: float
    p95_query_latency_ms: float
    p99_query_latency_ms: float
    recall_at_10: float
    stale_vector_percentage: float
    embedding_model_version: str
    search_ocu_utilization: float = 0.0
    indexing_ocu_utilization: float = 0.0
    last_reindex_time: Optional[datetime] = None
    last_compaction_time: Optional[datetime] = None
    recommendations: List[str] = field(default_factory=list)


@dataclass
class EmbeddingFreshnessRecord:
    """Tracks freshness of a single document's embedding vs source data."""
    document_id: str
    collection_name: str
    source_updated_at: datetime
    embedding_created_at: datetime
    embedding_model_version: str
    source_hash: str = ""
    staleness_hours: float = 0.0
    needs_refresh: bool = False


@dataclass
class HNSWConfig:
    """Tracks HNSW configuration for a collection with tuning history."""
    collection_name: str
    ef_construction: int
    m: int
    ef_search_default: int
    dimension: int
    distance_metric: str = "cosine"
    last_tuned: Optional[datetime] = None
    tuning_history: List[Dict] = field(default_factory=list)


@dataclass
class CompactionEvent:
    """Record of a segment compaction operation."""
    collection_name: str
    started_at: datetime
    completed_at: Optional[datetime] = None
    segments_before: int = 0
    segments_after: int = 0
    size_before_bytes: int = 0
    size_after_bytes: int = 0
    trigger_reason: str = "scheduled"
    success: bool = False

import boto3
import time
import json
import statistics
from datetime import datetime, timedelta
from typing import Dict, List, Optional


class VectorStoreOpsMonitor:
    """
    Monitors OpenSearch Serverless vector store health and performance.

    Publishes custom CloudWatch metrics for query latency, embedding freshness,
    index health, and triggers automated remediation via alarms.
    """

    def __init__(self, config: dict):
        self.cloudwatch = boto3.client('cloudwatch')
        self.opensearch = boto3.client('opensearchserverless')
        self.sns = boto3.client('sns')
        self.namespace = config.get("namespace", "MangaAssist/VectorStore")
        self.collection_name = config["collection_name"]
        self.collection_endpoint = config["collection_endpoint"]
        self.stale_threshold_hours = config.get("stale_threshold_hours", 24)
        self.alert_topic_arn = config.get("alert_topic_arn")
        self._latency_window: List[float] = []
        self._quality_scores: List[float] = []
        self._max_window = 1000

    # ------------------------------------------------------------------
    # Query Performance Monitoring
    # ------------------------------------------------------------------

    def record_query(self, query_id: str, latency_ms: float, results_count: int,
                     top_score: float, min_score: float = 0.0,
                     ef_search_used: int = 256, cache_hit: bool = False):
        """Record a vector query execution and publish CloudWatch metrics."""
        self._latency_window.append(latency_ms)
        if len(self._latency_window) > self._max_window:
            self._latency_window.pop(0)

        dimensions = [{"Name": "Collection", "Value": self.collection_name}]
        metrics = [
            {
                "MetricName": "QueryLatency",
                "Value": latency_ms,
                "Unit": "Milliseconds",
                "Dimensions": dimensions,
                "Timestamp": datetime.utcnow(),
            },
            {
                "MetricName": "QueryCount",
                "Value": 1,
                "Unit": "Count",
                "Dimensions": dimensions,
            },
            {
                "MetricName": "ResultsCount",
                "Value": results_count,
                "Unit": "Count",
                "Dimensions": dimensions,
            },
            {
                "MetricName": "TopScore",
                "Value": top_score,
                "Unit": "None",
                "Dimensions": dimensions,
            },
            {
                "MetricName": "ScoreSpread",
                "Value": top_score - min_score if min_score else 0,
                "Unit": "None",
                "Dimensions": dimensions,
            },
            {
                "MetricName": "CacheHit",
                "Value": 1 if cache_hit else 0,
                "Unit": "Count",
                "Dimensions": dimensions,
            },
        ]
        self.cloudwatch.put_metric_data(Namespace=self.namespace, MetricData=metrics)

        # Anomaly detection: spike relative to recent p95
        if len(self._latency_window) >= 50:
            p95 = sorted(self._latency_window)[int(len(self._latency_window) * 0.95)]
            if latency_ms > p95 * 2:
                self._emit_alert(
                    "query_latency_spike",
                    f"Query {query_id} latency {latency_ms:.0f}ms > 2x p95 ({p95:.0f}ms)",
                    severity="WARNING",
                )

    def get_latency_percentiles(self) -> Dict[str, float]:
        """Calculate current latency percentiles from the sliding window."""
        if not self._latency_window:
            return {"p50": 0, "p95": 0, "p99": 0, "count": 0}
        sorted_latencies = sorted(self._latency_window)
        n = len(sorted_latencies)
        return {
            "p50": sorted_latencies[int(n * 0.50)],
            "p95": sorted_latencies[int(n * 0.95)],
            "p99": sorted_latencies[int(n * 0.99)],
            "count": n,
        }

    # ------------------------------------------------------------------
    # Embedding Freshness Monitoring
    # ------------------------------------------------------------------

    def check_embedding_freshness(self, documents: List[dict]) -> dict:
        """
        Check freshness of embeddings against source data timestamps.

        Each document dict must contain:
          - document_id: str
          - source_updated_at: datetime
          - embedding_created_at: datetime
        """
        stale_count = 0
        total = len(documents)
        stale_docs = []

        for doc in documents:
            source_updated = doc.get("source_updated_at")
            embedding_created = doc.get("embedding_created_at")
            if source_updated and embedding_created:
                staleness = (source_updated - embedding_created).total_seconds() / 3600
                if staleness > self.stale_threshold_hours:
                    stale_count += 1
                    stale_docs.append({
                        "document_id": doc["document_id"],
                        "staleness_hours": round(staleness, 1),
                    })

        stale_pct = (stale_count / total * 100) if total > 0 else 0

        self.cloudwatch.put_metric_data(
            Namespace=self.namespace,
            MetricData=[
                {
                    "MetricName": "StaleEmbeddingPercentage",
                    "Value": stale_pct,
                    "Unit": "Percent",
                    "Dimensions": [
                        {"Name": "Collection", "Value": self.collection_name}
                    ],
                },
                {
                    "MetricName": "StaleEmbeddingCount",
                    "Value": stale_count,
                    "Unit": "Count",
                    "Dimensions": [
                        {"Name": "Collection", "Value": self.collection_name}
                    ],
                },
            ],
        )

        if stale_pct > 10:
            self._emit_alert(
                "stale_embeddings",
                f"{stale_pct:.1f}% embeddings stale (>{self.stale_threshold_hours}h) — "
                f"{stale_count}/{total} documents need re-embedding",
                severity="WARNING" if stale_pct <= 25 else "CRITICAL",
            )

        return {
            "total": total,
            "stale_count": stale_count,
            "stale_pct": round(stale_pct, 2),
            "stale_docs": stale_docs[:50],  # Cap to avoid oversized responses
        }

    # ------------------------------------------------------------------
    # Index Health Evaluation
    # ------------------------------------------------------------------

    def evaluate_index_health(self, recall_at_10: float, avg_latency_ms: float,
                               p95_latency_ms: float, segment_count: int,
                               stale_pct: float, ocu_utilization: float = 0.0) -> dict:
        """Evaluate overall index health and generate prioritized recommendations."""
        recommendations = []
        status = "healthy"

        # Latency checks
        if avg_latency_ms > 100:
            recommendations.append(
                f"HIGH: Avg latency {avg_latency_ms:.0f}ms exceeds 100ms target — "
                "consider increasing OCU capacity or running segment compaction"
            )
            status = "degraded"
        if p95_latency_ms > 200:
            recommendations.append(
                f"HIGH: p95 latency {p95_latency_ms:.0f}ms exceeds 200ms target — "
                "check segment count and ef_search setting"
            )
            status = "degraded"

        # Recall check
        if recall_at_10 < 0.85:
            recommendations.append(
                f"HIGH: Recall@10 {recall_at_10:.2f} below 0.85 target — "
                "increase m parameter (requires re-index) or increase ef_search"
            )
            status = "degraded"

        # Segment count
        if segment_count > 20:
            recommendations.append(
                f"MEDIUM: {segment_count} segments exceeds 20 threshold — "
                "schedule segment compaction to reduce search overhead"
            )

        # Freshness
        if stale_pct > 10:
            recommendations.append(
                f"MEDIUM: {stale_pct:.1f}% embeddings stale — "
                "trigger targeted re-embedding pipeline"
            )
            if status == "healthy":
                status = "stale"

        # OCU utilization
        if ocu_utilization > 85:
            recommendations.append(
                f"HIGH: OCU utilization at {ocu_utilization:.0f}% — "
                "increase max OCU capacity policy to prevent throttling"
            )

        # Critical thresholds
        if recall_at_10 < 0.7 or avg_latency_ms > 500:
            status = "unhealthy"
            recommendations.insert(0,
                "CRITICAL: Consider full re-index with updated HNSW parameters — "
                "current performance severely degraded"
            )

        # Publish health status metric
        status_value = {"healthy": 1, "degraded": 2, "stale": 3, "unhealthy": 4}
        self.cloudwatch.put_metric_data(
            Namespace=self.namespace,
            MetricData=[{
                "MetricName": "IndexHealthStatus",
                "Value": status_value.get(status, 0),
                "Unit": "None",
                "Dimensions": [
                    {"Name": "Collection", "Value": self.collection_name}
                ],
            }],
        )

        return {"status": status, "recommendations": recommendations}

    # ------------------------------------------------------------------
    # OCU Monitoring
    # ------------------------------------------------------------------

    def get_ocu_metrics(self, period_minutes: int = 60) -> dict:
        """Retrieve current OCU utilization from CloudWatch."""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(minutes=period_minutes)

        search_ocu = self.cloudwatch.get_metric_statistics(
            Namespace="AWS/AOSS",
            MetricName="SearchOCU",
            StartTime=start_time,
            EndTime=end_time,
            Period=300,
            Statistics=["Average", "Maximum"],
            Dimensions=[
                {"Name": "CollectionName", "Value": self.collection_name}
            ],
        )
        indexing_ocu = self.cloudwatch.get_metric_statistics(
            Namespace="AWS/AOSS",
            MetricName="IndexingOCU",
            StartTime=start_time,
            EndTime=end_time,
            Period=300,
            Statistics=["Average", "Maximum"],
            Dimensions=[
                {"Name": "CollectionName", "Value": self.collection_name}
            ],
        )
        return {
            "search_ocu": search_ocu.get("Datapoints", []),
            "indexing_ocu": indexing_ocu.get("Datapoints", []),
        }

    # ------------------------------------------------------------------
    # Alerting
    # ------------------------------------------------------------------

    def _emit_alert(self, alert_type: str, message: str, severity: str = "WARNING"):
        """Emit alert metric and optionally publish to SNS."""
        self.cloudwatch.put_metric_data(
            Namespace=self.namespace,
            MetricData=[{
                "MetricName": "Alert",
                "Value": 1,
                "Unit": "Count",
                "Dimensions": [
                    {"Name": "Collection", "Value": self.collection_name},
                    {"Name": "AlertType", "Value": alert_type},
                    {"Name": "Severity", "Value": severity},
                ],
            }],
        )

        if self.alert_topic_arn:
            self.sns.publish(
                TopicArn=self.alert_topic_arn,
                Subject=f"[{severity}] MangaAssist Vector Store: {alert_type}",
                Message=json.dumps({
                    "alert_type": alert_type,
                    "severity": severity,
                    "collection": self.collection_name,
                    "message": message,
                    "timestamp": datetime.utcnow().isoformat(),
                }),
            )