MCP Tools for Trino — Coordinator health, worker state, query execution stages, and memory pool management

Trino architecture for MCP tool authors

Trino has a coordinator (single node — the query planner, parser, scheduler, and REST API server) and workers (distributed nodes that execute query tasks and hold memory pools). The coordinator itself does not execute query tasks — it only plans, schedules, and monitors. In high-availability setups, a load balancer routes traffic to the active coordinator; failover is not automatic in OSS Trino (TrinoDb Enterprise and Starburst Galaxy add HA).

Trino's memory management is pooled: each worker has a configurable query.max-memory-per-node (default 1 GB) and a cluster-wide query.max-memory limit. When a query exceeds its memory reservation, it is killed with INSUFFICIENT_RESOURCES. When the cluster runs out of query memory across all workers, new queries are blocked in the queue rather than killed. Blocked queries appear in GET /v1/cluster as blockedQueries > 0 and in GET /v1/query as state: "BLOCKED".

async function createTrinoClient(coordinatorUrl, trinoUser = 'trino', trinoPassword = null) {
  const base = coordinatorUrl.replace(/\/$/, '');
  const headers = {
    'X-Trino-User': trinoUser,
    'Accept': 'application/json',
  };
  if (trinoPassword) {
    headers['Authorization'] = `Basic ${Buffer.from(`${trinoUser}:${trinoPassword}`).toString('base64')}`;
  }

  async function get(path) {
    const res = await fetch(`${base}${path}`, {
      headers,
      signal: AbortSignal.timeout(15_000),
    });
    if (!res.ok) {
      throw new Error(`Trino API ${path} → ${res.status}: ${await res.text().catch(() => '')}`);
    }
    return res.json();
  }

  // Coordinator liveness and version
  async function getInfo() {
    return get('/v1/info');
  }

  // Cluster aggregate metrics
  async function getCluster() {
    return get('/v1/cluster');
  }

  // All current queries (active + recently completed)
  async function getQueries(state = null) {
    const qs = state ? `?state=${state}` : '';
    return get(`/v1/query${qs}`);
  }

  // Full detail for a specific query
  async function getQueryDetail(queryId) {
    return get(`/v1/query/${queryId}`);
  }

  // Worker node list
  async function getWorkers() {
    return get('/v1/node');
  }

  // Failed worker nodes
  async function getFailedWorkers() {
    return get('/v1/node/failed');
  }

  return { getInfo, getCluster, getQueries, getQueryDetail, getWorkers, getFailedWorkers };
}

Coordinator status and startup detection

The GET /v1/info endpoint returns the coordinator's basic state. The starting boolean field is true while Trino is initializing — during startup, the coordinator is reachable on the HTTP port but not yet accepting queries. An MCP tool that only checks HTTP 200 on /v1/info will incorrectly report "healthy" during the startup period.

The nodeVersion.version field reports the Trino version. Track this across probes to detect unplanned version changes (a restart with a different Trino image). The environment field identifies the cluster name (useful when routing probes to multiple clusters). There is no "shutting down" signal in the REST API — a coordinator that is gracefully draining before shutdown will continue returning starting: false while rejecting new query submissions.

async function checkCoordinatorHealth(trino) {
  let info;
  try {
    info = await trino.getInfo();
  } catch (err) {
    return {
      reachable: false,
      healthy: false,
      error: err.message,
    };
  }

  const isStarting = info.starting === true;
  const version = info.nodeVersion?.version ?? null;
  const environment = info.environment ?? null;
  const uptime = info.uptime ?? null;

  return {
    reachable: true,
    healthy: !isStarting,
    isStarting,
    version,
    environment,
    uptime,
    alerts: isStarting ? ['Trino coordinator is still starting — not accepting queries'] : [],
  };
}

Cluster metrics and query queue health

The GET /v1/cluster endpoint provides aggregate cluster metrics including activeWorkers, runningQueries, queuedQueries, blockedQueries, and resource utilization. Blocked queries are the most important signal: they indicate the cluster has exhausted memory and new queries cannot start. Unlike queued queries (which are waiting because all concurrency slots are full), blocked queries have already been allocated resources but cannot proceed.

The cluster-level reservedMemory and totalMemory fields show the proportion of cluster memory reserved by active queries. When reservedMemory / totalMemory > 0.85, the cluster is approaching memory exhaustion. The next large query may push it into the blocked state. Alert at 70% utilization to give time to either scale workers or kill runaway queries before the cluster stalls.

