Apache Doris is a massively parallel processing (MPP) analytical database that supports both batch and real-time analytics with MySQL-compatible SQL. It's used by companies like Meituan, JD.com, and Xiaomi for user behavior analytics, financial reporting, and operational dashboards. But its FE/BE split architecture introduces monitoring challenges: FE leader elections can cause brief query outages, BE tablet compaction errors grow silently, and Stream Load failures stop data from arriving without surfacing as obvious errors.
Vigilmon gives you external visibility into every Doris service layer through HTTP probe monitoring and heartbeat monitors for your ingestion pipelines. This tutorial covers FE leader election health, BE tablet health, query execution metrics, schema change job status, stream load success rate, and Vigilmon endpoint checks.
Why Doris Needs External Monitoring
Doris's web UI and SHOW PROC system tables give rich internal visibility — but require an active FE connection to query. External monitoring with Vigilmon adds:
- FE leader election health probes that detect when the active leader has changed or is unavailable
- BE tablet health monitoring that catches compaction errors and disk pressure before they cause query failures
- Query execution health checks that validate end-to-end query path, not just process liveness
- Schema change job failure detection for DDL operations running in the background
- Stream Load success rate monitoring for real-time ingestion pipelines
The most dangerous Doris failure mode is silent: a BE node loses a disk and tablets start erroring one by one, reducing query coverage until a threshold is crossed. By the time queries fail, tablet loss has been accumulating for hours.
Step 1: Build Doris Health Endpoints
Apache Doris FE and BE nodes expose HTTP APIs on dedicated ports for health checking.
Native Doris Health Endpoints
# FE node HTTP health (default port 8030)
curl -i http://doris-fe:8030/api/health
# Check FE leader status
curl -i http://doris-fe:8030/rest/v2/manager/node/frontends
# BE node health (default port 8040)
curl -i http://doris-be:8040/api/health
The FE /api/health returns {"status":"OK","msg":"Success"} when the FE process is healthy and the leader is reachable. Configure Vigilmon monitors against these endpoints directly for basic liveness.
FE Leader Election Health Sidecar
In a multi-FE deployment, monitor whether the MASTER node is alive and followers haven't fallen too far behind:
# doris_health.py
from flask import Flask, jsonify
import requests
import pymysql
import os
import time
app = Flask(__name__)
FE_HTTP_URL = os.environ.get('DORIS_FE_HTTP_URL', 'http://localhost:8030')
DORIS_HOST = os.environ.get('DORIS_HOST', 'localhost')
DORIS_PORT = int(os.environ.get('DORIS_PORT', '9030'))
DORIS_USER = os.environ.get('DORIS_USER', 'root')
DORIS_PASSWORD = os.environ.get('DORIS_PASSWORD', '')
def get_connection():
return pymysql.connect(
host=DORIS_HOST,
port=DORIS_PORT,
user=DORIS_USER,
password=DORIS_PASSWORD,
connect_timeout=5
)
@app.route('/health/doris/fe')
def fe_leader_health():
try:
# Check native health endpoint
resp = requests.get(f'{FE_HTTP_URL}/api/health', timeout=5)
if resp.status_code != 200:
return jsonify({'status': 'down', 'error': f'HTTP {resp.status_code}'}), 503
health_data = resp.json()
if health_data.get('status') != 'OK':
return jsonify({'status': 'down', 'reason': 'fe_health_not_ok', 'data': health_data}), 503
# Check that a master is elected
conn = get_connection()
with conn.cursor() as cur:
cur.execute("SHOW PROC '/frontends'")
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
conn.close()
fe_nodes = [dict(zip(cols, row)) for row in rows]
masters = [fe for fe in fe_nodes if fe.get('IsMaster', '').lower() == 'true']
alive_nodes = [fe for fe in fe_nodes if fe.get('Alive', '').lower() == 'true']
if not masters:
return jsonify({
'status': 'down',
'reason': 'no_fe_master_elected',
'fe_count': len(fe_nodes)
}), 503
if len(alive_nodes) < len(fe_nodes):
dead = [fe.get('Name') for fe in fe_nodes if fe.get('Alive', '').lower() != 'true']
return jsonify({
'status': 'degraded',
'reason': 'fe_nodes_down',
'dead_nodes': dead,
'master': masters[0].get('Name')
}), 503
return jsonify({
'status': 'ok',
'fe_count': len(fe_nodes),
'master': masters[0].get('Name')
}), 200
except Exception as e:
return jsonify({'status': 'down', 'error': str(e)}), 503
BE Tablet Health
BE tablet errors grow silently. Monitor for tablets in error state before they affect query availability:
@app.route('/health/doris/be/tablets')
def tablet_health():
try:
conn = get_connection()
with conn.cursor() as cur:
# Check BE node status
cur.execute("SHOW PROC '/backends'")
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
conn.close()
backends = [dict(zip(cols, row)) for row in rows]
dead_backends = [be for be in backends if be.get('Alive', '').lower() != 'true']
decommissioning = [be for be in backends if be.get('SystemDecommissioned', '').lower() == 'true']
if dead_backends:
return jsonify({
'status': 'down',
'reason': 'be_nodes_down',
'dead': [be.get('BackendId') for be in dead_backends]
}), 503
if decommissioning:
return jsonify({
'status': 'degraded',
'reason': 'be_decommissioning',
'nodes': [be.get('BackendId') for be in decommissioning]
}), 503
return jsonify({'status': 'ok', 'backends': len(backends)}), 200
except Exception as e:
return jsonify({'status': 'down', 'error': str(e)}), 503
Query Execution Health
@app.route('/health/doris/query')
def query_health():
latency_threshold_ms = int(os.environ.get('DORIS_QUERY_THRESHOLD_MS', '2000'))
try:
conn = get_connection()
start = time.time()
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM information_schema.TABLES LIMIT 1")
cur.fetchone()
latency_ms = int((time.time() - start) * 1000)
conn.close()
if latency_ms > latency_threshold_ms:
return jsonify({
'status': 'degraded',
'reason': 'high_query_latency',
'latency_ms': latency_ms,
'threshold_ms': latency_threshold_ms
}), 503
return jsonify({'status': 'ok', 'latency_ms': latency_ms}), 200
except Exception as e:
return jsonify({'status': 'down', 'error': str(e)}), 503
Schema Change Job Status
Long-running schema changes (ALTER TABLE) block writes on the affected table. Monitor for stuck jobs:
@app.route('/health/doris/schema-changes')
def schema_change_health():
max_running_hours = int(os.environ.get('DORIS_SCHEMA_CHANGE_MAX_HOURS', '4'))
try:
conn = get_connection()
with conn.cursor() as cur:
cur.execute("""
SELECT JobId, TableName, State, Progress, CreateTime
FROM information_schema.SCHEMA_CHANGE_JOBS
WHERE State IN ('PENDING', 'WAITING_TXN', 'RUNNING')
""")
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
conn.close()
jobs = [dict(zip(cols, row)) for row in rows]
long_running = []
for job in jobs:
# Check if job has been running too long
if job.get('State') == 'RUNNING':
long_running.append({
'job_id': job.get('JobId'),
'table': job.get('TableName'),
'progress': job.get('Progress')
})
if long_running:
return jsonify({
'status': 'degraded',
'reason': 'long_running_schema_changes',
'jobs': long_running
}), 503
return jsonify({'status': 'ok', 'active_jobs': len(jobs)}), 200
except Exception as e:
return jsonify({'status': 'down', 'error': str(e)}), 503
Stream Load Success Rate
@app.route('/health/doris/stream-load')
def stream_load_health():
try:
conn = get_connection()
with conn.cursor() as cur:
# Check recent stream load transactions
cur.execute("""
SELECT Label, State, FailMsg, CreateTime
FROM information_schema.LOAD_JOBS
WHERE JobType = 'MINI_LOAD'
AND CreateTime > NOW() - INTERVAL 30 MINUTE
ORDER BY CreateTime DESC
LIMIT 20
""")
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
conn.close()
jobs = [dict(zip(cols, row)) for row in rows]
failed = [j for j in jobs if j.get('State') == 'CANCELLED']
if len(failed) > 3:
return jsonify({
'status': 'degraded',
'reason': 'high_stream_load_failure_rate',
'failed_last_30m': len(failed),
'total_last_30m': len(jobs)
}), 503
return jsonify({
'status': 'ok',
'jobs_last_30m': len(jobs),
'failed_last_30m': len(failed)
}), 200
except Exception as e:
return jsonify({'status': 'down', 'error': str(e)}), 503
if __name__ == '__main__':
app.run(port=9092)
Step 2: Configure Vigilmon HTTP Monitors for Doris
- Log in to vigilmon.online and go to Monitors → New Monitor
- Choose HTTP / HTTPS
- Configure monitors for each health surface:
| Monitor Name | URL | Expected | Interval |
|---|---|---|---|
| Doris FE liveness | http://doris-fe:8030/api/health | 200, body "OK" | 1 min |
| Doris FE leader election | https://your-sidecar/health/doris/fe | 200, body "ok" | 2 min |
| Doris BE liveness | http://doris-be:8040/api/health | 200 | 1 min |
| Doris BE tablet health | https://your-sidecar/health/doris/be/tablets | 200, body "ok" | 5 min |
| Doris query execution | https://your-sidecar/health/doris/query | 200, body "ok" | 1 min |
| Doris schema changes | https://your-sidecar/health/doris/schema-changes | 200, body "ok" | 10 min |
| Doris stream load rate | https://your-sidecar/health/doris/stream-load | 200, body "ok" | 5 min |
For the query execution monitor, set the response time alert threshold to 1500ms — early warning before the 2000ms P99 threshold fires.
For the schema change monitor, a 10-minute interval is sufficient — long-running DDL is a slow-burn issue, not an instant alert.
Step 3: Heartbeat Monitoring for Doris Ingestion Pipelines
Doris internal monitoring can tell you that Stream Load succeeded for the last batch — but it can't tell you whether your upstream data pipeline is still producing data. If your Flink job, Python ETL, or Kafka connector that feeds Doris stops running, the stream load endpoint stays healthy while data freshness silently degrades.
Vigilmon heartbeat monitors solve this by requiring your pipeline to actively prove it's alive. If pings stop arriving, Vigilmon fires an alert.
Set Up the Heartbeat Monitor
- In Vigilmon, go to Monitors → New Monitor → Heartbeat
- Set the name:
doris-stream-load-etl - Set the expected interval: 15 minutes
- Set the grace period: 30 minutes
- Save — copy the heartbeat URL
Python ETL Pipeline with Heartbeat
# doris_etl_pipeline.py
import os
import requests
import json
import time
import logging
DORIS_FE_HOST = os.environ['DORIS_FE_HOST']
DORIS_FE_HTTP_PORT = os.environ.get('DORIS_FE_HTTP_PORT', '8030')
DORIS_USER = os.environ['DORIS_USER']
DORIS_PASSWORD = os.environ['DORIS_PASSWORD']
DORIS_DATABASE = os.environ['DORIS_DATABASE']
DORIS_TABLE = os.environ['DORIS_TABLE']
VIGILMON_HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
logger = logging.getLogger(__name__)
def stream_load(records: list, label: str) -> bool:
"""Load a batch of JSON records into Doris via Stream Load."""
data = '\n'.join(json.dumps(r) for r in records)
response = requests.put(
f'http://{DORIS_FE_HOST}:{DORIS_FE_HTTP_PORT}'
f'/api/{DORIS_DATABASE}/{DORIS_TABLE}/_stream_load',
auth=(DORIS_USER, DORIS_PASSWORD),
headers={
'Expect': '100-continue',
'format': 'json',
'label': label,
'Content-Type': 'application/json',
},
data=data,
timeout=300
)
result = response.json()
status = result.get('Status', '')
if status == 'Success':
logger.info(f"Stream load {label}: {result.get('NumberLoadedRows', 0)} rows")
return True
else:
logger.error(f"Stream load {label} failed: {result.get('Message', '')}")
return False
def run_etl():
last_heartbeat = 0
HEARTBEAT_INTERVAL = 15 * 60 # 15 minutes
batch_number = 0
while True:
try:
records = fetch_records_from_source()
if records:
batch_number += 1
label = f"etl_batch_{int(time.time())}_{batch_number}"
success = stream_load(records, label)
if success:
now = time.time()
if now - last_heartbeat >= HEARTBEAT_INTERVAL:
requests.get(VIGILMON_HEARTBEAT_URL, timeout=5)
last_heartbeat = now
logger.info("Vigilmon heartbeat sent")
else:
logger.warning("Batch failed — skipping heartbeat")
except Exception as e:
logger.error(f"ETL error: {e}")
time.sleep(60)
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
run_etl()
Flink → Doris Pipeline Heartbeat (via Doris Flink Connector)
For Flink streaming jobs using the official Doris Flink connector, send the heartbeat from a side-output operator:
// DorisHeartbeatOperator.java
public class DorisHeartbeatOperator extends AbstractStreamOperator<String>
implements OneInputStreamOperator<String, String> {
private final String vigilmonUrl;
private long lastHeartbeat = 0;
private static final long HEARTBEAT_INTERVAL_MS = 15 * 60 * 1000L;
private transient OkHttpClient httpClient;
public DorisHeartbeatOperator(String vigilmonUrl) {
this.vigilmonUrl = vigilmonUrl;
}
@Override
public void open() {
this.httpClient = new OkHttpClient.Builder()
.connectTimeout(5, TimeUnit.SECONDS)
.readTimeout(5, TimeUnit.SECONDS)
.build();
}
@Override
public void processElement(StreamRecord<String> element) throws Exception {
output.collect(element);
long now = System.currentTimeMillis();
if (now - lastHeartbeat >= HEARTBEAT_INTERVAL_MS) {
sendHeartbeat();
lastHeartbeat = now;
}
}
private void sendHeartbeat() {
Request request = new Request.Builder().url(vigilmonUrl).get().build();
try (Response response = httpClient.newCall(request).execute()) {
LOG.info("Vigilmon heartbeat: HTTP {}", response.code());
} catch (IOException e) {
LOG.warn("Vigilmon heartbeat failed: {}", e.getMessage());
}
}
}
Step 4: Alert Routing for Doris Failures
Doris failure severity maps directly to which layer is affected:
- FE leader loss: queries begin failing within seconds; no new DDL or DML possible
- BE node failure: tablet rebalancing begins but reduces immediate query capacity
- Tablet errors: queries return incorrect or incomplete results without explicit errors
- Stream load failures: data freshness degrades silently; dashboards show stale data
- Schema change stuck: writes to the affected table may block
Configure alert routing in Vigilmon:
| Monitor | Alert Channel | Priority | |---|---|---| | Doris FE liveness | Slack + PagerDuty | P1 | | Doris FE leader election | Slack + PagerDuty | P1 | | Doris query execution | Slack + PagerDuty | P1 | | Doris BE liveness | Slack | P2 | | Doris BE tablet health | Slack | P2 | | Doris stream load rate | Slack + email | P2 | | Doris schema changes | Email | P3 | | Doris ETL heartbeat | Slack | P2 |
Add a response time alert on the FE liveness monitor at 500ms — FE response time above 500ms is an early indicator of election activity or Bdbje log replay overhead, which precedes outages.
For multi-region Doris deployments, create a separate monitor per region using Vigilmon's regional probe selection to distinguish region-specific failures from global cluster issues.
Summary
Apache Doris's reliability depends on FE leadership stability, BE tablet health, and continuous ingestion pipeline liveness — all of which need external validation that internal metrics can't provide. Vigilmon covers every layer:
| Monitor Type | What It Covers |
|---|---|
| HTTP probe on FE/BE /api/health | Service process liveness |
| HTTP probe on FE leader sidecar | Leader election and follower health |
| HTTP probe on query sidecar | End-to-end query execution latency |
| HTTP probe on tablet sidecar | BE tablet health and compaction status |
| HTTP probe on load sidecar | Stream Load success rate |
| HTTP probe on schema sidecar | Long-running DDL job detection |
| Heartbeat monitor | ETL pipeline and Flink job liveness |
Get started free at vigilmon.online — your first Doris monitor is live in under two minutes.