tutorial

How to Monitor Netflix Conductor with Vigilmon

Netflix Conductor workflow engines can fail silently when the task queue backs up, workers stop polling, or the server loses database connectivity. Learn how to monitor Conductor server health, task queues, and workflow completion rates with Vigilmon.

Netflix Conductor is a microservices orchestration engine that manages complex, long-running workflows by coordinating distributed workers through a central task queue. Originally built by Netflix to handle media encoding pipelines, Conductor is now widely used for order processing, ML pipelines, and data transformation workflows. But Conductor's distributed architecture has multiple failure surfaces: the server can be healthy while no workers are polling, a task type can have no healthy executors, or a workflow can spin in retry loops indefinitely without completing.

Vigilmon provides external observability for Conductor through HTTP endpoint monitoring of the Conductor API and heartbeat monitoring for workflow completion. This tutorial shows you how to detect Conductor failures before they become workflow backlogs that affect real users.

Why Conductor Needs External Monitoring

Conductor exposes a Swagger UI and a comprehensive REST API — but these are passive. External monitoring with Vigilmon adds:

  • Server availability checks from outside your network perimeter every minute
  • Task queue depth monitoring — a growing queue with no decrease means workers have stopped
  • Worker liveness checks via the Conductor task poll health endpoint
  • Workflow completion heartbeats — verify that your most critical workflow types complete end-to-end
  • Retry storm detection — workflows stuck in infinite retry loops inflate queue depth silently

Conductor's most dangerous failure mode is a running server with no active workers. The UI loads, the API responds, and workflows are created — but nothing executes. Workflow definitions sit in IN_PROGRESS state indefinitely while your system appears healthy.

Step 1: Probe the Conductor Server Health Endpoint

Conductor exposes a health endpoint via Spring Boot Actuator (versions ≥ 3.x) and a metadata endpoint that serves as an availability indicator in older versions:

# Conductor 3.x+ health (Spring Boot Actuator)
curl -s http://localhost:8080/health | jq .

# Metadata endpoint (availability indicator in all versions)
curl -s http://localhost:8080/api/metadata/workflow | jq .length

# Task queue summary
curl -s http://localhost:8080/api/tasks/queue/all | jq .

Health Proxy with Queue Depth Monitoring (Node.js)

A thin proxy gives you a single Vigilmon-compatible endpoint that checks server health and task queue depth:

// conductor-health.js
const express = require('express');
const axios = require('axios');

const app = express();
const CONDUCTOR_URL = process.env.CONDUCTOR_URL || 'http://localhost:8080';
const QUEUE_DEPTH_THRESHOLD = parseInt(process.env.QUEUE_DEPTH_THRESHOLD || '100');