async function checkClusterHealth(trino) {
  const cluster = await trino.getCluster();

  const activeWorkers = cluster.activeWorkers ?? 0;
  const runningQueries = cluster.runningQueries ?? 0;
  const queuedQueries = cluster.queuedQueries ?? 0;
  const blockedQueries = cluster.blockedQueries ?? 0;

  // Memory utilization
  const reservedMemoryBytes = cluster.reservedMemory ?? 0;
  const totalMemoryBytes = cluster.totalMemory ?? 0;
  const memoryUtilization = totalMemoryBytes > 0 ? reservedMemoryBytes / totalMemoryBytes : null;

  // Query concurrency ratio: running queries per worker (> 4 = overloaded for most workloads)
  const queriesPerWorker = activeWorkers > 0 ? runningQueries / activeWorkers : null;

  const alerts = [];
  if (blockedQueries > 0) {
    alerts.push({
      severity: 'critical',
      type: 'blocked_queries',
      detail: `${blockedQueries} queries blocked — cluster out of memory; new queries cannot start`,
    });
  }
  if (memoryUtilization !== null && memoryUtilization > 0.85) {
    alerts.push({
      severity: 'warning',
      type: 'memory_pressure',
      detail: `${(memoryUtilization * 100).toFixed(1)}% of cluster memory reserved — approaching exhaustion`,
    });
  }
  if (queuedQueries > activeWorkers * 2) {
    alerts.push({
      severity: 'warning',
      type: 'query_queue_buildup',
      detail: `${queuedQueries} queries queued (${activeWorkers} workers) — cluster at capacity`,
    });
  }
  if (queriesPerWorker !== null && queriesPerWorker > 8) {
    alerts.push({
      severity: 'warning',
      type: 'overloaded',
      detail: `${queriesPerWorker.toFixed(1)} running queries per worker — likely contention`,
    });
  }

  return {
    activeWorkers,
    runningQueries,
    queuedQueries,
    blockedQueries,
    memoryUtilization,
    queriesPerWorker,
    alerts,
    healthy: alerts.filter(a => a.severity === 'critical').length === 0,
  };
}

Query execution stage health: detecting stuck queries

Trino decomposes each query into a tree of stages, each stage into tasks distributed across workers, and each task into splits (the unit of parallel I/O). A query that is RUNNING at the top level can have individual stages stuck if: a worker holding pending tasks is slow (network I/O bound, GC paused), splits are not being assigned because the connector's split generation is slow, or a shuffle (data exchange between stages) is blocked waiting for memory.

The GET /v1/query/{queryId} endpoint returns the full query plan and per-stage statistics: stages[].state (RUNNING|BLOCKED|FINISHED|FAILED|CANCELED|ABORTED|FLUSHING|PLANNED|SCHEDULING|SCHEDULED), stages[].subStages, and per-stage stats including completedSplits, totalSplits, processedBytes, and wallTime. A stage that has been RUNNING for more than 5 minutes with zero completedSplits progression is stuck.

