Unstructured Monitoring Tutorial

title: "How to Monitor Unstructured with Vigilmon" published: true description: Unstructured powers the document parsing stage of your RAG pipeline — PDFs, Word docs, PowerPoint, emails, and images. Here's how to monitor Unstructured API availability, detect pipeline failures, and alert when document processing stalls. tags: unstructured, rag, ai, monitoring cover_image:

Unstructured handles the hardest part of any RAG pipeline: turning heterogeneous documents — PDFs, Word files, PowerPoint presentations, HTML pages, emails, and scanned images — into clean, structured elements that language models can reason over. When Unstructured is unavailable or returns partition errors, your entire document ingestion pipeline stalls and your knowledge base stops growing. Vigilmon gives you HTTP health checks, cron heartbeats for document processing jobs, and instant alerts so you catch Unstructured failures before your RAG application serves stale content.

What You'll Set Up

• HTTP monitor for the Unstructured API endpoint
• Health wrapper endpoint that probes Unstructured document parsing
• Cron heartbeat for scheduled document processing pipelines
• Alert channels for on-call notification

Prerequisites

• An Unstructured account with an API key from unstructured.io
• A service that calls the Unstructured API (hosted or local)
• A free Vigilmon account

Step 1: Add a Health Endpoint to Your Document Processing Service

Your application calls the Unstructured API to partition documents before indexing. Add a /health endpoint with a lightweight process check and a deeper Unstructured probe:

import os
import httpx
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

UNSTRUCTURED_API_KEY = os.environ["UNSTRUCTURED_API_KEY"]
UNSTRUCTURED_API_URL = os.environ.get(
    "UNSTRUCTURED_API_URL", "https://api.unstructured.io/general/v0/general"
)


@app.get("/health")
async def health():
    """Fast check — no external calls."""
    return {"status": "ok", "service": "document-processor"}


@app.get("/health/deep")
async def deep_health():
    """Deep check — verifies Unstructured API is reachable and partitioning works."""
    import base64

    # Minimal valid PDF bytes (empty 1-page PDF)
    TINY_PDF_B64 = (
        "JVBERi0xLjQKMSAwIG9iago8PCAvVHlwZSAvQ2F0YWxvZyAvUGFnZXMgMiAwIFIgPj4KZW5k"
        "b2JqCjIgMCBvYmoKPDwgL1R5cGUgL1BhZ2VzIC9LaWRzIFszIDAgUl0gL0NvdW50IDEgPj4K"
        "ZW5kb2JqCjMgMCBvYmoKPDwgL1R5cGUgL1BhZ2UgL1BhcmVudCAyIDAgUiAvTWVkaWFCb3gg"
        "WzAgMCA2MTIgNzkyXSA+PgplbmRvYmoKeHJlZgowIDQKMDAwMDAwMDAwMCA2NTUzNSBmIAow"
        "MDAwMDAwMDA5IDAwMDAwIG4gCjAwMDAwMDAwNTggMDAwMDAgbiAKMDAwMDAwMDExNSAwMDAw"
        "MCBuIAp0cmFpbGVyCjw8IC9TaXplIDQgL1Jvb3QgMSAwIFIgPj4Kc3RhcnR4cmVmCjE5MAol"
        "JUVPRQ=="
    )
    pdf_bytes = base64.b64decode(TINY_PDF_B64)

    try:
        async with httpx.AsyncClient(timeout=20.0) as client:
            resp = await client.post(
                UNSTRUCTURED_API_URL,
                headers={"unstructured-api-key": UNSTRUCTURED_API_KEY},
                files={"files": ("health_check.pdf", pdf_bytes, "application/pdf")},
                data={"strategy": "fast"},
            )
            if resp.status_code == 200:
                elements = resp.json()
                return {
                    "status": "ok",
                    "unstructured_reachable": True,
                    "elements_returned": len(elements),
                }
            return JSONResponse(
                status_code=503,
                content={"status": "error", "http_status": resp.status_code},
            )
    except httpx.TimeoutException:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "detail": "Unstructured API timeout"},
        )
    except Exception as e:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "detail": str(e)},
        )

Use /health for 1-minute Vigilmon probes and /health/deep for 5-minute checks that validate Unstructured can actually partition a document.

Step 2: Monitor the Unstructured API Endpoint

With the health endpoints deployed, add Vigilmon monitors:

1. Log in to vigilmon.online and click Add Monitor.
2. Set Type to HTTP / HTTPS.
3. URL: https://api.unstructured.io/healthcheck
4. Check interval: 5 minutes
5. Expected HTTP status: 200
6. Keyword check: "healthcheck" or "ok"
7. Click Save.

Add a second monitor for your application's deep health endpoint:

1. Click Add Monitor → HTTP / HTTPS.
2. URL: https://your-service.com/health/deep
3. Check interval: 5 minutes
4. Response timeout: 30 seconds — Unstructured's hi-res partition strategy can take time on large documents.
5. Keyword check: "ok"
6. Click Save.

Step 3: Heartbeat Monitoring for Document Processing Pipelines

Document processing pipelines are batch jobs — a nightly run that partitions new PDFs into chunks for your vector store. These jobs have no persistent HTTP listener, so HTTP monitoring cannot catch silent failures. Use a Vigilmon heartbeat:

1. In Vigilmon, click Add Monitor → Cron Heartbeat.
2. Set the expected ping interval to match your processing schedule (e.g. 1440 minutes for a daily job).
3. Copy the heartbeat URL.

