Cápsula 04: Backup y Disaster Recovery

Descripción de la cápsula

Un sistema RAG sin backup probado no está listo para producción. Esta cápsula te guía para definir respaldo, retención y recuperación con objetivos de tiempo claros, scripts de automatización para ChromaDB y procedimientos paso a paso que puedas ejecutar en un incidente real.

Tiempo estimado: 45-60 minutos

RTO y RPO para sistemas RAG

¿Qué significan?

• RPO (Recovery Point Objective): cuánto dato máximo puedes perder sin comprometer la integridad del negocio. En RAG significa: documentos, embeddings y metadatos que no estarían en el último backup.
• RTO (Recovery Time Objective): cuánto tiempo máximo aceptas que el sistema esté caído o degradado hasta recuperarse.

Objetivos sugeridos por criticidad

Ejemplo inicial razonable

• RPO: 24 horas (backup diario)
• RTO: 2 horas (restore documentado y probado)

Estructura de datos ChromaDB

Antes de hacer backup, conviene saber qué estás copiando:

./chroma_db/
├── chroma.sqlite3       # Metadata, IDs, config de collections
├── index/               # Índices HNSW (búsqueda rápida)
│   └── <collection_id>.bin
└── data/                # Vectores y documentos
    └── <collection_id>.parquet

Importante: ChromaDB requiere que el cliente esté cerrado para backups por copia de archivos consistentes. Si no, puedes obtener un backup corrupto.

Scripts de backup y export para ChromaDB

1. Backup por copia (offline)

Ideal para restauración completa y rápida. El cliente debe estar cerrado.

# scripts/backup_chromadb.py
import os
import shutil
from datetime import datetime
from pathlib import Path


def backup_chromadb(
    source_path: str,
    backup_dir: str,
    prefix: str = "chroma_backup",
) -> str:
    """
    Copia toda la carpeta ChromaDB. Usar cuando el proceso
    que usa ChromaDB está detenido.
    """
    source = Path(source_path)
    if not source.exists():
        raise FileNotFoundError(f"No existe: {source_path}")

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"{prefix}_{timestamp}"
    backup_path = Path(backup_dir) / backup_name

    Path(backup_dir).mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, backup_path)

    print(f"Backup creado: {backup_path}")
    return str(backup_path)


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--source", default="./chroma_db")
    parser.add_argument("--backup-dir", default="./backups")
    args = parser.parse_args()
    backup_chromadb(args.source, args.backup_dir)

2. Export por colección (online)

Útil para migrar una colección o hacer backup sin detener el servicio (los datos se leen vía API).

# scripts/export_collection.py
import json
import chromadb
from pathlib import Path


def export_collection(
    chroma_path: str,
    collection_name: str,
    output_file: str,
    include_embeddings: bool = True,
) -> int:
    """
    Exporta una colección a JSON. Funciona con el servicio en marcha.
    Retorna el número de documentos exportados.
    """
    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_collection(collection_name)

    include = ["documents", "metadatas"]
    if include_embeddings:
        include.append("embeddings")

    data = collection.get(include=include)

    payload = {
        "collection": collection_name,
        "count": len(data["ids"]),
        "ids": data["ids"],
        "documents": data.get("documents", []),
        "metadatas": data.get("metadatas", []),
    }
    if include_embeddings and data.get("embeddings"):
        payload["embeddings"] = data["embeddings"]

    Path(output_file).parent.mkdir(parents=True, exist_ok=True)
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)

    print(f"Exportados {len(data['ids'])} docs a {output_file}")
    return len(data["ids"])


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", default="./chroma_db")
    parser.add_argument("--collection", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    export_collection(args.chroma_path, args.collection, args.output)

3. Import desde JSON

# scripts/import_collection.py
import json
import chromadb
from pathlib import Path


def import_collection(
    chroma_path: str,
    collection_name: str,
    json_file: str,
    batch_size: int = 500,
) -> int:
    """
    Importa una colección desde un backup JSON.
    Retorna el número de documentos importados.
    """
    with open(json_file, "r", encoding="utf-8") as f:
        data = json.load(f)

    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_or_create_collection(collection_name)

    ids = data["ids"]
    documents = data.get("documents", [])
    metadatas = data.get("metadatas", [])
    embeddings = data.get("embeddings")

    n = 0
    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i : i + batch_size]
        batch_docs = documents[i : i + batch_size] if documents else None
        batch_meta = metadatas[i : i + batch_size] if metadatas else None
        batch_emb = embeddings[i : i + batch_size] if embeddings else None

        kwargs = {"ids": batch_ids}
        if batch_docs:
            kwargs["documents"] = batch_docs
        if batch_meta:
            kwargs["metadatas"] = batch_meta
        if batch_emb:
            kwargs["embeddings"] = batch_emb

        collection.add(**kwargs)
        n += len(batch_ids)

    print(f"Importados {n} docs a {collection_name}")
    return n

