Line data Source code
1 : import {
2 : quoteLiteral,
3 : type HyperDBExtractColumn,
4 : type HyperDBExtractTable,
5 : type HyperDBIncrementalExtractSource,
6 : type HyperDBSourceConnection,
7 : } from './hyperdb_incremental_extract.js';
8 : import { quoteIdentifier } from './hyperdb_wasm.js';
9 :
10 : /// The columns of hyper_log(), with ctx and v as JSON text.
11 1 : export const HYPER_LOG_COLUMNS: readonly HyperDBExtractColumn[] = [
12 1 : { name: 'ts', type: 'TIMESTAMP' },
13 1 : { name: 'pid', type: 'BIGINT' },
14 1 : { name: 'tid', type: 'TEXT' },
15 1 : { name: 'sev', type: 'TEXT' },
16 1 : { name: 'req', type: 'TEXT' },
17 1 : { name: 'sess', type: 'TEXT' },
18 1 : { name: 'coordinating-req', type: 'TEXT' },
19 1 : { name: 'coordinating-sess', type: 'TEXT' },
20 1 : { name: 'ctx', type: 'TEXT' },
21 1 : { name: 'k', type: 'TEXT' },
22 1 : { name: 'v', type: 'TEXT' },
23 1 : ];
24 :
25 : /// An extract table with the unchanged log events.
26 1 : export const HYPER_LOG_EVENTS_TABLE: HyperDBExtractTable = {
27 1 : name: 'events',
28 1 : query: 'SELECT * FROM log',
29 1 : };
30 :
31 : /// A value at a JSON path of a JSON column of log events, v by default, cast to a type.
32 : /// It is NULL when the path is missing, and for the gauges of resource-metrics events that the platform does not
33 : /// record and logs as `{}`.
34 1 : export function hyperLogValue(path: readonly string[], type: string, column = 'v'): string {
35 1 : const parents = path.slice(0, -1).map(key => `->${quoteLiteral(key)}`).join('');
36 1 : return `nullif(${column}::json${parents}->>${quoteLiteral(path[path.length - 1])}, '{}')::${type}`;
37 1 : }
38 :
39 : /// The sum of a number of the compilation and the execution of a query-end event, 0 when missing
40 1 : function phaseSum(path: readonly string[]): string {
41 1 : const phase = (name: string) => `coalesce(${hyperLogValue([name, ...path], 'double precision')}, 0)`;
42 1 : return `${phase('pre-execution')} + ${phase('execution')}`;
43 1 : }
44 :
45 : /// A field of the client context of the queries, which Hyper logs as ctx.client, as a text column of the query ends
46 : export interface HyperLogClientColumn {
47 : /// The name of the column
48 : name: string;
49 : /// The path of the field in the client context
50 : path: readonly string[];
51 : }
52 :
53 : /// An extract table with one row per finished query, with the metrics of its query-end event.
54 : /// The outcome is the event key, e.g. query-end or query-end-user-error. The times are in seconds.
55 : ///
56 : /// A query-hash like `bfe4.387e195b3a02.24573b` has three parts: the second hashes the text with normalized literals,
57 : /// the third hashes the literal values, and the first varies with the literals too. Queries that only differ in their
58 : /// literals or comments share the second part, the query_shape. Queries without a hash are shaped by their text.
59 1 : export function hyperLogQueryEndsTable(clientColumns: readonly HyperLogClientColumn[] = []): HyperDBExtractTable {
60 1 : const hash = `v::json->>'query-hash'`;
61 1 : return {
62 1 : name: 'query_ends',
63 1 : query: `
64 : SELECT
65 : ts,
66 : sess,
67 : req,
68 : k AS outcome,
69 : k = 'query-end' AS succeeded,
70 : v::json->>'error-code' AS error_code,
71 : v::json->>'error-message' AS error_message,
72 : v::json->>'statement' AS statement,
73 : v::json->>'query-trunc' AS query,
74 1 : ${hash} AS query_hash,
75 1 : CASE WHEN ${hash} LIKE '%.%.%' THEN split_part(${hash}, '.', 2)
76 1 : ELSE coalesce(${hash}, 'text:' || coalesce(v::json->>'query-trunc', '')) END AS query_shape,
77 1 : ${hyperLogValue(['elapsed'], 'double precision')} AS elapsed,
78 1 : ${phaseSum(['threads', 'thread-time'])} AS thread_time,
79 1 : ${phaseSum(['threads', 'cpu-time'])} AS cpu_time,
80 1 : ${phaseSum(['threads', 'wait-time'])} AS wait_time,
81 1 : (${phaseSum(['processed-rows', 'total'])})::bigint AS processed_rows,
82 1 : ${hyperLogValue(['time-to-schedule'], 'double precision')} AS time_to_schedule,
83 : greatest(
84 1 : ${hyperLogValue(['pre-execution', 'peak-transaction-memory-mb'], 'double precision')},
85 1 : ${hyperLogValue(['execution', 'peak-transaction-memory-mb'], 'double precision')}
86 : ) AS peak_memory_mb,
87 1 : ${hyperLogValue(['rows'], 'bigint')} AS rows${clientColumns.map(column => `,
88 1 : ${hyperLogValue(['client', ...column.path], 'text', 'ctx')} AS ${quoteIdentifier(column.name)}`).join('')}
89 : FROM log
90 : WHERE k LIKE 'query-end%'
91 : `,
92 1 : };
93 1 : }
94 :
95 : /// The query ends without client columns
96 1 : export const HYPER_LOG_QUERY_ENDS_TABLE: HyperDBExtractTable = hyperLogQueryEndsTable();
97 :
98 : /// An extract table with one row per resource-metrics event, which Hyper writes every log_resource_usage_interval.
99 : /// The process memory is the size of the Wasm memory, which never shrinks. The transaction memory is the memory that
100 : /// the running queries hold. The tracked memory is the sum of all memory trackers, e.g. the table data of the attached
101 : /// databases, temporary tables, the running queries and the plan cache, and shrinks again when they are freed. The
102 : /// loads are between 0 and 1.
103 1 : export const HYPER_LOG_RESOURCE_METRICS_TABLE: HyperDBExtractTable = {
104 1 : name: 'resource_metrics',
105 1 : query: `
106 : SELECT
107 : ts,
108 1 : ${hyperLogValue(['memory', 'process_physical_memory_mb'], 'double precision')} AS process_memory_mb,
109 1 : ${hyperLogValue(['memory', 'total_physical_memory_mb'], 'double precision')} AS total_memory_mb,
110 1 : ${hyperLogValue(['mem-trackers', 'memory_tracked_current_usage_mb', 'global_transactions'], 'double precision')} AS transaction_memory_mb,
111 1 : ${hyperLogValue(['mem-trackers', 'memory_tracked_current_usage_mb', 'global'], 'double precision')} AS tracked_memory_mb,
112 1 : ${hyperLogValue(['load', 'overall_load'], 'double precision')} AS overall_load,
113 1 : ${hyperLogValue(['load', 'scheduler_load'], 'double precision')} AS scheduler_load,
114 1 : ${hyperLogValue(['load', 'memory_load'], 'double precision')} AS memory_load,
115 1 : ${hyperLogValue(['scheduler-thread-count', 'scheduler_thread_count', 'active'], 'integer')} AS active_threads,
116 1 : ${hyperLogValue(['scheduler-thread-count', 'scheduler_waiting_jobs_count'], 'integer')} AS waiting_jobs
117 : FROM log
118 : WHERE k = 'resource-metrics'
119 : `,
120 1 : };
121 :
122 : /// The error of a Hyper that does not let the app read its log, e.g. without experimental_hyper_introspection_functions.
123 1 : export class HyperLogUnreadableError extends Error {
124 1 : constructor(message: string) {
125 1 : super(`the Hyper log cannot be read: ${message}`);
126 1 : this.name = 'HyperLogUnreadableError';
127 1 : }
128 1 : }
129 :
130 : /// Did a read of the log time out? hyper_log() flushes the structured log when a query is planned, which times out on a
131 : /// Hyper under load. The log can be read, a later read succeeds.
132 1 : export function isHyperLogFlushTimeout(error: unknown): boolean {
133 1 : return (error instanceof Error ? error.message : String(error)).includes('Timed out while flushing the structured log');
134 1 : }
135 :
136 : /// Did a read of the log race with the rotation of the log files? These errors come with rotating log files, probably
137 : /// because a read plans its files before it scans them: a file that was deleted meanwhile cannot be read, and the invalid
138 : /// JSON may be a partial event. A read right after it succeeds.
139 1 : export function isHyperLogRotationRace(error: unknown): boolean {
140 1 : const message = error instanceof Error ? error.message : String(error);
141 1 : return message.includes('could not read from external source') || message.includes('invalid JSON document');
142 1 : }
143 :
144 : /// The time zones are offset from UTC by multiples of 15 minutes
145 1 : const CLOCK_OFFSET_STEP_MS = 15 * 60_000;
146 : /// How long readClockOffset() waits for the end of its marker query in the log, e.g. on a Hyper under load
147 1 : const CLOCK_QUERY_LOG_TIMEOUT_MS = 10_000;
148 : /// The pauses between the reads of readClockOffset(), they double up to the maximum
149 1 : const CLOCK_QUERY_LOG_FIRST_RETRY_MS = 50;
150 1 : const CLOCK_QUERY_LOG_MAX_RETRY_MS = 1_000;
151 :
152 : /// The Hyper log as the source of an incremental extract, the relation `log` of its queries.
153 : ///
154 : /// A scoped source reads the log scope of the session of its reader with every poll, the events since the session
155 : /// started. The extract connects the next reader before every read, see HyperDBIncrementalExtract. Its backfills read
156 : /// the retained log files from the watermark, e.g. after the cursor died with the connection.
157 : ///
158 : /// Without sessions, e.g. over gRPC or HTTP where every query runs in a session of its own, the scope of a query only
159 : /// holds its own events. A source that is not scoped backfills every poll from the watermark, reading the whole
160 : /// retained log.
161 : ///
162 : /// Limits:
163 : /// - Log files that were rotated out, or written by an earlier Hyper process, cannot be backfilled.
164 : /// - The source query only sees the events of one update or backfill, e.g. not the earlier events of a session.
165 1 : export class HyperDBLogSource implements HyperDBIncrementalExtractSource {
166 1 : readonly relation = 'log';
167 1 : readonly columns = HYPER_LOG_COLUMNS;
168 1 : readonly watermarkColumn = 'ts';
169 1 : readonly isTransientReadError = isHyperLogRotationRace;
170 : /// The query of the events in the scope of the reader, only for a scoped source
171 1 : readonly updateQuery?: () => string;
172 :
173 1 : constructor(
174 : /// A query over the relation `log` with the source events that selects the events to extract,
175 : /// e.g. `SELECT * FROM log WHERE sess <> '-' AND k LIKE 'query-%'`. It returns the columns of hyper_log().
176 1 : private readonly sourceQuery = 'SELECT * FROM log',
177 : /// Do the connections to the Hyper keep their sessions, so that a poll reads the scope of its reader?
178 1 : private readonly scoped = true,
179 1 : ) {
180 1 : if (scoped) this.updateQuery = () => this.query(true, null);
181 1 : }
182 :
183 : /// Check that the log can be read
184 1 : async open(connection: HyperDBSourceConnection): Promise<void> {
185 1 : try {
186 1 : await connection.query(`EXPLAIN (FORMAT SCHEMA) SELECT * FROM hyper_log(current_session := true, last_log_scope := ${this.scoped})`);
187 1 : } catch (error) {
188 1 : if (isHyperLogFlushTimeout(error)) throw error;
189 1 : throw new HyperLogUnreadableError(error instanceof Error ? error.message : String(error));
190 1 : }
191 1 : }
192 :
193 1 : backfillQuery(since: string | null): string {
194 1 : return this.query(false, since);
195 1 : }
196 :
197 : /// Read the offset of the wall clock of the log from UTC in milliseconds, e.g. 7,200,000 for a Hyper in UTC+2.
198 : /// The log writes the wall-clock time of the Hyper host as timestamp without time zone.
199 : ///
200 : /// The offset is the logged end of a marker query read as UTC, minus its end on the clock of the app, rounded to
201 : /// 15 minutes. It tolerates a clock skew and a latency of a few minutes. A scoped source reads the marker from the
202 : /// scope of a new connection, otherwise it reads the whole retained log.
203 : ///
204 : /// Hyper may return the result of a query before it logs the end of the query, and a read may time out, so the read
205 : /// is repeated until the log holds the marker.
206 1 : async readClockOffset(connection: HyperDBSourceConnection, timeoutMs = CLOCK_QUERY_LOG_TIMEOUT_MS): Promise<number> {
207 1 : const marker = `dashql_clock_${crypto.randomUUID().replace(/-/g, '_')}`;
208 1 : await connection.query(`SELECT 1 AS ${marker}`);
209 1 : const endedAt = Date.now();
210 1 : const deadline = performance.now() + timeoutMs;
211 1 : for (let retryMs = CLOCK_QUERY_LOG_FIRST_RETRY_MS; ; retryMs = Math.min(retryMs * 2, CLOCK_QUERY_LOG_MAX_RETRY_MS)) {
212 1 : let loggedAt: unknown = null;
213 1 : try {
214 1 : const result = await connection.query(`
215 : SELECT (extract(epoch from max(ts)) * 1000)::double precision AS ended_at
216 1 : FROM hyper_log(current_session := ${this.scoped}, last_log_scope := ${this.scoped})
217 1 : WHERE k = 'query-end' AND v::json->>'query-trunc' LIKE ${quoteLiteral(`%${marker}%`)}
218 1 : `);
219 1 : loggedAt = result.getChild('ended_at')?.get(0);
220 1 : } catch (error) {
221 1 : if (!isHyperLogFlushTimeout(error) && !isHyperLogRotationRace(error)) throw error;
222 1 : }
223 1 : if (loggedAt != null) {
224 : // + 0 turns the -0 of a log just before the clock of the app into 0
225 1 : return Math.round((Number(loggedAt) - endedAt) / CLOCK_OFFSET_STEP_MS) * CLOCK_OFFSET_STEP_MS + 0;
226 : }
227 1 : if (performance.now() + retryMs > deadline) {
228 1 : throw new Error(`the Hyper log does not hold the end of the clock query after ${timeoutMs}ms`);
229 : }
230 1 : await new Promise(resolve => setTimeout(resolve, retryMs));
231 1 : }
232 1 : }
233 :
234 : /// The source query over the scoped or watermarked log
235 1 : private query(scoped: boolean, since: string | null): string {
236 1 : const columns = HYPER_LOG_COLUMNS.map(({ name, type }) => `${quoteIdentifier(name)}::${type} AS ${quoteIdentifier(name)}`).join(', ');
237 1 : return `
238 : WITH log AS (
239 1 : SELECT ${columns}
240 1 : FROM hyper_log(current_session := false, last_log_scope := ${scoped})
241 1 : ${since == null ? '' : `WHERE ${since}`}
242 : ),
243 1 : source_query AS (${this.sourceQuery})
244 : SELECT * FROM source_query
245 : `;
246 1 : }
247 1 : }
|
