Skip to content
This repository was archived by the owner on Aug 2, 2025. It is now read-only.

Commit 6ef8769

Browse files
Its4NikIts4Nik
committed
Feat: Websocket server for relaying stats
1 parent 8ba8823 commit 6ef8769

13 files changed

Lines changed: 158 additions & 132 deletions

File tree

‎data/.gitignore‎

Lines changed: 0 additions & 1 deletion
This file was deleted.

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@
4040
"@types/bun": "latest",
4141
"@types/dockerode": "^3.3.42",
4242
"@types/js-yaml": "^4.0.9",
43-
"@types/node": "^22.15.32",
43+
"@types/node": "^22.16.0",
4444
"@types/split2": "^4.2.3",
4545
"bun-types": "latest",
4646
"cross-env": "^7.0.3",

‎src/core/database/backup.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -60,9 +60,9 @@ export async function backupDatabase(): Promise<string> {
6060
copyFileSync(`${backupDir}dockstatapi.db`, backupFilename);
6161
logger.info(`Backup created successfully: ${backupFilename}`);
6262
logger.debug("File copy operation completed without errors");
63-
} catch (e) {
64-
logger.error(`Failed to create backup file: ${(e as Error).message}`);
65-
throw e;
63+
} catch (error) {
64+
logger.error(`Failed to create backup file: ${(error as Error).message}`);
65+
throw new Error(error as string);
6666
}
6767

6868
return backupFilename;
@@ -97,9 +97,9 @@ export function restoreDatabase(backupFilename: string): void {
9797
copyFileSync(backupFile, `${backupDir}dockstatapi.db`);
9898
logger.info(`Database restored successfully from: ${backupFilename}`);
9999
logger.debug("Database file replacement completed");
100-
} catch (e) {
101-
logger.error(`Restore failed: ${(e as Error).message}`);
102-
throw e;
100+
} catch (error) {
101+
logger.error(`Restore failed: ${(error as Error).message}`);
102+
throw new Error(error as string);
103103
}
104104
},
105105
() => {

‎src/core/docker/scheduler.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ async function setSchedules() {
117117
logger.info("Schedules have been set successfully.");
118118
} catch (error) {
119119
logger.error("Error setting schedules:", error);
120-
throw error;
120+
throw new Error(error as string);
121121
}
122122
}
123123

