tutorial

How to Monitor Apache Metron Health and Threat Intelligence Pipeline Availability with Vigilmon

Metron pipeline failures and enrichment topology stalls silently break your cybersecurity analytics. Learn how to monitor Metron REST API health, topology liveness, and client application health with Vigilmon HTTP probes and heartbeat monitors.

Apache Metron is the open-source cybersecurity analytics platform that ingests, enriches, and stores security telemetry at enterprise scale — combining stream processing, threat intelligence, and anomaly detection into a unified pipeline. But when an enrichment topology stalls, a Kafka consumer group falls behind, or the Metron REST API goes down, your security analytics pipeline degrades silently. Missed threat alerts, stale enrichments, and gaps in SIEM data may not surface for hours — long after an attacker has moved laterally.

Vigilmon gives you external visibility into Metron platform health through HTTP probe monitoring (via Metron's REST API or a custom health sidecar) and heartbeat monitoring for your Metron pipeline client applications. This tutorial covers both.

Why Metron Needs External Monitoring

Metron's built-in tooling (Metron REST API on port 8082, Storm UI, Ambari metrics) provides rich internal diagnostics — but only if someone is watching. External monitoring with Vigilmon adds:

  • Proactive alerting when the Metron REST API becomes unreachable (signals a management plane failure)
  • Pipeline liveness detection by verifying enrichment and indexing topologies are ACTIVE via the Storm UI API
  • Heartbeat monitoring so you know immediately when a custom sensor or client application stops shipping events
  • Multi-region availability checking from outside your Metron cluster network

These layers work together: the REST API can be reachable while enrichment topologies are stuck in REBALANCING state, and topology metrics can look healthy while alert generation is silently failing due to a misconfigured threat intelligence feed.

Step 1: Build a Metron Health Endpoint

Metron exposes a REST API (default port 8082) for managing sensors, alerts, and topology status. Use it directly or wrap it in a custom health sidecar.

Using the Metron REST API Directly

# Check Metron REST API is reachable (requires auth)
curl -i -u admin:password http://metron.example.com:8082/api/v1/sensor/parser/list

# List active Storm topologies via Storm UI
curl -i http://storm-ui.example.com:8080/api/v1/topology/summary

# Check Kafka consumer group lag for Metron parsers
/opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server kafka.example.com:9092 \
  --describe --group metron-parsers

Point a Vigilmon monitor at http://metron.example.com:8082/api/v1/sensor/parser/list for REST API reachability.

Custom Python Health Sidecar

For deeper health checks — REST API status, Storm topology health, and Kafka consumer lag — build a lightweight sidecar:

# metron_health.py
from flask import Flask, jsonify
import requests
import subprocess
import os

app = Flask(__name__)

METRON_REST_URL = os.environ.get('METRON_REST_URL', 'http://localhost:8082')
STORM_UI_URL = os.environ.get('STORM_UI_URL', 'http://localhost:8080')
METRON_USER = os.environ.get('METRON_USER', 'admin')
METRON_PASS = os.environ.get('METRON_PASS', 'password')
REQUIRED_TOPOLOGIES = os.environ.get(
    'METRON_TOPOLOGIES',
    'enrichment,indexing,bro,snort'
).split(',')

@app.route('/health/metron')
def metron_health():
    issues = []

    # Check Metron REST API
    try:
        resp = requests.get(
            f'{METRON_REST_URL}/api/v1/sensor/parser/list',
            auth=(METRON_USER, METRON_PASS),
            timeout=5
        )
        resp.raise_for_status()
    except requests.RequestException as e:
        return jsonify({'status': 'down', 'reason': 'rest_api_unreachable',
                        'error': str(e)}), 503

    # Check Storm topologies for each required Metron topology
    try:
        topo_resp = requests.get(
            f'{STORM_UI_URL}/api/v1/topology/summary', timeout=5
        )
        topo_resp.raise_for_status()
        active_names = {
            t['name'] for t in topo_resp.json().get('topologies', [])
            if t.get('status') == 'ACTIVE'
        }

        missing = [t for t in REQUIRED_TOPOLOGIES if t not in active_names]
        if missing:
            issues.append(f'topologies_not_active: {",".join(missing)}')

    except requests.RequestException as e:
        issues.append(f'storm_ui_unreachable: {str(e)}')

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

    return jsonify({
        'status': 'ok',
        'rest_api': 'reachable',
        'topologies_active': REQUIRED_TOPOLOGIES,
    }), 200

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

Java Health Sidecar (Metron Client)

// MetronHealthController.java
@RestController
public class MetronHealthController {

    @Value("${metron.rest.url:http://localhost:8082}")
    private String metronRestUrl;

    @Value("${storm.ui.url:http://localhost:8080}")
    private String stormUiUrl;

    @Value("${metron.user:admin}")
    private String metronUser;

    @Value("${metron.password}")
    private String metronPassword;

    @Value("${metron.required.topologies:enrichment,indexing}")
    private String requiredTopologiesCsv;

    private final RestTemplate restTemplate = new RestTemplate();

    @GetMapping("/health/metron")
    public ResponseEntity<Map<String, Object>> checkHealth() {
        Map<String, Object> status = new LinkedHashMap<>();
        List<String> issues = new ArrayList<>();

        // Check REST API
        try {
            HttpHeaders headers = new HttpHeaders();
            headers.setBasicAuth(metronUser, metronPassword);
            HttpEntity<Void> entity = new HttpEntity<>(headers);
            restTemplate.exchange(
                metronRestUrl + "/api/v1/sensor/parser/list",
                HttpMethod.GET, entity, List.class);
        } catch (Exception e) {
            status.put("status", "down");
            status.put("reason", "rest_api_unreachable");
            return ResponseEntity.status(503).body(status);
        }

        // Check Storm topologies
        try {
            ResponseEntity<Map> topoResp = restTemplate.getForEntity(
                stormUiUrl + "/api/v1/topology/summary", Map.class);
            List<Map<String, Object>> topologies =
                (List<Map<String, Object>>) topoResp.getBody().get("topologies");
            Set<String> activeNames = topologies.stream()
                .filter(t -> "ACTIVE".equals(t.get("status")))
                .map(t -> (String) t.get("name"))
                .collect(java.util.stream.Collectors.toSet());

            List<String> required = Arrays.asList(requiredTopologiesCsv.split(","));
            List<String> missing = required.stream()
                .filter(t -> !activeNames.contains(t))
                .collect(java.util.stream.Collectors.toList());

            if (!missing.isEmpty()) {
                issues.add("topologies_not_active: " + String.join(",", missing));
            }
        } catch (Exception e) {
            issues.add("storm_ui_unreachable: " + e.getMessage());
        }

        if (!issues.isEmpty()) {
            status.put("status", "degraded");
            status.put("issues", issues);
            return ResponseEntity.status(503).body(status);
        }

        status.put("status", "ok");
        status.put("rest_api", "reachable");
        return ResponseEntity.ok(status);
    }
}

Step 2: Configure Vigilmon HTTP Monitor for Metron

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Metron health endpoint: https://your-app.example.com/health/metron
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 8000ms (Metron health checks aggregate REST, Storm, and Kafka lookups)
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for the Metron REST API directly:

  • URL: http://metron.example.com:8082/api/v1/sensor/parser/list
  • Expected: 200
  • Interval: 2 minutes
  • Alert channel: security-ops channel

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

Step 3: Heartbeat Monitoring for Metron Pipeline Clients

REST API and topology health checks are necessary — but not sufficient. A Metron enrichment topology can process events and produce output without triggering alerts when threat intelligence feeds are stale or misconfigured. A custom sensor can appear healthy in Storm while shipping no raw telemetry. Vigilmon heartbeat monitors detect these silent processing stalls: your sensor or pipeline application pings Vigilmon after successfully shipping or processing each batch of events.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: metron-sensor-pipeline
  3. Set the expected interval: 5 minutes (adjust to your event throughput)
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into Your Metron Kafka Producer (Java)

// MonitoredMetronProducer.java
import org.apache.kafka.clients.producer.*;
import java.net.http.*;
import java.net.URI;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;

public class MonitoredMetronProducer {

    private final KafkaProducer<String, String> producer;
    private final HttpClient httpClient = HttpClient.newHttpClient();
    private final String heartbeatUrl = System.getenv("VIGILMON_HEARTBEAT_URL");
    private final AtomicLong eventCount = new AtomicLong(0);
    private static final long HEARTBEAT_EVERY = 500;

    public MonitoredMetronProducer(Properties kafkaProps) {
        this.producer = new KafkaProducer<>(kafkaProps);
    }

    public void sendEvent(String topic, String sensorEvent) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, sensorEvent);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                long count = eventCount.incrementAndGet();
                if (count % HEARTBEAT_EVERY == 0) {
                    pingVigilmon();
                }
            }
        });
    }

    private void pingVigilmon() {
        try {
            HttpRequest req = HttpRequest.newBuilder()
                .uri(URI.create(heartbeatUrl))
                .GET().build();
            httpClient.sendAsync(req, HttpResponse.BodyHandlers.discarding());
        } catch (Exception e) {
            // Non-fatal: don't interrupt event shipping
        }
    }
}

