Skip to content

Commit 13bbc09

Browse files
lukepololukepolo
committed
feature: download logs
1 parent 9e86c82 commit 13bbc09

File tree

3 files changed

+217
-28
lines changed

3 files changed

+217
-28
lines changed

‎src/k8s/logging/logging.service.ts‎

Lines changed: 108 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -47,21 +47,23 @@ export class LoggingService {
4747

4848
archive.on("error", (err) => {
4949
this.logger.error("Archive stream error", err);
50-
5150
try {
52-
stream.destroy(err);
53-
} catch {
54-
stream.end();
51+
if (!stream.destroyed) {
52+
stream.destroy(err);
53+
}
54+
} catch (error) {
55+
this.logger.error("Error destroying stream", error);
5556
}
5657
});
5758

5859
stream.on("error", (err) => {
5960
this.logger.error("Output stream error", err);
60-
6161
try {
62-
archive.abort();
63-
} catch {
64-
archive.end();
62+
if (archive) {
63+
archive.abort();
64+
}
65+
} catch (error) {
66+
this.logger.error("Error handling archive abort", error);
6567
}
6668
});
6769

@@ -78,7 +80,61 @@ export class LoggingService {
7880
pods = await this.getPodsFromService(service);
7981
}
8082

83+
if (pods.length === 0) {
84+
if (download && archive) {
85+
void archive.finalize();
86+
return;
87+
}
88+
stream.end();
89+
return;
90+
}
91+
8192
const podLogs: Promise<void>[] = [];
93+
let totalContainers = 0;
94+
let completedContainers = 0;
95+
let archiveFinalizePromise: Promise<void> | null = null;
96+
let archiveFinalizeResolve: (() => void) | null = null;
97+
98+
// Count total containers across all pods
99+
for (const pod of pods) {
100+
totalContainers += pod.spec.containers.length;
101+
}
102+
103+
// Set up archive finalization promise if in download mode
104+
if (download && archive) {
105+
archiveFinalizePromise = new Promise<void>((resolve) => {
106+
archiveFinalizeResolve = resolve;
107+
});
108+
109+
let resolved = false;
110+
const resolveOnce = () => {
111+
if (!resolved) {
112+
resolved = true;
113+
archiveFinalizeResolve?.();
114+
}
115+
};
116+
117+
// Resolve when archive finishes (this is the main event)
118+
archive.on("finish", resolveOnce);
119+
120+
// Also resolve on end as backup
121+
archive.on("end", resolveOnce);
122+
123+
// Resolve when stream finishes (archive should have finished by then)
124+
stream.on("finish", resolveOnce);
125+
}
126+
127+
const finalizeArchive = () => {
128+
completedContainers++;
129+
if (download && archive && completedContainers === totalContainers) {
130+
try {
131+
void archive.finalize();
132+
} catch (error) {
133+
this.logger.error("Error finalizing archive", error);
134+
archiveFinalizeResolve?.();
135+
}
136+
}
137+
};
82138

83139
for (const pod of pods) {
84140
podLogs.push(
@@ -90,14 +146,17 @@ export class LoggingService {
90146
archive,
91147
download ? undefined : tailLines,
92148
since,
149+
totalContainers,
150+
finalizeArchive,
93151
),
94152
);
95153
}
96154

97155
await Promise.all(podLogs);
98156

99-
if (pods.length === 0) {
100-
stream.end();
157+
// If in download mode, wait for archive to finish
158+
if (download && archiveFinalizePromise) {
159+
await archiveFinalizePromise;
101160
}
102161
}
103162

@@ -177,7 +236,9 @@ export class LoggingService {
177236

178237
stream.on("error", () => {
179238
podLogs?.abort();
180-
stream.end();
239+
if (!download) {
240+
stream.end();
241+
}
181242
});
182243

183244
podLogs = await logApi.log(
@@ -212,7 +273,9 @@ export class LoggingService {
212273
error,
213274
);
214275

215-
stream.end();
276+
if (!download) {
277+
stream.end();
278+
}
216279
}
217280
}
218281

@@ -227,6 +290,8 @@ export class LoggingService {
227290
start: string;
228291
until: string;
229292
},
293+
totalContainers?: number,
294+
onContainerComplete?: () => void,
230295
) {
231296
let totalAdded = 0;
232297
let streamEnded = false;
@@ -235,16 +300,20 @@ export class LoggingService {
235300
const until = since ? new Date(since.until) : undefined;
236301

237302
const endStream = () => {
238-
if (oldestTimestamp) {
239-
stream.emit(
240-
"data",
241-
JSON.stringify({
242-
oldest_timestamp: oldestTimestamp.toISOString(),
243-
}),
244-
);
303+
if (!archive) {
304+
// Only emit metadata when not in download mode
305+
if (oldestTimestamp) {
306+
stream.emit(
307+
"data",
308+
JSON.stringify({
309+
oldest_timestamp: oldestTimestamp.toISOString(),
310+
}),
311+
);
312+
}
245313
}
246314

247-
if (!streamEnded) {
315+
if (!streamEnded && !archive) {
316+
// Don't end stream directly when using archive - let archive finalize
248317
streamEnded = true;
249318
stream.end();
250319
}
@@ -257,7 +326,7 @@ export class LoggingService {
257326
logStream.on("end", async () => {
258327
++totalAdded;
259328

260-
if (totalLines < tailLines) {
329+
if (totalLines < tailLines && tailLines !== undefined) {
261330
this.logger.log(
262331
`loading more logs from service ${pod.metadata.name}...`,
263332
);
@@ -274,7 +343,10 @@ export class LoggingService {
274343
oldestTimestamp.setMinutes(oldestTimestamp.getMinutes() - 60);
275344

276345
if (oldestTimestamp < firstLogTimestamp) {
277-
endStream();
346+
if (!archive) {
347+
endStream();
348+
}
349+
onContainerComplete?.();
278350
return;
279351
}
280352

@@ -289,14 +361,17 @@ export class LoggingService {
289361
start: oldestTimestamp.toISOString(),
290362
until: until.toISOString(),
291363
},
364+
totalContainers,
365+
onContainerComplete,
292366
);
293367
return;
294368
}
295369

296-
if (archive && totalAdded == pod.spec.containers.length) {
297-
void archive.finalize();
370+
onContainerComplete?.();
371+
372+
if (!archive && totalAdded === pod.spec.containers.length) {
373+
endStream();
298374
}
299-
endStream();
300375
});
301376

302377
logStream.on("data", (chunk: Buffer) => {
@@ -349,11 +424,17 @@ export class LoggingService {
349424

350425
logStream.on("error", (error) => {
351426
this.logger.error("Log stream error", error);
352-
endStream();
427+
if (!archive) {
428+
endStream();
429+
return;
430+
}
431+
onContainerComplete?.();
353432
});
354433

355434
logStream.on("close", () => {
356-
endStream();
435+
if (!archive) {
436+
endStream();
437+
}
357438
});
358439

359440
if (archive) {

‎src/main.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,13 @@ async function bootstrap() {
7070

7171
const appConfig = configService.get<AppConfig>("app");
7272

73+
if (process.env.DEV) {
74+
app.enableCors({
75+
origin: ["http://localhost:3000", "http://0.0.0.0:3000"],
76+
credentials: true,
77+
});
78+
}
79+
7380
app.use(
7481
session({
7582
rolling: true,

‎src/system/system.controller.ts‎

Lines changed: 102 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
1-
import { Controller } from "@nestjs/common";
1+
import { Controller, Post, Req, Res } from "@nestjs/common";
2+
import { Request, Response } from "express";
23
import { SystemService } from "./system.service";
34
import { HasuraAction } from "src/hasura/hasura.controller";
45
import { Get } from "@nestjs/common";
@@ -9,6 +10,9 @@ import { HasuraEvent } from "src/hasura/hasura.controller";
910
import { HasuraEventData } from "src/hasura/types/HasuraEventData";
1011
import { settings_set_input } from "generated/schema";
1112
import { GameServerNodeService } from "src/game-server-node/game-server-node.service";
13+
import { LoggingService } from "src/k8s/logging/logging.service";
14+
import { isRoleAbove } from "src/utilities/isRoleAbove";
15+
import { PassThrough } from "stream";
1216

1317
@Controller("system")
1418
export class SystemController {
@@ -17,13 +21,110 @@ export class SystemController {
1721
private readonly hasura: HasuraService,
1822
private readonly notifications: NotificationsService,
1923
private readonly gameServerNodeService: GameServerNodeService,
24+
private readonly loggingService: LoggingService,
2025
) {}
2126

2227
@Get("healthz")
2328
public async status() {
2429
return;
2530
}
2631

32+
@Post("logs/download")
33+
public async logs(@Req() request: Request, @Res() response: Response): Promise<void> {
34+
const user = request.user;
35+
36+
if (!user || !isRoleAbove(user.role, "administrator")) {
37+
response.status(403).json({
38+
message: "Forbidden",
39+
});
40+
return;
41+
}
42+
43+
const {
44+
service,
45+
previous,
46+
tailLines,
47+
since,
48+
}: {
49+
service: string;
50+
previous?: boolean;
51+
tailLines?: number;
52+
since?: {
53+
start: string;
54+
until: string;
55+
};
56+
} = request.body;
57+
58+
if (!service) {
59+
response.status(400).json({
60+
message: "service is required",
61+
});
62+
return;
63+
}
64+
65+
const isJob =
66+
service.startsWith("cs-update:") || service.startsWith("m-");
67+
68+
const stream = new PassThrough();
69+
70+
try {
71+
const filename = `${service}-logs.zip`;
72+
73+
response.setHeader("Content-Type", "application/zip");
74+
response.setHeader(
75+
"Content-Disposition",
76+
`attachment; filename="${filename}"`,
77+
);
78+
79+
stream.pipe(response);
80+
81+
stream.on("error", (error) => {
82+
if (!response.headersSent) {
83+
response.status(500).json({
84+
message: error?.message || "Stream error",
85+
});
86+
return;
87+
}
88+
response.destroy();
89+
});
90+
91+
response.on("close", () => {
92+
if (!stream.destroyed) {
93+
stream.destroy();
94+
}
95+
});
96+
97+
await this.loggingService.getServiceLogs(
98+
service.startsWith("cs-update:")
99+
? GameServerNodeService.GET_UPDATE_JOB_NAME(
100+
service.replace("cs-update:", ""),
101+
)
102+
: service,
103+
stream,
104+
tailLines,
105+
!!previous,
106+
true,
107+
isJob,
108+
since,
109+
);
110+
111+
if (stream.readableEnded || stream.destroyed) {
112+
if (!response.writableEnded) {
113+
response.end();
114+
}
115+
}
116+
} catch (error) {
117+
if (!response.headersSent) {
118+
response.status(500).json({
119+
message: error?.body?.message || error.message || "Unable to get logs",
120+
});
121+
return;
122+
}
123+
stream.destroy();
124+
response.destroy();
125+
}
126+
}
127+
27128
@HasuraAction()
28129
public async updateServices() {
29130
await this.system.updateServices();

0 commit comments

Comments
 (0)