feat(sqlserver): add activity trace
This commit is contained in:
24 files changed
+2108
-10
No files matched your search
@@ -0,0 +1,434 @@
|
||||
import { reactive } from "vue";
|
||||
import { uuid } from "@/lib/common/utils";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import { safeLocalStorageGet, safeLocalStorageSet } from "@/lib/backend/safeStorage";
|
||||
import type { QueryResult } from "@/types/database";
|
||||
import {
|
||||
SQLSERVER_TRACE_DEFAULT_DURATION_MINUTES,
|
||||
SQLSERVER_TRACE_DEFAULT_MAX_EVENTS,
|
||||
SQLSERVER_TRACE_MAX_DURATION_MINUTES,
|
||||
SQLSERVER_TRACE_READ_BATCH_SIZE,
|
||||
SQLSERVER_TRACE_STALE_GRACE_MINUTES,
|
||||
buildCleanupSqlServerTraceSessionsSql,
|
||||
buildCreateSqlServerTraceSessionSql,
|
||||
buildDropSqlServerTraceSessionSql,
|
||||
buildListSqlServerTraceSessionsSql,
|
||||
buildReadSqlServerTraceEventsSql,
|
||||
buildSqlServerTraceCapabilitiesSql,
|
||||
buildSqlServerTraceSessionName,
|
||||
buildStopSqlServerTraceSessionSql,
|
||||
isValidSqlServerTraceSessionName,
|
||||
missingSqlServerTraceCapabilities,
|
||||
normalizeSqlServerTraceDurationMinutes,
|
||||
normalizeSqlServerTraceMaxEvents,
|
||||
parseSqlServerTraceCapabilities,
|
||||
parseSqlServerTraceEvents,
|
||||
sqlServerTraceCapabilityProblem,
|
||||
staleSqlServerTraceSessionNames,
|
||||
type SqlServerTraceEvent,
|
||||
type SqlServerTraceCapabilityProblem,
|
||||
} from "@/lib/sqlserver/sqlServerActivityTrace";
|
||||
|
||||
export type SqlServerTraceStatus = "idle" | "starting" | "running" | "paused" | "stopping" | "stopped" | "error";
|
||||
export type SqlServerTraceAutoStopReason = "duration" | "events";
|
||||
|
||||
export interface SqlServerActivityTraceState {
|
||||
status: SqlServerTraceStatus;
|
||||
selectedDatabase: string;
|
||||
includeStatements: boolean;
|
||||
maxEvents: number;
|
||||
durationMinutes: number;
|
||||
events: SqlServerTraceEvent[];
|
||||
selectedEvent: SqlServerTraceEvent | null;
|
||||
sqlFilter: string;
|
||||
loginFilter: string;
|
||||
clientFilter: string;
|
||||
sessionFilter: string;
|
||||
error: string;
|
||||
sessionName: string;
|
||||
startedAt: number | null;
|
||||
elapsedSeconds: number;
|
||||
polling: boolean;
|
||||
autoStopReason: SqlServerTraceAutoStopReason | null;
|
||||
autoStopRevision: number;
|
||||
capabilityProblem: SqlServerTraceCapabilityProblem | null;
|
||||
capabilityVersion: string;
|
||||
missingCapabilities: string[];
|
||||
}
|
||||
|
||||
interface SqlServerTraceCursor {
|
||||
timestamp: string;
|
||||
counts: Map<string, number>;
|
||||
}
|
||||
|
||||
interface PendingSqlServerTraceSession {
|
||||
connectionId: string;
|
||||
database: string;
|
||||
sessionName: string;
|
||||
expiresAt: number;
|
||||
}
|
||||
|
||||
interface PollOptions {
|
||||
final?: boolean;
|
||||
allowWhileStopping?: boolean;
|
||||
}
|
||||
|
||||
export interface SqlServerActivityTraceRuntime {
|
||||
readonly tabId: string;
|
||||
readonly connectionId: string;
|
||||
readonly state: SqlServerActivityTraceState;
|
||||
start(): Promise<void>;
|
||||
pause(): void;
|
||||
resume(): Promise<void>;
|
||||
stop(reason?: SqlServerTraceAutoStopReason): Promise<void>;
|
||||
clearEvents(): void;
|
||||
}
|
||||
|
||||
const PENDING_SESSIONS_STORAGE_KEY = "dbx:sqlserver-trace:pending-sessions:v1";
|
||||
const runtimes = new Map<string, InternalSqlServerActivityTraceRuntime>();
|
||||
const activeSessionNames = new Set<string>();
|
||||
const staleCleanupTasks = new Map<string, Promise<number>>();
|
||||
|
||||
interface InternalSqlServerActivityTraceRuntime extends SqlServerActivityTraceRuntime {
|
||||
pollTimer?: ReturnType<typeof setInterval>;
|
||||
elapsedTimer?: ReturnType<typeof setInterval>;
|
||||
pollPromise?: Promise<void>;
|
||||
activeReadExecutionId?: string;
|
||||
generation: number;
|
||||
cursor?: SqlServerTraceCursor;
|
||||
startPromise?: Promise<void>;
|
||||
stopPromise?: Promise<void>;
|
||||
}
|
||||
|
||||
function readPendingSessions(): PendingSqlServerTraceSession[] {
|
||||
try {
|
||||
const value = JSON.parse(safeLocalStorageGet(PENDING_SESSIONS_STORAGE_KEY) || "[]") as unknown;
|
||||
if (!Array.isArray(value)) return [];
|
||||
return value.filter(
|
||||
(entry): entry is PendingSqlServerTraceSession =>
|
||||
!!entry &&
|
||||
typeof entry === "object" &&
|
||||
typeof (entry as PendingSqlServerTraceSession).connectionId === "string" &&
|
||||
typeof (entry as PendingSqlServerTraceSession).database === "string" &&
|
||||
typeof (entry as PendingSqlServerTraceSession).sessionName === "string" &&
|
||||
isValidSqlServerTraceSessionName((entry as PendingSqlServerTraceSession).sessionName) &&
|
||||
Number.isFinite((entry as PendingSqlServerTraceSession).expiresAt),
|
||||
);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function writePendingSessions(sessions: PendingSqlServerTraceSession[]) {
|
||||
safeLocalStorageSet(PENDING_SESSIONS_STORAGE_KEY, JSON.stringify(sessions));
|
||||
}
|
||||
|
||||
function registerPendingSession(runtime: InternalSqlServerActivityTraceRuntime, sessionName: string) {
|
||||
activeSessionNames.add(sessionName);
|
||||
const sessions = readPendingSessions().filter((entry) => entry.sessionName !== sessionName);
|
||||
sessions.push({
|
||||
connectionId: runtime.connectionId,
|
||||
database: runtime.state.selectedDatabase,
|
||||
sessionName,
|
||||
expiresAt: Date.now() + (normalizeSqlServerTraceDurationMinutes(runtime.state.durationMinutes) + SQLSERVER_TRACE_STALE_GRACE_MINUTES) * 60_000,
|
||||
});
|
||||
writePendingSessions(sessions);
|
||||
}
|
||||
|
||||
function unregisterPendingSession(sessionName: string) {
|
||||
activeSessionNames.delete(sessionName);
|
||||
writePendingSessions(readPendingSessions().filter((entry) => entry.sessionName !== sessionName));
|
||||
}
|
||||
|
||||
async function executeControlQuery(connectionId: string, database: string, sql: string, maxRows = 1): Promise<QueryResult> {
|
||||
return api.executeQuery(connectionId, database || "master", sql, undefined, uuid(), { maxRows });
|
||||
}
|
||||
|
||||
export function cleanupStaleSqlServerTraceSessions(connectionId: string, database = "master"): Promise<number> {
|
||||
const existing = staleCleanupTasks.get(connectionId);
|
||||
if (existing) return existing;
|
||||
const task = (async () => {
|
||||
const now = Date.now();
|
||||
const pending = readPendingSessions().filter((entry) => entry.connectionId === connectionId && entry.expiresAt <= now && !activeSessionNames.has(entry.sessionName));
|
||||
const listResult = await executeControlQuery(connectionId, database, buildListSqlServerTraceSessionsSql(), 500);
|
||||
const expired = staleSqlServerTraceSessionNames(listResult, now);
|
||||
const names = [...new Set([...pending.map((entry) => entry.sessionName), ...expired])].filter((name) => !activeSessionNames.has(name));
|
||||
if (names.length === 0) return 0;
|
||||
await executeControlQuery(connectionId, database, buildCleanupSqlServerTraceSessionsSql(names));
|
||||
const cleanedNames = new Set(names);
|
||||
writePendingSessions(readPendingSessions().filter((entry) => !cleanedNames.has(entry.sessionName)));
|
||||
return names.length;
|
||||
})().finally(() => staleCleanupTasks.delete(connectionId));
|
||||
staleCleanupTasks.set(connectionId, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
function stopTimers(runtime: InternalSqlServerActivityTraceRuntime) {
|
||||
if (runtime.pollTimer) clearInterval(runtime.pollTimer);
|
||||
if (runtime.elapsedTimer) clearInterval(runtime.elapsedTimer);
|
||||
runtime.pollTimer = undefined;
|
||||
runtime.elapsedTimer = undefined;
|
||||
}
|
||||
|
||||
function startTimers(runtime: InternalSqlServerActivityTraceRuntime) {
|
||||
stopTimers(runtime);
|
||||
runtime.pollTimer = setInterval(() => {
|
||||
if (runtime.state.status === "running") void pollEvents(runtime);
|
||||
}, 1000);
|
||||
runtime.elapsedTimer = setInterval(() => {
|
||||
if (!runtime.state.startedAt || (runtime.state.status !== "running" && runtime.state.status !== "paused")) return;
|
||||
runtime.state.elapsedSeconds = Math.max(0, Math.floor((Date.now() - runtime.state.startedAt) / 1000));
|
||||
if (runtime.state.elapsedSeconds >= normalizeSqlServerTraceDurationMinutes(runtime.state.durationMinutes) * 60) {
|
||||
void stopRuntime(runtime, "duration");
|
||||
}
|
||||
}, 1000);
|
||||
}
|
||||
|
||||
function mergeIncrementalBatch(runtime: InternalSqlServerActivityTraceRuntime, batch: SqlServerTraceEvent[]): SqlServerTraceEvent[] {
|
||||
const cursor = runtime.cursor;
|
||||
const additions: SqlServerTraceEvent[] = [];
|
||||
const countsByTimestamp = new Map<string, Map<string, number>>();
|
||||
for (const event of batch) {
|
||||
const timestampCounts = countsByTimestamp.get(event.timestamp) ?? new Map<string, number>();
|
||||
const occurrence = (timestampCounts.get(event.key) ?? 0) + 1;
|
||||
timestampCounts.set(event.key, occurrence);
|
||||
countsByTimestamp.set(event.timestamp, timestampCounts);
|
||||
if (!cursor || event.timestamp > cursor.timestamp || (event.timestamp === cursor.timestamp && occurrence > (cursor.counts.get(event.key) ?? 0))) {
|
||||
additions.push({ ...event, key: `${event.key}\u0001${occurrence}` });
|
||||
}
|
||||
}
|
||||
const latestTimestamp = batch[batch.length - 1]?.timestamp;
|
||||
if (latestTimestamp) runtime.cursor = { timestamp: latestTimestamp, counts: countsByTimestamp.get(latestTimestamp) ?? new Map() };
|
||||
return additions;
|
||||
}
|
||||
|
||||
async function cancelActiveRead(runtime: InternalSqlServerActivityTraceRuntime) {
|
||||
const executionId = runtime.activeReadExecutionId;
|
||||
if (!executionId) return;
|
||||
runtime.activeReadExecutionId = undefined;
|
||||
await api.cancelQuery(executionId).catch(() => false);
|
||||
await runtime.pollPromise?.catch(() => undefined);
|
||||
}
|
||||
|
||||
async function cleanupSession(runtime: InternalSqlServerActivityTraceRuntime) {
|
||||
const name = runtime.state.sessionName;
|
||||
if (!name) return;
|
||||
try {
|
||||
await executeControlQuery(runtime.connectionId, runtime.state.selectedDatabase, buildStopSqlServerTraceSessionSql(name));
|
||||
} catch {
|
||||
// The session may already be stopped or the connection may be recovering.
|
||||
}
|
||||
try {
|
||||
await executeControlQuery(runtime.connectionId, runtime.state.selectedDatabase, buildDropSqlServerTraceSessionSql(name));
|
||||
runtime.state.sessionName = "";
|
||||
unregisterPendingSession(name);
|
||||
} catch (error) {
|
||||
runtime.state.error = error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
}
|
||||
|
||||
async function pollEvents(runtime: InternalSqlServerActivityTraceRuntime, options: PollOptions = {}) {
|
||||
if (!runtime.state.sessionName || runtime.state.polling) return;
|
||||
if (!options.allowWhileStopping && runtime.state.status !== "running") return;
|
||||
const generation = runtime.generation;
|
||||
const executionId = uuid();
|
||||
runtime.activeReadExecutionId = executionId;
|
||||
runtime.state.polling = true;
|
||||
let autoStopReason: SqlServerTraceAutoStopReason | undefined;
|
||||
const promise = api
|
||||
.executeQuery(runtime.connectionId, runtime.state.selectedDatabase, buildReadSqlServerTraceEventsSql(runtime.state.sessionName, runtime.cursor?.timestamp), undefined, executionId, { maxRows: SQLSERVER_TRACE_READ_BATCH_SIZE })
|
||||
.then((result) => {
|
||||
if (runtime.generation !== generation) return;
|
||||
const additions = mergeIncrementalBatch(runtime, parseSqlServerTraceEvents(result));
|
||||
const remaining = Math.max(0, normalizeSqlServerTraceMaxEvents(runtime.state.maxEvents) - runtime.state.events.length);
|
||||
if (remaining > 0) runtime.state.events.push(...additions.slice(0, remaining));
|
||||
if (!options.final && runtime.state.events.length >= normalizeSqlServerTraceMaxEvents(runtime.state.maxEvents)) {
|
||||
autoStopReason = "events";
|
||||
}
|
||||
})
|
||||
.catch(async (error) => {
|
||||
if (options.final || runtime.generation !== generation || runtime.state.status === "stopping") return;
|
||||
runtime.state.error = error instanceof Error ? error.message : String(error);
|
||||
runtime.state.status = "error";
|
||||
stopTimers(runtime);
|
||||
await cleanupSession(runtime);
|
||||
})
|
||||
.finally(() => {
|
||||
if (runtime.activeReadExecutionId === executionId) runtime.activeReadExecutionId = undefined;
|
||||
runtime.state.polling = false;
|
||||
if (runtime.pollPromise === promise) runtime.pollPromise = undefined;
|
||||
});
|
||||
runtime.pollPromise = promise;
|
||||
await promise;
|
||||
if (autoStopReason) void stopRuntime(runtime, autoStopReason);
|
||||
}
|
||||
|
||||
async function startRuntime(runtime: InternalSqlServerActivityTraceRuntime) {
|
||||
const state = runtime.state;
|
||||
if (!state.selectedDatabase || state.status === "running" || state.status === "paused" || state.status === "starting" || state.status === "stopping") return;
|
||||
state.error = "";
|
||||
state.capabilityProblem = null;
|
||||
state.capabilityVersion = "";
|
||||
state.missingCapabilities = [];
|
||||
state.status = "starting";
|
||||
state.maxEvents = normalizeSqlServerTraceMaxEvents(state.maxEvents);
|
||||
state.durationMinutes = normalizeSqlServerTraceDurationMinutes(state.durationMinutes);
|
||||
state.events = [];
|
||||
state.selectedEvent = null;
|
||||
state.elapsedSeconds = 0;
|
||||
runtime.cursor = undefined;
|
||||
runtime.generation += 1;
|
||||
const generation = runtime.generation;
|
||||
try {
|
||||
await cleanupStaleSqlServerTraceSessions(runtime.connectionId, state.selectedDatabase);
|
||||
const capabilities = parseSqlServerTraceCapabilities(await executeControlQuery(runtime.connectionId, state.selectedDatabase, buildSqlServerTraceCapabilitiesSql(state.selectedDatabase)));
|
||||
const problem = sqlServerTraceCapabilityProblem(capabilities, state.includeStatements);
|
||||
if (problem) {
|
||||
state.capabilityProblem = problem;
|
||||
state.capabilityVersion = capabilities.productVersion;
|
||||
state.missingCapabilities = missingSqlServerTraceCapabilities(capabilities, state.includeStatements);
|
||||
throw new Error(problem);
|
||||
}
|
||||
const name = buildSqlServerTraceSessionName(Date.now() + (state.durationMinutes + SQLSERVER_TRACE_STALE_GRACE_MINUTES) * 60_000);
|
||||
state.sessionName = name;
|
||||
registerPendingSession(runtime, name);
|
||||
await executeControlQuery(
|
||||
runtime.connectionId,
|
||||
state.selectedDatabase,
|
||||
buildCreateSqlServerTraceSessionSql({
|
||||
sessionName: name,
|
||||
databaseId: capabilities.databaseId,
|
||||
maxEvents: state.maxEvents,
|
||||
includeStatements: state.includeStatements,
|
||||
}),
|
||||
);
|
||||
if (runtime.generation !== generation) {
|
||||
await cleanupSession(runtime);
|
||||
return;
|
||||
}
|
||||
state.startedAt = Date.now();
|
||||
state.status = "running";
|
||||
startTimers(runtime);
|
||||
void pollEvents(runtime);
|
||||
} catch (error) {
|
||||
if (!state.capabilityProblem) state.error = error instanceof Error ? error.message : String(error);
|
||||
state.status = "error";
|
||||
stopTimers(runtime);
|
||||
await cleanupSession(runtime);
|
||||
}
|
||||
}
|
||||
|
||||
async function stopRuntime(runtime: InternalSqlServerActivityTraceRuntime, reason?: SqlServerTraceAutoStopReason, final = true) {
|
||||
if (runtime.stopPromise) return runtime.stopPromise;
|
||||
const promise = (async () => {
|
||||
if (!runtime.state.sessionName && runtime.state.status !== "starting") return;
|
||||
runtime.state.status = "stopping";
|
||||
stopTimers(runtime);
|
||||
runtime.generation += 1;
|
||||
await runtime.startPromise?.catch(() => undefined);
|
||||
await cancelActiveRead(runtime);
|
||||
if (final && runtime.state.sessionName) await pollEvents(runtime, { final: true, allowWhileStopping: true });
|
||||
await cleanupSession(runtime);
|
||||
runtime.state.status = runtime.state.sessionName ? "error" : "stopped";
|
||||
if (reason && !runtime.state.sessionName) {
|
||||
runtime.state.autoStopReason = reason;
|
||||
runtime.state.autoStopRevision += 1;
|
||||
}
|
||||
})().finally(() => {
|
||||
if (runtime.stopPromise === promise) runtime.stopPromise = undefined;
|
||||
});
|
||||
runtime.stopPromise = promise;
|
||||
await promise;
|
||||
}
|
||||
|
||||
export function getSqlServerActivityTraceRuntime(tabId: string, connectionId: string, database = ""): SqlServerActivityTraceRuntime {
|
||||
const existing = runtimes.get(tabId);
|
||||
if (existing) return existing;
|
||||
const runtime: InternalSqlServerActivityTraceRuntime = {
|
||||
tabId,
|
||||
connectionId,
|
||||
generation: 0,
|
||||
state: reactive<SqlServerActivityTraceState>({
|
||||
status: "idle",
|
||||
selectedDatabase: database,
|
||||
includeStatements: false,
|
||||
maxEvents: SQLSERVER_TRACE_DEFAULT_MAX_EVENTS,
|
||||
durationMinutes: SQLSERVER_TRACE_DEFAULT_DURATION_MINUTES,
|
||||
events: [],
|
||||
selectedEvent: null,
|
||||
sqlFilter: "",
|
||||
loginFilter: "",
|
||||
clientFilter: "",
|
||||
sessionFilter: "",
|
||||
error: "",
|
||||
sessionName: "",
|
||||
startedAt: null,
|
||||
elapsedSeconds: 0,
|
||||
polling: false,
|
||||
autoStopReason: null,
|
||||
autoStopRevision: 0,
|
||||
capabilityProblem: null,
|
||||
capabilityVersion: "",
|
||||
missingCapabilities: [],
|
||||
}),
|
||||
start: () => {
|
||||
if (runtime.startPromise) return runtime.startPromise;
|
||||
const promise = startRuntime(runtime).finally(() => {
|
||||
if (runtime.startPromise === promise) runtime.startPromise = undefined;
|
||||
});
|
||||
runtime.startPromise = promise;
|
||||
return promise;
|
||||
},
|
||||
pause: () => {
|
||||
if (runtime.state.status !== "running") return;
|
||||
runtime.state.status = "paused";
|
||||
if (runtime.pollTimer) clearInterval(runtime.pollTimer);
|
||||
runtime.pollTimer = undefined;
|
||||
},
|
||||
resume: async () => {
|
||||
if (runtime.state.status !== "paused") return;
|
||||
runtime.state.status = "running";
|
||||
startTimers(runtime);
|
||||
await pollEvents(runtime);
|
||||
},
|
||||
stop: (reason) => stopRuntime(runtime, reason),
|
||||
clearEvents: () => {
|
||||
runtime.state.events = [];
|
||||
runtime.state.selectedEvent = null;
|
||||
},
|
||||
};
|
||||
runtimes.set(tabId, runtime);
|
||||
return runtime;
|
||||
}
|
||||
|
||||
export async function disposeSqlServerActivityTrace(tabId: string): Promise<void> {
|
||||
const runtime = runtimes.get(tabId);
|
||||
if (!runtime) return;
|
||||
runtime.generation += 1;
|
||||
stopTimers(runtime);
|
||||
await cancelActiveRead(runtime);
|
||||
await runtime.startPromise?.catch(() => undefined);
|
||||
if (runtime.state.sessionName) await stopRuntime(runtime, undefined, false);
|
||||
if (runtimes.get(tabId) === runtime) runtimes.delete(tabId);
|
||||
}
|
||||
|
||||
export async function disposeAllSqlServerActivityTraces(): Promise<void> {
|
||||
await Promise.all([...runtimes.keys()].map(disposeSqlServerActivityTrace));
|
||||
}
|
||||
|
||||
export async function disposeSqlServerActivityTracesForConnection(connectionId: string, database?: string): Promise<void> {
|
||||
const tabIds = [...runtimes.values()].filter((runtime) => runtime.connectionId === connectionId && (!database || runtime.state.selectedDatabase === database)).map((runtime) => runtime.tabId);
|
||||
await Promise.all(tabIds.map(disposeSqlServerActivityTrace));
|
||||
}
|
||||
|
||||
export function activeSqlServerActivityTraceRuntimeCount(): number {
|
||||
return runtimes.size;
|
||||
}
|
||||
|
||||
export function hasSqlServerActivityTraceForConnection(connectionId: string, database?: string): boolean {
|
||||
return [...runtimes.values()].some((runtime) => runtime.connectionId === connectionId && (!database || runtime.state.selectedDatabase === database));
|
||||
}
|
||||
|
||||
export function sqlServerTracePendingSessionLifetimeMs(): number {
|
||||
return (SQLSERVER_TRACE_MAX_DURATION_MINUTES + SQLSERVER_TRACE_STALE_GRACE_MINUTES) * 60_000;
|
||||
}
|
||||
Reference in new issue
Block a user