tutorial

Monitoring Logfire (Pydantic) with Vigilmon: Uptime, Heartbeat, and Alerting for Observability Pipelines

How to monitor Logfire (Pydantic's observability platform) with Vigilmon — HTTP uptime checks for Logfire-instrumented services, heartbeat monitors for telemetry pipelines, and alerting when observability coverage goes dark.

Logfire is Pydantic's production-ready observability platform that brings OpenTelemetry-compatible tracing, structured logging, and metrics to Python applications — with deep integration into FastAPI, Django, SQLAlchemy, and other popular frameworks. Logfire instruments your application code and ships telemetry to Logfire's cloud backend (or a self-hosted OpenTelemetry Collector). The irony of observability tools is that they can go dark themselves: if the Logfire SDK silently stops exporting, your dashboards look healthy while actual failures go undetected. Vigilmon adds an external layer — HTTP uptime checks for your Logfire-instrumented services, heartbeat monitors for telemetry export jobs, and alerting when observability coverage gaps appear.

What You'll Build

  • A Vigilmon HTTP monitor for your Logfire-instrumented service's health endpoint
  • A heartbeat monitor that signals when your telemetry export pipeline is active
  • An HTTP monitor for a custom /health/logfire endpoint that verifies the SDK is connected
  • Alert thresholds for when Logfire instrumentation goes silent

Prerequisites

Step 1: Verify Your Logfire Instrumentation

Confirm Logfire is initialized and exporting before setting up monitoring:

# main.py — Basic Logfire setup
import logfire

logfire.configure(
    service_name="my-api",
    service_version="1.0.0",
    send_to_logfire=True,  # Set to False for local-only dev
)

# Test that spans are emitted
with logfire.span("startup-check"):
    logfire.info("Service starting up", env="production")

Run it and confirm you see traces in the Logfire dashboard at logfire.pydantic.dev.

Check the SDK version and configuration:

python -c "import logfire; print(logfire.__version__)"

For FastAPI with automatic instrumentation:

import logfire
from fastapi import FastAPI

logfire.configure(service_name="my-fastapi-app")
app = FastAPI()
logfire.instrument_fastapi(app)

@app.get("/health")
def health():
    return {"status": "ok"}

Step 2: Expose a Health Endpoint That Reports Logfire Status

Add an endpoint to your service that checks whether the Logfire exporter is connected and flushing spans. This gives Vigilmon something concrete to poll.

FastAPI Health Endpoint

# health.py — Logfire SDK connectivity check
import logfire
import time
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# Track last successful span export time
_last_export_ts: float = 0.0

MAX_EXPORT_GAP_SECONDS = 300  # alert if no export in 5 minutes


def record_export():
    """Call this after any successful instrumented operation."""
    global _last_export_ts
    _last_export_ts = time.time()


@app.get("/health")
def health():
    """Basic liveness probe — no Logfire dependency."""
    return {"status": "ok"}


@app.get("/health/logfire")
def logfire_health():
    """
    Checks that the Logfire exporter is active by emitting a span
    and verifying the export timestamp is recent.
    """
    try:
        with logfire.span("health-check", _tags=["vigilmon"]):
            record_export()
            elapsed = time.time() - _last_export_ts

        if elapsed > MAX_EXPORT_GAP_SECONDS:
            return JSONResponse(status_code=503, content={
                "status": "degraded",
                "reason": "no successful export in last 5 minutes",
                "last_export_seconds_ago": round(elapsed, 1),
            })

        return {
            "status": "ok",
            "last_export_seconds_ago": round(elapsed, 1),
            "logfire_configured": logfire.DEFAULT_LOGFIRE_INSTANCE is not None,
        }
    except Exception as e:
        return JSONResponse(status_code=503, content={
            "status": "down",
            "error": str(e),
        })

# Start the service and verify the endpoint
uvicorn health:app --host 0.0.0.0 --port 8000

curl http://localhost:8000/health/logfire
# {"status":"ok","last_export_seconds_ago":0.2,"logfire_configured":true}

Django Health View

# monitoring/views.py
import logfire
import time
from django.http import JsonResponse

_last_export_ts: float = 0.0


def logfire_health(request):
    global _last_export_ts
    try:
        with logfire.span("health-check"):
            _last_export_ts = time.time()

        elapsed = time.time() - _last_export_ts
        if elapsed > 300:
            return JsonResponse(
                {"status": "degraded", "last_export_seconds_ago": round(elapsed, 1)},
                status=503,
            )
        return JsonResponse({"status": "ok", "last_export_seconds_ago": round(elapsed, 1)})
    except Exception as e:
        return JsonResponse({"status": "down", "error": str(e)}, status=503)

# urls.py
from django.urls import path
from monitoring.views import logfire_health

urlpatterns = [
    path("health/logfire/", logfire_health),
]

Step 3: Create Vigilmon HTTP Monitors

Monitor the Primary Health Endpoint

  1. Log in to Vigilmon → Add Monitor → HTTP.
  2. URL: https://your-service.example.com/health.
  3. Check interval: 60 seconds.
  4. Response timeout: 10 seconds.
  5. Expected status: 200.
  6. Keyword: "status":"ok".
  7. Click Save.

This catches service-level outages regardless of Logfire status.

Monitor the Logfire-Specific Endpoint

  1. Add Monitor → HTTP.
  2. URL: https://your-service.example.com/health/logfire.
  3. Check interval: 5 minutes (aligned with the MAX_EXPORT_GAP_SECONDS threshold).
  4. Response timeout: 15 seconds (span export can take a moment).
  5. Expected status: 200.
  6. Keyword: "status":"ok".
  7. Click Save.

What This Catches

| Failure Mode | Primary Health | Logfire Health | |---|---|---| | Service process crash | ✓ | ✓ | | Logfire SDK not initialized | ✗ | ✓ | | Network block to logfire.pydantic.dev | ✗ | ✓ | | Logfire token expired or revoked | ✗ | ✓ | | OpenTelemetry exporter silently dropped | ✗ | ✓ | | Service alive but emitting zero spans | ✗ | ✓ |

Step 4: Heartbeat Monitoring for Telemetry Pipelines

If you run a batch telemetry pipeline — for example, a script that reads events and emits structured logs to Logfire on a schedule — use a Vigilmon heartbeat monitor to detect when it stops running.

Create the Heartbeat Monitor

  1. Add Monitor → Heartbeat.
  2. Name: logfire-telemetry-pipeline.
  3. Expected interval: Match your pipeline cadence (e.g., 3600 seconds for hourly).
  4. Grace period: 15 minutes.
  5. Copy the unique heartbeat URL: https://vigilmon.online/heartbeat/your-unique-id.

Wire Into a Batch Pipeline

# telemetry_pipeline.py
import os
import logfire
import requests

VIGILMON_HEARTBEAT = os.environ["VIGILMON_HEARTBEAT_URL"]

logfire.configure(service_name="telemetry-pipeline")


def process_batch():
    """Fetch events and emit structured logs to Logfire."""
    events = fetch_events()  # your data source
    with logfire.span("process-batch", event_count=len(events)):
        for event in events:
            logfire.info(
                "Processing event",
                event_id=event["id"],
                event_type=event["type"],
                _tags=["batch"],
            )
    return len(events)


def main():
    with logfire.span("pipeline-run"):
        count = process_batch()
        logfire.info("Pipeline complete", events_processed=count)

    # Ping heartbeat only after successful completion
    try:
        requests.get(VIGILMON_HEARTBEAT, timeout=10)
    except Exception as e:
        logfire.warn("Heartbeat ping failed", error=str(e))


if __name__ == "__main__":
    main()

Add to crontab:

0 * * * * /usr/bin/python3 /opt/pipeline/telemetry_pipeline.py >> /var/log/telemetry-pipeline.log 2>&1

Wire Into a Celery Task

# tasks.py
import logfire
import requests
import os
from celery import Celery

app = Celery("tasks", broker=os.environ["CELERY_BROKER_URL"])
logfire.configure(service_name="celery-worker")
logfire.instrument_celery(app)

VIGILMON_HEARTBEAT = os.environ["VIGILMON_HEARTBEAT_URL"]


@app.task(name="telemetry.sync")
def sync_telemetry():
    with logfire.span("celery-sync-telemetry"):
        events = fetch_events()
        for event in events:
            logfire.info("Syncing event", event_id=event["id"])

    # Ping heartbeat after successful task completion
    try:
        requests.get(VIGILMON_HEARTBEAT, timeout=10)
    except Exception:
        pass


# Schedule in celery beat:
# app.conf.beat_schedule = {
#     "sync-telemetry-hourly": {
#         "task": "telemetry.sync",
#         "schedule": 3600.0,
#     }
# }

Step 5: Monitoring Logfire With a Self-Hosted OpenTelemetry Collector

If you route telemetry through a self-hosted OpenTelemetry Collector before forwarding to Logfire, monitor the collector's health endpoint too:

# OpenTelemetry Collector exposes a health check on port 13133 by default
curl http://otel-collector.example.com:13133/

# Returns: {"status":"Server available","upSince":"2026-07-03T02:00:00Z","uptime":"2h15m"}

Add a Vigilmon HTTP monitor:

  1. Add Monitor → HTTP.
  2. URL: http://otel-collector.example.com:13133/.
  3. Check interval: 60 seconds.
  4. Expected status: 200.
  5. Keyword: Server available.
  6. Click Save.

Collector Configuration With Health Check

# otel-collector-config.yaml
extensions:
  health_check:
    endpoint: "0.0.0.0:13133"

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"
      http:
        endpoint: "0.0.0.0:4318"

exporters:
  otlphttp:
    endpoint: "https://logfire-api.pydantic.dev/v1/traces"
    headers:
      authorization: "Bearer ${LOGFIRE_TOKEN}"

service:
  extensions: [health_check]
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [otlphttp]

Step 6: Configure Alerting

In Vigilmon under Settings → Notifications, configure alert routing:

| Monitor | Trigger | Recommended Action | |---|---|---| | HTTP (primary health) | 503 or keyword absent | Check service process; inspect application logs | | HTTP (Logfire health endpoint) | 503 or status not ok | Check Logfire SDK init; verify LOGFIRE_TOKEN; check network to logfire.pydantic.dev | | Heartbeat (batch pipeline) | Ping not received | Check cron/Celery logs; confirm pipeline process is running | | HTTP (OTel Collector) | Health check failed | Restart collector; inspect collector logs for export errors |

Recommended thresholds:

  • Confirmation period: 2 failures for the primary health endpoint (restarts are brief); 1 failure for Logfire-specific checks (a silent exporter is unambiguous)
  • Response time alert: 2000ms for the /health/logfire endpoint (span export should be fast; slowness indicates exporter backpressure)
  • Recovery notification: Enable so your team knows when telemetry is flowing again

Common Logfire / Observability Pipeline Failure Modes

| Scenario | What Vigilmon Catches | |---|---| | Service process crash | Primary health check fails | | logfire.configure() not called at startup | Logfire health endpoint returns 503 | | LOGFIRE_TOKEN expired or rotated | SDK silently stops exporting; health endpoint detects gap | | Firewall blocks logfire.pydantic.dev | Spans buffered indefinitely; health endpoint detects gap | | OTel Collector OOM-killed | Collector health check fails; traces stop reaching Logfire | | Celery worker not running | Heartbeat goes silent | | Cron job misconfigured after server reboot | Heartbeat goes silent | | Logfire cloud incident | All Logfire health endpoints degrade; primary service health still ok |

Distinguishing Service Health From Observability Health

A key insight when monitoring Logfire-instrumented services: service health and observability health are independent failure modes. Your service can be perfectly healthy while emitting zero telemetry, and it can be emitting rich telemetry while serving 503s to real users. Vigilmon monitors both:

  • Primary health endpoint (/health) → catches service-level failures
  • Logfire health endpoint (/health/logfire) → catches observability blind spots
  • Heartbeat monitors → catch pipeline-level gaps in telemetry ingestion

Run both monitors simultaneously so you always know whether a gap in your Logfire dashboards means the service is down or the instrumentation is broken.

Logfire adds powerful observability to Python applications, but observability tools need their own monitoring layer. Vigilmon's HTTP and heartbeat monitors give you external validation that your Logfire SDK is active, your telemetry pipelines are running, and your OTel Collector is healthy — so you're never flying blind about whether you're flying blind.

Get started free at vigilmon.online — your Logfire observability monitor is running in under 5 minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles