Line data Source code
1 : import * as React from 'react';
2 :
3 : import { useEmbeddedDatabaseSetup } from '../../platform/database/embedded_database_provider.js';
4 : import { useStorageReader } from '../../storage/storage_provider.js';
5 : import { connectionMonitoringDatabasePath, isSessionRegistryBackend } from '../../storage/storage_backend.js';
6 : import { stringifyError } from '../../platform/logger/logger.js';
7 : import { useLogger } from '../../platform/logger/logger_provider.js';
8 : import { useConnectionRegistryStore } from '../connection_registry.js';
9 : import {
10 : CLOSED_LOG_STATE,
11 : HyperLogMetrics,
12 : lastExtractRow,
13 : queryEndsQuery,
14 : readQueryEnds,
15 : readResourceSamples,
16 : resourceSamplesQuery,
17 : type HyperLogState,
18 : type HyperQueryEnd,
19 : type HyperResourceSample,
20 : } from './hyper_log_metrics.js';
21 :
22 1 : const LOG_CTX = 'hyper_log_metrics';
23 :
24 1 : const LOG_METRICS_CTX = React.createContext<HyperLogMetrics | null>(null);
25 :
26 : /// The time that the rows of a connection reach back, the largest window of the connection metrics
27 1 : const RETAINED_MS = 60 * 60_000;
28 :
29 : interface Props {
30 : /// The children
31 : children: React.ReactElement;
32 : }
33 :
34 : /// Provides the extracts of the Hyper logs of the connections, in the monitoring databases in the folders of the
35 : /// connections. The Hyper connections start their monitors, the monitors of deleted connections are stopped.
36 : /// Must be nested in the connection registry and the embedded database, which holds the monitoring databases.
37 1 : export const HyperLogMetricsProvider: React.FC<Props> = (props: Props) => {
38 1 : const logger = useLogger();
39 1 : const storage = useStorageReader();
40 1 : const registryStore = useConnectionRegistryStore();
41 1 : const setupEmbeddedDatabase = useEmbeddedDatabaseSetup();
42 1 : const [logMetrics] = React.useState(() => new HyperLogMetrics({
43 1 : logger,
44 1 : sink: () => setupEmbeddedDatabase('hyper_log_metrics'),
45 1 : prepareDatabase: async connectionId => {
46 1 : const backend = storage.backend;
47 1 : if (!isSessionRegistryBackend(backend)) throw new Error('the storage keeps no connection folders');
48 1 : await backend.createConnectionFolder(connectionId);
49 1 : return connectionMonitoringDatabasePath(connectionId);
50 1 : },
51 1 : }));
52 1 : React.useEffect(() => {
53 1 : const retain = () => logMetrics.retain(new Set(registryStore.current.connections.keys()));
54 1 : retain();
55 1 : return registryStore.subscribe(retain);
56 1 : }, [logMetrics, registryStore]);
57 1 : React.useEffect(() => () => { void logMetrics.stop(); }, [logMetrics]);
58 1 : return <LOG_METRICS_CTX.Provider value={logMetrics}>{props.children}</LOG_METRICS_CTX.Provider>;
59 1 : };
60 :
61 : /// The Hyper log metrics, null outside of a HyperLogMetricsProvider
62 1 : export function useHyperLogMetrics(): HyperLogMetrics | null {
63 1 : return React.useContext(LOG_METRICS_CTX);
64 1 : }
65 :
66 1 : const NO_STATES: ReadonlyMap<string, HyperLogState> = new Map();
67 1 : const NO_SUBSCRIPTION = () => () => {};
68 :
69 : /// The state of the extract of the Hyper log of a connection, it changes with every transfer of new rows
70 0 : export function useHyperLogState(connectionId: string): HyperLogState {
71 0 : const logMetrics = useHyperLogMetrics();
72 0 : const states = React.useSyncExternalStore(logMetrics?.state.subscribe ?? NO_SUBSCRIPTION, () => logMetrics?.state.current ?? NO_STATES);
73 0 : return states.get(connectionId) ?? CLOSED_LOG_STATE;
74 0 : }
75 :
76 : /// The rows of a connection in the extract of the Hyper log
77 : export interface HyperLogRows {
78 : /// The finished queries of the connection, ordered by their end, see queryEndsQuery()
79 : queries: HyperQueryEnd[];
80 : /// The resource samples of the Hyper process, ordered by time
81 : samples: HyperResourceSample[];
82 : }
83 :
84 1 : const NO_ROWS: HyperLogRows = { queries: [], samples: [] };
85 :
86 : /// Append rows that are mostly newer than the rows, and drop the rows before a time
87 0 : function appendRows<T>(rows: T[], added: T[], time: (row: T) => number, since: number): T[] {
88 0 : if (added.length === 0 && (rows.length === 0 || time(rows[0]) >= since)) return rows;
89 0 : const all = rows.concat(added);
90 : // The events of a transfer can end before the events of the last one, e.g. the queries that ran during its read
91 0 : if (rows.length > 0 && added.length > 0 && time(added[0]) < time(rows[rows.length - 1])) {
92 0 : all.sort((l, r) => time(l) - time(r));
93 : }
94 0 : const first = all.findIndex(row => time(row) >= since);
95 0 : return first < 0 ? [] : all.slice(first);
96 0 : }
97 :
98 : /// The rows of a connection of the last hour, read from its extract with the reader of its monitor.
99 : /// Every transfer of the extract reads the rows after the last read row and appends them.
100 0 : export function useHyperLogRows(connectionId: string): HyperLogRows {
101 0 : const logger = useLogger();
102 0 : const logMetrics = useHyperLogMetrics();
103 0 : const state = useHyperLogState(connectionId);
104 0 : const [rows, setRows] = React.useState<HyperLogRows>(NO_ROWS);
105 0 : const reader = React.useRef({
106 : /// The connection of the read rows
107 0 : connectionId: null as string | null,
108 : /// Changes with the connection, the reads of an earlier generation are dropped
109 0 : generation: 0,
110 : /// The last read row of the query ends and the resource metrics, the row numbers of an extract never repeat
111 0 : queryEndsRow: 0,
112 0 : resourceMetricsRow: 0,
113 : /// The running reads, one after another
114 0 : reading: Promise.resolve(),
115 : /// Is a read queued that did not start yet?
116 0 : queued: false,
117 0 : });
118 :
119 0 : React.useEffect(() => {
120 0 : const { queryEnds, resourceMetrics, clockOffsetMs } = state;
121 0 : if (logMetrics == null || queryEnds == null || resourceMetrics == null || clockOffsetMs == null) return;
122 0 : const current = reader.current;
123 : // Start over for another connection
124 0 : if (current.connectionId !== connectionId) {
125 0 : current.connectionId = connectionId;
126 0 : current.generation += 1;
127 0 : current.queryEndsRow = 0;
128 0 : current.resourceMetricsRow = 0;
129 0 : current.queued = false;
130 0 : setRows(NO_ROWS);
131 0 : }
132 : // A queued read reads the rows of this transfer too
133 0 : if (current.queued) return;
134 0 : current.queued = true;
135 0 : const generation = current.generation;
136 0 : current.reading = current.reading.then(async () => {
137 0 : if (current.generation !== generation) return;
138 0 : current.queued = false;
139 0 : const since = Date.now() - RETAINED_MS;
140 0 : const queryTable = await logMetrics.query(connectionId, queryEndsQuery(queryEnds, clockOffsetMs, { afterRow: current.queryEndsRow, since }));
141 0 : const sampleTable = await logMetrics.query(connectionId, resourceSamplesQuery(resourceMetrics, clockOffsetMs, { afterRow: current.resourceMetricsRow, since }));
142 0 : if (current.generation !== generation) return;
143 0 : current.queryEndsRow = Math.max(current.queryEndsRow, lastExtractRow(queryTable));
144 0 : current.resourceMetricsRow = Math.max(current.resourceMetricsRow, lastExtractRow(sampleTable));
145 0 : const queries = readQueryEnds(queryTable);
146 0 : const samples = readResourceSamples(sampleTable);
147 0 : setRows(previous => ({
148 0 : queries: appendRows(previous.queries, queries, query => query.endedAt, since),
149 0 : samples: appendRows(previous.samples, samples, sample => sample.sampledAt, since),
150 0 : }));
151 0 : }).catch(error => {
152 0 : logger.warn('Reading the Hyper log extract failed', { error: stringifyError(error) }, LOG_CTX);
153 0 : });
154 0 : }, [state, connectionId, logMetrics, logger]);
155 0 : return rows;
156 0 : }
|
