fix: fixes swallowed error when bad headers are found and fixing #2416 (#2423)

* fix: fixes swallowed error when bad headers are found and fixing #2416

* minor clean up

* fixes broken on first connect

* fixes on reconnect

assets/components/LogViewer/LogEventSource.vue

@@ -6,8 +6,7 @@
 <script lang="ts" setup>
 const loadingMore = defineEmit<[value: boolean]>();
 
-const { container, streamConfig } = useContainerContext();
-const { messages, loadOlderLogs } = useLogStream(container, streamConfig);
+const { messages, loadOlderLogs } = useLogStream();
 
 const beforeLoading = () => loadingMore(true);
 const afterLoading = () => loadingMore(false);

assets/composables/eventsource.ts

@@ -9,7 +9,6 @@ import {
   DockerEventLogEntry,
   SkippedLogsEntry,
 } from "@/models/LogEntry";
-import { Container } from "@/models/Container";
 
 function parseMessage(data: string): LogEntry<string | JSONObject> {
   const e = JSON.parse(data, (key, value) => {
@@ -21,12 +20,8 @@ function parseMessage(data: string): LogEntry<string | JSONObject> {
   return asLogEntry(e);
 }
 
-type LogStreamConfig = {
-  stdout: boolean;
-  stderr: boolean;
-};
-
-export function useLogStream(container: Ref<Container>, streamConfig: LogStreamConfig) {
+export function useLogStream() {
+  const { container, streamConfig } = useContainerContext();
   let messages: LogEntry<string | JSONObject>[] = $ref([]);
   let buffer: LogEntry<string | JSONObject>[] = $ref([]);
   const scrollingPaused = $ref(inject("scrollingPaused") as Ref<boolean>);
@@ -74,12 +69,12 @@ export function useLogStream(container: Ref<Container>, streamConfig: LogStreamC
     console.debug(`Clearing messages for ${containerId}`);
   }
 
-  function connect({ clear } = { clear: true }) {
+  watchEffect(() => {
     close();
-
-    if (clear) {
+    if (containerId != container.value.id) {
       clearMessages();
     }
+    containerId = container.value.id;
 
     const params = {} as { stdout?: string; stderr?: string };
 
@@ -89,8 +84,6 @@ export function useLogStream(container: Ref<Container>, streamConfig: LogStreamC
     if (streamConfig.stderr) {
       params.stderr = "1";
     }
-    containerId = container.value.id;
-
     console.debug(`Connecting to ${containerId} with params`, params);
 
     es = new EventSource(
@@ -103,17 +96,14 @@ export function useLogStream(container: Ref<Container>, streamConfig: LogStreamC
       flushBuffer();
       flushBuffer.flush();
     });
-    es.onerror = (e) => {
-      console.error(`Unexpected error for eventsource container-id:${containerId}. Clearing logs and reconnecting.`);
-      clearMessages();
-    };
     es.onmessage = (e) => {
       if (e.data) {
         buffer.push(parseMessage(e.data));
         flushBuffer();
       }
     };
-  }
+    es.onerror = () => clearMessages();
+  });
 
   async function loadOlderLogs({ beforeLoading, afterLoading } = { beforeLoading: () => {}, afterLoading: () => {} }) {
     if (messages.length < 300) return;
@@ -157,20 +147,11 @@ export function useLogStream(container: Ref<Container>, streamConfig: LogStreamC
       console.log("LogEventSource: container changed", newValue, oldValue);
       if (newValue == "running" && newValue != oldValue) {
         buffer.push(new DockerEventLogEntry("Container started", new Date(), "container-started"));
-        connect({ clear: false });
       }
     },
   );
 
   onUnmounted(() => close());
 
-  watch(
-    () => container.value.id,
-    () => connect(),
-    { immediate: true },
-  );
-
-  watch(streamConfig, () => connect());
-
   return { ...$$({ messages }), loadOlderLogs };
 }

docker/event_generator.go

@@ -31,7 +31,7 @@ var bufPool = sync.Pool{
 	},
 }
 
-var BadHeaderErr = fmt.Errorf("dozzle/docker: unable to read header")
+var ErrBadHeader = fmt.Errorf("dozzle/docker: unable to read header")
 
 func NewEventGenerator(reader io.Reader, tty bool) *EventGenerator {
 	generator := &EventGenerator{
@@ -83,11 +83,12 @@ func (g *EventGenerator) consumeReader() {
 		}
 
 		if readerError != nil {
-			if readerError != BadHeaderErr {
+			log.Debugf("reader error: %v", readerError)
+			if readerError != ErrBadHeader {
 				g.Errors <- readerError
+				close(g.buffer)
+				break
 			}
-			close(g.buffer)
-			break
 		}
 	}
 	g.wg.Done()
@@ -124,7 +125,7 @@ func readEvent(reader *bufio.Reader, tty bool) (string, StdType, error) {
 			return "", streamType, err
 		}
 		if n != 8 {
-			return "", streamType, BadHeaderErr
+			return "", streamType, ErrBadHeader
 		}
 
 		switch header[0] {
@@ -133,7 +134,7 @@ func readEvent(reader *bufio.Reader, tty bool) (string, StdType, error) {
 		case 2:
 			streamType = STDERR
 		default:
-			log.Warnf("unknown stream type %d", header[0])
+			return "", streamType, ErrBadHeader
 		}
 
 		count := binary.BigEndian.Uint32(header[4:])