app.get('/health/conductor', async (req, res) => {
  try {
    // Check if server is responsive
    const healthRes = await axios.get(`${CONDUCTOR_URL}/health`, { timeout: 5000 });
    const status = healthRes.data?.status;

    if (status && status !== 'UP') {
      return res.status(503).json({ status: 'down', reason: 'actuator_status_not_up', serverStatus: status });
    }

    // Get task queue depths
    const queueRes = await axios.get(`${CONDUCTOR_URL}/api/tasks/queue/all`, { timeout: 8000 });
    const queues = queueRes.data || {};
    const totalQueueDepth = Object.values(queues).reduce((sum, size) => sum + (size || 0), 0);

    // Find individual overloaded queues
    const overloadedQueues = Object.entries(queues)
      .filter(([, size]) => size > QUEUE_DEPTH_THRESHOLD)
      .map(([name, size]) => ({ name, size }));

    if (overloadedQueues.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'task_queues_overloaded',
        totalQueueDepth,
        overloadedQueues,
        threshold: QUEUE_DEPTH_THRESHOLD,
      });
    }

    return res.status(200).json({
      status: 'ok',
      totalQueueDepth,
      queueCount: Object.keys(queues).length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3007);

Python Health Proxy (FastAPI)

# conductor_health.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import httpx, os

app = FastAPI()

CONDUCTOR_URL = os.environ.get("CONDUCTOR_URL", "http://localhost:8080")
QUEUE_DEPTH_THRESHOLD = int(os.environ.get("QUEUE_DEPTH_THRESHOLD", "100"))

@app.get("/health/conductor")
async def conductor_health():
    async with httpx.AsyncClient(timeout=8.0) as client:
        try:
            # Check server health
            health = await client.get(f"{CONDUCTOR_URL}/health")
            if health.json().get("status") not in ("UP", None):
                return JSONResponse(status_code=503, content={
                    "status": "down", "serverStatus": health.json().get("status")
                })

            # Check queue depths
            queues = await client.get(f"{CONDUCTOR_URL}/api/tasks/queue/all")
            queue_data = queues.json()
            total_depth = sum(queue_data.values())
            overloaded = [
                {"name": k, "size": v}
                for k, v in queue_data.items()
                if v > QUEUE_DEPTH_THRESHOLD
            ]

            if overloaded:
                return JSONResponse(status_code=503, content={
                    "status": "degraded",
                    "reason": "task_queues_overloaded",
                    "totalQueueDepth": total_depth,
                    "overloadedQueues": overloaded,
                })

            return {"status": "ok", "totalQueueDepth": total_depth}

        except Exception as e:
            return JSONResponse(status_code=503, content={"status": "down", "error": str(e)})

Step 2: Monitor Worker Liveness

Conductor workers are separate processes that poll the server for tasks. They can go down independently of the server. Conductor provides an API to check which workers are active for each task type:

// worker-health endpoint added to conductor-health.js
const REQUIRED_TASK_TYPES = (process.env.REQUIRED_TASK_TYPES || '').split(',').filter(Boolean);

app.get('/health/conductor/workers', async (req, res) => {
  try {
    const results = {};
    let anyMissing = false;

    for (const taskType of REQUIRED_TASK_TYPES) {
      // GET /api/tasks/queue/polldata returns poll data for a task type
      const pollRes = await axios.get(
        `${CONDUCTOR_URL}/api/tasks/queue/polldata?taskType=${taskType}`,
        { timeout: 5000 }
      );

      const pollData = pollRes.data || [];
      const recentPollers = pollData.filter(pd => {
        const lastPoll = new Date(pd.lastPollTime);
        const ageMs = Date.now() - lastPoll.getTime();
        return ageMs < 5 * 60 * 1000; // Active within 5 minutes
      });

      results[taskType] = {
        totalPollers: pollData.length,
        recentPollers: recentPollers.length,
        healthy: recentPollers.length > 0,
      };

      if (recentPollers.length === 0) {
        anyMissing = true;
      }
    }

    if (anyMissing) {
      const unhealthyTypes = Object.entries(results)
        .filter(([, v]) => !v.healthy)
        .map(([k]) => k);

      return res.status(503).json({
        status: 'degraded',
        reason: 'no_active_workers_for_task_types',
        unhealthyTaskTypes: unhealthyTypes,
        workers: results,
      });
    }

    return res.status(200).json({ status: 'ok', workers: results });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

Set REQUIRED_TASK_TYPES=encode_video,send_notification,process_payment to match your critical task types.

Step 3: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Conductor health proxy: https://your-app.example.com/health/conductor
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 10000ms (queue depth checks can be slow under load)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for worker liveness:

  • URL: https://your-app.example.com/health/conductor/workers
  • Expected: 200, body contains "status":"ok"
  • Interval: 2 minutes
  • Alert channel: workflow platform ops Slack channel

If your Conductor server is directly reachable, also monitor the raw health endpoint:

  • URL: https://conductor.example.com/health
  • Expected: 200, body contains "status":"UP"
  • Interval: 30 seconds

Step 4: Heartbeat Monitoring for Workflow Completion

Server health and worker liveness checks confirm the infrastructure is operational — but the ultimate validation is that workflows actually complete. A misconfigured workflow definition, a retry loop on a failing task, or a deadlock between two workflow types can prevent completion without any infrastructure alert.

Vigilmon heartbeat monitors provide end-to-end workflow completion proof: a completion callback or result consumer sends a ping to Vigilmon each time a workflow instance of a critical type finishes. If completions stop, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: conductor-media-encoding-completions
  3. Set the expected interval: 15 minutes (adjust to your workflow frequency)
  4. Set the grace period: 30 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

Wire Heartbeat Into a Conductor Worker

Add heartbeat pinging to the final task worker in your workflow:

// final-task-worker.js
const { ConductorClient } = require('@io-orkes/conductor-javascript');
const axios = require('axios');

const client = new ConductorClient({
  serverUrl: process.env.CONDUCTOR_URL || 'http://localhost:8080/api',
});

const HEARTBEAT_URL = process.env.VIGILMON_HEARTBEAT_URL;

async function pollAndExecute() {
  while (true) {
    try {
      // Poll for the finalization task
      const task = await client.taskResource.poll('WORKFLOW_FINALIZE', 'finalize-worker-1');

      if (!task) {
        await new Promise(r => setTimeout(r, 1000));
        continue;
      }

      // Execute your finalization logic
      const result = await finalizeWorkflow(task);

      // Mark task as completed
      await client.taskResource.updateTask({
        taskId: task.taskId,
        workflowInstanceId: task.workflowInstanceId,
        status: 'COMPLETED',
        outputData: result,
      });

      // Ping Vigilmon — workflow reached the final task successfully
      if (HEARTBEAT_URL) {
        await axios.get(HEARTBEAT_URL, { timeout: 3000 }).catch(() => {});
      }
    } catch (err) {
      console.error('Worker error:', err.message);
      await new Promise(r => setTimeout(r, 5000));
    }
  }
}

pollAndExecute();

Java Worker with Heartbeat

// FinalizeWorkflowWorker.java
import com.netflix.conductor.client.worker.Worker;
import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskResult;

import java.net.HttpURLConnection;
import java.net.URL;

public class FinalizeWorkflowWorker implements Worker {

    private static final String HEARTBEAT_URL = System.getenv("VIGILMON_HEARTBEAT_URL");

    @Override
    public String getTaskDefName() {
        return "WORKFLOW_FINALIZE";
    }

    @Override
    public TaskResult execute(Task task) {
        TaskResult result = new TaskResult(task);
        try {
            // Your finalization logic here
            Object output = finalizeWorkflow(task.getInputData());
            result.addOutputData("result", output);
            result.setStatus(TaskResult.Status.COMPLETED);

            // Ping Vigilmon on successful completion
            pingVigilmon();
        } catch (Exception e) {
            result.setStatus(TaskResult.Status.FAILED);
            result.setReasonForIncompletion(e.getMessage());
        }
        return result;
    }

    private void pingVigilmon() {
        if (HEARTBEAT_URL == null) return;
        try {
            HttpURLConnection conn = (HttpURLConnection) new URL(HEARTBEAT_URL).openConnection();
            conn.setRequestMethod("GET");
            conn.setConnectTimeout(3000);
            conn.setReadTimeout(3000);
            conn.getResponseCode();
            conn.disconnect();
        } catch (Exception ignored) {}
    }
}

Using Conductor's Workflow Status Polling (Alternative)

If you cannot modify workers, poll the Conductor API for recently completed workflows and ping Vigilmon from a sidecar process:

# completion-monitor.py
import requests, os, time

CONDUCTOR_URL = os.environ['CONDUCTOR_URL']
HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
WORKFLOW_TYPE = os.environ.get('WORKFLOW_TYPE', 'MediaEncoding')

last_seen_completion = None

while True:
    try:
        # Search for recently completed workflows
        resp = requests.get(
            f"{CONDUCTOR_URL}/api/workflow/search",
            params={
                'query': f'workflowType="{WORKFLOW_TYPE}" AND status="COMPLETED"',
                'start': 0,
                'size': 1,
                'sort': 'endTime:DESC',
            },
            timeout=10
        )
        results = resp.json().get('results', [])

        if results:
            latest = results[0]['workflowId']
            if latest != last_seen_completion:
                last_seen_completion = latest
                requests.get(HEARTBEAT_URL, timeout=5)
    except Exception as e:
        print(f"Monitor error: {e}")

    time.sleep(60)

Step 5: Alert Routing for Conductor Failures

| Monitor | Alert Channel | Priority | |---|---|---| | /health/conductor (server down or queue backlog) | Slack + PagerDuty | P1 | | /health/conductor/workers (no active workers) | Slack + PagerDuty | P1 | | Heartbeat: media-encoding-completions | Slack + email | P2 | | Heartbeat: payment-processing-completions | Slack + PagerDuty | P1 |

Worker absence is often P1 because it means nothing in the workflow is executing. Queue backlog can be P2 initially (it takes time to build up) but should escalate to P1 if the queue depth doubles between checks.

Configure Vigilmon's response time threshold on the queue health endpoint at 5000ms — slow queue API responses often precede queue depth explosions as the backend database struggles under load.

Summary

Netflix Conductor makes microservice orchestration reliable — but the orchestrator itself needs external monitoring. Vigilmon gives you the visibility layer that Conductor's UI and Swagger docs cannot:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/conductor | Server availability, task queue depth backlogs | | HTTP monitor on /health/conductor/workers | Worker liveness per task type | | Heartbeat monitor | End-to-end workflow completion rates |

Get started free at vigilmon.online — your first Conductor monitor is live in under two 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