Backup automático diario con retención de 30 días

Script completo de automatización

# scripts/backup_automation.py
"""
Backup diario de ChromaDB con política de retención.
Ejecutar vía cron o systemd timer.
"""
import argparse
import logging
import os
import shutil
import sys
from datetime import datetime, timedelta
from pathlib import Path

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.StreamHandler(sys.stdout),
    ],
)
logger = logging.getLogger(__name__)

# Configuración por defecto
DEFAULT_SOURCE = os.environ.get("CHROMA_DB_PATH", "./chroma_db")
DEFAULT_BACKUP_DIR = os.environ.get("CHROMA_BACKUP_DIR", "./backups")
DEFAULT_RETENTION_DAYS = int(os.environ.get("CHROMA_BACKUP_RETENTION_DAYS", "30"))


def run_backup(source: str, backup_dir: str) -> str:
    """Ejecuta backup por copia de directorio."""
    source_path = Path(source)
    if not source_path.exists():
        raise FileNotFoundError(f"Directorio ChromaDB no existe: {source}")

    backup_root = Path(backup_dir)
    backup_root.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_name = f"chroma_backup_{timestamp}"
    backup_path = backup_root / backup_name

    shutil.copytree(source_path, backup_path)
    logger.info("Backup creado: %s", backup_path)
    return str(backup_path)


def apply_retention(backup_dir: str, keep_days: int) -> int:
    """
    Elimina backups más antiguos que keep_days.
    Retorna el número de backups eliminados.
    """
    backup_root = Path(backup_dir)
    if not backup_root.exists():
        return 0

    cutoff = datetime.now() - timedelta(days=keep_days)
    cutoff_ts = cutoff.timestamp()
    deleted = 0

    for item in backup_root.iterdir():
        if item.is_dir() and item.name.startswith("chroma_backup_"):
            if item.stat().st_mtime < cutoff_ts:
                shutil.rmtree(item)
                logger.info("Backup antiguo eliminado: %s", item.name)
                deleted += 1

    return deleted


def main():
    parser = argparse.ArgumentParser(description="Backup ChromaDB con retención")
    parser.add_argument("--source", default=DEFAULT_SOURCE)
    parser.add_argument("--backup-dir", default=DEFAULT_BACKUP_DIR)
    parser.add_argument("--retention-days", type=int, default=DEFAULT_RETENTION_DAYS)
    parser.add_argument("--retention-only", action="store_true", help="Solo limpiar, no hacer backup")
    args = parser.parse_args()

    try:
        if not args.retention_only:
            run_backup(args.source, args.backup_dir)
        deleted = apply_retention(args.backup_dir, args.retention_days)
        logger.info("Retención: %d backups eliminados", deleted)
    except Exception as e:
        logger.exception("Error en backup: %s", e)
        sys.exit(1)


if __name__ == "__main__":
    main()

Configuración de cron

Ejecutar backup diario a las 02:00 y aplicar retención:

