1
/*
2
 *  This file is part of CoCalc: Copyright © 2020 Sagemath, Inc.
3
 *  License: MS-RSL – see LICENSE.md for details
4
 */
5

6
/*
7
Project status server, doing the heavy lifting of telling the client
8
what's going on in the project, especially if there is a problem.
9

10
Under the hood, it subscribes to the ProjectInfoServer, which updates
11
various statistics at a high-frequency. Therefore, this here filters
12
that information to a low-frequency low-volume stream of important
13
status updates.
14

15
Hence in particular, information like cpu, memory and disk are smoothed out and throttled.
16
*/
17

18
import { getLogger } from "@cocalc/project/logger";
19
import { how_long_ago_m, round1 } from "@cocalc/util/misc";
20
import { version as smcVersion } from "@cocalc/util/smc-version";
21
import { delay } from "awaiting";
22
import { EventEmitter } from "events";
23
import { isEqual } from "lodash";
24
import { get_ProjectInfoServer, ProjectInfoServer } from "../project-info";
25
import { ProjectInfo } from "@cocalc/util/types/project-info/types";
26
import {
27
  ALERT_DISK_FREE,
28
  ALERT_HIGH_PCT /* ALERT_MEDIUM_PCT */,
29
  RAISE_ALERT_AFTER_MIN,
30
  STATUS_UPDATES_INTERVAL_S,
31
} from "@cocalc/comm/project-status/const";
32
import {
33
  Alert,
34
  AlertType,
35
  ComponentName,
36
  ProjectStatus,
37
} from "@cocalc/comm/project-status/types";
38
import { cgroup_stats } from "@cocalc/comm/project-status/utils";
39
import { createPublisher } from "@cocalc/conat/project/project-status";
40
import { compute_server_id, project_id } from "@cocalc/project/data";
41

42
// TODO: only return the "next" value, if it is significantly different from "prev"
43
//function threshold(prev?: number, next?: number): number | undefined {
44
//  return next;
45
//}
46

47
const logger = getLogger("project-status:server");
48

49
function quantize(val, order) {
50
  const q = Math.round(Math.pow(10, order));
51
  return Math.round(q * Math.ceil(val / q));
52
}
53

54
// tracks, when for the first time we saw an elevated value
55
// we clear it if we're below a threshold (in the clear)
56
interface Elevated {
57
  cpu: number | null; // timestamps
58
  memory: number | null; // timestamps
59
  disk: number | null; // timestamps
60
}
61