Python Metron Event Shipper with Heartbeat

# monitored_sensor.py
from kafka import KafkaProducer
import requests
import os
import time
import json

HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
KAFKA_BOOTSTRAP = os.environ.get('KAFKA_BOOTSTRAP', 'localhost:9092')
METRON_TOPIC = os.environ.get('METRON_TOPIC', 'bro')

producer = KafkaProducer(
    bootstrap_servers=KAFKA_BOOTSTRAP,
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

events_sent = 0
last_heartbeat = time.time()
HEARTBEAT_INTERVAL = 60  # seconds

def ship_event(event: dict):
    global events_sent, last_heartbeat
    producer.send(METRON_TOPIC, event)
    events_sent += 1

    if time.time() - last_heartbeat > HEARTBEAT_INTERVAL:
        try:
            requests.get(HEARTBEAT_URL, timeout=5)
        except Exception:
            pass
        last_heartbeat = time.time()

Step 4: Alert Routing for Metron Failures

Metron failures cascade: a Kafka broker becoming unreachable causes consumer groups to stall, enrichment topologies to back-pressure, and indexing pipelines to drop events. Security alert generation can stop entirely without any visible error. Alert routing should reflect this cascade.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Metron health /health/metron | Slack + PagerDuty | P1 | | Metron REST API /api/v1/sensor/parser/list | Slack | P2 | | Heartbeat: sensor event producer | Slack + email | P2 | | Heartbeat: enrichment pipeline output | Email | P3 |

Set response time thresholds for early warning:

  • Alert at 3000ms for the health endpoint (slow REST responses signal management plane pressure)
  • Alert at 5000ms for the REST API endpoint (slow sensor list aggregation signals backend persistence issues)

For security-critical pipelines, enable two consecutive failures before alerting — Metron's enrichment topologies can briefly report degraded status during scheduled threat intelligence feed refreshes without representing a full outage.

Summary

Metron failures are silent at the security analytics layer. External monitoring catches REST API crashes, enrichment topology stalls, and sensor pipeline gaps before they create blind spots in your threat detection:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/metron | REST API liveness, topology status, pipeline integrity | | HTTP monitor on Metron REST API | Management plane availability | | Heartbeat monitor | Sensor liveness, event shipping throughput |

Get started free at vigilmon.online — your first Metron 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