Line data Source code
1 : import type * as arrow from 'apache-arrow';
2 :
3 : import type { EmbeddedComputeDatabase, EmbeddedConnection } from '../../platform/database/embedded_database.js';
4 : import {
5 : HyperDBIncrementalExtract,
6 : HYPER_EXTRACT_ROW_COLUMN,
7 : quoteLiteral,
8 : type HyperDBExtractTable,
9 : type HyperDBExtractTransfer,
10 : type HyperDBSourceDatabase,
11 : } from '../../platform/hyperdb/hyperdb_incremental_extract.js';
12 : import {
13 : HYPER_LOG_RESOURCE_METRICS_TABLE,
14 : HyperDBLogSource,
15 : HyperLogUnreadableError,
16 : hyperLogQueryEndsTable,
17 : isHyperLogFlushTimeout,
18 : } from '../../platform/hyperdb/hyperdb_log_extract.js';
19 : import { quoteIdentifier } from '../../platform/hyperdb/hyperdb_wasm.js';
20 : import type { Logger } from '../../platform/logger/logger.js';
21 : import { stringifyError } from '../../platform/logger/logger.js';
22 : import { Store } from '../../utils/store.js';
23 :
24 1 : const LOG_CTX = 'hyper_log_metrics';
25 :
26 : /// The polling interval of the Hyper log in milliseconds, the smallest bucket of the charts.
27 : /// A poll reads all events since the last one, e.g. the resource usage that the embedded Hyper logs every second.
28 : /// Every poll is about 15 queries of the monitor itself, which the extract of an embedded Hyper keeps too.
29 1 : const DEFAULT_POLL_INTERVAL_MS = 5_000;
30 : /// The alias of the monitoring database of a connection on the connections of its monitor
31 1 : const MONITORING_ALIAS = 'monitoring';
32 : /// The attempts to open an extract while reading the log times out, e.g. on a Hyper under load, and the pause between them
33 1 : const OPEN_ATTEMPTS = 5;
34 1 : const OPEN_RETRY_MS = 1_000;
35 : /// The time that the extract reaches back, twice the largest window of the connection metrics
36 1 : const RETENTION_MS = 2 * 60 * 60_000;
37 : /// The finished queries that the extract keeps at most, e.g. after a burst of queries. The oldest are deleted first.
38 : /// The log monitor finishes about 3 queries per second itself, about 22,000 in the retention.
39 1 : const MAX_QUERY_ENDS = 20_000;
40 : /// The resource samples that the extract keeps at most, the retention with a resource usage interval of 1s
41 1 : const MAX_RESOURCE_SAMPLES = 7_200;
42 :
43 : /// A query that a Hyper session finished, from a row of the query ends of the extract.
44 : export interface HyperQueryEnd {
45 : /// The Hyper session that ran the query
46 : sessionId: string;
47 : /// The end of the query in milliseconds since the epoch
48 : endedAt: number;
49 : /// Did the query succeed?
50 : succeeded: boolean;
51 : /// The SQLSTATE of a failed query
52 : errorCode: string | null;
53 : /// The error message of a failed query
54 : errorMessage: string | null;
55 : /// The statement type, e.g. SELECT
56 : statement: string | null;
57 : /// The query text, truncated by Hyper
58 : query: string | null;
59 : /// The elapsed time of the query in seconds
60 : elapsed: number | null;
61 : /// The thread time of the compilation and the execution in seconds
62 : threadTime: number;
63 : /// The CPU time of the threads of the compilation and the execution in seconds
64 : cpuTime: number;
65 : /// The time that the threads of the compilation and the execution waited in seconds
66 : waitTime: number;
67 : /// The time until a worker thread picked up the query in seconds
68 : timeToSchedule: number | null;
69 : /// The peak transaction memory of the compilation or the execution in MB
70 : peakMemoryMb: number | null;
71 : /// The result rows
72 : rows: number | null;
73 : /// The rows that the compilation and the execution processed
74 : processedRows: number;
75 : /// The kind of the query from its client context, see DashQLClientContext
76 : queryType: string | null;
77 : /// Did the app run the query and not the user? E.g. a computation (SQLFrame) on a pooled compute connection, or a
78 : /// read of the Hyper log.
79 : sentByApp: boolean;
80 : }
81 :
82 : /// The client context of the queries of a connection that Hyper logs as ctx.client
83 : interface DashQLClientContext {
84 : /// The fields of the app
85 : dashql: {
86 : /// The connection that the queries are attributed to, none for the queries of the log monitor
87 : connectionId?: string;
88 : /// The kind of the queries: of the user on the sessions of the connection, of computations, of the
89 : /// connection overview, or of the log monitor
90 : queryType: 'user' | 'sqlframe' | 'overview' | 'monitor';
91 : };
92 : }
93 :
94 : /// The client context of the queries of the user on the sessions of a connection
95 1 : export function connectionClientContext(connectionId: string): DashQLClientContext & Record<string, unknown> {
96 1 : return { dashql: { connectionId, queryType: 'user' } };
97 1 : }
98 :
99 : /// The client context that attributes the queries of a computation to the connection of its table.
100 : /// The queries run on pooled compute connections, whose sessions are shared by all connections.
101 1 : export function sqlframeClientContext(connectionId: string): DashQLClientContext & Record<string, unknown> {
102 1 : return { dashql: { connectionId, queryType: 'sqlframe' } };
103 1 : }
104 :
105 : /// The client context that attributes the queries of the overview of a connection to the connection
106 1 : export function overviewClientContext(connectionId: string): DashQLClientContext & Record<string, unknown> {
107 1 : return { dashql: { connectionId, queryType: 'overview' } };
108 1 : }
109 :
110 : /// The client context of the connections of the log monitor of a connection, they read the Hyper log and write the
111 : /// extract
112 1 : function monitorClientContext(connectionId: string): DashQLClientContext & Record<string, unknown> {
113 1 : return { dashql: { connectionId, queryType: 'monitor' } };
114 1 : }
115 :
116 : /// A sample of the resource usage of a Hyper process, from a row of the resource metrics of the extract.
117 : /// Hyper writes one every log_resource_usage_interval.
118 : export interface HyperResourceSample {
119 : /// The time of the sample in milliseconds since the epoch
120 : sampledAt: number;
121 : /// The physical memory of the process in MB, the size of the Wasm memory, which never shrinks
122 : processMemoryMb: number | null;
123 : /// The memory that the running queries hold in MB
124 : transactionMemoryMb: number | null;
125 : /// The memory of all memory trackers in MB, e.g. table data, temporary tables, the running queries and caches
126 : trackedMemoryMb: number | null;
127 : /// The overall load between 0 and 1
128 : overallLoad: number | null;
129 : /// The load of the scheduler between 0 and 1
130 : schedulerLoad: number | null;
131 : /// The memory load between 0 and 1
132 : memoryLoad: number | null;
133 : }
134 :
135 : /// The query ends with the fields of the client context of the app
136 1 : const QUERY_ENDS_TABLE: HyperDBExtractTable = {
137 1 : ...hyperLogQueryEndsTable([
138 1 : { name: 'client_connection_id', path: ['dashql', 'connectionId'] },
139 1 : { name: 'client_query_type', path: ['dashql', 'queryType'] },
140 1 : ]),
141 1 : maxRows: MAX_QUERY_ENDS,
142 1 : maxAgeMs: RETENTION_MS,
143 1 : };
144 :
145 : /// The resource samples of the Hyper process
146 1 : const RESOURCE_METRICS_TABLE: HyperDBExtractTable = {
147 1 : ...HYPER_LOG_RESOURCE_METRICS_TABLE,
148 1 : maxRows: MAX_RESOURCE_SAMPLES,
149 1 : maxAgeMs: RETENTION_MS,
150 1 : };
151 :
152 : /// The events of the extract of a connection: the query-end events that its client context names, and the
153 : /// resource-metrics events of the Hyper process
154 1 : function metricsSourceQuery(connectionId: string): string {
155 1 : return `
156 : SELECT * FROM log
157 : WHERE k = 'resource-metrics'
158 1 : OR (k LIKE 'query-end%' AND ctx::json->'client'->'dashql'->>'connectionId' = ${quoteLiteral(connectionId)})
159 : `;
160 1 : }
161 :
162 : /// The rows of an extract table that a read selects
163 : export interface ExtractRowRange {
164 : /// The rows after this row number, see HYPER_EXTRACT_ROW_COLUMN
165 : afterRow: number;
166 : /// The rows logged since this time in milliseconds since the epoch
167 : since: number;
168 : }
169 :
170 : /// The time of a log event in milliseconds since the epoch.
171 : /// The log writes the wall-clock time of the Hyper host as timestamp without time zone, see HyperLogState.clockOffsetMs.
172 1 : function epochMs(clockOffsetMs: number): string {
173 1 : return `(extract(epoch from ts) * 1000 - ${clockOffsetMs})::double precision`;
174 1 : }
175 :
176 : /// The time of a log event in the time zone of the app, as timestamp without time zone
177 1 : function localTimestamp(clockOffsetMs: number): string {
178 1 : const shiftMs = -new Date().getTimezoneOffset() * 60_000 - clockOffsetMs;
179 1 : return shiftMs === 0 ? 'ts' : `ts + interval '${shiftMs / 1000} seconds'`;
180 1 : }
181 :
182 : /// The predicate of the rows of a range
183 1 : function rangePredicate(clockOffsetMs: number, range: ExtractRowRange): string {
184 1 : return `${HYPER_EXTRACT_ROW_COLUMN} > ${range.afterRow} AND ${epochMs(clockOffsetMs)} >= ${range.since}`;
185 1 : }
186 :
187 : /// The finished queries in the query ends of the extract of a connection, with their sent_by_app, epoch_ms (the end in
188 : /// milliseconds since the epoch) and local_ts (the end in the time zone of the app)
189 1 : export function queryEndsQuery(queryEnds: string, clockOffsetMs: number, range?: ExtractRowRange): string {
190 1 : return `
191 : SELECT
192 : *,
193 1 : ${epochMs(clockOffsetMs)} AS epoch_ms,
194 1 : ${localTimestamp(clockOffsetMs)} AS local_ts,
195 : client_query_type IS DISTINCT FROM 'user' AS sent_by_app
196 1 : FROM ${queryEnds}
197 1 : ${range == null ? '' : `WHERE ${rangePredicate(clockOffsetMs, range)}`}
198 : `;
199 1 : }
200 :
201 : /// The resource samples in the resource metrics of the extract, with their epoch_ms
202 1 : export function resourceSamplesQuery(resourceMetrics: string, clockOffsetMs: number, range: ExtractRowRange): string {
203 1 : return `SELECT *, ${epochMs(clockOffsetMs)} AS epoch_ms FROM ${resourceMetrics} WHERE ${rangePredicate(clockOffsetMs, range)}`;
204 1 : }
205 :
206 1 : function nullableNumber(value: unknown): number | null {
207 1 : return value == null ? null : Number(value);
208 1 : }
209 :
210 1 : function nullableString(value: unknown): string | null {
211 1 : return value == null ? null : String(value);
212 1 : }
213 :
214 : /// The highest row number of a result of extract rows, 0 without rows
215 1 : export function lastExtractRow(table: arrow.Table): number {
216 1 : const rows = table.getChild(HYPER_EXTRACT_ROW_COLUMN);
217 1 : let last = 0;
218 1 : for (let i = 0; i < table.numRows; ++i) last = Math.max(last, Number(rows?.get(i) ?? 0));
219 1 : return last;
220 1 : }
221 :
222 : /// Read the finished queries of a result of queryEndsQuery(), ordered by their end.
223 : /// The log times of Hyper in Wasm have millisecond resolution, the request number orders the queries of a session that
224 : /// end within the same millisecond.
225 1 : export function readQueryEnds(table: arrow.Table): HyperQueryEnd[] {
226 1 : const queries: { query: HyperQueryEnd, request: number }[] = [];
227 1 : for (const row of table) {
228 1 : queries.push({ request: Number(row.req ?? 0), query: {
229 1 : sessionId: String(row.sess),
230 1 : endedAt: Number(row.epoch_ms),
231 1 : succeeded: Boolean(row.succeeded),
232 1 : errorCode: nullableString(row.error_code),
233 1 : errorMessage: nullableString(row.error_message),
234 1 : statement: nullableString(row.statement),
235 1 : query: nullableString(row.query),
236 1 : elapsed: nullableNumber(row.elapsed),
237 1 : threadTime: Number(row.thread_time ?? 0),
238 1 : cpuTime: Number(row.cpu_time ?? 0),
239 1 : waitTime: Number(row.wait_time ?? 0),
240 1 : timeToSchedule: nullableNumber(row.time_to_schedule),
241 1 : peakMemoryMb: nullableNumber(row.peak_memory_mb),
242 1 : rows: nullableNumber(row.rows),
243 1 : processedRows: Number(row.processed_rows ?? 0),
244 1 : queryType: nullableString(row.client_query_type),
245 1 : sentByApp: Boolean(row.sent_by_app),
246 1 : } });
247 : }
248 1 : return queries
249 1 : .sort((l, r) => l.query.endedAt - r.query.endedAt || l.request - r.request)
250 1 : .map(({ query }) => query);
251 1 : }
252 :
253 : /// Read the resource samples of a result of resourceSamplesQuery(), ordered by time
254 1 : export function readResourceSamples(table: arrow.Table): HyperResourceSample[] {
255 1 : const samples: HyperResourceSample[] = [];
256 1 : for (const row of table) {
257 1 : samples.push({
258 1 : sampledAt: Number(row.epoch_ms),
259 1 : processMemoryMb: nullableNumber(row.process_memory_mb),
260 1 : transactionMemoryMb: nullableNumber(row.transaction_memory_mb),
261 1 : trackedMemoryMb: nullableNumber(row.tracked_memory_mb),
262 1 : overallLoad: nullableNumber(row.overall_load),
263 1 : schedulerLoad: nullableNumber(row.scheduler_load),
264 1 : memoryLoad: nullableNumber(row.memory_load),
265 1 : });
266 : }
267 1 : return samples.sort((l, r) => l.sampledAt - r.sampledAt);
268 1 : }
269 :
270 : /// The status of the monitor of the Hyper log of a connection
271 1 : export enum HyperLogStatus {
272 : /// The log of the connection is not monitored, e.g. before the setup of the connection
273 1 : Closed = 'closed',
274 : /// The monitor opens the extract
275 1 : Opening = 'opening',
276 : /// The extract is open and polled
277 1 : Open = 'open',
278 : /// The log cannot be read, or a poll failed, see HyperLogState.error
279 1 : Unavailable = 'unavailable',
280 1 : }
281 :
282 : /// Why the Hyper log of a connection cannot be read
283 : export interface HyperLogError {
284 : /// Does the Hyper refuse to read its log, e.g. without experimental_hyper_introspection_functions?
285 : /// Otherwise reading or extracting the log failed, e.g. after the Hyper was terminated.
286 : unreadable: boolean;
287 : /// The message of the error
288 : message: string;
289 : }
290 :
291 : /// The state of the extract of the Hyper log of a connection
292 : export interface HyperLogState {
293 : /// The status of the monitor
294 : status: HyperLogStatus;
295 : /// Why the log cannot be read, set when the monitor is unavailable
296 : error: HyperLogError | null;
297 : /// The qualified table of the finished queries on the reader, see queryEndsQuery(), null until the extract is open
298 : queryEnds: string | null;
299 : /// The qualified table of the resource samples on the reader, see resourceSamplesQuery()
300 : resourceMetrics: string | null;
301 : /// The offset of the wall clock of the log from UTC in milliseconds, e.g. 7,200,000 for a Hyper in UTC+2.
302 : /// The log writes the wall-clock time of the Hyper host, which is not the time zone of the app for a remote Hyper.
303 : clockOffsetMs: number | null;
304 : /// The transfers that appended rows to the extract, it changes with every new row
305 : transfers: number;
306 : /// Does the Hyper log its resource usage? Null until a poll after the open tells, false while no transfer found a
307 : /// resource sample, e.g. without log_resource_usage_interval.
308 : resourceUsageLogged: boolean | null;
309 : }
310 :
311 : /// The state of a connection without an open extract
312 1 : export const CLOSED_LOG_STATE: HyperLogState = {
313 1 : status: HyperLogStatus.Closed,
314 1 : error: null,
315 1 : queryEnds: null,
316 1 : resourceMetrics: null,
317 1 : clockOffsetMs: null,
318 1 : transfers: 0,
319 1 : resourceUsageLogged: null,
320 1 : };
321 :
322 : /// The open extract of a connection and its reader
323 : interface OpenHyperLog {
324 : /// The extract
325 : extract: HyperDBIncrementalExtract;
326 : /// The connection that reads the extract for the overview, with the monitoring database attached
327 : reader: EmbeddedConnection;
328 : }
329 :
330 : /// The Hyper whose log is monitored
331 : export interface HyperLogOrigin {
332 : /// The Hyper, e.g. the embedded Hyper or a channel to a remote Hyper
333 : database: HyperDBSourceDatabase;
334 : /// Do the connections to the Hyper keep their sessions? A poll then reads the log scope of its session, otherwise
335 : /// the whole retained log, see HyperDBLogSource.
336 : sessions: boolean;
337 : }
338 :
339 : /// The monitor of the log of a connection
340 : interface HyperLogMonitor {
341 : /// The Hyper whose log is read
342 : origin: HyperLogOrigin;
343 : /// Opens the extract and its reader, resolves to null when the log cannot be read
344 : log: Promise<OpenHyperLog | null>;
345 : /// The last read of the reader, the reads run one after another
346 : reading: Promise<unknown>;
347 : }
348 :
349 : /// Options of the Hyper log metrics.
350 : export interface HyperLogMetricsOptions {
351 : /// The logger
352 : logger: Logger;
353 : /// The polling interval in milliseconds
354 : pollIntervalMs?: number;
355 : /// The pause between the attempts to open an extract while reading the log times out, in milliseconds
356 : openRetryMs?: number;
357 : /// Prepare the monitoring database of a connection, e.g. create the folder of the connection, and return its path
358 : prepareDatabase: (connectionId: string) => Promise<string>;
359 : /// The embedded Hyper that holds the monitoring databases, its compute database stages the transfers
360 : sink: () => Promise<EmbeddedComputeDatabase>;
361 : }
362 :
363 : /// The finished queries of the connections to a Hyper and the resource samples of the Hyper, extracted from its log
364 : /// into the monitoring database of every connection, every poll interval.
365 : ///
366 : /// The extract of a connection holds the queries whose client context names the connection, see DashQLClientContext:
367 : /// the queries of the user and the app, and the reads and writes of its own log monitor. Its reader reads the
368 : /// extract with the client context of the overview. The resource samples describe the whole Hyper process, which the
369 : /// connections to the same Hyper share.
370 : ///
371 : /// The monitoring databases live in the embedded Hyper, also for a remote Hyper.
372 : ///
373 : /// Limits:
374 : /// - The connections to the same Hyper monitor its log one by one, each reads the resource samples again.
375 : /// - A Hyper without sessions, e.g. over gRPC or HTTP, is read from the watermark with every poll, which reads its whole
376 : /// retained log.
377 : /// - The log files of earlier Hyper processes cannot be backfilled, the extract misses the queries while no app
378 : /// window monitored the connection.
379 : /// - Resource samples are only logged with log_resource_usage_interval set.
380 : /// - A Hyper that does not let the app read its log, e.g. without experimental_hyper_introspection_functions, leaves the
381 : /// monitor unavailable. The connection itself works as before, the monitor does not retry.
382 : /// - A read of the log that times out on a Hyper under load is retried, see isHyperLogFlushTimeout(). The extract then
383 : /// falls behind until the reads succeed again, also without a notice in the state.
384 1 : export class HyperLogMetrics {
385 : /// The states of the extracts by connection
386 1 : readonly state = new Store<ReadonlyMap<string, HyperLogState>>(new Map());
387 : /// The monitors by connection
388 1 : private readonly monitors = new Map<string, HyperLogMonitor>();
389 :
390 1 : constructor(private readonly options: HyperLogMetricsOptions) {}
391 :
392 : /// Extract the log of the Hyper of a connection, e.g. of the embedded Hyper. Another Hyper replaces the watched
393 : /// one, e.g. after the hand-over of the app to another tab. The backfill of the extract reads the retained log.
394 1 : watch(origin: HyperLogOrigin, connectionId: string): void {
395 1 : if (this.monitors.get(connectionId)?.origin.database === origin.database) return;
396 1 : void this.forget(connectionId);
397 1 : const monitor: HyperLogMonitor = { origin, log: Promise.resolve(null), reading: Promise.resolve() };
398 1 : this.monitors.set(connectionId, monitor);
399 1 : this.updateState(connectionId, { ...CLOSED_LOG_STATE, status: HyperLogStatus.Opening });
400 1 : monitor.log = this.open(connectionId, monitor);
401 1 : }
402 :
403 : /// Wait until the extract of a connection is open, false when it cannot be read
404 1 : async opened(connectionId: string): Promise<boolean> {
405 1 : const monitor = this.monitors.get(connectionId);
406 1 : return monitor != null && (await monitor.log) != null;
407 1 : }
408 :
409 : /// Read the extract of a connection on its reader, e.g. with queryEndsQuery(). The reads run one after another.
410 1 : async query(connectionId: string, sql: string): Promise<arrow.Table> {
411 1 : const monitor = this.monitors.get(connectionId);
412 1 : if (monitor == null) throw new Error(`the Hyper log of the connection ${connectionId} is not monitored`);
413 1 : const read = monitor.reading.catch(() => {}).then(async () => {
414 1 : const log = await monitor.log;
415 1 : if (log == null) throw new Error(`the Hyper log extract of the connection ${connectionId} cannot be read`);
416 1 : return await log.reader.query(sql);
417 1 : });
418 1 : monitor.reading = read;
419 1 : return await read;
420 1 : }
421 :
422 : /// Transfer the events that were logged since the last poll, e.g. in tests that do not wait for the interval
423 1 : async poll(connectionId: string): Promise<void> {
424 1 : const log = await this.monitors.get(connectionId)?.log;
425 1 : if (log == null) return;
426 1 : try {
427 1 : this.transferred(connectionId, await log.extract.poll());
428 1 : } catch (error) {
429 : // The next poll reads the same events
430 1 : if (!isHyperLogFlushTimeout(error)) throw error;
431 1 : this.options.logger.warn('Polling the Hyper log timed out', { error: stringifyError(error) }, LOG_CTX);
432 1 : }
433 1 : }
434 :
435 : /// Stop the monitors of all connections except the given ones, e.g. of deleted connections.
436 : /// A stopped monitor closes the monitoring database of its connection.
437 1 : retain(connectionIds: ReadonlySet<string>): void {
438 1 : for (const connectionId of [...this.monitors.keys()]) {
439 1 : if (!connectionIds.has(connectionId)) void this.forget(connectionId);
440 : }
441 : // The unavailable monitors were stopped already, but kept their state
442 1 : for (const connectionId of [...this.state.current.keys()]) {
443 1 : if (!connectionIds.has(connectionId)) this.updateState(connectionId, null);
444 : }
445 1 : }
446 :
447 : /// Stop the monitor of a connection and close its monitoring database
448 1 : async forget(connectionId: string): Promise<void> {
449 1 : const monitor = this.monitors.get(connectionId);
450 1 : if (monitor == null) return;
451 1 : this.monitors.delete(connectionId);
452 1 : this.updateState(connectionId, null);
453 1 : await monitor.reading.catch(() => {});
454 1 : const log = await monitor.log;
455 1 : await log?.reader.close().catch(() => {});
456 1 : await log?.extract.stop().catch(() => {});
457 1 : }
458 :
459 : /// Stop the monitors of all connections
460 1 : async stop(): Promise<void> {
461 1 : await Promise.all([...this.monitors.keys()].map(connectionId => this.forget(connectionId)));
462 1 : }
463 :
464 : /// Open the extract of a connection and its reader, and start polling.
465 : /// A failure leaves the connection unavailable with the error, e.g. for a Hyper that does not let the app read its log.
466 : /// A read of the log that times out is tried again first.
467 1 : private async open(connectionId: string, monitor: HyperLogMonitor): Promise<OpenHyperLog | null> {
468 1 : for (let attempt = 1; ; ++attempt) {
469 1 : try {
470 1 : return await this.openOnce(connectionId, monitor);
471 1 : } catch (error) {
472 1 : if (attempt < OPEN_ATTEMPTS && isHyperLogFlushTimeout(error)) {
473 1 : this.options.logger.info('Opening the Hyper log extract timed out', { connectionId, attempt: String(attempt) }, LOG_CTX);
474 1 : await new Promise(resolve => setTimeout(resolve, this.options.openRetryMs ?? OPEN_RETRY_MS));
475 1 : if (this.monitors.get(connectionId) !== monitor) return null;
476 1 : continue;
477 1 : }
478 1 : if (error instanceof HyperLogUnreadableError) {
479 1 : this.options.logger.info('The Hyper does not let the app read its log', { error: stringifyError(error) }, LOG_CTX);
480 : } else {
481 1 : this.options.logger.warn('Opening the Hyper log extract failed', { error: stringifyError(error) }, LOG_CTX);
482 : }
483 1 : if (this.monitors.get(connectionId) === monitor) this.fail(connectionId, error);
484 1 : return null;
485 1 : }
486 : }
487 1 : }
488 :
489 : /// Open the extract of a connection and its reader, and start polling. A failure closes them again and throws.
490 1 : private async openOnce(connectionId: string, monitor: HyperLogMonitor): Promise<OpenHyperLog | null> {
491 1 : const { origin } = monitor;
492 1 : const source = new HyperDBLogSource(metricsSourceQuery(connectionId), origin.sessions);
493 1 : let extract: HyperDBIncrementalExtract | null = null;
494 1 : let reader: EmbeddedConnection | null = null;
495 1 : let clockOffsetMs: number;
496 1 : try {
497 1 : const path = await this.options.prepareDatabase(connectionId);
498 1 : const sink = await this.options.sink();
499 1 : extract = await HyperDBIncrementalExtract.open({
500 1 : sourceDatabase: origin.database,
501 1 : source,
502 1 : sink,
503 1 : databasePath: path,
504 1 : databaseAlias: MONITORING_ALIAS,
505 1 : tables: [QUERY_ENDS_TABLE, RESOURCE_METRICS_TABLE],
506 1 : clientContext: monitorClientContext(connectionId),
507 1 : intervalMs: this.options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS,
508 1 : onTransfer: transfer => this.transferred(connectionId, transfer),
509 1 : onRebuild: reason => this.options.logger.info('Rebuilding the extract of the Hyper log', { connectionId, reason }, LOG_CTX),
510 1 : onError: error => {
511 : // A Hyper that was terminated, e.g. by a hand-over to another tab, fails every poll.
512 : // Setting up the connection again starts a new monitor.
513 1 : this.options.logger.warn('Polling the Hyper log failed', { error: stringifyError(error) }, LOG_CTX);
514 : // The next poll reads the events of a read that timed out
515 1 : if (isHyperLogFlushTimeout(error)) return;
516 0 : if (this.monitors.get(connectionId) === monitor) this.fail(connectionId, error);
517 1 : },
518 1 : });
519 1 : const clock = await origin.database.connect({ clientContext: monitorClientContext(connectionId) });
520 1 : try {
521 1 : clockOffsetMs = await source.readClockOffset(clock);
522 1 : } finally {
523 1 : await clock.close().catch(() => {});
524 1 : }
525 1 : reader = await sink.connect({ clientContext: overviewClientContext(connectionId) });
526 1 : await reader.execute(`ATTACH DATABASE ${quoteIdentifier(path)} AS ${quoteIdentifier(MONITORING_ALIAS)}`);
527 1 : } catch (error) {
528 1 : await reader?.close().catch(() => {});
529 1 : await extract?.stop().catch(() => {});
530 1 : throw error;
531 1 : }
532 1 : if (this.monitors.get(connectionId) !== monitor) {
533 1 : await reader.close().catch(() => {});
534 1 : await extract.stop().catch(() => {});
535 1 : return null;
536 1 : }
537 1 : this.updateState(connectionId, {
538 1 : status: HyperLogStatus.Open,
539 1 : error: null,
540 1 : queryEnds: extract.tableName(QUERY_ENDS_TABLE.name),
541 1 : resourceMetrics: extract.tableName(RESOURCE_METRICS_TABLE.name),
542 1 : clockOffsetMs,
543 1 : transfers: 1,
544 : // The backfill read the retained log, a Hyper that just started may not have logged a sample yet
545 1 : resourceUsageLogged: (extract.backfill.tables[RESOURCE_METRICS_TABLE.name] ?? 0) > 0 ? true : null,
546 1 : });
547 1 : extract.start();
548 1 : return { extract, reader };
549 1 : }
550 :
551 : /// Stop the monitor of a connection after an error, its state keeps the error until the connection is watched again
552 1 : private fail(connectionId: string, error: unknown): void {
553 1 : void this.forget(connectionId);
554 1 : this.updateState(connectionId, {
555 1 : ...CLOSED_LOG_STATE,
556 1 : status: HyperLogStatus.Unavailable,
557 1 : error: {
558 1 : unreadable: error instanceof HyperLogUnreadableError,
559 1 : message: error instanceof Error ? error.message : String(error),
560 1 : },
561 1 : });
562 1 : }
563 :
564 1 : private transferred(connectionId: string, transfer: HyperDBExtractTransfer): void {
565 1 : const state = this.state.current.get(connectionId);
566 1 : if (state == null || state.status !== HyperLogStatus.Open) return;
567 : // A Hyper that logs its resource usage logs a sample every interval, a poll without one marks it as not logged
568 1 : const resourceUsageLogged = state.resourceUsageLogged === true || (transfer.tables[RESOURCE_METRICS_TABLE.name] ?? 0) > 0;
569 1 : if (transfer.appended === 0 && resourceUsageLogged === state.resourceUsageLogged) return;
570 1 : this.updateState(connectionId, {
571 1 : ...state,
572 1 : transfers: state.transfers + (transfer.appended === 0 ? 0 : 1),
573 1 : resourceUsageLogged,
574 1 : });
575 1 : }
576 :
577 : /// Replace the state of a connection, null removes it
578 1 : private updateState(connectionId: string, state: HyperLogState | null): void {
579 1 : this.state.update(states => {
580 1 : const next = new Map(states);
581 1 : if (state == null) next.delete(connectionId);
582 1 : else next.set(connectionId, state);
583 1 : return next;
584 1 : });
585 1 : }
586 1 : }
|
