fix: admin/monitor: inconsistent results

Massimo Melina committed Aug 6, 2022 at 17:22 UTC 052ddd35f68f0dc79f5535e37e8d16167d16d042
3 files changed +32 -17
server/src/api.monitor.ts
+27 -5
@@ -21,29 +21,51 @@ const apis: ApiHandlers = {
21 const list = new SendListReadable( getConnections().map(c => serializeConnection(c)) )
22 type Change = Partial<Omit<Connection,'ip'>>
23 const throttledUpdate = _.throttle(update, 1000/20) // try to avoid clogging with updates
24 + const state = Symbol('state') // undefined=added, Timeout=add-pending, false=removed
25 return list.events(ctx, {
25 - connection: conn => list.add(serializeConnection(conn)),
26 + connection(conn: Connection) {
27 + conn[state] = setTimeout(() => add(conn), 100)
28 + },
29 connectionClosed(conn: Connection) {
30 + if (cancel(conn)) return
31 list.remove(serializeConnection(conn, true))
32 + conn[state] = false
33 },
34 connectionUpdated(conn: Connection, change: Change) {
35 if (!change.ctx)
36 return throttledUpdate(conn, change)
37 +
38 Object.assign(change, fromCtx(change.ctx))
33 - delete change.ctx
34 - throttledUpdate(conn, change)
39 + change.ctx = undefined
40 + if (!add(conn))
41 + throttledUpdate(conn, change)
42 },
43 })
44
45 + function add(conn: Connection) {
46 + if (!cancel(conn)) return
47 + list.add(serializeConnection(conn))
48 + return true
49 + }
50 +
51 + function cancel(conn: Connection) {
52 + if (!conn[state]) return
53 + clearTimeout(conn[state])
54 + conn[state] = undefined
55 + return true
56 + }
57 +
58 function update(conn: Connection, change: Change) {
59 + if (conn[state] === false) return
60 list.update(serializeConnection(conn, true), change)
61 }
62
63 function serializeConnection(conn: Connection, minimal?:true) {
43 - const { socket, started, secure, got } = conn
64 + const { socket, started, secure } = conn
65 return Object.assign(getConnAddress(conn), !minimal && {
66 v: (socket.remoteFamily?.endsWith('6') ? 6 : 4),
46 - got,
67 + got: socket.bytesRead,
68 + sent: socket.bytesWritten,
69 started,
70 secure: (secure || undefined) as boolean|undefined, // undefined will save some space once json-ed
71 ...fromCtx(conn.ctx),
server/src/connections.ts
+3 -9
@@ -7,9 +7,7 @@ import _ from 'lodash'
7
8 export class Connection {
9 readonly started = new Date()
10 - got = 0
10 sent = 0
12 - alreadyEmitted = false // already communicated to
11 outSpeed?: number
12 ctx?: Koa.Context
13 private _cachedIp?: string
@@ -17,14 +15,11 @@ export class Connection {
15
16 constructor(readonly socket: Socket) {
17 all.push(this)
20 - socket.on('data', data =>
21 - this.got += data.length )
18 socket.on('close', () => {
19 all.splice(all.indexOf(this), 1)
24 - if (this.alreadyEmitted)
25 - events.emit('connectionClosed', this)
20 + events.emit('connectionClosed', this)
21 })
27 - events.emit('socket', socket)
22 + events.emit('connection', this)
23 }
24
25 get ip() {
@@ -61,6 +56,5 @@ export function updateConnection(conn: Connection, change: Partial<Connection>)
56 if (!change.ctx && Object.entries(change).every(([k,v]) => _.isEqual(v, conn[k as keyof Connection]) ))
57 return
58 Object.assign(conn, change)
64 - events.emit(conn.alreadyEmitted ? 'connectionUpdated' : 'connection', conn, change)
65 - conn.alreadyEmitted = true
59 + events.emit('connectionUpdated', conn, change)
60 }
server/src/throttler.ts
+2 -3
@@ -8,7 +8,6 @@ import { getOrSet, isLocalHost } from './misc'
8 import { Connection, updateConnection } from './connections'
9 import _ from 'lodash'
10 import events from './events'
11 -import { Socket } from 'net'
11
12 const mainThrottleGroup = new ThrottleGroup(Infinity)
13
@@ -102,6 +101,6 @@ setInterval(() => {
101 totalInSpeed = roundKb(deltaGotKb / past)
102 }, 1000)
103
105 -events.on('socket', (socket: Socket) =>
106 - socket.on('data', data =>
104 +events.on('connection', (c: Connection) =>
105 + c.socket.on('data', data =>
106 totalGot += data.length ))