62
export class ProjectStatusServer extends EventEmitter {
63
  private readonly dbg: Function;
64
  private running = false;
65
  private readonly testing: boolean;
66
  private readonly project_info: ProjectInfoServer;
67
  private info?: ProjectInfo;
68
  private status?: ProjectStatus;
69
  private last?: ProjectStatus;
70
  private elevated: Elevated = {
71
    cpu: null,
72
    disk: null,
73
    memory: null,
74
  };
75
  private elevated_cpu_procs: { [pid: string]: number } = {};
76
  private disk_mb?: number;
77
  private cpu_pct?: number;
78
  private cpu_tot?: number; // total time in seconds
79
  private mem_pct?: number;
80
  private mem_rss?: number;
81
  private mem_tot?: number;
82
  private components: { [name in ComponentName]?: number | undefined } = {};
83
  private lastEmit: number = 0; // timestamp, when status was emitted last
84

85
  constructor(testing = false) {
86
    super();
87
    this.testing = testing;
88
    this.dbg = (...msg) => logger.debug(...msg);
89
    this.project_info = get_ProjectInfoServer();
90
  }
91

92
  private async init(): Promise<void> {
93
    this.project_info.start();
94
    this.project_info.on("info", (info) => {
95
      //this.dbg(`got info timestamp=${info.timestamp}`);
96
      this.info = info;
97
      this.update();
98
      this.emitInfo();
99
    });
100
  }
101

102
  // checks if there the current state (after update()) should be emitted
103
  private emitInfo(): void {
104
    if (this.lastEmit === 0) {
105
      this.dbg("emitInfo[last=0]", this.status);
106
      this.doEmit();
107
      return;
108
    }
109

110
    // if alert changed, emit immediately
111
    if (!isEqual(this.last?.alerts, this.status?.alerts)) {
112
      this.dbg("emitInfo[alert]", this.status);
113
      this.doEmit();
114
    } else {
115
      // deep comparison check via lodash and we rate limit
116
      const recent =
117
        this.lastEmit + 1000 * STATUS_UPDATES_INTERVAL_S > Date.now();
118
      const changed = !isEqual(this.status, this.last);
119
      if (!recent && changed) {
120
        this.dbg("emitInfo[changed]", this.status);
121
        this.doEmit();
122
      }
123
    }
124
  }
125

126
  private doEmit(): void {
127
    this.emit("status", this.status);
128
    this.lastEmit = Date.now();
129
  }
130

131
  public setComponentAlert(name: ComponentName) {
132
    // we set this to the time when we first got notified about the problem
133
    if (this.components[name] == null) {
134
      this.components[name] = Date.now();
135
    }
136
  }
137

138
  public clearComponentAlert(name: ComponentName) {
139
    delete this.components[name];
140
  }
141

142
  // this derives elevated levels from the project info object
143
  private update_alerts() {
144
    if (this.info == null) return;
145
    const du = this.info.disk_usage.project;
146
    const ts = this.info.timestamp;
147

148
    const do_alert = (type: AlertType, is_bad: boolean) => {
149
      if (is_bad) {
150
        // if it isn't fine, set it once to the timestamp (and let it age)
151
        if (this.elevated[type] == null) {
152
          this.elevated[type] = ts;
153
        }
154
      } else {
155
        // unless it's fine again, then remove the timestamp
156
        this.elevated[type] = null;
157
      }
158
    };
159

160
    do_alert("disk", du.free < ALERT_DISK_FREE);
161
    this.disk_mb = du.usage;
162

163
    const cg = this.info.cgroup;
164
    const du_tmp = this.info.disk_usage.tmp;
165
    if (cg != null) {
166
      // we round/quantisize values to reduce the number of updates
167
      // and also send less data with each update
168
      const cgStats = cgroup_stats(cg, du_tmp);
169
      this.mem_pct = Math.round(cgStats.mem_pct);
170
      this.cpu_pct = Math.round(cgStats.cpu_pct);
171
      this.cpu_tot = Math.round(cgStats.cpu_tot);
172
      this.mem_tot = quantize(cgStats.mem_tot, 1);
173
      this.mem_rss = quantize(cgStats.mem_rss, 1);
174
      do_alert("memory", cgStats.mem_pct > ALERT_HIGH_PCT);
175
      do_alert("cpu-cgroup", cgStats.cpu_pct > ALERT_HIGH_PCT);
176
    }
177
  }
178

179
  private alert_cpu_processes(): string[] {
180
    const pids: string[] = [];
181
    if (this.info == null) return [];
182
    const ts = this.info.timestamp;
183
    const ecp = this.elevated_cpu_procs;
184
    // we have to check if there aren't any processes left which no longer exist
185
    const leftovers = new Set(Object.keys(ecp));
186
    // bookkeeping of elevated process PIDS
187
    for (const [pid, proc] of Object.entries(this.info.processes ?? {})) {
188
      leftovers.delete(pid);
189
      if (proc.cpu.pct > ALERT_HIGH_PCT) {
190
        if (ecp[pid] == null) {
191
          ecp[pid] = ts;
192
        }
193
      } else {
194
        delete ecp[pid];
195
      }
196
    }
197
    for (const pid of leftovers) {
198
      delete ecp[pid];
199
    }
200
    // to actually fire alert when necessary
201
    for (const [pid, ts] of Object.entries(ecp)) {
202
      if (ts != null && how_long_ago_m(ts) > RAISE_ALERT_AFTER_MIN) {
203
        pids.push(pid);
204
      }
205
    }
206
    pids.sort(); // to make this stable across iterations
207
    //this.dbg("alert_cpu_processes", pids, ecp);
208
    return pids;
209
  }
210

211
  // update alert levels and set alert states if they persist to be active
212
  private alerts(): Alert[] {
213
    this.update_alerts();
214
    const alerts: Alert[] = [];
215
    const alert_keys: AlertType[] = ["cpu-cgroup", "disk", "memory"];
216
    for (const k of alert_keys) {
217
      const ts = this.elevated[k];
218
      if (ts != null && how_long_ago_m(ts) > RAISE_ALERT_AFTER_MIN) {
219
        alerts.push({ type: k } as Alert);
220
      }
221
    }
222
    const pids: string[] = this.alert_cpu_processes();
223
    if (pids.length > 0) alerts.push({ type: "cpu-process", pids });
224

225
    const componentNames: ComponentName[] = [];
226
    for (const [k, ts] of Object.entries(this.components)) {
227
      if (ts == null) continue;
228
      // we alert without a delay
229
      componentNames.push(k as ComponentName);
230
    }
231
    // only send any alert if there is actually a problem!
232
    if (componentNames.length > 0) {
233
      alerts.push({ type: "component", names: componentNames });
234
    }
235
    return alerts;
236
  }
237

238
  private fake_data(): ProjectStatus["usage"] {
239
    const lastUsage = this.last?.["usage"];
240

241
    const next = (key, max) => {
242
      const last = lastUsage?.[key] ?? max / 2;
243
      const dx = max / 50;
244
      const val = last + dx * Math.random() - dx / 2;
245
      return Math.round(Math.min(max, Math.max(0, val)));
246
    };
247

248
    const mem_tot = 3000;
249
    const mem_pct = next("mem_pct", 100);
250
    const mem_rss = Math.round((mem_tot * mem_pct) / 100);
251
    const cpu_tot = round1((lastUsage?.["cpu_tot"] ?? 0) + Math.random() / 10);
252

253
    return {
254
      disk_mb: next("disk", 3000),
255
      mem_tot,
256
      mem_pct,
257
      cpu_pct: next("cpu_pct", 100),
258
      cpu_tot,
259
      mem_rss,
260
    };
261
  }
262

263
  // this function takes the "info" we have (+ more maybe?)
264
  // and derives various states from it.
265
  // It shouldn't really matter how often it is being called,
266
  // but still only emit new objects if it is either really necessary (new alert)
267
  // or after some time. This must be a low-frequency and low-volume stream of data.
268
  private update(): void {
269
    this.last = this.status;
270

271
    // alerts must come first, it updates usage status fields
272
    const alerts = this.alerts();
273

274
    // set this to true if you're developing (otherwise you don't get any data)
275
    const fake_data = false;
276

277
    // collect status fields in usage object
278
    const usage = fake_data
279
      ? this.fake_data()
280
      : {
281
          disk_mb: this.disk_mb,
282
          mem_pct: this.mem_pct,
283
          cpu_pct: this.cpu_pct,
284
          cpu_tot: this.cpu_tot,
285
          mem_rss: this.mem_rss,
286
          mem_tot: this.mem_tot,
287
        };
288

289
    this.status = { alerts, usage, version: smcVersion };
290
  }
291

292
  private async get_status(): Promise<ProjectStatus | undefined> {
293
    this.update();
294
    return this.status;
295
  }
296

297
  public stop(): void {
298
    this.running = false;
299
  }
300

301
  public async start(): Promise<void> {
302
    if (!this.running) {
303
      await this._start();
304
    }
305
  }
306

307
  private async _start(): Promise<void> {
308
    this.dbg("start");
309
    if (this.running) {
310
      throw Error("Cannot start ProjectStatusServer twice");
311
    }
312
    this.running = true;
313
    await this.init();
314

315
    const status = await this.get_status();
316
    this.emit("status", status);
317

318
    while (this.testing) {
319
      await delay(5000);
320
      const status = await this.get_status();
321
      this.emit("status", status);
322
    }
323
  }
324
}
325

326
// singleton, we instantiate it when we need it
327
let status: ProjectStatusServer | undefined = undefined;
328

329
export function init() {
330
  logger.debug("initializing project status server, and enabling publishing");
331
  if (status == null) {
332
    status = new ProjectStatusServer();
333
  }
334
  createPublisher({
335
    projectStatusServer: status,
336
    compute_server_id,
337
    project_id,
338
  });
339
  status.start();
340
}
341

342
// testing: $ ts-node server.ts
343
if (require.main === module) {
344
  const pss = new ProjectStatusServer(true);
345
  pss.start();
346
  let cnt = 0;
347
  pss.on("status", (status) => {
348
    console.log(JSON.stringify(status, null, 2));
349
    cnt += 1;
350
    if (cnt >= 2) process.exit();
351
  });
352
}
353

354