tutorial

How to Monitor Apache SINGA Distributed Deep Learning Health and Training Pipeline Availability with Vigilmon

SINGA training jobs fail silently when workers crash or parameter servers stall. Learn how to monitor Apache SINGA model serving health, distributed training liveness, and pipeline availability with Vigilmon HTTP probes and heartbeat monitors.

Apache SINGA is a distributed deep learning training system designed for large-scale neural network training across multiple nodes — supporting synchronous and asynchronous SGD, model parallelism, and efficient tensor operations. But when a SINGA worker crashes during backpropagation, a parameter server stalls on gradient aggregation, or a training job silently diverges and produces garbage weights, your ML pipeline breaks without warning. Model endpoints serve stale predictions, training progress metrics stop updating, and downstream AI features degrade — all while your infrastructure monitoring shows healthy machines.

Vigilmon gives you external visibility into SINGA training health through HTTP probe monitoring (via a custom health sidecar wrapping SINGA's training state) and heartbeat monitoring for your deep learning pipeline applications. This tutorial covers both.

Why SINGA Needs External Monitoring

SINGA's built-in tooling (training logs, loss metrics, Python-level callbacks) provides rich internal diagnostics — but only if someone is watching. External monitoring with Vigilmon adds:

  • Proactive alerting when the SINGA training coordinator or serving endpoint becomes unreachable
  • Training progress detection by verifying epoch milestones are reached within expected time windows rather than letting jobs stall indefinitely
  • Heartbeat monitoring so you know immediately when a SINGA training job stops making forward progress
  • Multi-region availability checking from outside your SINGA training cluster

These layers work together: the training coordinator can be reachable while a worker has crashed and is retrying indefinitely, and loss metrics can look stable while gradient updates have stalled on a dead parameter server connection.

Step 1: Build a SINGA Health Endpoint

SINGA doesn't expose a built-in HTTP health API, so you need a lightweight sidecar that wraps training state. The pattern is a small Flask/FastAPI server that reads heartbeat files, checks process liveness, or queries a shared state store written by the SINGA training loop.

Python Health Sidecar (File-Based Training State)

The simplest approach: your SINGA training loop writes a timestamp file after each epoch. The health sidecar checks file freshness.

# singa_health.py
from flask import Flask, jsonify
import os
import time

app = Flask(__name__)

CHECKPOINT_DIR = os.environ.get('SINGA_CHECKPOINT_DIR', '/tmp/singa_checkpoints')
MAX_STALE_SECONDS = int(os.environ.get('MAX_STALE_SECONDS', '3600'))
STATE_FILE = os.environ.get('SINGA_STATE_FILE', '/tmp/singa_last_epoch')

@app.route('/health/singa')
def singa_health():
    issues = []

    # Check checkpoint directory accessibility
    if not os.path.isdir(CHECKPOINT_DIR):
        return jsonify({
            'status': 'down',
            'reason': 'checkpoint_dir_missing',
            'path': CHECKPOINT_DIR,
        }), 503

    # Check training state file freshness
    if os.path.exists(STATE_FILE):
        last_modified = os.path.getmtime(STATE_FILE)
        stale_seconds = int(time.time() - last_modified)

        if stale_seconds > MAX_STALE_SECONDS:
            issues.append(f'training_stale_{stale_seconds}s')

        with open(STATE_FILE) as f:
            state_data = {}
            for line in f.read().strip().split('\n'):
                if '=' in line:
                    k, v = line.split('=', 1)
                    state_data[k.strip()] = v.strip()

        last_epoch = int(state_data.get('epoch', -1))
        last_loss = state_data.get('loss', 'unknown')
    else:
        issues.append('training_state_file_missing')
        stale_seconds = -1
        last_epoch = -1
        last_loss = 'unknown'

    # Check for recent checkpoint files
    try:
        checkpoints = sorted([
            f for f in os.listdir(CHECKPOINT_DIR)
            if f.endswith('.pkl') or f.endswith('.bin')
        ])
        latest_checkpoint = checkpoints[-1] if checkpoints else None
    except OSError as e:
        issues.append(f'checkpoint_dir_unreadable: {str(e)}')
        latest_checkpoint = None

    if issues:
        return jsonify({
            'status': 'degraded',
            'issues': issues,
            'last_epoch': last_epoch,
            'stale_seconds': stale_seconds,
        }), 503

    return jsonify({
        'status': 'ok',
        'last_epoch': last_epoch,
        'last_loss': last_loss,
        'stale_seconds': stale_seconds,
        'latest_checkpoint': latest_checkpoint,
    }), 200

if __name__ == '__main__':
    app.run(port=8095)

FastAPI Health Sidecar with SINGA Model Serving

For teams also serving models from SINGA, extend the health check to verify inference availability:

# singa_serving_health.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import requests
import os
import time

app = FastAPI()

SERVING_URL = os.environ.get('SINGA_SERVING_URL', 'http://localhost:8090')
STATE_FILE = os.environ.get('SINGA_STATE_FILE', '/tmp/singa_last_epoch')
MAX_STALE_SECONDS = int(os.environ.get('MAX_STALE_SECONDS', '3600'))

@app.get('/health/singa')
def singa_health():
    issues = []

    # Check SINGA model server reachability
    try:
        resp = requests.get(f'{SERVING_URL}/health', timeout=5)
        resp.raise_for_status()
        serving_status = 'reachable'
    except requests.RequestException as e:
        serving_status = 'unreachable'
        issues.append(f'serving_endpoint_down: {str(e)}')

    # Check training state freshness
    stale_seconds = -1
    last_epoch = -1
    if os.path.exists(STATE_FILE):
        stale_seconds = int(time.time() - os.path.getmtime(STATE_FILE))
        if stale_seconds > MAX_STALE_SECONDS:
            issues.append(f'training_stale_{stale_seconds}s')
        try:
            with open(STATE_FILE) as f:
                for line in f.read().strip().split('\n'):
                    if line.startswith('epoch='):
                        last_epoch = int(line.split('=')[1])
        except (ValueError, OSError):
            pass
    else:
        issues.append('training_state_file_missing')

    if issues:
        return JSONResponse(
            status_code=503,
            content={
                'status': 'degraded',
                'issues': issues,
                'serving_status': serving_status,
                'last_epoch': last_epoch,
                'stale_seconds': stale_seconds,
            }
        )

    return JSONResponse(
        status_code=200,
        content={
            'status': 'ok',
            'serving_status': serving_status,
            'last_epoch': last_epoch,
            'stale_seconds': stale_seconds,
        }
    )

Step 2: Configure Vigilmon HTTP Monitor for SINGA

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your SINGA health endpoint: https://your-app.example.com/health/singa
  4. Set the check interval to 2 minutes
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 10000ms (training state checks involve filesystem operations and optional inference probes)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for the SINGA model serving endpoint:

  • URL: http://singa-server.example.com:8090/health
  • Expected: 200
  • Interval: 1 minute
  • Alert channel: ml-ops channel

Vigilmon's multi-region probe consensus prevents false positives from transient network issues between probe nodes and your SINGA training cluster.

Step 3: Heartbeat Monitoring for SINGA Training Pipelines

Health endpoint checks are necessary — but not sufficient. A SINGA training job can be technically alive (process running, checkpoint directory accessible) while training loss has plateaued or gradient updates have halted due to a dead worker connection. Vigilmon heartbeat monitors detect these silent stalls: your SINGA training loop pings Vigilmon at the end of each epoch. If pings stop, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: singa-training-pipeline
  3. Set the expected interval: 30 minutes (adjust to your epoch duration)
  4. Set the grace period: 60 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into Your SINGA Training Script

# singa_training.py
import singa
from singa import tensor, autograd, opt
import requests
import os
import time

HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
STATE_FILE = os.environ.get('SINGA_STATE_FILE', '/tmp/singa_last_epoch')
CHECKPOINT_DIR = os.environ.get('SINGA_CHECKPOINT_DIR', '/tmp/singa_checkpoints')

def ping_vigilmon():
    try:
        requests.get(HEARTBEAT_URL, timeout=5)
    except Exception:
        pass  # Non-fatal: don't interrupt training

def write_training_state(epoch: int, loss: float):
    with open(STATE_FILE, 'w') as f:
        f.write(f'epoch={epoch}\n')
        f.write(f'loss={loss:.6f}\n')
        f.write(f'timestamp={int(time.time())}\n')

def train_model(model, train_data, num_epochs: int = 50, lr: float = 0.01):
    optimizer = opt.SGD(lr=lr, momentum=0.9)

    for epoch in range(num_epochs):
        total_loss = 0.0
        for batch_x, batch_y in train_data:
            x = tensor.Tensor(device=singa.cpu())
            x.copy_from_numpy(batch_x)
            y = tensor.Tensor(device=singa.cpu())
            y.copy_from_numpy(batch_y)

            with autograd.train():
                out = model(x)
                loss = autograd.cross_entropy(out, y)
                total_loss += tensor.to_numpy(loss).mean()

            for p, gp in autograd.backward():
                optimizer.apply(p, gp)

        avg_loss = total_loss / len(train_data)

        # Save checkpoint
        model.save(f'{CHECKPOINT_DIR}/checkpoint_epoch_{epoch:04d}.bin')

        # Update training state file (read by health sidecar)
        write_training_state(epoch, avg_loss)

        # Ping Vigilmon after each epoch
        ping_vigilmon()
        print(f'Epoch {epoch}/{num_epochs}: loss={avg_loss:.6f}')

SINGA Distributed Training with Heartbeat

# singa_distributed_training.py
import singa
from singa import device as singa_device
import requests
import os

HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
WORKER_RANK = int(os.environ.get('SINGA_WORKER_RANK', '0'))
NUM_WORKERS = int(os.environ.get('SINGA_NUM_WORKERS', '1'))

def ping_vigilmon():
    try:
        requests.get(HEARTBEAT_URL, timeout=5)
    except Exception:
        pass

def train_distributed(model, train_data, num_epochs: int = 50):
    # Initialize multi-process device
    cuda_dev = singa_device.create_cuda_gpu_on(WORKER_RANK)

    for epoch in range(num_epochs):
        for batch in train_data:
            # ... training step ...
            pass

        # Only rank 0 pings to avoid duplicate heartbeats
        if WORKER_RANK == 0:
            ping_vigilmon()
            print(f'Distributed epoch {epoch} complete on all {NUM_WORKERS} workers')

Step 4: Alert Routing for SINGA Failures

SINGA failures cascade: a crashed worker causes gradient updates to stall across the whole training run; a stale model checkpoint causes downstream serving to use an out-of-date model; a training divergence causes AI features to degrade over time without any visible infrastructure error. Alert routing should reflect this cascade.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | SINGA health /health/singa | Slack + PagerDuty | P1 | | Model serving /health | Slack | P1 | | Heartbeat: training pipeline | Slack + email | P2 | | Heartbeat: model export job | Email | P3 |

Set response time thresholds for early warning:

  • Alert at 5000ms for the health endpoint (slow filesystem checks signal disk pressure or NFS mount issues)
  • Alert at 30000ms for the inference probe (GPU-bound inference can spike under load, but this catches hangs)

For bursty training workloads with variable epoch lengths, enable two consecutive failures before alerting — SINGA workers can briefly become unresponsive during checkpoint serialization without representing a full outage.

Summary

SINGA failures are silent at the distributed training and model serving layer. External monitoring catches coordinator crashes, worker stalls, and training pipeline timeouts before they degrade AI model quality or leave production services running on stale model weights:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/singa | Training state freshness, checkpoint availability, serving endpoint liveness | | HTTP monitor on serving endpoint | Model inference availability for downstream consumers | | Heartbeat monitor | Training epoch progress, checkpoint write liveness |

Get started free at vigilmon.online — your first SINGA monitor is running 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