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 information server, doing the heavy lifting of telling the client
8
about what's going on in a project.
9

10
This is an event emitter that emits a ProjectInfo object periodically when running.
11
*/
12

13
import { delay } from "awaiting";
14
import type { DiskUsage as DF_DiskUsage } from "diskusage";
15
import { check as df } from "diskusage";
16
import { EventEmitter } from "node:events";
17
import { readFile } from "node:fs/promises";
18
import { ProcessStats } from "@cocalc/backend/process-stats";
19
import { pidToPath as terminalPidToPath } from "@cocalc/project/conat/terminal/manager";
20
import type {
21
  CGroup,
22
  CoCalcInfo,
23
  DiskUsage,
24
  Process,
25
  Processes,
26
  ProjectInfo,
27
} from "@cocalc/util/types/project-info/types";
28
import { get_path_for_pid as x11_pid2path } from "../x11/server";
29
import { getLogger } from "../logger";
30

31
const L = getLogger("project-info:server").debug;
32

33
const bytes2MiB = (bytes) => bytes / (1024 * 1024);
34

35
export class ProjectInfoServer extends EventEmitter {
36
  private last?: ProjectInfo = undefined;
37
  private readonly dbg: Function;
38
  private running = false;
39
  private readonly testing: boolean;
40
  private delay_s: number;
41
  private cgroupFilesAreMissing: boolean = false;
42
  private processStats: ProcessStats;
43

44
  constructor(testing = false) {
45
    super();
46
    this.delay_s = 2;
47
    this.testing = testing;
48
    this.dbg = L;
49
  }
50

51
  private async processes(timestamp: number) {
52
    return await this.processStats.processes(timestamp);
53
  }
54

55
  // delta-time for this and the previous process information
56
  private dt(timestamp) {
57
    return (timestamp - (this.last?.timestamp ?? 0)) / 1000;
58
  }
59

60
  public latest(): ProjectInfo | undefined {
61
    return this.last;
62
  }
63

64
  // for a process we know (pid, etc.) we try to map to cocalc specific information
65
  private async cocalc({
66
    pid,
67
    cmdline,
68
  }: Pick<Process, "pid" | "cmdline">): Promise<CoCalcInfo | undefined> {
69
    //this.dbg("classify", { pid, exe, cmdline });
70
    if (pid === process.pid) {
71
      return { type: "project" };
72
    }
73
    // SPEED: importing @cocalc/jupyter/kernel is slow, so it MUST NOT BE DONE
74
    // on the top level, especially not in any code that is loaded during
75
    // project startup
76
    const { get_kernel_by_pid } = await import("@cocalc/jupyter/kernel");
77
    const jupyter_kernel = get_kernel_by_pid(pid);
78
    if (jupyter_kernel != null) {
79
      return { type: "jupyter", path: jupyter_kernel.get_path() };
80
    }
81
    const termpath = terminalPidToPath(pid);
82
    if (termpath != null) {
83
      return { type: "terminal", path: termpath };
84
    }
85
    const x11_path = x11_pid2path(pid);
86
    if (x11_path != null) {
87
      return { type: "x11", path: x11_path };
88
    }
89
    // SSHD: strangely, just one long string in cmdline[0]
90
    if (
91
      cmdline.length === 1 &&
92
      cmdline[0].startsWith("sshd:") &&
93
      cmdline[0].indexOf("-p 2222") != -1
94
    ) {
95
      return { type: "sshd" };
96
    }
97
  }
98

99
  private async lookupCoCalcInfo(processes: Processes) {
100
    // iterate over all processes keys (pid) and call this.cocalc({pid, cmdline})
101
    // to update the processes coclc field
102
    for (const pid in processes) {
103
      processes[pid].cocalc = await this.cocalc({
104
        pid: parseInt(pid),
105
        cmdline: processes[pid].cmdline,
106
      });
107
    }
108
  }
109

110
  // this is specific to running a project in a CGroup container
111
  // Harald: however, even without a container this shouldn't fail … just tells
112
  // you what the whole system is doing, all your processes.
113
  // William: it's constantly failing in cocalc-docker every second, so to avoid
114
  // clogging logs and wasting CPU, if the files are missing once, it stops updating.
115
  private async cgroup({ timestamp }): Promise<CGroup | undefined> {
116
    if (this.cgroupFilesAreMissing) {
117
      return;
118
    }
119
    try {
120
      const [mem_stat_raw, cpu_raw, oom_raw, cfs_quota_raw, cfs_period_raw] =
121
        await Promise.all([
122
          readFile("/sys/fs/cgroup/memory/memory.stat", "utf8"),
123
          readFile("/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage", "utf8"),
124
          readFile("/sys/fs/cgroup/memory/memory.oom_control", "utf8"),
125
          readFile("/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_quota_us", "utf8"),
126
          readFile("/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_period_us", "utf8"),
127
        ]);
128
      const mem_stat_keys = [
129
        "total_rss",
130
        "total_cache",
131
        "hierarchical_memory_limit",
132
      ];
133
      const cpu_usage = parseFloat(cpu_raw) / Math.pow(10, 9);
134
      const dt = this.dt(timestamp);
135
      const cpu_usage_rate =
136
        this.last?.cgroup != null
137
          ? (cpu_usage - this.last.cgroup.cpu_usage) / dt
138
          : 0;
139
      const [cfs_quota, cfs_period] = [
140
        parseInt(cfs_quota_raw),
141
        parseInt(cfs_period_raw),
142
      ];
143
      const mem_stat = mem_stat_raw
144
        .split("\n")
145
        .map((line) => line.split(" "))
146
        .filter(([k, _]) => mem_stat_keys.includes(k))
147
        .reduce((stat, [key, val]) => {
148
          stat[key] = bytes2MiB(parseInt(val));
149
          return stat;
150
        }, {});
151
      const oom_kills = oom_raw
152
        .split("\n")
153
        .filter((val) => val.startsWith("oom_kill "))
154
        .map((val) => parseInt(val.slice("oom_kill ".length)))[0];
155
      return {
156
        mem_stat,
157
        cpu_usage,
158
        cpu_usage_rate,
159
        cpu_cores_limit: cfs_quota / cfs_period,
160
        oom_kills,
161
      };
162
    } catch (err) {
163
      this.dbg("cgroup: error", err);
164
      if (err.code == "ENOENT") {
165
        // TODO: instead of shutting this down, we could maybe do a better job
166
        // figuring out what the correct cgroups files are on a given system.
167
        // E.g., in my cocalc-docker, I do NOT have /sys/fs/cgroup/memory/memory.stat
168
        // but I do have /sys/fs/cgroup/memory.stat
169
        this.cgroupFilesAreMissing = true;
170
        this.dbg(
171
          "cgroup: files are missing so cgroups info will no longer be updated",
172
        );
173
      }
174
      return undefined;
175
    }
176
  }
177

178
  // for cocalc/kucalc we want to know the disk usage + limits of the
179
  // users home dir and /tmp. /tmp is a ram disk, which will count against
180
  // the overall memory limit!
181
  private async disk_usage(): Promise<DiskUsage> {
182
    const convert = function (val: DF_DiskUsage) {
183
      return {
184
        total: bytes2MiB(val.total),
185
        free: bytes2MiB(val.free),
186
        available: bytes2MiB(val.available),
187
        usage: bytes2MiB(val.total - val.free),
188
      };
189
    };
190
    const [tmp, project] = await Promise.all([
191
      df("/tmp"),
192
      df(process.env.HOME ?? "/home/user"),
193
    ]);
194
    return { tmp: convert(tmp), project: convert(project) };
195
  }
196

197
  // orchestrating where all the information is bundled up for an update
198
  private async get_info(): Promise<ProjectInfo | undefined> {
199
    try {
200
      const timestamp = Date.now();
201
      const [processes, cgroup, disk_usage] = await Promise.all([
202
        this.processes(timestamp),
203
        this.cgroup({ timestamp }),
204
        this.disk_usage(),
205
      ]);
206
      const { procs, boottime, uptime } = processes;
207
      await this.lookupCoCalcInfo(procs);
208
      const info: ProjectInfo = {
209
        timestamp,
210
        processes: procs,
211
        uptime,
212
        boottime,
213
        cgroup,
214
        disk_usage,
215
      };
216
      return info;
217
    } catch (err) {
218
      this.dbg("get_info: error", err);
219
    }
220
  }
221

222
  public stop() {
223
    this.running = false;
224
  }
225

226
  close = () => {
227
    this.stop();
228
  };
229

230
  public async start(): Promise<void> {
231
    if (this.running) {
232
      this.dbg("project-info/server: already running, cannot be started twice");
233
    } else {
234
      await this._start();
235
    }
236
  }
237

238
  private async _start(): Promise<void> {
239
    this.dbg("start");
240
    if (this.running) {
241
      throw Error("Cannot start ProjectInfoServer twice");
242
    }
243
    this.running = true;
244
    this.processStats = new ProcessStats({
245
      testing: this.testing,
246
      dbg: this.dbg,
247
    });
248
    await this.processStats.init();
249
    while (true) {
250
      //this.dbg(`listeners on 'info': ${this.listenerCount("info")}`);
251
      const info = await this.get_info();
252
      if (info != null) this.last = info;
253
      this.emit("info", info ?? this.last);
254
      if (this.running) {
255
        await delay(1000 * this.delay_s);
256
      } else {
257
        this.dbg("start: no longer running → stopping loop");
258
        this.last = undefined;
259
        return;
260
      }
261
      // in test mode just one more, that's enough
262
      if (this.last != null && this.testing) {
263
        const info = await this.get_info();
264
        this.dbg(JSON.stringify(info, null, 2));
265
        return;
266
      }
267
    }
268
  }
269
}
270

271
// testing: $ ts-node server.ts
272
if (require.main === module) {
273
  const pis = new ProjectInfoServer(true);
274
  pis.start().then(() => process.exit());
275
}
276

277