How to Monitor Flyte with Vigilmon

Flyte is a production-grade workflow orchestration platform for machine learning and data engineering. It runs typed, versioned workflows on Kubernetes with native support for Python, Spark, Dask, and Ray — providing reproducibility, artifact lineage tracking, and complex DAGs with dynamic workflows and parallel map tasks.

Production ML pipelines fail in ways that are hard to observe: a model training run that silently completes but produces a corrupted artifact, a feature engineering DAG that hangs on one task while the rest succeed, or a data movement workflow that stops triggering because the Flyte scheduler lost state. Monitoring with Vigilmon gives you an external view of Flyte's control plane health and your pipeline execution status.

Why Monitor Flyte

Flyte is a distributed system running on Kubernetes. Its failure modes span infrastructure, orchestration, and application layers:

1. FlyteAdmin unavailability — the control plane API is unreachable, preventing workflow launches and status queries
2. Workflow execution failures — tasks error out, exceed timeout, or hit resource limits without triggering alerts to the right people
3. Silent hangs — a workflow stalls on a blocking task and never completes, but FlyteAdmin shows it as RUNNING indefinitely
4. Kubernetes pod scheduling failures — tasks queue but never start due to node resource exhaustion or eviction
5. Data lineage gaps — artifact registration succeeds but downstream consumers fail to find the artifact due to catalog issues
6. Propeller degradation — FlytePropeller (the workflow execution engine) can lag or stall under load, causing execution delays across all pipelines
7. Scheduled workflow drift — launchplan schedules can drift or stop firing after infrastructure changes

Without external monitoring, many of these failures are invisible until a downstream consumer (a model serving endpoint, a report, a data sync) produces wrong results.

Key Metrics to Monitor

| Metric | Why it matters | |--------|---------------| | FlyteAdmin API availability | Is the control plane reachable? | | Workflow execution success rate | What fraction of launched workflows complete successfully? | | Task failure rate | Individual task failures before they cascade to workflow failures | | Execution age (hung workflow detection) | Workflows in RUNNING state for longer than expected | | Pod scheduling latency | Time from task ready to pod Running on Kubernetes | | Scheduled launchplan firing | Did the schedule trigger on time? | | Artifact catalog health | Are output artifacts registered and retrievable? |

Setup Guide

1. Monitor the FlyteAdmin API

FlyteAdmin exposes a health endpoint. Monitor it to confirm the control plane is reachable:

1. Monitors → New Monitor
2. Type: HTTP
3. Method: GET
4. URL: https://flyte.yourdomain.com/healthcheck
5. Keyword check: {"status":"healthy"}
6. Interval: 2 minutes
7. Response timeout: 10000ms
8. Save

If you use the gRPC gateway or have a custom ingress, the path may differ. Check your FlyteAdmin deployment configuration for the correct health endpoint.

For self-hosted Flyte using the default Helm chart, the admin service health route is typically:

GET https://flyte.yourdomain.com/healthcheck
# or via the gRPC-gateway
GET https://flyte.yourdomain.com/api/v1/health

2. Monitor the Flyte UI

The Flyte console is the primary operational interface. Monitor it as a secondary signal for platform availability:

1. Type: HTTP
2. URL: https://flyte.yourdomain.com/console
3. Keyword check: Flyte
4. Interval: 5 minutes
5. Save

3. Heartbeat Monitoring for Scheduled Workflows

Flyte's launchplan scheduler fires workflows on a cron schedule. If the scheduler drifts or a launchplan is suspended, workflows stop running silently.

The heartbeat pattern: add a final task to each critical scheduled workflow that pings a Vigilmon heartbeat URL. If the workflow doesn't complete within its expected interval, Vigilmon alerts you.

# workflows/daily_feature_pipeline.py
import flytekit as fl
import httpx
import os

@fl.task
def extract_features() -> fl.FlyteDirectory:
    # ... your feature extraction logic ...
    return fl.FlyteDirectory("/tmp/features")

@fl.task
def transform_features(features: fl.FlyteDirectory) -> fl.FlyteDirectory:
    # ... your transformation logic ...
    return fl.FlyteDirectory("/tmp/transformed")

@fl.task
def ping_heartbeat() -> None:
    """Signal successful workflow completion to Vigilmon."""
    heartbeat_url = os.environ.get("VIGILMON_DAILY_PIPELINE_HEARTBEAT")
    if heartbeat_url:
        try:
            httpx.get(heartbeat_url, timeout=5)
        except Exception:
            pass  # Don't fail the workflow on monitoring ping failure

@fl.workflow
def daily_feature_pipeline() -> None:
    features = extract_features()
    transformed = transform_features(features=features)
    # ping_heartbeat runs last, after all pipeline tasks succeed
    ping_heartbeat()

In Vigilmon:

1. New Monitor → Heartbeat
2. Expected interval: match your launchplan schedule (e.g., 24 hours for a daily pipeline)
3. Grace period: 30 minutes (allows for normal execution time variation)
4. Copy the ping URL into your Flyte task environment as VIGILMON_DAILY_PIPELINE_HEARTBEAT

Now if the launchplan stops firing, a workflow fails before reaching the ping task, or execution time grows beyond the grace period, Vigilmon alerts you.

4. Build a Workflow Health API

For pipelines that run on a schedule, expose a health endpoint that checks recent execution status via the Flyte API:

