feat(postgres): add process list viewer

This commit is contained in:
Bagus Wahyu Aprianto authored and GitHub committed 2026-07-14 18:03:50 +08:00
1 parent 9105df9a04
commit 7f096653db
8 files changed
+514 -39

No files matched your search

@@ -0,0 +1,133 @@
import { describe, expect, it } from "vitest";
import type { QueryResult } from "@/types/database";
import { buildPgKillSql, isPgProcessListCompatibilityError, mapPgProcessRows, pgKillResultError, PG_PROCESS_LIST_LEGACY_SQL, PG_PROCESS_LIST_SQL, supportsPgProcessList } from "@/lib/database/postgresProcessList";
import { connectionSupportsProcessList, resolveProcessListDriver, resolveProcessListDriverForConnection, supportsProcessList } from "@/lib/database/processListDrivers";
import type { ConnectionConfig } from "@/types/database";
function result(columns: string[], rows: (string | number | boolean | null)[][]): QueryResult {
return { columns, rows, affected_rows: 0, execution_time_ms: 0 };
}
describe("mapPgProcessRows", () => {
it("maps a pg_stat_activity result into typed rows", () => {
const rows = mapPgProcessRows(result(["pid", "user", "db", "client", "app", "state", "wait", "time", "query"], [[4211, "app", "shop", "10.0.0.4", "psql", "active", "Client:ClientRead", 12, "SELECT * FROM orders"]]));
expect(rows).toEqual([
{
id: 4211,
user: "app",
db: "shop",
client: "10.0.0.4",
app: "psql",
state: "active",
wait: "Client:ClientRead",
time: 12,
query: "SELECT * FROM orders",
},
]);
});
it("tolerates NULL columns and case-variant names", () => {
const rows = mapPgProcessRows(result(["PID", "USER", "DB", "CLIENT", "APP", "STATE", "WAIT", "TIME", "QUERY"], [["4200", "postgres", null, "local", null, "idle", null, "340", null]]));
expect(rows[0]).toMatchObject({ id: 4200, user: "postgres", db: null, app: null, state: "idle", wait: null, time: 340, query: null });
});
it("returns an empty array for empty or malformed input", () => {
expect(mapPgProcessRows(null)).toEqual([]);
expect(mapPgProcessRows(undefined)).toEqual([]);
expect(mapPgProcessRows(result([], []))).toEqual([]);
});
});
describe("buildPgKillSql", () => {
it("builds pg_terminate_backend for a valid pid", () => {
expect(buildPgKillSql(4211)).toBe("SELECT pg_terminate_backend(4211)");
});
it("rejects non-integer, zero, or negative pids", () => {
expect(() => buildPgKillSql(1.5)).toThrow();
expect(() => buildPgKillSql(0)).toThrow();
expect(() => buildPgKillSql(-1)).toThrow();
expect(() => buildPgKillSql(Number.NaN)).toThrow();
});
});
describe("Postgres compatibility", () => {
it("provides a pre-9.6 query and only falls back for missing wait-event columns", () => {
expect(PG_PROCESS_LIST_SQL).toContain("wait_event_type");
expect(PG_PROCESS_LIST_LEGACY_SQL).toContain("CASE WHEN waiting THEN 'Lock'");
expect(PG_PROCESS_LIST_LEGACY_SQL).not.toContain("wait_event_type");
expect(isPgProcessListCompatibilityError(new Error('column "wait_event_type" does not exist'))).toBe(true);
expect(isPgProcessListCompatibilityError({ code: "42703" })).toBe(true);
expect(isPgProcessListCompatibilityError(new Error("permission denied for view pg_stat_activity"))).toBe(false);
});
it("requires pg_terminate_backend to confirm termination", () => {
expect(pgKillResultError([result(["pg_terminate_backend"], [[true]])])).toBeNull();
expect(pgKillResultError([result(["pg_terminate_backend"], [["t"]])])).toBeNull();
expect(pgKillResultError([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
expect(pgKillResultError([result(["pg_terminate_backend"], [])])).toContain("did not terminate");
});
});
describe("supportsPgProcessList", () => {
it("covers the Postgres-kernel family and excludes divergent wire-protocol engines", () => {
expect(supportsPgProcessList("postgres")).toBe(true);
// Postgres-kernel forks are unverified for now, so they stay excluded.
expect(supportsPgProcessList("opengauss")).toBe(false);
expect(supportsPgProcessList("kingbase")).toBe(false);
expect(supportsPgProcessList("redshift")).toBe(false);
expect(supportsPgProcessList("questdb")).toBe(false);
expect(supportsPgProcessList("mysql")).toBe(false);
expect(supportsPgProcessList(undefined)).toBe(false);
});
});
describe("resolveProcessListDriver", () => {
it("routes MySQL and Postgres engines to distinct drivers", () => {
const mysql = resolveProcessListDriver("mysql");
const postgres = resolveProcessListDriver("postgres");
expect(mysql?.buildKillSql(7)).toBe("KILL CONNECTION 7");
expect(postgres?.buildKillSql(7)).toBe("SELECT pg_terminate_backend(7)");
expect(postgres?.fallbackListSql).toBe(PG_PROCESS_LIST_LEGACY_SQL);
expect(postgres?.shouldUseFallbackListSql?.(new Error('column "wait_event" does not exist'))).toBe(true);
expect(postgres?.killResultError?.([result(["pg_terminate_backend"], [[false]])])).toContain("did not terminate");
expect(resolveProcessListDriver("sqlite")).toBeNull();
});
it("unifies process-list support across both families", () => {
expect(supportsProcessList("mysql")).toBe(true);
expect(supportsProcessList("postgres")).toBe(true);
expect(supportsProcessList("sqlite")).toBe(false);
expect(supportsProcessList(undefined)).toBe(false);
});
});
function conn(partial: Partial<ConnectionConfig>): ConnectionConfig {
return { db_type: "mysql", ...partial } as ConnectionConfig;
}
describe("connectionSupportsProcessList", () => {
it("gates on the real connection profile", () => {
expect(connectionSupportsProcessList(conn({ db_type: "mysql" }))).toBe(true);
expect(connectionSupportsProcessList(conn({ db_type: "postgres" }))).toBe(true);
expect(connectionSupportsProcessList(conn({ db_type: "sqlite" }))).toBe(false);
expect(connectionSupportsProcessList(undefined)).toBe(false);
});
it("resolves JDBC connections by their effective engine", () => {
expect(connectionSupportsProcessList(conn({ db_type: "jdbc", connection_string: "jdbc:mysql://db:3306/app" }))).toBe(true);
expect(connectionSupportsProcessList(conn({ db_type: "jdbc", connection_string: "jdbc:postgresql://db:5432/app" }))).toBe(true);
});
it("excludes Kyuubi / Hive2 JDBC that infer as MySQL but cannot serve SHOW PROCESSLIST", () => {
expect(connectionSupportsProcessList(conn({ db_type: "jdbc", connection_string: "jdbc:kyuubi://gw:10009/" }))).toBe(false);
expect(connectionSupportsProcessList(conn({ db_type: "jdbc", connection_string: "jdbc:hive2://hs2:10000/default" }))).toBe(false);
});
});
describe("resolveProcessListDriverForConnection", () => {
it("returns the engine driver for a JDBC MySQL connection and null for lookalikes", () => {
expect(resolveProcessListDriverForConnection(conn({ db_type: "jdbc", connection_string: "jdbc:mysql://db/app" }))?.buildKillSql(9)).toBe("KILL CONNECTION 9");
expect(resolveProcessListDriverForConnection(conn({ db_type: "jdbc", connection_string: "jdbc:hive2://hs2/default" }))).toBeNull();
});
});
@@ -0,0 +1,146 @@
import type { DatabaseType, QueryResult } from "@/types/database";
/**
* PostgreSQL "current activity / process list" helpers. Pure and framework-free
* so they can be unit-tested in isolation; the generic panel component wires them
* to the SQL bridge and the production-safety guard via the driver registry.
*
* The MySQL family lives in `./mysqlProcessList`; the generic, engine-agnostic
* bits (coordinator, interval clamping, session counting) are shared from there.
*/
const PG_PROCESS_LIST_DB_TYPES = new Set<DatabaseType>(["postgres"]);
/**
* One row per server-side backend. `now() - query_start` gives the age of the
* currently running (or last) statement; we fall back to the transaction and
* backend start so idle sessions still report a sensible age. Own session is
* kept in the result set so the panel can dim it rather than hide it.
*/
export const PG_PROCESS_LIST_SQL = `SELECT pid,
usename AS "user",
datname AS db,
coalesce(host(client_addr), client_hostname, 'local') AS client,
application_name AS app,
state,
coalesce(nullif(wait_event_type, '') || ':' || wait_event, wait_event_type, '') AS wait,
floor(extract(epoch FROM (now() - coalesce(query_start, xact_start, backend_start))))::bigint AS time,
query
FROM pg_stat_activity
ORDER BY time DESC NULLS LAST`;
/** PostgreSQL 9.2-9.5 expose `waiting` instead of wait-event detail columns. */
export const PG_PROCESS_LIST_LEGACY_SQL = `SELECT pid,
usename AS "user",
datname AS db,
coalesce(host(client_addr), client_hostname, 'local') AS client,
application_name AS app,
state,
CASE WHEN waiting THEN 'Lock' ELSE '' END AS wait,
floor(extract(epoch FROM (now() - coalesce(query_start, xact_start, backend_start))))::bigint AS time,
query
FROM pg_stat_activity
ORDER BY time DESC NULLS LAST`;
/** Scalar query that returns the viewer's own backend pid. */
export const PG_OWN_SESSION_SQL = "SELECT pg_backend_pid()";
export interface PgProcessRow {
id: number;
user: string;
db: string | null;
client: string;
app: string | null;
state: string | null;
wait: string | null;
time: number;
query: string | null;
}
function columnIndex(columns: string[], name: string): number {
const target = name.toLowerCase();
return columns.findIndex((column) => column.toLowerCase() === target);
}
function asString(value: unknown): string {
if (value === null || value === undefined) return "";
return String(value);
}
function asNullableString(value: unknown): string | null {
if (value === null || value === undefined) return null;
const text = String(value);
return text.length === 0 ? null : text;
}
function asNumber(value: unknown): number {
if (typeof value === "number") return value;
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : 0;
}
/**
* Map a `pg_stat_activity` result into typed rows. Column names are matched
* case-insensitively and any missing column degrades to an empty value rather
* than throwing, so forks that rename or drop a column still render.
*/
export function mapPgProcessRows(result: QueryResult | null | undefined): PgProcessRow[] {
if (!result || !Array.isArray(result.columns) || !Array.isArray(result.rows)) return [];
const columns = result.columns;
const pidIdx = columnIndex(columns, "pid");
const userIdx = columnIndex(columns, "user");
const dbIdx = columnIndex(columns, "db");
const clientIdx = columnIndex(columns, "client");
const appIdx = columnIndex(columns, "app");
const stateIdx = columnIndex(columns, "state");
const waitIdx = columnIndex(columns, "wait");
const timeIdx = columnIndex(columns, "time");
const queryIdx = columnIndex(columns, "query");
const cell = (row: (string | number | boolean | null)[], idx: number) => (idx >= 0 ? row[idx] : null);
return result.rows.map((row) => ({
id: asNumber(cell(row, pidIdx)),
user: asString(cell(row, userIdx)),
db: asNullableString(cell(row, dbIdx)),
client: asString(cell(row, clientIdx)),
app: asNullableString(cell(row, appIdx)),
state: asNullableString(cell(row, stateIdx)),
wait: asNullableString(cell(row, waitIdx)),
time: asNumber(cell(row, timeIdx)),
query: asNullableString(cell(row, queryIdx)),
}));
}
/**
* Build a `SELECT pg_terminate_backend(<pid>)` statement. `pid` is validated as a
* finite positive integer (never interpolated as free text) so there is no
* injection path.
*/
export function buildPgKillSql(pid: number): string {
if (!Number.isInteger(pid) || pid <= 0) {
throw new Error(`Invalid backend pid: ${pid}`);
}
return `SELECT pg_terminate_backend(${pid})`;
}
/** Return an error when PostgreSQL declines to terminate the target backend. */
export function pgKillResultError(results: QueryResult[]): string | null {
const result = results.find((item) => item.execution_error !== true);
const value = result?.rows?.[0]?.[0];
if (value === true || value === 1 || String(value).toLowerCase() === "t" || String(value).toLowerCase() === "true") return null;
return "pg_terminate_backend did not terminate the backend";
}
/** Detect the undefined-column failure produced by pre-9.6 pg_stat_activity. */
export function isPgProcessListCompatibilityError(error: unknown): boolean {
const code = typeof error === "object" && error !== null && "code" in error ? String((error as { code?: unknown }).code ?? "") : "";
if (code === "42703") return true;
const message = error instanceof Error ? error.message : String(error);
return /(?:wait_event_type|wait_event).*(?:does not exist|42703)|(?:does not exist|42703).*(?:wait_event_type|wait_event)/i.test(message);
}
/** Whether the given database type exposes the Postgres process-list viewer. */
export function supportsPgProcessList(dbType: DatabaseType | undefined): boolean {
return !!dbType && PG_PROCESS_LIST_DB_TYPES.has(dbType);
}
@@ -0,0 +1,133 @@
import type { ConnectionConfig, DatabaseType, QueryResult } from "@/types/database";
import { effectiveDatabaseTypeForConnection } from "@/lib/database/jdbcDialect";
import { buildKillSql as buildMysqlKillSql, mapProcessRows as mapMysqlProcessRows, PROCESS_LIST_SQL as MYSQL_PROCESS_LIST_SQL, supportsProcessList as supportsMysqlProcessList } from "./mysqlProcessList";
import { buildPgKillSql, isPgProcessListCompatibilityError, mapPgProcessRows, pgKillResultError, PG_OWN_SESSION_SQL, PG_PROCESS_LIST_LEGACY_SQL, PG_PROCESS_LIST_SQL, supportsPgProcessList } from "./postgresProcessList";
/**
* Engine-agnostic process-list model. Each supported engine contributes a driver
* describing how to list sessions, identify the caller's own session, render the
* columns, and kill a session. The panel component stays entirely generic.
*/
/** A displayable session row. `id` is the value passed to the driver's kill SQL. */
export type ProcessRow = { id: number } & Record<string, string | number | null>;
export interface ProcessColumn {
/** Key into the mapped row. */
key: string;
/** i18n key for the header label. */
labelKey: string;
/** Render the cell in a monospace font. */
mono?: boolean;
/** Sort numerically and default to descending on first click. */
numeric?: boolean;
/** Long free text (SQL statement) — truncate with a hover title. */
wide?: boolean;
}
export interface ProcessListDriver {
/** SQL that lists current sessions, one row each. */
listSql: string;
/** Compatibility query used when the primary list SQL references newer columns. */
fallbackListSql?: string;
/** Restrict fallback attempts to known version-compatibility failures. */
shouldUseFallbackListSql?(error: unknown): boolean;
/** Scalar SQL returning the caller's own session id (nullable path tolerated). */
ownSessionSql: string;
/** Columns to render, in display order. */
columns: ProcessColumn[];
/** Column key used for the initial sort. */
defaultSortKey: string;
/** Upper bound on rows fetched per refresh. */
maxRows: number;
/** Map a raw list result into typed rows. */
mapRows(result: QueryResult | null | undefined): ProcessRow[];
/** Build the validated statement that kills the given session id. */
buildKillSql(id: number): string;
/** Validate any engine-specific success value returned by the kill statement. */
killResultError?(results: QueryResult[]): string | null;
}
const MYSQL_COLUMNS: ProcessColumn[] = [
{ key: "id", labelKey: "processList.colId", mono: true, numeric: true },
{ key: "user", labelKey: "processList.colUser" },
{ key: "host", labelKey: "processList.colHost" },
{ key: "db", labelKey: "processList.colDb" },
{ key: "command", labelKey: "processList.colCommand" },
{ key: "time", labelKey: "processList.colTime", mono: true, numeric: true },
{ key: "state", labelKey: "processList.colState" },
{ key: "info", labelKey: "processList.colInfo", mono: true, wide: true },
];
const POSTGRES_COLUMNS: ProcessColumn[] = [
{ key: "id", labelKey: "processList.colPid", mono: true, numeric: true },
{ key: "user", labelKey: "processList.colUser" },
{ key: "db", labelKey: "processList.colDb" },
{ key: "client", labelKey: "processList.colClient" },
{ key: "app", labelKey: "processList.colApp" },
{ key: "state", labelKey: "processList.colState" },
{ key: "wait", labelKey: "processList.colWait" },
{ key: "time", labelKey: "processList.colTime", mono: true, numeric: true },
{ key: "query", labelKey: "processList.colQuery", mono: true, wide: true },
];
const MYSQL_DRIVER: ProcessListDriver = {
listSql: MYSQL_PROCESS_LIST_SQL,
ownSessionSql: "SELECT CONNECTION_ID()",
columns: MYSQL_COLUMNS,
defaultSortKey: "time",
maxRows: 5000,
// Typed structs carry no index signature; they are plain string-keyed objects at runtime.
mapRows: (result) => mapMysqlProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildMysqlKillSql,
};
const POSTGRES_DRIVER: ProcessListDriver = {
listSql: PG_PROCESS_LIST_SQL,
fallbackListSql: PG_PROCESS_LIST_LEGACY_SQL,
shouldUseFallbackListSql: isPgProcessListCompatibilityError,
ownSessionSql: PG_OWN_SESSION_SQL,
columns: POSTGRES_COLUMNS,
defaultSortKey: "time",
maxRows: 5000,
mapRows: (result) => mapPgProcessRows(result) as unknown as ProcessRow[],
buildKillSql: buildPgKillSql,
killResultError: pgKillResultError,
};
/** Resolve the process-list driver for a connection, or null if unsupported. */
export function resolveProcessListDriver(dbType: DatabaseType | undefined): ProcessListDriver | null {
if (supportsMysqlProcessList(dbType)) return MYSQL_DRIVER;
if (supportsPgProcessList(dbType)) return POSTGRES_DRIVER;
return null;
}
/** Whether any process-list viewer (MySQL or Postgres family) covers this engine. */
export function supportsProcessList(dbType: DatabaseType | undefined): boolean {
return resolveProcessListDriver(dbType) !== null;
}
/**
* JDBC profiles that only borrow MySQL SQL syntax (Kyuubi / HiveServer2) infer as
* `mysql` but are Spark/Hive engines that cannot serve `SHOW FULL PROCESSLIST`.
*/
const MYSQL_LOOKALIKE_JDBC = /(?:kyuubi|hive2|org\.apache\.hive\.jdbc\.HiveDriver|hive-jdbc)/i;
/**
* Resolve the process-list driver from the real connection profile. Uses the
* effective engine (so JDBC connections that resolve to MySQL/Postgres work) and
* excludes MySQL-lookalike JDBC engines that cannot serve the process list.
*/
export function resolveProcessListDriverForConnection(connection: ConnectionConfig | undefined): ProcessListDriver | null {
if (!connection) return null;
if (connection.db_type === "jdbc") {
const profile = [connection.driver_profile, connection.connection_string, connection.jdbc_driver_class, ...(connection.jdbc_driver_paths ?? [])].filter(Boolean).join("\n");
if (MYSQL_LOOKALIKE_JDBC.test(profile)) return null;
}
return resolveProcessListDriver(effectiveDatabaseTypeForConnection(connection));
}
/** Connection-aware process-list gate (mirrors the server-dashboard gate). */
export function connectionSupportsProcessList(connection: ConnectionConfig | undefined): boolean {
return resolveProcessListDriverForConnection(connection) !== null;
}