# /etc/cron.d/chromadb-backup
# Backup diario de ChromaDB a las 02:00, retención 30 días
0 2 * * * cd /opt/rag-app && /usr/bin/python3 scripts/backup_automation.py --source /opt/rag-app/chroma_db --backup-dir /opt/backups/chroma --retention-days 30 >> /var/log/chromadb-backup.log 2>&1

Con variables de entorno (recomendado):

# En /opt/rag-app/.env o systemd
export CHROMA_DB_PATH=/opt/rag-app/chroma_db
export CHROMA_BACKUP_DIR=/opt/backups/chroma
export CHROMA_BACKUP_RETENTION_DAYS=30

0 2 * * * source /opt/rag-app/.env && /usr/bin/python3 /opt/rag-app/scripts/backup_automation.py >> /var/log/chromadb-backup.log 2>&1

systemd timer (alternativa a cron)

# /etc/systemd/system/chromadb-backup.service
[Unit]
Description=ChromaDB daily backup
After=network.target

[Service]
Type=oneshot
User=rag-app
WorkingDirectory=/opt/rag-app
Environment="CHROMA_DB_PATH=/opt/rag-app/chroma_db"
Environment="CHROMA_BACKUP_DIR=/opt/backups/chroma"
Environment="CHROMA_BACKUP_RETENTION_DAYS=30"
ExecStart=/usr/bin/python3 /opt/rag-app/scripts/backup_automation.py

# /etc/systemd/system/chromadb-backup.timer
[Unit]
Description=ChromaDB backup diario
Requires=chromadb-backup.service

[Timer]
OnCalendar=*-*-* 02:00:00
Persistent=true

[Install]
WantedBy=timers.target

sudo systemctl enable chromadb-backup.timer
sudo systemctl start chromadb-backup.timer

Validación de backups

Un backup sin validación puede fallar cuando más lo necesitas. Este script verifica integridad básica:

# scripts/validate_backup.py
"""
Valida que un backup de ChromaDB sea restaurable.
"""
import argparse
import sys
import chromadb
from pathlib import Path


def validate_backup(backup_path: str) -> bool:
    """
    Intenta cargar el backup como cliente ChromaDB.
    Retorna True si es válido.
    """
    path = Path(backup_path)
    if not path.exists():
        print(f"❌ No existe: {backup_path}")
        return False

    # Archivos mínimos esperados
    sqlite = path / "chroma.sqlite3"
    if not sqlite.exists():
        print(f"❌ Falta chroma.sqlite3")
        return False

    try:
        client = chromadb.PersistentClient(path=str(backup_path))
        collections = client.list_collections()
        total_docs = 0
        for col in collections:
            count = col.count()
            total_docs += count
            print(f"  - {col.name}: {count} documentos")
        print(f"✅ Backup válido: {len(collections)} collections, {total_docs} documentos")
        return True
    except Exception as e:
        print(f"❌ Backup corrupto: {e}")
        return False


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("backup_path")
    args = parser.parse_args()
    ok = validate_backup(args.backup_path)
    sys.exit(0 if ok else 1)

Integrar validación en el flujo de backup:

# Añadir al final de run_backup() en backup_automation.py
def run_backup_with_validation(source: str, backup_dir: str) -> str:
    backup_path = run_backup(source, backup_dir)
    if not validate_backup(backup_path):
        Path(backup_path).rmdir()
        raise RuntimeError("Backup falló validación, eliminado")
    return backup_path

Procedimientos de recuperación paso a paso

Escenario 1: Pérdida total del directorio ChromaDB

Tiempo estimado: 15-30 min (depende del tamaño del backup)

1. Detener el servicio RAG (2 min)

   sudo systemctl stop rag-api
   # o pm2 stop rag-api, o docker stop rag-container

2. Identificar el backup más reciente (1 min)

   ls -lt /opt/backups/chroma/ | head -5

3. Hacer copia de seguridad del estado actual (1 min)

   mv /opt/rag-app/chroma_db /opt/rag-app/chroma_db.corrupt.$(date +%Y%m%d)

