fix: terminate Docker logs process when client disconnects (#9091)

* Update container.go

* Update container.go

agent/app/service/container.go

@@ -880,84 +880,116 @@ func (u *ContainerService) ContainerLogClean(req dto.OperationWithName) error {
 }
 
 func (u *ContainerService) StreamLogs(ctx *gin.Context, params dto.StreamLog) {
-	messageChan := make(chan string, 1024)
+    messageChan := make(chan string, 1024)
-	errorChan := make(chan error, 1)
+    errorChan := make(chan error, 1)
-
+    doneChan := make(chan struct{})
-	go collectLogs(params, messageChan, errorChan)
+    
-
+    // 监听客户端连接状态
-	ctx.Stream(func(w io.Writer) bool {
+    go func() {
-		select {
+        <-ctx.Request.Context().Done()
-		case msg, ok := <-messageChan:
+        close(doneChan) // 通知 collectLogs 停止
-			if !ok {
+    }()
-				return msg == ""
+    // 启动日志收集协程
-			}
+    go collectLogs(doneChan, params, messageChan, errorChan)
-			_, err := fmt.Fprintf(w, "data: %v\n\n", msg)
+    // 流式发送日志到客户端
-			if err != nil {
+    ctx.Stream(func(w io.Writer) bool {
-				return false
+        select {
-			}
+        case msg, ok := <-messageChan:
-			return true
+            if !ok {
-		case err := <-errorChan:
+                return false
-			if err != nil {
+            }
-				_, _ = fmt.Fprintf(w, "event: error\ndata: %v\n\n", err.Error())
+            _, err := fmt.Fprintf(w, "data: %s\n\n", msg)
-			}
+            if err != nil {
-			return true
+                return false
-		case <-ctx.Request.Context().Done():
+            }
-			return false
+            return true
-		}
+        case err := <-errorChan:
-	})
+            if err != nil {
+                _, _ = fmt.Fprintf(w, "event: error\ndata: %v\n\n", err.Error())
+            }
+            return false
+        }
+    })
 }
 
-func collectLogs(params dto.StreamLog, messageChan chan<- string, errorChan chan<- error) {
+func collectLogs(
-	defer close(messageChan)
+    done <-chan struct{},
-	defer close(errorChan)
+    params dto.StreamLog,
-
+    messageChan chan<- string,
-	var cmdArgs []string
+    errorChan chan<- error,
-	if params.Type == "compose" {
+) {
-		cmdArgs = []string{"compose", "-f", params.Compose}
+    defer close(messageChan)
-	}
+    defer close(errorChan)
-	cmdArgs = append(cmdArgs, "logs")
+    var cmdArgs []string
-	if params.Follow {
+    if params.Type == "compose" {
-		cmdArgs = append(cmdArgs, "-f")
+        cmdArgs = []string{"compose", "-f", params.Compose}
-	}
+    }
-	if params.Tail != "0" {
+    cmdArgs = append(cmdArgs, "logs")
-		cmdArgs = append(cmdArgs, "--tail", params.Tail)
+    if params.Follow {
-	}
+        cmdArgs = append(cmdArgs, "-f")
-	if params.Since != "all" {
+    }
-		cmdArgs = append(cmdArgs, "--since", params.Since)
+    if params.Tail != "0" {
-	}
+        cmdArgs = append(cmdArgs, "--tail", params.Tail)
-	if params.Container != "" {
+    }
-		cmdArgs = append(cmdArgs, params.Container)
+    if params.Since != "all" {
-	}
+        cmdArgs = append(cmdArgs, "--since", params.Since)
-	dockerCmd := exec.Command("docker", cmdArgs...)
+    }
-
+    if params.Container != "" {
-	stdout, err := dockerCmd.StdoutPipe()
+        cmdArgs = append(cmdArgs, params.Container)
-	if err != nil {
+    }
-		errorChan <- fmt.Errorf("failed to get stdout pipe: %v", err)
+    
-		return
+    dockerCmd := exec.Command("docker", cmdArgs...)
-	}
+    
-	dockerCmd.Stderr = dockerCmd.Stdout
+    stdout, err := dockerCmd.StdoutPipe()
-	if err = dockerCmd.Start(); err != nil {
+    if err != nil {
-		errorChan <- fmt.Errorf("failed to start command: %v", err)
+        errorChan <- fmt.Errorf("failed to get stdout pipe: %v", err)
-		return
+        return
-	}
+    }
-	scanner := bufio.NewScanner(stdout)
+    
-	for scanner.Scan() {
+    dockerCmd.Stderr = dockerCmd.Stdout
-		message := scanner.Text()
+    
-		select {
+    if err = dockerCmd.Start(); err != nil {
-		case messageChan <- message:
+        errorChan <- fmt.Errorf("failed to start docker logs command: %v", err)
-		case <-time.After(5 * time.Second):
+        return
-			errorChan <- fmt.Errorf("message channel blocked")
+    }
-			return
+    
-		}
+    // 确保在函数退出时清理进程
-	}
+    defer func() {
-	if err = scanner.Err(); err != nil {
+        if dockerCmd.Process != nil {
-		errorChan <- fmt.Errorf("scanner error: %v", err)
+            _ = dockerCmd.Process.Kill()
-		return
+        }
-	}
+    }()
-	if err = dockerCmd.Wait(); err != nil {
+    
-		errorChan <- fmt.Errorf("%v", err)
+    // 创建一个scanner来读取stdout
-		return
+    scanner := bufio.NewScanner(stdout)
-	}
+    
+    // 启动一个goroutine监听done信号
+    processKilled := false
+    go func() {
+        <-done
+        if !processKilled && dockerCmd.Process != nil {
+            processKilled = true
+            _ = dockerCmd.Process.Kill()
+        }
+    }()
+    
+    // 读取日志输出
+    for scanner.Scan() {
+        message := scanner.Text()
+        select {
+        case messageChan <- message:
+            // 消息发送成功
+        case <-done:
+            return
+        }
+    }
+    
+    if err = scanner.Err(); err != nil && err != io.EOF {
+        errorChan <- fmt.Errorf("scanner error: %v", err)
+    }
+    
+    // 等待命令完成
+    _ = dockerCmd.Wait()
 }
 
 func (u *ContainerService) DownloadContainerLogs(containerType, container, since, tail string, c *gin.Context) error {