admin/logs: up to 50x faster loading
Massimo Melina committed
Aug 9, 2022 at 17:15 UTC
ce7e7909b7119c1ec1072e071feac6c9083edf9b
7 files changed
+158
-131
admin/src/LogsPage.ts
+2
-2
@@ -17,11 +17,11 @@ export default function LogsPage() {
17
}
18
19
function LogFile({ file }: { file: string }) {
20
- const { list, error, initializing } = useApiList('get_log', { file }, { addId: true })
20
+ const { list, error, connecting } = useApiList('get_log', { file }, { addId: true })
21
if (error)
22
return error
23
return h(DataGrid, {
24
- loading: initializing,
24
+ loading: connecting,
25
rows: list as any,
26
componentsProps: {
27
pagination: {
admin/src/api.ts
+51
-46
@@ -147,6 +147,7 @@ export function useApiEvents(cmd: string, params: Dict={}) {
147
export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { addId=false, map=((x:any)=>x) }={}) {
148
const [list, setList] = useStateMounted<T[]>([])
149
const [error, setError] = useStateMounted<any>(undefined)
150
+ const [connecting, setConnecting] = useStateMounted(true)
151
const [loading, setLoading] = useStateMounted(false)
152
const [initializing, setInitializing] = useStateMounted(true)
153
const idRef = useRef(0)
@@ -160,11 +161,13 @@ export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { addId=fal
161
}, 1000, { maxWait: 1000 })
162
setError(undefined)
163
setLoading(true)
164
+ setConnecting(true)
165
setInitializing(true)
166
setList([])
167
const src = apiEvents(cmd, params, (type, data) => {
168
switch (type) {
169
case 'connected':
170
+ setConnecting(false)
171
return setTimeout(() => apply.flush()) // this trick we'll cause first entries to be rendered almost immediately, while the rest will be subject to normal debouncing
172
case 'error':
173
setError("Connection error")
@@ -172,54 +175,56 @@ export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { addId=fal
175
case 'closed':
176
return stop()
177
case 'msg':
175
- if (src?.readyState === src?.CLOSED)
176
- return stop()
177
- if (data === 'ready') {
178
- apply.flush()
179
- setInitializing(false)
180
- return
181
- }
182
- if (data.error)
183
- return setError(err2msg(data.error))
184
- if (data.add) {
185
- const rec = map(data.add)
186
- if (addId)
187
- rec.id = ++idRef.current
188
- buffer.push(rec)
189
- apply()
190
- return
191
- }
192
- if (data.remove) {
193
- const matchOnList: ReturnType<typeof _.matches>[] = []
194
- // first remove from the buffer
195
- for (const key of data.remove) {
196
- const match1 = _.matches(key)
197
- if (_.isEmpty(_.remove(buffer, match1)))
198
- matchOnList.push(match1)
178
+ wantArray(data).forEach(data => {
179
+ if (data === 'ready') {
180
+ apply.flush()
181
+ setInitializing(false)
182
+ return
183
}
200
- // then work the hooked state
201
- if (_.isEmpty(matchOnList))
184
+ if (data.error)
185
+ return setError(err2msg(data.error))
186
+ if (data.add) {
187
+ const rec = map(data.add)
188
+ if (addId)
189
+ rec.id = ++idRef.current
190
+ buffer.push(rec)
191
+ apply()
192
return
203
- setList(list => {
204
- const filtered = list.filter(rec => !matchOnList.some(match1 => match1(rec)))
205
- return filtered.length < list.length ? filtered : list // avoid unnecessary changes
206
- })
207
- return
208
- }
209
- if (data.update) {
210
- apply.flush() // avoid treating buffer
211
- setList(list => {
212
- const modified = [...list]
213
- for (const { search, change } of data.update) {
214
- const idx = modified.findIndex(_.matches(search))
215
- if (idx >= 0)
216
- modified[idx] = { ...modified[idx], ...change }
193
+ }
194
+ if (data.remove) {
195
+ const matchOnList: ReturnType<typeof _.matches>[] = []
196
+ // first remove from the buffer
197
+ for (const key of data.remove) {
198
+ const match1 = _.matches(key)
199
+ if (_.isEmpty(_.remove(buffer, match1)))
200
+ matchOnList.push(match1)
201
}
218
- return modified
219
- })
220
- return
221
- }
222
- console.debug('unknown api event', type, data)
202
+ // then work the hooked state
203
+ if (_.isEmpty(matchOnList))
204
+ return
205
+ setList(list => {
206
+ const filtered = list.filter(rec => !matchOnList.some(match1 => match1(rec)))
207
+ return filtered.length < list.length ? filtered : list // avoid unnecessary changes
208
+ })
209
+ return
210
+ }
211
+ if (data.update) {
212
+ apply.flush() // avoid treating buffer
213
+ setList(list => {
214
+ const modified = [...list]
215
+ for (const { search, change } of data.update) {
216
+ const idx = modified.findIndex(_.matches(search))
217
+ if (idx >= 0)
218
+ modified[idx] = { ...modified[idx], ...change }
219
+ }
220
+ return modified
221
+ })
222
+ return
223
+ }
224
+ console.debug('unknown api event', type, data)
225
+ })
226
+ if (src?.readyState === src?.CLOSED)
227
+ stop()
228
}
229
})
230
@@ -231,7 +236,7 @@ export function useApiList<T=any>(cmd:string|Falsy, params: Dict={}, { addId=fal
236
apply.flush()
237
}
238
}, [cmd, JSON.stringify(params)]) //eslint-disable-line
234
- return { list, loading, error, initializing, setList, updateList }
239
+ return { list, loading, error, initializing, connecting, setList, updateList }
240
241
function updateList(cb: (toModify: Draft<typeof list>) => void) {
242
setList(produce(list, x => {
admin/src/misc.ts
+5
@@ -134,3 +134,8 @@ export function err2msg(code: string) {
134
ENOTDIR: "Not a folder",
135
}[code] || code
136
}
137
+
138
+export function wantArray<T>(x?: void | T | T[]) {
139
+ return x == null ? [] : Array.isArray(x) ? x : [x]
140
+}
141
+
server/src/adminApis.ts
+27
-24
@@ -90,30 +90,33 @@ export const adminApis: ApiHandlers = {
90
},
91
92
async get_log({ file }, ctx) {
93
- return new SendListReadable(list => {
94
- const logger = loggers.find(l => l.name === file)
95
- if (!logger)
96
- return list.error(404)
97
- const input = createReadStream(logger.path)
98
- input.on('error', async (e: any) => {
99
- if (e.code !== 'ENOENT') // ignore ENOENT, consider it an empty log
100
- return list.error(e.code || e.message)
101
- })
102
- input.on('ready', () => {
103
- list.ready()
104
- readline.createInterface({ input }).on('line', line => {
105
- if (ctx.aborted)
106
- return input.close()
107
- const obj = parse(line)
108
- if (obj)
109
- list.add(obj)
110
- }).on('close', () => // file is automatically closed, so we continue by events
111
- ctx.res.once('close', onOff(events, { // unsubscribe when connection is interrupted
112
- [logger.name](entry) {
113
- list.add(entry)
114
- }
115
- })) )
116
- })
93
+ return new SendListReadable({
94
+ bufferTime: 10,
95
+ doAtStart(list) {
96
+ const logger = loggers.find(l => l.name === file)
97
+ if (!logger)
98
+ return list.error(404)
99
+ const input = createReadStream(logger.path)
100
+ input.on('error', async (e: any) => {
101
+ if (e.code !== 'ENOENT') // ignore ENOENT, consider it an empty log
102
+ return list.error(e.code || e.message)
103
+ })
104
+ input.on('ready', () => {
105
+ readline.createInterface({ input }).on('line', line => {
106
+ if (ctx.aborted)
107
+ return input.close()
108
+ const obj = parse(line)
109
+ if (obj)
110
+ list.add(obj)
111
+ }).on('close', () => { // file is automatically closed, so we continue by events
112
+ ctx.res.once('close', onOff(events, { // unsubscribe when connection is interrupted
113
+ [logger.name](entry) {
114
+ list.add(entry)
115
+ }
116
+ }))
117
+ })
118
+ })
119
+ }
120
})
121
122
function parse(line: string) {
server/src/api.monitor.ts
+1
-1
@@ -19,7 +19,7 @@ const apis: ApiHandlers = {
19
},
20
21
get_connections({}, ctx) {
22
- const list = new SendListReadable( getConnections().map(c => serializeConnection(c)) )
22
+ const list = new SendListReadable({ addAtStart: getConnections().map(c => serializeConnection(c)) })
23
type Change = Partial<Omit<Connection,'ip'>>
24
const throttledUpdate = _.throttle(update, 1000/20) // try to avoid clogging with updates
25
const state = Symbol('state') // undefined=added, Timeout=add-pending, false=removed
server/src/api.plugins.ts
+36
-37
@@ -18,7 +18,7 @@ import { downloadPlugin, getFolder2repo, getRepoInfo, readOnlinePlugin, searchPl
18
const apis: ApiHandlers = {
19
20
get_plugins({}, ctx) {
21
- const list = new SendListReadable([ ...mapPlugins(serialize), ...getAvailablePlugins() ])
21
+ const list = new SendListReadable({ addAtStart: [ ...mapPlugins(serialize), ...getAvailablePlugins() ] })
22
return list.events(ctx, {
23
pluginInstalled: p => list.add(serialize(p)),
24
'pluginStarted pluginStopped pluginUpdated': p => {
@@ -74,45 +74,44 @@ const apis: ApiHandlers = {
74
},
75
76
search_online_plugins({ text }, ctx) {
77
- const list = new SendListReadable()
78
- setTimeout(async () => {
79
- try {
80
- const folder2repo = getFolder2repo()
81
- for await (const pl of searchPlugins(text)) {
82
- const repo = pl.id
83
- const folder = _.findKey(folder2repo, x => x === repo)
84
- const installed = folder && getPluginInfo(folder)
85
- Object.assign(pl, {
86
- installed: _.includes(folder2repo, repo),
87
- update: installed && installed.version < pl.version!,
88
- })
89
- list.add(pl)
90
- // watch for events about this plugin, until this request is closed
91
- ctx.req.on('close', onOff(events, {
92
- pluginInstalled: p => {
93
- if (p.repo === repo)
94
- list.update({ id: repo }, { installed: true })
95
- },
96
- pluginUninstalled: folder => {
97
- if (repo === getFolder2repo()[folder])
98
- list.update({ id: repo }, { installed: false })
99
- },
100
- pluginUpdated: p => {
101
- if (p.repo === repo)
102
- list.update({ id: repo }, { update: p.version < pl.version! })
103
- },
104
- ['pluginDownload_'+repo](status) {
105
- list.update({ id: repo }, { downloading: status ?? null })
106
- }
107
- }) )
77
+ return new SendListReadable({
78
+ async doAtStart(list) {
79
+ try {
80
+ const folder2repo = getFolder2repo()
81
+ for await (const pl of searchPlugins(text)) {
82
+ const repo = pl.id
83
+ const folder = _.findKey(folder2repo, x => x === repo)
84
+ const installed = folder && getPluginInfo(folder)
85
+ Object.assign(pl, {
86
+ installed: _.includes(folder2repo, repo),
87
+ update: installed && installed.version < pl.version!,
88
+ })
89
+ list.add(pl)
90
+ // watch for events about this plugin, until this request is closed
91
+ ctx.req.on('close', onOff(events, {
92
+ pluginInstalled: p => {
93
+ if (p.repo === repo)
94
+ list.update({ id: repo }, { installed: true })
95
+ },
96
+ pluginUninstalled: folder => {
97
+ if (repo === getFolder2repo()[folder])
98
+ list.update({ id: repo }, { installed: false })
99
+ },
100
+ pluginUpdated: p => {
101
+ if (p.repo === repo)
102
+ list.update({ id: repo }, { update: p.version < pl.version! })
103
+ },
104
+ ['pluginDownload_' + repo](status) {
105
+ list.update({ id: repo }, { downloading: status ?? null })
106
+ }
107
+ }))
108
+ }
109
+ } catch (err: any) {
110
+ list.error(err.code || err.message)
111
}
112
+ list.ready()
113
}
110
- catch (err: any) {
111
- list.error(err.code || err.message)
112
- }
113
- list.ready()
114
})
115
- return list
115
},
116
117
async download_plugin(pl) {
server/src/apiMiddleware.ts
+36
-21
@@ -4,9 +4,10 @@ import { IncomingMessage } from 'http'
4
import Koa from 'koa'
5
import createSSE from './sse'
6
import { Readable } from 'stream'
7
-import { asyncGeneratorToReadable, objSameKeys, onOff, tryJson } from './misc'
7
+import { asyncGeneratorToReadable, objSameKeys, onOff, tryJson, wantArray } from './misc'
8
import events from './events'
9
import { UNAUTHORIZED } from './const'
10
+import _, { DebouncedFunc } from 'lodash'
11
12
export class ApiError extends Error {
13
constructor(public status:number, message?:string | Error) {
@@ -76,44 +77,58 @@ async function getJsonFromReq(req: IncomingMessage): Promise<any> {
77
type SendListFunc<T> = (list:SendListReadable<T>) => void
78
export class SendListReadable<T> extends Readable {
79
protected lastError: string | number | undefined
79
- constructor(addOrDoAtStart?: T[] | SendListFunc<T>) {
80
+ protected buffer: any[] = []
81
+ protected processBuffer: DebouncedFunc<any>
82
+ constructor({ addAtStart, doAtStart, bufferTime }:{ bufferTime?: number, addAtStart?: T[], doAtStart?: SendListFunc<T> }={}) {
83
super({ objectMode: true, read(){} })
84
+ if (!bufferTime)
85
+ bufferTime = 100
86
+ this.processBuffer = _.debounce(() => {
87
+ this.push(this.buffer)
88
+ this.buffer = []
89
+ }, bufferTime, { maxWait: bufferTime })
90
this.on('end', () =>
91
this.destroy())
83
- if (!addOrDoAtStart)
84
- return
85
- if (typeof addOrDoAtStart === 'function') {
86
- setTimeout(() => addOrDoAtStart(this))
87
- return
92
+ if (doAtStart)
93
+ setTimeout(() => doAtStart(this)) // work later, when list object has been received by Koa
94
+ if (addAtStart) {
95
+ for (const x of addAtStart)
96
+ this.add(x)
97
+ this.ready()
98
}
89
- for (const x of addOrDoAtStart)
90
- this.add(x)
91
- this.ready()
99
}
93
- add(rec: T) {
94
- this.push({ add: rec })
100
+ protected _push(rec: any) {
101
+ this.buffer.push(rec)
102
+ if (this.buffer.length > 10_000) // hard limit
103
+ this.processBuffer.flush()
104
+ else
105
+ this.processBuffer()
106
+ }
107
+ add(rec: T | T[]) {
108
+ this._push({ add: rec })
109
}
110
remove(key: Partial<T>) {
97
- this.push({ remove: [key] })
111
+ this._push({ remove: [key] })
112
}
113
update(search: Partial<T>, change: Partial<T>) {
100
- this.push({ update:[{ search, change }] })
101
- }
102
- close() {
103
- this.push(null)
114
+ this._push({ update:[{ search, change }] })
115
}
116
ready() { // useful to indicate the end of an initial phase, but we leave open for updates
106
- this.push('ready')
117
+ this._push('ready')
118
+ }
119
+ custom(data: any) {
120
+ this._push(data)
121
}
122
error(msg: NonNullable<typeof this.lastError>) {
109
- this.push({ error: msg })
123
+ this._push({ error: msg })
124
this.lastError = msg
125
}
126
getLastError() {
127
return this.lastError
128
}
115
- custom(data: any) {
116
- this.push(data)
129
+ close() {
130
+ this.processBuffer.flush()
131
+ this.push(null)
132
}
133
events(ctx: Koa.Context, eventMap: Parameters<typeof onOff>[1]) {
134
const off = onOff(events, eventMap)