Instrument your pipeline to ping the heartbeat after each successful run:

import os
import httpx
import logging
from pathlib import Path

logger = logging.getLogger(__name__)

UNSTRUCTURED_API_KEY = os.environ["UNSTRUCTURED_API_KEY"]
UNSTRUCTURED_API_URL = os.environ.get(
    "UNSTRUCTURED_API_URL", "https://api.unstructured.io/general/v0/general"
)
VIGILMON_HEARTBEAT_URL = os.environ.get("VIGILMON_HEARTBEAT_URL")


async def partition_document(doc_path: Path, strategy: str = "hi_res") -> list[dict]:
    async with httpx.AsyncClient(timeout=120.0) as client:
        with open(doc_path, "rb") as f:
            resp = await client.post(
                UNSTRUCTURED_API_URL,
                headers={"unstructured-api-key": UNSTRUCTURED_API_KEY},
                files={"files": (doc_path.name, f, "application/octet-stream")},
                data={"strategy": strategy, "chunking_strategy": "by_title"},
            )
        resp.raise_for_status()
        return resp.json()


async def run_nightly_document_pipeline():
    """Partition all pending documents and ping heartbeat on success."""
    pending_dir = Path(os.environ["PENDING_DOCS_DIR"])
    processed_dir = Path(os.environ["PROCESSED_DOCS_DIR"])
    failed = []

    for doc_path in pending_dir.iterdir():
        if not doc_path.is_file():
            continue
        try:
            elements = await partition_document(doc_path)
            await index_elements(elements, source=doc_path.name)
            doc_path.rename(processed_dir / doc_path.name)
            logger.info(f"Processed {doc_path.name}: {len(elements)} elements")
        except Exception as e:
            logger.error(f"Failed to process {doc_path.name}: {e}")
            failed.append(doc_path.name)

    if failed:
        logger.error(f"Pipeline completed with failures: {failed}")
        # Do NOT ping heartbeat — absence of ping is the alert
        return

    if VIGILMON_HEARTBEAT_URL:
        async with httpx.AsyncClient() as client:
            await client.get(VIGILMON_HEARTBEAT_URL, timeout=10)
        logger.info("Heartbeat pinged — pipeline succeeded")


async def index_elements(elements: list[dict], source: str):
    """Stub: push parsed elements to your vector store."""
    pass

If a document fails to parse — a corrupted PDF, an oversized image that times out OCR — the pipeline skips the heartbeat ping and Vigilmon alerts your team after the expected interval.

Step 4: Track Partition Strategy Errors

Unstructured supports multiple partition strategies: fast, hi_res, and ocr_only. The hi_res and ocr_only strategies call document intelligence models and can fail on specific file types. Track per-strategy error rates:

import time
from collections import defaultdict

partition_metrics = defaultdict(lambda: {"success": 0, "failure": 0, "total_ms": 0})


async def partition_with_metrics(doc_path: Path, strategy: str) -> list[dict]:
    start = time.monotonic()
    try:
        result = await partition_document(doc_path, strategy=strategy)
        elapsed_ms = int((time.monotonic() - start) * 1000)
        partition_metrics[strategy]["success"] += 1
        partition_metrics[strategy]["total_ms"] += elapsed_ms
        return result
    except Exception:
        partition_metrics[strategy]["failure"] += 1
        raise


@app.get("/health/metrics")
async def pipeline_metrics():
    return {
        "partition_metrics": dict(partition_metrics),
        "status": "ok",
    }

Point a Vigilmon monitor at /health/metrics with a keyword check for "ok". When failure counts rise, your partition pipeline is degrading.

Step 5: Monitor the Unstructured Status Page

Unstructured publishes service status for API incidents. Add a status page monitor:

1. In Vigilmon, click Add Monitor → HTTP / HTTPS.
2. URL: https://status.unstructured.io
3. Keyword check: All Systems Operational
4. Check interval: 5 minutes
5. Click Save.

When Unstructured posts an incident — OCR service issues, API latency — the keyword will no longer match and Vigilmon alerts your team.

Step 6: Configure Alert Channels

1. Go to Alert Channels in Vigilmon and add Slack, email, or a webhook.
2. For the HTTP deep health monitor, set Consecutive failures before alert to 2 — partition requests can occasionally time out on large documents without indicating a true outage.
3. For the heartbeat monitor, keep the default 1 missed interval — a missed ingest means stale data in your knowledge base.

Use the Vigilmon API to suppress alerts during planned reindexing operations:

curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{"monitor_id": "abc123", "duration_minutes": 60}'

python reindex_all_documents.py

Summary

| Monitor | Target | What It Catches | |---|---|---| | HTTP endpoint | /health | Service crash, startup failure | | Deep health check | /health/deep | Unstructured API unreachable, key invalid, partition failure | | Cron heartbeat | Heartbeat URL | Pipeline crash, missed schedule, partial batch failure | | Status page | status.unstructured.io | Unstructured provider-side incidents | | Pipeline metrics | /health/metrics | Rising partition error rates by strategy |

Unstructured eliminates the engineering burden of multi-format document parsing — Vigilmon ensures your document processing pipeline is monitored just as rigorously as the RAG retrieval it feeds. With heartbeats on every ingestion job and health checks on every API call, you'll catch failures before your knowledge base goes stale.

Get started free at vigilmon.online — your first monitor is running in under a minute.