Various prebuild fixes (#20057)

filiptronicek · web-flow · commit 10810cee500d · 2024-07-24T15:48:33.000+02:00
* log it all * fix failing image logs prebuild updates * make sure we dispose of requests properly * minor cleanup * handle aborts gracefully * Try using a tailing library for reading supervisor task logs * Fix caching requests on Firefox * Revert "Try using a tailing library for reading supervisor task logs" This reverts commit fdc3183. * Change log order Make sure we wait for the line to be written first * Tweak abortion error message * Remove extra log line
diff --git a/components/dashboard/src/data/prebuilds/prebuild-logs-emitter.ts b/components/dashboard/src/data/prebuilds/prebuild-logs-emitter.ts
@@ -145,29 +145,40 @@ function streamPrebuildLogs(
         };
 
         let response: Response | undefined = undefined;
+        let abortController: AbortController | undefined = undefined;
         let reader: ReadableStreamDefaultReader<Uint8Array> | undefined = undefined;
         try {
+            abortController = new AbortController();
+            disposables.push({
+                dispose: async () => {
+                    abortController?.abort();
+                    await reader?.cancel();
+                },
+            });
             console.debug("fetching from streamUrl: " + streamUrl);
             response = await fetch(streamUrl, {
                 method: "GET",
-                cache: "no-cache",
+                cache: "reload",
                 credentials: "include",
-                keepalive: true,
                 headers: {
                     TE: "trailers", // necessary to receive stream status code
                 },
+                signal: abortController.signal,
                 redirect: "follow",
             });
             reader = response.body?.getReader();
             if (!reader) {
                 await retryBackoff("no reader");
                 return;
             }
-            disposables.push({ dispose: () => reader?.cancel() });
 
             const decoder = new TextDecoder("utf-8");
             let chunk = await reader.read();
             while (!chunk.done) {
+                if (disposables.disposed) {
+                    // stop reading when disposed
+                    return;
+                }
                 const msg = decoder.decode(chunk.value, { stream: true });
 
                 // In an ideal world, we'd use res.addTrailers()/response.trailer here. But despite being introduced with HTTP/1.1 in 1999, trailers are not supported by popular proxies (nginx, for example).
@@ -199,6 +210,10 @@ function streamPrebuildLogs(
                 return;
             }
         } catch (err) {
+            if (err instanceof DOMException && err.name === "AbortError") {
+                console.debug("stopped watching headless logs, not retrying: method got disposed of");
+                return;
+            }
             reader?.cancel().catch(console.debug);
             if (err.code === 400) {
                 // sth is really off, and we _should not_ retry
diff --git a/components/public-api/typescript-common/src/prebuild-utils.ts b/components/public-api/typescript-common/src/prebuild-utils.ts
@@ -4,7 +4,6 @@
  * See License.AGPL.txt in the project root for license information.
  */
 
-import { Disposable, DisposableCollection, HEADLESS_LOG_STREAM_STATUS_CODE_REGEX } from "@gitpod/gitpod-protocol";
 import { ApplicationError, ErrorCode, ErrorCodes } from "@gitpod/gitpod-protocol/lib/messaging/error";
 
 /**
@@ -44,134 +43,3 @@ export function getPrebuildErrorMessage(err: any) {
     }
     return `${PREBUILD_LOG_STREAM_ERROR}#${code}#${message}#${PREBUILD_LOG_STREAM_ERROR}`;
 }
-
-const defaultBackoffTimes = 3;
-interface Options {
-    includeCredentials: boolean;
-    maxBackoffTimes?: number;
-    onEnd?: () => void;
-}
-
-/**
- * backoff fetch prebuild logs
- * @returns a function to cancel fetching
- */
-export function onDownloadPrebuildLogsUrl(
-    streamUrl: string,
-    onLog: (message: string) => void,
-    options: Options,
-): () => void {
-    const disposables = new DisposableCollection();
-
-    // initializing non-empty here to use this as a stopping signal for the retries down below
-    disposables.push(Disposable.NULL);
-
-    // retry configuration goes here
-    const initialDelaySeconds = 1;
-    const backoffFactor = 1.2;
-    const maxBackoffSeconds = 5;
-    let delayInSeconds = initialDelaySeconds;
-    let currentBackoffTimes = 0;
-
-    const startWatchingLogs = async () => {
-        const retryBackoff = async (reason: string, err?: Error) => {
-            delayInSeconds = Math.min(delayInSeconds * backoffFactor, maxBackoffSeconds);
-            console.debug("re-trying headless-logs because: " + reason, err);
-            await new Promise((resolve) => {
-                setTimeout(resolve, delayInSeconds * 1000);
-            });
-            if (disposables.disposed) {
-                return; // and stop retrying
-            }
-            startWatchingLogs().catch(console.error);
-        };
-
-        let response: Response | undefined = undefined;
-        let reader: ReadableStreamDefaultReader<Uint8Array> | undefined = undefined;
-        try {
-            currentBackoffTimes += 1;
-            console.debug("fetching from streamUrl: " + streamUrl);
-            response = await fetch(streamUrl, {
-                method: "GET",
-                cache: "no-cache",
-                credentials: options.includeCredentials ? "include" : undefined,
-                keepalive: true,
-                headers: {
-                    TE: "trailers", // necessary to receive stream status code
-                },
-                redirect: "follow",
-            });
-            reader = response.body?.getReader();
-            if (!reader) {
-                await retryBackoff("no reader");
-                return;
-            }
-            disposables.push({ dispose: () => reader?.cancel() });
-
-            const decoder = new TextDecoder("utf-8");
-            let chunk = await reader.read();
-            while (!chunk.done) {
-                const msg = decoder.decode(chunk.value, { stream: true });
-
-                // In an ideal world, we'd use res.addTrailers()/response.trailer here. But despite being introduced with HTTP/1.1 in 1999, trailers are not supported by popular proxies (nginx, for example).
-                // So we resort to this hand-written solution:
-                const matches = msg.match(HEADLESS_LOG_STREAM_STATUS_CODE_REGEX);
-                const prebuildMatches = matchPrebuildError(msg);
-                if (matches) {
-                    if (matches.length < 2) {
-                        console.debug("error parsing log stream status code. msg: " + msg);
-                    } else {
-                        const code = parseStatusCode(matches[1]);
-                        if (code !== 200) {
-                            throw new ApplicationError(
-                                ErrorCodes.INTERNAL_SERVER_ERROR,
-                                `prebuild log download status code: ${code}`,
-                            );
-                        }
-                    }
-                } else if (prebuildMatches && prebuildMatches.code === ErrorCodes.HEADLESS_LOG_NOT_YET_AVAILABLE) {
-                    // reset backoff because this error is expected
-                    delayInSeconds = initialDelaySeconds;
-                    currentBackoffTimes = 0;
-                    throw prebuildMatches;
-                } else {
-                    onLog(msg);
-                }
-
-                chunk = await reader.read();
-            }
-        } catch (err) {
-            if (currentBackoffTimes > (options.maxBackoffTimes ?? defaultBackoffTimes)) {
-                console.debug("stopped watching headless logs, max backoff reached", err);
-                return;
-            }
-            if (err.code === 400) {
-                // sth is really off, and we _should not_ retry
-                console.debug("stopped watching headless logs", err);
-                return;
-            }
-            await retryBackoff("error while listening to stream", err);
-        } finally {
-            reader?.cancel().catch(console.debug);
-            if (options.onEnd) {
-                options.onEnd();
-            }
-        }
-    };
-    startWatchingLogs().catch(console.error);
-
-    return () => {
-        disposables.dispose();
-    };
-}
-
-function parseStatusCode(code: string | undefined): number | undefined {
-    try {
-        if (!code) {
-            return undefined;
-        }
-        return Number.parseInt(code);
-    } catch (err) {
-        return undefined;
-    }
-}
diff --git a/components/server/src/workspace/headless-log-controller.ts b/components/server/src/workspace/headless-log-controller.ts
@@ -69,15 +69,14 @@ export class HeadlessLogController {
                 const user = req.user as User; // verified by authenticateAndAuthorize
                 await runWithSubjectId(SubjectId.fromUserId(user.id), async () => {
                     try {
-                        const instanceId = req.params.instanceId;
-                        const terminalId = req.params.terminalId;
+                        const { instanceId, terminalId } = req.params;
 
                         const logCtx = { userId: user.id, instanceId };
                         try {
                             const head = {
                                 "Content-Type": "text/html; charset=utf-8", // is text/plain, but with that node.js won't stream...
                                 "Transfer-Encoding": "chunked",
-                                "Cache-Control": "no-cache, no-store, must-revalidate", // make sure stream are not re-used on reconnect
+                                "Cache-Control": "no-cache, no-store, must-revalidate", // make sure streams are not re-used on reconnect
                             };
                             res.writeHead(200, head);
 
@@ -291,12 +290,11 @@ export class HeadlessLogController {
                         return;
                     }
 
-                    const prebuildId = req.params.prebuildId;
+                    const { prebuildId, taskId } = req.params;
                     if (!uuidValidate(prebuildId)) {
                         res.status(400).send("prebuildId is invalid");
                         return;
                     }
-                    const { taskId } = req.params;
                     const logCtx = { userId: user.id, prebuildId, taskId };
 
                     let firstChunk = true;
diff --git a/components/server/src/workspace/workspace-starter.ts b/components/server/src/workspace/workspace-starter.ts
@@ -851,7 +851,7 @@ export class WorkspaceStarter {
         try {
             if (workspace.type === "prebuild") {
                 const prebuild = await this.workspaceDb.trace({ span }).findPrebuildByWorkspaceID(workspace.id);
-                if (prebuild && prebuild.state !== "failed") {
+                if (prebuild && prebuild.state !== "failed" && prebuild.projectId) {
                     prebuild.state = "failed";
                     prebuild.error = err.toString();
 
@@ -860,6 +860,12 @@ export class WorkspaceStarter {
                         type: HeadlessWorkspaceEventType.Failed,
                         workspaceID: workspace.id,
                     });
+                    await this.publisher.publishPrebuildUpdate({
+                        status: "failed",
+                        prebuildID: prebuild.id,
+                        projectID: prebuild.projectId,
+                        workspaceID: workspace.id,
+                    });
                 }
             }
         } catch (err) {
diff --git a/components/supervisor/pkg/supervisor/services.go b/components/supervisor/pkg/supervisor/services.go
@@ -1133,6 +1133,7 @@ func (s *taskService) ListenToOutput(req *api.ListenToOutputRequest, srv api.Tas
 		if err == io.EOF {
 			if isClosed.Load() {
 				// We are done
+				log.Println("[prebuildlog]: closing because task state is done and we are done reading")
 				return nil
 			}
 		}
@@ -1149,8 +1150,10 @@ func (s *taskService) ListenToOutput(req *api.ListenToOutputRequest, srv api.Tas
 
 		select {
 		case <-srv.Context().Done():
+			log.Println("[prebuildlog]: closing because context done")
 			return nil
 		case <-s.willShutdownCtx.Done():
+			log.Println("[prebuildlog]: closing because willShutdownCtx fired")
 			return nil
 		case <-closedChannel:
 			continue
diff --git a/components/supervisor/pkg/supervisor/tasks.go b/components/supervisor/pkg/supervisor/tasks.go
@@ -323,8 +323,13 @@ func (tm *tasksManager) Run(ctx context.Context, wg *sync.WaitGroup, successChan
 			return true
 		})
 
+		taskWatchWg := &sync.WaitGroup{}
+
 		go func(t *task, term *terminal.Term) {
 			state, err := term.Wait()
+			taskLog.Info("task terminal has been closed. Waiting for watch() to finish...")
+			taskWatchWg.Wait()
+			taskLog.Info("watch() has finished, setting task state to closed")
 			if state != nil {
 				if state.Success() {
 					t.successChan <- taskSuccessful
@@ -341,11 +346,10 @@ func (tm *tasksManager) Run(ctx context.Context, wg *sync.WaitGroup, successChan
 
 				t.successChan <- taskFailed(fmt.Sprintf("%s: %s", msg, t.lastOutput))
 			}
-			taskLog.Info("task terminal has been closed")
 			tm.setTaskState(t, api.TaskState_closed)
 		}(t, term)
 
-		tm.watch(t, term)
+		tm.watch(t, term, taskWatchWg)
 
 		if t.command != "" {
 			term.PTY.Write([]byte(t.command + "\n"))
@@ -446,7 +450,7 @@ func prebuildLogFileName(task *task, storeLocation string) string {
 	return logs.PrebuildLogFileName(storeLocation, task.Id)
 }
 
-func (tm *tasksManager) watch(task *task, term *terminal.Term) {
+func (tm *tasksManager) watch(task *task, term *terminal.Term, wg *sync.WaitGroup) {
 	if !tm.config.isPrebuild() {
 		return
 	}
@@ -462,6 +466,9 @@ func (tm *tasksManager) watch(task *task, term *terminal.Term) {
 	go func() {
 		defer stdout.Close()
 
+		wg.Add(1)
+		defer wg.Done()
+
 		var (
 			fileName    = prebuildLogFileName(task, tm.storeLocation)
 			oldFileName = fileName + "-old"