‎src/core/plugins/plugin-manager.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,8 @@ class PluginManager extends EventEmitter {
3030

3131
async start() {
3232
try {
33-
return await loadPlugins("./server/src/plugins");
33+
await loadPlugins("./server/src/plugins");
34+
return;
3435
} catch (error) {
3536
logger.error(`Failed to init plugin manager: ${error}`);
3637
return;
@@ -65,7 +66,7 @@ class PluginManager extends EventEmitter {
6566
const plugins: PluginInfo[] = [];
6667

6768
for (const plugin of this.plugins.values()) {
68-
logger.debug(`Loaded plugin: ${plugin}`);
69+
logger.debug(`Loaded plugin: ${JSON.stringify(plugin)}`);
6970
const hooks = getHooks(plugin);
7071
plugins.push({
7172
name: plugin.name,
@@ -76,7 +77,7 @@ class PluginManager extends EventEmitter {
7677
}
7778

7879
for (const plugin of this.failedPlugins.values()) {
79-
logger.debug(`Loaded plugin: ${plugin}`);
80+
logger.debug(`Loaded plugin: ${JSON.stringify(plugin)}`);
8081
const hooks = getHooks(plugin);
8182
plugins.push({
8283
name: plugin.name,

‎src/core/utils/logger.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import type { log_message } from "~/typings/database";
1212

1313
import { backupInProgress } from "../database/_dbState";
1414

15-
const padNewlines = process.env.PAD_NEW_LINES !== "false";
15+
const padNewlines = true; //process.env.PAD_NEW_LINES !== "false";
1616

1717
type LogLevel =
1818
| "error"

‎src/handlers/config.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ class apiHandler {
4444

4545
getPlugins(): PluginInfo[] {
4646
try {
47+
logger.debug("Gathering plugins");
4748
return pluginManager.getPlugins();
4849
} catch (error) {
4950
const errMsg = error instanceof Error ? error.message : String(error);

‎src/handlers/index.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
import { setSchedules } from "~/core/docker/scheduler";
22
import { pluginManager } from "~/core/plugins/plugin-manager";
3+
import { logger } from "~/core/utils/logger";
34
import { ApiHandler } from "./config";
45
import { DatabaseHandler } from "./database";
56
import { BasicDockerHandler } from "./docker";
67
import { LogHandler } from "./logs";
8+
import { startDockerStatsBroadcast } from "./modules/docker-socket";
9+
import { Starter } from "./modules/starter";
710
import { Sockets } from "./sockets";
811
import { StackHandler } from "./stacks";
912
import { CheckHealth } from "./utils";
@@ -16,6 +19,6 @@ export const handlers = {
1619
LogHandler,
1720
CheckHealth,
1821
Sockets: Sockets,
19-
StartServer: setSchedules(),
20-
ImportPlugins: await pluginManager.start(),
2122
};
23+
24+
Starter.startAll();

‎src/handlers/modules/docker-socket.ts‎

Lines changed: 98 additions & 113 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { Readable, type Transform } from "node:stream";
1+
import { serve } from "bun";
22
import split2 from "split2";
33
import { dbFunctions } from "~/core/database";
44
import { getDockerClient } from "~/core/docker/client";
@@ -9,134 +9,119 @@ import {
99
import { logger } from "~/core/utils/logger";
1010
import type { DockerStatsEvent } from "~/typings/docker";
1111

12-
export function createDockerStatsStream(): Readable {
13-
const stream = new Readable({
14-
objectMode: true,
15-
read() {},
16-
});
12+
// Track all connected WebSocket clients
13+
const clients = new Set<Bun.ServerWebSocket<unknown>>();
1714

18-
const substreams: Array<{
19-
statsStream: Readable;
20-
splitStream: Transform;
21-
}> = [];
22-
23-
const cleanup = () => {
24-
for (const { statsStream, splitStream } of substreams) {
25-
try {
26-
statsStream.unpipe(splitStream);
27-
statsStream.destroy();
28-
splitStream.destroy();
29-
} catch (error) {
30-
logger.error(`Cleanup error: ${error}`);
31-
}
15+
// Broadcast a DockerStatsEvent to every connected client
16+
function broadcast(event: DockerStatsEvent) {
17+
const message = JSON.stringify(event);
18+
for (const ws of clients) {
19+
if (ws.readyState === 1) {
20+
ws.send(message);
3221
}
33-
substreams.length = 0;
34-
};
35-
36-
stream.on("close", cleanup);
37-
stream.on("error", cleanup);
38-
39-
(async () => {
40-
try {
41-
const hosts = dbFunctions.getDockerHosts();
42-
logger.debug(`Retrieved ${hosts.length} docker host(s)`);
43-
44-
for (const host of hosts) {
45-
if (stream.destroyed) break;
22+
}
23+
}
4624

47-
try {
48-
const docker = getDockerClient(host);
49-
await docker.ping();
50-
const containers = await docker.listContainers({
51-
all: true,
52-
});
25+
// Start Docker stats polling and broadcasting
26+
export async function startDockerStatsBroadcast() {
27+
logger.debug("Starting Docker stats broadcast...");
5328

54-
logger.debug(
55-
`Found ${containers.length} containers on ${host.name} (id: ${host.id})`,
56-
);
29+
try {
30+
const hosts = dbFunctions.getDockerHosts();
31+
logger.debug(`Retrieved ${hosts.length} Docker host(s)`);
5732

58-
for (const containerInfo of containers) {
59-
if (stream.destroyed) break;
33+
for (const host of hosts) {
34+
try {
35+
const docker = getDockerClient(host);
36+
await docker.ping();
37+
const containers = await docker.listContainers({ all: true });
38+
logger.debug(
39+
`Host ${host.name} contains ${containers.length} containers`,
40+
);
6041

42+
for (const info of containers) {
43+
// Kick off one independent async task per container
44+
(async () => {
6145
try {
62-
const container = docker.getContainer(containerInfo.Id);
63-
const statsStream = (await container.stats({
64-
stream: true,
65-
})) as Readable;
66-
const splitStream = split2();
67-
68-
substreams.push({ statsStream, splitStream });
46+
const statsStream = await docker
47+
.getContainer(info.Id)
48+
.stats({ stream: true });
49+
const splitter = split2();
50+
statsStream.pipe(splitter);
6951

70-
statsStream
71-
.on("close", () => splitStream.destroy())
72-
.pipe(splitStream)
73-
.on("data", (line: string) => {
74-
if (stream.destroyed || !line) return;
75-
76-
try {
77-
const stats = JSON.parse(line);
78-
const event: DockerStatsEvent = {
79-
type: "stats",
80-
id: containerInfo.Id,
81-
hostId: host.id,
82-
name: containerInfo.Names[0].replace(/^\//, ""),
83-
image: containerInfo.Image,
84-
status: containerInfo.Status,
85-
state: containerInfo.State,
86-
cpuUsage: calculateCpuPercent(stats) ?? 0,
87-
memoryUsage: calculateMemoryUsage(stats) ?? 0,
88-
};
89-
stream.push(event);
90-
} catch (error) {
91-
stream.push({
92-
type: "error",
93-
hostId: host.id,
94-
containerId: containerInfo.Id,
95-
error: `Parse error: ${
96-
error instanceof Error ? error.message : String(error)
97-
}`,
98-
});
99-
}
100-
})
101-
.on("error", (error: Error) => {
102-
stream.push({
52+
for await (const line of splitter) {
53+
if (!line) continue;
54+
try {
55+
const stats = JSON.parse(line);
56+
broadcast({
57+
type: "stats",
58+
id: info.Id,
59+
hostId: host.id,
60+
name: info.Names[0].replace(/^\//, ""),
61+
image: info.Image,
62+
status: info.Status,
63+
state: stats.state || info.State,
64+
cpuUsage: calculateCpuPercent(stats) ?? 0,
65+
memoryUsage: calculateMemoryUsage(stats) ?? 0,
66+
});
67+
} catch (err) {
68+
broadcast({
10369
type: "error",
10470
hostId: host.id,
105-
containerId: containerInfo.Id,
106-
error: `Stream error: ${error.message}`,
71+
containerId: info.Id,
72+
error: `Parse error: ${(err as Error).message}`,
10773
});
108-
});
109-
} catch (error) {
110-
stream.push({
74+
}
75+
}
76+
} catch (err) {
77+
broadcast({
11178
type: "error",
11279
hostId: host.id,
113-
containerId: containerInfo.Id,
114-
error: `Container error: ${
115-
error instanceof Error ? error.message : String(error)
116-
}`,
80+
containerId: info.Id,
81+
error: `Stats stream error: ${(err as Error).message}`,
11782
});
11883
}
119-
}
120-
} catch (error) {
121-
stream.push({
122-
type: "error",
123-
hostId: host.id,
124-
error: `Host connection error: ${
125-
error instanceof Error ? error.message : String(error)
126-
}`,
127-
});
84+
})();
12885
}
86+
} catch (err) {
87+
broadcast({
88+
type: "error",
89+
hostId: host.id,
90+
error: `Host connection error: ${(err as Error).message}`,
91+
});
12992
}
130-
} catch (error) {
131-
stream.push({
132-
type: "error",
133-
error: `Initialization error: ${
134-
error instanceof Error ? error.message : String(error)
135-
}`,
136-
});
137-
stream.destroy();
13893
}
139-
})();
140-
141-
return stream;
94+
} catch (err) {
95+
broadcast({
96+
type: "error",
97+
hostId: 0,
98+
error: `Initialization error: ${(err as Error).message}`,
99+
});
100+
}
142101
}
102+
103+
serve({
104+
port: 4837,
105+
reusePort: true,
106+
fetch(req, server) {
107+
// Upgrade requests to WebSocket
108+
if (req.url.endsWith("/ws/docker")) {
109+
if (server.upgrade(req)) {
110+
return; // auto 101 Switching Protocols
111+
}
112+
}
113+
return new Response("Expected WebSocket upgrade", { status: 426 });
114+
},
115+
116+
websocket: {
117+
open(ws) {
118+
logger.debug("Client connected via WebSocket");
119+
clients.add(ws);
120+
},
121+
close(ws, code, reason) {
122+
logger.debug(`Client disconnected (${code}): ${reason}`);
123+
clients.delete(ws);
124+
},
125+
message() {},
126+
},
127+
});

‎src/handlers/modules/starter.ts‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
import { setSchedules } from "~/core/docker/scheduler";
2+
import { pluginManager } from "~/core/plugins/plugin-manager";
3+
import { startDockerStatsBroadcast } from "./docker-socket";
4+
5+
function banner(msg: string) {
6+
const fenced = `= ${msg} =`;
7+
const lines = msg.length;
8+
console.info("=".repeat(fenced.length));
9+
console.info(fenced);
10+
console.info("=".repeat(fenced.length));
11+
}
12+
13+
class starter {
14+
public started = false;
15+
async startAll() {
16+
try {
17+
if (!this.started) {
18+
banner("Setting schedules");
19+
await setSchedules();
20+
banner("Importing plugins");
21+
await startDockerStatsBroadcast();
22+
banner("Started DockStatAPI succesfully");
23+
await pluginManager.start();
24+
banner("Starting WebSocket server");
25+
this.started = true;
26+
return;
27+
}
28+
console.info("Already started");
29+
} catch (error) {
30+
throw new Error(`Could not start DockStatAPI: ${error}`);
31+
}
32+
}
33+
}
34+
35+
export const Starter = new starter();

0 commit comments

Comments
 (0)