async function assessQueryHealth(trino, queryId) {
  const query = await trino.getQueryDetail(queryId);

  const state = query.state;  // 'RUNNING' | 'QUEUED' | 'BLOCKED' | 'FINISHED' | 'FAILED'
  const queryStats = query.queryStats;

  // Memory usage
  const userMemoryBytes = queryStats?.userMemoryReservation
    ? parseTrinoSize(queryStats.userMemoryReservation)
    : null;
  const totalMemoryBytes = queryStats?.totalMemoryReservation
    ? parseTrinoSize(queryStats.totalMemoryReservation)
    : null;
  const peakMemoryBytes = queryStats?.peakUserMemoryReservation
    ? parseTrinoSize(queryStats.peakUserMemoryReservation)
    : null;

  // Spill to disk (if configured)
  const spilledBytes = queryStats?.spilledDataSize
    ? parseTrinoSize(queryStats.spilledDataSize)
    : null;

  // Stage health
  const stages = flattenStages(query.outputStage ?? null);
  const stuckStages = stages.filter(s => {
    if (s.state !== 'RUNNING') return false;
    // Stuck if no splits completed in last minute (approximate — compare totalSplits vs completedSplits progress)
    return s.completedSplits === 0 && s.totalSplits > 0 && parseWallTime(s.wallTime) > 60_000;
  });

  const failedStages = stages.filter(s => s.state === 'FAILED' || s.state === 'ABORTED');
  const blockedStages = stages.filter(s => s.state === 'BLOCKED');

  // Error info
  const failureInfo = query.failureInfo;

  return {
    queryId,
    state,
    isRunning: state === 'RUNNING',
    isBlocked: state === 'BLOCKED',
    isFailed: state === 'FAILED',
    elapsedMs: queryStats?.elapsedTime ? parseWallTime(queryStats.elapsedTime) : null,
    cpuMs: queryStats?.totalCpuTime ? parseWallTime(queryStats.totalCpuTime) : null,
    memory: { userBytes: userMemoryBytes, totalBytes: totalMemoryBytes, peakBytes: peakMemoryBytes },
    spilledBytes,
    splits: {
      completed: queryStats?.completedSplits ?? 0,
      total: queryStats?.totalSplits ?? 0,
      queued: queryStats?.queuedSplits ?? 0,
      running: queryStats?.runningSplits ?? 0,
    },
    stages: { total: stages.length, stuck: stuckStages.length, failed: failedStages.length, blocked: blockedStages.length },
    error: failureInfo ? { type: failureInfo.type, message: failureInfo.message } : null,
    healthy: !['BLOCKED', 'FAILED', 'ABORTED', 'CANCELED'].includes(state) && stuckStages.length === 0,
  };
}

function flattenStages(stage) {
  if (!stage) return [];
  const stats = stage.stageStats ?? {};
  const result = [{
    stageId: stage.stageId,
    state: stage.state,
    completedSplits: stats.completedSplits ?? 0,
    totalSplits: stats.totalSplits ?? 0,
    wallTime: stats.totalScheduledTime ?? '0.00ms',
  }];
  for (const sub of (stage.subStages ?? [])) {
    result.push(...flattenStages(sub));
  }
  return result;
}

function parseTrinoSize(sizeStr) {
  // Trino sizes: "1.23GB", "456MB", "789kB"
  if (!sizeStr) return null;
  const match = sizeStr.match(/^([\d.]+)([TGMK]?B)$/i);
  if (!match) return null;
  const [, num, unit] = match;
  const multipliers = { B: 1, KB: 1024, MB: 1_048_576, GB: 1_073_741_824, TB: 1_099_511_627_776 };
  return parseFloat(num) * (multipliers[unit.toUpperCase()] ?? 1);
}

function parseWallTime(timeStr) {
  // Trino times: "1.23m", "45.6s", "789ms"
  if (!timeStr) return 0;
  const match = timeStr.match(/^([\d.]+)(m|s|ms|us|ns)$/);
  if (!match) return 0;
  const [, num, unit] = match;
  const multipliers = { ns: 0.000001, us: 0.001, ms: 1, s: 1000, m: 60_000 };
  return parseFloat(num) * (multipliers[unit] ?? 1);
}

Worker health and failed node detection

The GET /v1/node endpoint returns all active workers. The GET /v1/node/failed endpoint returns recently failed workers. A worker appears in /v1/node/failed when the coordinator has not received a heartbeat from it for more than node.heartbeat-interval (default 500 ms) × query.client.timeout. Failed workers do not cause running queries to fail immediately — Trino retries tasks from failed workers on other available workers. If there are no other workers to retry on, the query fails with WORKER_NODE_FAILURE.

async function checkWorkerHealth(trino) {
  const [activeNodes, failedNodes] = await Promise.all([
    trino.getWorkers(),
    trino.getFailedWorkers(),
  ]);

  const active = Array.isArray(activeNodes) ? activeNodes : [];
  const failed = Array.isArray(failedNodes) ? failedNodes : [];

  // Workers with low free memory (may cause OOM-induced failures)
  const memoryPressureNodes = active.filter(n => {
    const freeBytes = n.memoryInfo?.pool?.freeBytes ?? null;
    const maxBytes = n.memoryInfo?.pool?.maxBytes ?? null;
    if (freeBytes === null || maxBytes === null || maxBytes === 0) return false;
    return freeBytes / maxBytes < 0.1;  // < 10% free
  });

  const alerts = [];

  if (failed.length > 0) {
    const failedRatio = failed.length / (active.length + failed.length);
    alerts.push({
      severity: failedRatio > 0.3 ? 'critical' : 'warning',
      type: 'worker_failures',
      detail: `${failed.length} worker(s) recently failed (${active.length} active)`,
      failedNodes: failed.map(n => n.uri ?? n.nodeId),
    });
  }

  if (active.length === 0) {
    alerts.push({
      severity: 'critical',
      type: 'no_workers',
      detail: 'No active Trino workers — cluster cannot execute queries',
    });
  }

  if (memoryPressureNodes.length > active.length * 0.5) {
    alerts.push({
      severity: 'warning',
      type: 'worker_memory_pressure',
      detail: `${memoryPressureNodes.length}/${active.length} workers have < 10% memory free`,
    });
  }

  return {
    activeWorkers: active.length,
    failedWorkers: failed.length,
    memoryPressureWorkers: memoryPressureNodes.length,
    alerts,
    healthy: alerts.filter(a => a.severity === 'critical').length === 0,
  };
}