4. Restaurar el backup (5-20 min según tamaño)

   cp -r /opt/backups/chroma/chroma_backup_20240313_020000 /opt/rag-app/chroma_db

5. Validar el restore (2 min)

   python scripts/validate_backup.py /opt/rag-app/chroma_db

6. Reiniciar el servicio (1 min)

   sudo systemctl start rag-api

7. Smoke test (3 min)

   • Ejecutar 3-5 queries de prueba contra el API
   • Verificar que las respuestas sean coherentes

Escenario 2: Restaurar en entorno de staging/pruebas

Tiempo estimado: 10-20 min

1. Crear directorio de pruebas
2. Copiar backup a ese directorio
3. Validar con validate_backup.py
4. Apuntar el servicio de staging al directorio restaurado
5. Ejecutar smoke tests automatizados

Escenario 3: Restaurar una colección desde JSON

Tiempo estimado: 5-15 min (según tamaño de la colección)

1. Identificar el JSON de la colección
2. Crear o vaciar la colección destino
3. Ejecutar import_collection.py
4. Re-indexar si usas embeddings on-the-fly (caso menos frecuente)

Game day: simulación de desastre

Un game day es un ejercicio donde simulas un fallo y ejecutas el runbook real.

Script de simulación

# scripts/game_day_simulation.py
"""
Simula pérdida del índice y recuperación.
EJECUTAR SOLO EN ENTORNO DE PRUEBAS.
"""
import os
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path


