How to Monitor CocoIndex Connector Status: Real-Time Observability Guide

CocoIndex exposes connector health through asynchronous update snapshots that stream via UpdateHandle.watch(), providing real-time visibility into processing status, per-component statistics, and error counts.

Monitoring data connector health is critical for production pipelines. In the cocoindex-io/cocoindex framework, every connector reports its status through structured update snapshots emitted during execution. This guide shows you how to stream live statistics, detect failures, and inspect granular performance data for any connected data source.

Understanding Update Snapshots and Status Types

The cocoindex engine reports health metrics through immutable update snapshots that contain three key fields: stats (hierarchical statistics), status (current execution state), and result (final output value). These snapshots are defined in cocoindex/_internal/update_stats.py and yielded continuously while an App runs.

Core Data Structures

According to the source code in cocoindex/_internal/update_stats.py, each snapshot contains:

• UpdateStatus: An enum with values RUNNING (work is in progress) and READY (the root component has caught up)
• UpdateStats: A hierarchical object aggregating counters across all components
• ComponentStats: Per-connector statistics including num_adds, num_deletes, num_processed, and num_errors
• by_component: A dictionary mapping component names (e.g., "localfs", "postgres") to their respective ComponentStats

Status Lifecycle Flow

When you initiate an update, the status transitions from RUNNING to READY once the root component completes its current batch. Connectors register their component names internally when mounted (e.g., via coco.mount()), allowing you to isolate health metrics for specific data sources within the by_component map.

Real-Time Async Monitoring with UpdateHandle.watch()

The primary method for monitoring connector status is the async iterator returned by UpdateHandle.watch(), implemented in cocoindex/_internal/app.py. This yields UpdateSnapshot objects continuously as the pipeline executes.

import cocoindex as coco
from cocoindex.connectors import localfs
from pathlib import Path

async def monitor_connectors():
    # Declare and mount your connectors

    target = await coco.use_mount(localfs.declare_dir_target, Path("./out"))
    await coco.mount(process_file, source_path, target)
    
    # Initialize the app and get update handle

    app = coco.App(coco.AppConfig(name="monitor-demo"), main)
    handle = app.update()
    
    # Stream live status updates

    async for snap in handle.watch():
        print(f"Status: {snap.status.value}")
        print(f"Total processed: {snap.stats.total.num_processed}")
        print(f"Errors so far: {snap.stats.total.num_errors}")
        
        # Drill into specific connector stats

        if "localfs" in snap.stats.by_component:
            ls = snap.stats.by_component["localfs"]
            print(f"  localfs adds={ls.num_adds} deletes={ls.num_deletes}")

    # Access final result after completion

    result = await handle.result()

Key implementation details:

• Component isolation: Use snap.stats.by_component["connector_name"] to access connector-specific counters
• Error detection: Check num_errors within component stats to identify failing connectors without halting the pipeline
• Real-time updates: The engine emits new snapshots whenever any connector changes state (new file detected, row inserted, error occurred)

Synchronous Monitoring Using update_blocking()

For scripts or CI pipelines that don't use async/await, use App.update_blocking() with the report_to_stdout=True flag. This method wraps the async watch iterator and prints formatted progress to the console.

import cocoindex as coco
from cocoindex.connectors import postgres

def run_sync_monitoring():
    app = coco.App(
        coco.AppConfig(name="postgres-sync"),
        my_root_fn,
        db_url="postgresql://user:pwd@localhost/db",
    )
    
    # Runs update and prints progress bar + live stats

    app.update_blocking(report_to_stdout=True)

if __name__ == "__main__":
    run_sync_monitoring()

The CLI provides identical functionality through the --progress flag, as implemented in cocoindex/cli.py:

coco run my_app.py --progress

Both approaches internally call UpdateHandle.watch() and render the same UpdateStatus and UpdateStats structures to stdout.

Accessing Connector Statistics After Completion

For post-run analysis or CI gate checks, access the final statistics snapshot via handle.stats() after awaiting handle.result(). This returns the last UpdateStats object containing cumulative counts across all components.

handle = app.update()
await handle.result()  # Block until completion

# Retrieve final snapshot

final_stats = handle.stats()
if final_stats:
    pg_stats = final_stats.by_component.get("postgres")
    if pg_stats:
        print("Postgres rows added:", pg_stats.num_adds)
        print("Postgres errors:", pg_stats.num_errors)
        
        # Fail CI if connector reported errors

        if pg_stats.num_errors > 0:
            raise RuntimeError("Connector encountered errors during sync")

This pattern is particularly useful for validating that connector targets (defined in files like cocoindex/connectors/postgres/_target.py or cocoindex/connectors/localfs/_target.py) successfully processed all expected records.

Summary

• Update snapshots provide real-time visibility into connector health through UpdateHandle.watch()
• snap.stats.by_component isolates statistics (adds, deletes, errors) per connector using the component names registered during mounting
• UpdateStatus transitions from RUNNING to READY when the root component catches up
• Synchronous monitoring is available via update_blocking(report_to_stdout=True) or the CLI --progress flag
• Post-run inspection via handle.stats() enables CI/CD validation of connector success

Frequently Asked Questions

How do I check if a specific connector has errors?

Access the num_errors counter within the component-specific statistics. After obtaining a snapshot from handle.watch() or handle.stats(), retrieve the connector's stats via snap.stats.by_component["connector_name"] (e.g., "postgres" or "localfs") and verify component_stats.num_errors == 0.

What is the difference between RUNNING and READY status?

RUNNING indicates active processing is occurring within the pipeline, while READY signals that the root component has caught up with the source data and no immediate work remains. The status transitions to READY only on the final snapshot before completion.

Can I monitor connectors without using async code?

Yes. Use App.update_blocking(report_to_stdout=True) to run the update synchronously while printing live statistics to the console. Alternatively, use the CLI command coco run my_app.py --progress, which invokes the same underlying monitoring logic defined in cocoindex/cli.py.

Where are connector component names defined?

Component names are registered internally when you call mounting functions like coco.mount() or connector-specific targets. The names appear as keys in the by_component dictionary (e.g., "localfs" for filesystem connectors or "postgres" for PostgreSQL targets), sourced from the connector target modules located in paths like cocoindex/connectors/postgres/_target.py.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}