AliveMCP integration for Trino cluster monitoring

Register three AliveMCP probes for a Trino cluster. A coordinator liveness probe (30s interval, 5s timeout, checks /v1/info) catches coordinator restarts and startup hangs. A cluster health probe (60s interval, 15s timeout, checks /v1/cluster + /v1/node/failed) catches memory exhaustion, query queue buildup, and worker failures. A long-running query probe (5-minute interval, 20s timeout, scans all RUNNING queries for wall time > 30 minutes) catches runaway queries consuming memory and starving other workloads.

async function trinoClusterHealthProbe(coordinatorUrl, trinoUser = 'monitoring') {
  const trino = createTrinoClient(coordinatorUrl, trinoUser);

  // Layer 1: Coordinator availability
  const coordHealth = await checkCoordinatorHealth(trino);
  if (!coordHealth.reachable) {
    return { healthy: false, layer: 'coordinator_unreachable', error: coordHealth.error };
  }
  if (!coordHealth.healthy) {
    return { healthy: false, layer: 'coordinator', alerts: coordHealth.alerts };
  }

  // Layer 2: Cluster metrics + worker health (parallel)
  const [clusterHealth, workerHealth] = await Promise.allSettled([
    checkClusterHealth(trino),
    checkWorkerHealth(trino),
  ]);

  // Layer 3: Long-running query detection
  let longRunningQueries = [];
  try {
    const allQueries = await trino.getQueries('RUNNING');
    const MAX_WALL_TIME_MS = 30 * 60 * 1000;  // 30 minutes
    for (const q of (Array.isArray(allQueries) ? allQueries : [])) {
      const wallMs = q.queryStats?.elapsedTime ? parseWallTime(q.queryStats.elapsedTime) : 0;
      if (wallMs > MAX_WALL_TIME_MS) {
        longRunningQueries.push({
          queryId: q.queryId,
          user: q.session?.user ?? 'unknown',
          elapsedMinutes: Math.round(wallMs / 60000),
          memoryBytes: q.queryStats?.userMemoryReservation
            ? parseTrinoSize(q.queryStats.userMemoryReservation)
            : null,
          queryText: (q.query ?? '').slice(0, 100),
        });
      }
    }
  } catch (_) {
    // Non-critical
  }

  const allAlerts = [
    ...(clusterHealth.status === 'fulfilled' ? clusterHealth.value.alerts : []),
    ...(workerHealth.status === 'fulfilled' ? workerHealth.value.alerts : []),
    ...(longRunningQueries.length > 0 ? [{
      severity: 'warning',
      type: 'long_running_queries',
      detail: `${longRunningQueries.length} queries running > 30 minutes`,
    }] : []),
  ];

  return {
    healthy: allAlerts.filter(a => a.severity === 'critical').length === 0,
    coordinator: coordHealth,
    cluster: clusterHealth.status === 'fulfilled' ? clusterHealth.value : null,
    workers: workerHealth.status === 'fulfilled' ? workerHealth.value : null,
    longRunningQueries,
    allAlerts,
  };
}

Related guides

• MCP tools for ClickHouse — ReplicatedMergeTree health, system.parts, and replication queue monitoring
• MCP tools for Apache Spark — SparkUI REST API, stage health, executor state, and shuffle spill
• MCP tools for Apache Flink — streaming job health, checkpoint failures, and backpressure detection
• MCP tools for dbt — model run status, test failures, source freshness, and artifact monitoring
• MCP tools for Apache Kafka — consumer group lag, partition health, and broker availability
• MCP server health check patterns — composite endpoints and failure classification