def game_day_simulation(
    chroma_path: str,
    backup_dir: str,
    validate_script: str = "scripts/validate_backup.py",
) -> bool:
    """
    1. Backup actual
    2. Borra chroma_path (simula pérdida)
    3. Restaura desde último backup
    4. Valida
    Retorna True si todo ok.
    """
    chroma = Path(chroma_path)
    if not chroma.exists():
        print("❌ No existe chroma_path, no hay nada que simular")
        return False

    # 1. Backup adicional de seguridad
    safe_backup = Path(backup_dir) / f"game_day_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    shutil.copytree(chroma, safe_backup)
    print(f"✓ Backup de seguridad: {safe_backup}")

    # 2. Simular pérdida
    shutil.rmtree(chroma)
    chroma.mkdir(parents=True)
    print("✓ Directorio vaciado (simulación de pérdida)")

    # 3. Encontrar último backup
    backups = sorted(
        Path(backup_dir).glob("chroma_backup_*"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    if not backups:
        print("❌ No hay backups para restaurar")
        return False

    latest = backups[0]
    print(f"✓ Restaurando desde: {latest}")

    # 4. Restaurar
    for item in latest.iterdir():
        dest = chroma / item.name
        if item.is_dir():
            shutil.copytree(item, dest)
        else:
            shutil.copy2(item, dest)

    # 5. Validar
    result = subprocess.run(
        [sys.executable, validate_script, str(chroma)],
        capture_output=True,
        text=True,
    )
    print(result.stdout)
    if result.returncode != 0:
        print(result.stderr)
        return False

    print("✅ Game day completado: restauración exitosa")
    return True


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", default="./chroma_db")
    parser.add_argument("--backup-dir", default="./backups")
    args = parser.parse_args()
    ok = game_day_simulation(args.chroma_path, args.backup_dir)
    sys.exit(0 if ok else 1)

Smoke tests post-recuperación

# scripts/smoke_test_restore.py
"""
Queries de smoke test tras un restore.
"""
import chromadb
import sys


def smoke_test(chroma_path: str, collection_name: str = None) -> bool:
    client = chromadb.PersistentClient(path=chroma_path)
    collections = client.list_collections()

    if not collections:
        print("❌ No hay colecciones")
        return False

    col_name = collection_name or collections[0].name
    col = client.get_collection(col_name)

    # Query simple
    results = col.query(query_texts=["test"], n_results=min(3, col.count()))
    print(f"✓ Query smoke: {len(results['ids'][0])} resultados")

    # Verificar que hay documentos
    if col.count() == 0:
        print("❌ Colección vacía")
        return False

    print("✅ Smoke test pasado")
    return True


if __name__ == "__main__":
    path = sys.argv[1] if len(sys.argv) > 1 else "./chroma_db"
    ok = smoke_test(path)
    sys.exit(0 if ok else 1)

Niveles de criticidad sugeridos

Troubleshooting de DR

1. "Tenemos backup, pero nunca restauramos"

Sin restore test no hay garantía. Los backups pueden estar corruptos, en rutas incorrectas o con permisos erróneos. Acción: calendarizar restore de prueba mensual mínimo y documentar el tiempo real.

2. "El restore tarda demasiado"

Posibles causas: backup muy grande, disco lento, red lenta (si el backup está en almacenamiento remoto). Acción: medir tiempos por etapa (copiar, validar, levantar servicio) y optimizar el cuello de botella. Considerar backups incrementales o por colección si solo falla una.

3. "El equipo no sabe operar el runbook"

Los runbooks se oxidan si no se usan. Acción: hacer game day guiado trimestral, rotar quien lidera el ejercicio y actualizar el runbook con los hallazgos.

4. "El backup falla por disco lleno"

La retención debe ir ligada al espacio disponible. Acción: monitorear espacio en /opt/backups, alertar cuando quede <20% libre, y ajustar retention_days o añadir limpieza más agresiva.

5. "Backup corrupto tras copia con servicio activo"

ChromaDB escribe en SQLite y archivos; una copia con el proceso activo puede dejar datos inconsistentes. Acción: para backup por copia, detener el servicio o usar export por colección (que lee vía API de forma consistente).

Checklist de validación

• Puedo listar backups disponibles con fechas claras.
• Puedo restaurar en entorno de prueba en menos del RTO definido.
• El sistema recuperado responde queries de smoke test correctamente.
• El equipo conoce el runbook y ha practicado al menos una vez.
• Los backups se validan automáticamente (o manualmente de forma periódica).
• La retención se aplica correctamente (30 días o lo definido).

Ejercicios

Ejercicio 1: Primer backup manual

Objetivo: Hacer tu primer backup de una instancia ChromaDB.

1. Crea una colección con 5 documentos de prueba.
2. Detén cualquier proceso que use ChromaDB.
3. Ejecuta el script backup_chromadb.py con --source y --backup-dir.
4. Verifica que exista el directorio de backup con chroma.sqlite3 y index/.

# Crear datos de prueba
import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
col = client.get_or_create_collection("test_backup")
col.add(
    documents=["Doc A", "Doc B", "Doc C", "Doc D", "Doc E"],
    ids=["1", "2", "3", "4", "5"]
)
# Cerrar Python para liberar el lock

# En terminal:
# python scripts/backup_chromadb.py --source ./chroma_db --backup-dir ./backups
# ls -la ./backups/chroma_backup_*/

Ejercicio 2: Restore en directorio nuevo

Objetivo: Restaurar un backup en un directorio distinto y validar.

1. Copia un backup a ./chroma_restored.
2. Ejecuta validate_backup.py sobre ese directorio.
3. Abre un cliente ChromaDB apuntando a ./chroma_restored y ejecuta una query.

cp -r ./backups/chroma_backup_20240313_020000 ./chroma_restored
python scripts/validate_backup.py ./chroma_restored

import chromadb
client = chromadb.PersistentClient(path="./chroma_restored")
col = client.get_collection("test_backup")
print(col.query(query_texts=["Doc"], n_results=3))

Ejercicio 3: Retención de 7 días

Objetivo: Modificar el script de retención para conservar solo 7 días y probarlo.

1. Crea manualmente varios directorios chroma_backup_* con fechas antiguas (usa touch -d para simular).
2. Ejecuta backup_automation.py --retention-only --retention-days 7.
3. Comprueba que se hayan eliminado los backups antiguos.

# Crear backups falsos con fechas antiguas
mkdir -p backups/chroma_backup_20240301_020000
mkdir -p backups/chroma_backup_20240305_020000
touch -d "2024-02-28" backups/chroma_backup_20240301_020000
touch -d "2024-03-02" backups/chroma_backup_20240305_020000

# Ejecutar solo retención
python scripts/backup_automation.py --backup-dir ./backups --retention-days 7 --retention-only

# Verificar: los de febrero/marzo antiguo deberían haberse eliminado
ls -la backups/

Ejercicio 4: Export e import de una colección

Objetivo: Exportar una colección a JSON e importarla en una base nueva.

1. Exporta la colección test_backup a ./exports/test_backup.json.
2. Crea un cliente apuntando a ./chroma_new.
3. Importa el JSON en una colección con el mismo nombre.
4. Compara el count() de la colección original y la nueva.

python scripts/export_collection.py --chroma-path ./chroma_db --collection test_backup --output ./exports/test_backup.json

# import_collection usa el script o:
import scripts.import_collection as imp
imp.import_collection("./chroma_new", "test_backup", "./exports/test_backup.json")

# Verificación
import chromadb
c_orig = chromadb.PersistentClient(path="./chroma_db").get_collection("test_backup")
c_new = chromadb.PersistentClient(path="./chroma_new").get_collection("test_backup")
assert c_orig.count() == c_new.count()

Ejercicio 5: Game day en entorno local

Objetivo: Ejecutar el game day sin tocar producción.

1. Usa un directorio ./chroma_test con datos de prueba.
2. Haz un backup manual previo.
3. Ejecuta game_day_simulation.py con --chroma-path ./chroma_test.
4. Ejecuta smoke_test_restore.py sobre el directorio restaurado.
5. Registra el tiempo total y compáralo con tu RTO.

# Preparar entorno
cp -r ./chroma_db ./chroma_test

# Ejecutar simulación
time python scripts/game_day_simulation.py --chroma-path ./chroma_test --backup-dir ./backups

# Smoke test
python scripts/smoke_test_restore.py ./chroma_test

Ejercicio 6: Cron local (solo Linux/Mac)

Objetivo: Configurar un cron que ejecute el backup diario.

1. Edita tu crontab: crontab -e.
2. Añade la línea del backup a las 02:00 (o a los 5 minutos para prueba).
3. Para prueba rápida, usa */5 * * * * (cada 5 min) y comprueba que se creen backups.
4. Cambia a 0 2 * * * cuando confirmes que funciona.

# Prueba cada 5 minutos (cambiar después)
*/5 * * * * cd /Users/tu/proyecto && python scripts/backup_automation.py >> /tmp/chroma-backup.log 2>&1

# Producción
0 2 * * * cd /opt/rag-app && python scripts/backup_automation.py >> /var/log/chromadb-backup.log 2>&1

Resumen

• RPO define cuánto dato puedes perder; RTO, cuánto tiempo puedes tardar en recuperar. Para RAG, un RPO de 24h y RTO de 2h es un punto de partida razonable.
• ChromaDB se puede respaldar por copia de directorio (offline) o por export/import JSON por colección (online).
• Implementa backup diario automatizado con retención de 30 días usando un script Python y cron o systemd timer.
• Valida los backups tras crearlos o periódicamente; un backup no validado puede fallar en el peor momento.
• Documenta procedimientos de recuperación paso a paso con tiempos estimados y ejecuta game days para practicar.
• El runbook debe estar accesible y el equipo debe haberlo practicado al menos una vez.
• Sin restore test no hay garantía de que el plan funcione; calendarízalo y ajústalo según los resultados.

Recursos adicionales

• Disaster recovery planning - Google Cloud
• ChromaDB persistence docs
• AWS Backup best practices
• SQLite backup (ChromaDB internals)
• Cron tutorial
• systemd timers
• RTO/RPO in practice - Backblaze
• Chaos engineering - principles

Tiempo estimado: 45-60 minutos
Siguiente: 05-seguridad-rag.md