# health/flyte_health.py
import os
import datetime
from flytekit.remote import FlyteRemote
from flytekit.configuration import Config

def check_workflow_health(
    project: str,
    domain: str,
    workflow_name: str,
    max_age_hours: int = 26,
) -> dict:
    """
    Check that a workflow has completed successfully within the expected window.
    max_age_hours: alert if no successful execution in this many hours.
    """
    remote = FlyteRemote(
        config=Config.auto(),
        default_project=project,
        default_domain=domain,
    )

    try:
        executions = remote.recent_executions(
            project=project,
            domain=domain,
            limit=10,
        )

        now = datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc)
        cutoff = now - datetime.timedelta(hours=max_age_hours)

        recent_successes = [
            e for e in executions
            if e.closure.phase.name == "SUCCEEDED"
            and e.closure.updated_at.replace(tzinfo=datetime.timezone.utc) > cutoff
            and workflow_name in str(e.spec.launch_plan.name)
        ]

        if not recent_successes:
            return {
                "ok": False,
                "workflow": workflow_name,
                "reason": f"No successful execution in the past {max_age_hours}h",
            }

        latest = recent_successes[0]
        return {
            "ok": True,
            "workflow": workflow_name,
            "last_success": latest.closure.updated_at.isoformat(),
        }

    except Exception as e:
        return {
            "ok": False,
            "workflow": workflow_name,
            "error": str(e),
        }

Expose this as a FastAPI health endpoint:

# app/api/health/flyte/route.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from health.flyte_health import check_workflow_health

app = FastAPI()

CRITICAL_WORKFLOWS = [
    {"project": "myproject", "domain": "production", "workflow": "daily_feature_pipeline", "max_age_hours": 26},
    {"project": "myproject", "domain": "production", "workflow": "model_retraining", "max_age_hours": 170},
]

@app.get("/health/flyte")
def flyte_health():
    results = [check_workflow_health(**w) for w in CRITICAL_WORKFLOWS]
    all_ok = all(r["ok"] for r in results)

    if not all_ok:
        return JSONResponse(content={"ok": False, "workflows": results}, status_code=503)

    return {"ok": True, "workflows": results}

Monitor this in Vigilmon:

1. URL: https://yourapp.com/health/flyte
2. Keyword check: "ok":true
3. Interval: 15 minutes
4. Response time warning: 5000ms
5. Save

5. Monitor Kubernetes Infrastructure

Flyte runs on Kubernetes. Infrastructure-level failures affect all workflows. Monitor your Kubernetes API server and the Flyte namespace separately:

# health/k8s_flyte_probe.sh
# Check that FlytePropeller pods are running
kubectl get pods -n flyte -l app=flytepropeller --field-selector=status.phase=Running -o json \
  | jq '.items | length'
# Returns count of running propeller pods — alert if 0

Wrap this in a health endpoint and monitor with Vigilmon:

import subprocess
import json

def check_flyte_pods() -> dict:
    try:
        result = subprocess.run(
            [
                "kubectl", "get", "pods",
                "-n", "flyte",
                "-l", "app=flytepropeller",
                "--field-selector=status.phase=Running",
                "-o", "json",
            ],
            capture_output=True,
            text=True,
            timeout=10,
        )
        data = json.loads(result.stdout)
        running_count = len(data.get("items", []))
        return {
            "ok": running_count > 0,
            "flytepropeller_running": running_count,
        }
    except Exception as e:
        return {"ok": False, "error": str(e)}

Alerting

Configure alert channels for your Flyte monitors:

1. Open monitor → Alerts → Add Channel
2. Recommended setup:
   • Slack #ml-ops: immediate notification on control plane failure or pipeline miss
   • Email: on-call ML engineer for sustained failures
   • PagerDuty: if a Flyte pipeline feeds a live production model serving endpoint

Recommended thresholds:

| Monitor | Alert condition | Escalation | |---------|----------------|------------| | FlyteAdmin API | 1 failure | 2 consecutive → email | | Workflow health endpoint | 1 failure | Immediate → Slack + email | | Scheduled pipeline heartbeat | 1 missed interval | Immediate → Slack + PagerDuty | | Kubernetes propeller pods | 0 running | Immediate → PagerDuty |

Alert routing by severity

• FlyteAdmin down: page on-call immediately — no workflows can be launched or monitored
• Workflow health endpoint 503: Slack alert with workflow name — investigate recent execution logs in FlyteConsole
• Missed heartbeat: PagerDuty alert — launchplan may be suspended or execution may be hung

Conclusion

Flyte's typed workflows, versioning, and Kubernetes-native execution make it a powerful foundation for production ML pipelines. But the same distributed architecture that makes it powerful also means failures can be silent and hard to observe without external monitoring.

Vigilmon's heartbeat monitors are the most valuable tool for Flyte: they confirm that scheduled pipelines actually complete successfully, not just that the control plane is reachable. Layer in the FlyteAdmin health check and the workflow health API endpoint for full coverage.

Start with a heartbeat monitor for your most critical scheduled pipeline (5 minutes to set up), then add FlyteAdmin availability monitoring as a foundation.

Further Reading

• Monitoring Kubernetes with Vigilmon
• Monitoring Meltano with Vigilmon
• Monitoring Docker with Vigilmon