fix: improve command output handling (#12837)

agent/app/service/backup_redis.go

@@ -97,7 +97,8 @@ func handleRedisBackup(redisInfo *repo.RootInfo, parentTask *task.Task, recordID
 			}
 		}
 
-		if err := cmd.NewCommandMgr().Run("docker", "exec", redisInfo.ContainerName, "redis-cli", "-a", redisInfo.Password, "--no-auth-warning", "save"); err != nil {
+		cmdMgr := cmd.NewCommandMgr(cmd.WithTimeout(30 * time.Minute))
+		if err := cmdMgr.Run("docker", "exec", redisInfo.ContainerName, "redis-cli", "-a", redisInfo.Password, "--no-auth-warning", "save"); err != nil {
 			return err
 		}
 
@@ -109,13 +110,13 @@ func handleRedisBackup(redisInfo *repo.RootInfo, parentTask *task.Task, recordID
 			return nil
 		}
 		if strings.HasSuffix(fileName, ".aof") {
-			if err := cmd.NewCommandMgr().Run("docker", "cp", redisInfo.ContainerName+":/data/appendonly.aof", path.Join(backupDir, fileName)); err != nil {
+			if err := cmdMgr.Run("docker", "cp", redisInfo.ContainerName+":/data/appendonly.aof", path.Join(backupDir, fileName)); err != nil {
 				return err
 			}
 			return nil
 		}
 
-		if err := cmd.NewCommandMgr().Run("docker", "cp", redisInfo.ContainerName+":/data/dump.rdb", path.Join(backupDir, fileName)); err != nil {
+		if err := cmdMgr.Run("docker", "cp", redisInfo.ContainerName+":/data/dump.rdb", path.Join(backupDir, fileName)); err != nil {
 			return err
 		}
 		return nil

agent/app/service/snapshot_create.go

@@ -406,8 +406,8 @@ func snapAppImage(snap snapHelper, req dto.SnapshotCreate, targetDir string) err
 		snap.Task.Log(strings.Join(imageList, " "))
 		snap.Task.Logf("docker save %s | gzip -c > %s", strings.Join(imageList, " "), path.Join(targetDir, "images.tar.gz"))
 		outputPath := path.Join(targetDir, "images.tar.gz")
-		cmdMgr := cmd.NewCommandMgr(cmd.WithOutputFile(outputPath))
-		if _, err := cmdMgr.RunPipe(
+		cmdMgr := cmd.NewCommandMgr()
+		if _, err := cmdMgr.RunPipeToFile(outputPath,
 			cmd.PipeCommand{Name: "docker", Args: append([]string{"save"}, imageList...)},
 			cmd.PipeCommand{Name: "gzip", Args: []string{"-c"}},
 		); err != nil {

agent/utils/cmd/cmdx.go

@@ -117,35 +117,27 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 		return "", nil
 	}
 
-	ctx, cancel := c.pipeContext()
+	ctx, cancel, cmds := c.preparePipeCommands(commands)
 	if cancel != nil {
 		defer cancel()
 	}
 
-	cmds := c.buildPipeCommands(ctx, commands)
 	customWriter := &CustomWriter{taskItem: c.taskItem}
 	var outputFile *os.File
-	limitOutputCapture := c.taskItem != nil || c.logger != nil || len(c.outputFile) != 0
 	stdout, stderr := &lockedBuffer{}, &lockedBuffer{}
+	limitOutputCapture := c.taskItem != nil || c.logger != nil || len(c.outputFile) != 0
 	if limitOutputCapture {
 		stdout.limit = maxStreamOutputCapture
 		stderr.limit = maxStreamOutputCapture
 	}
-	if commands[0].Stdin != nil {
-		cmds[0].Stdin = commands[0].Stdin
-	}
 	var pipeStderr io.Writer = stderr
 	var lastStdout io.Writer = stdout
 	var lastStderr io.Writer = stderr
 	var streamWriter io.Writer
-	var streamClosers []io.Closer
 	if c.taskItem != nil {
 		streamWriter = customWriter
 	} else if c.logger != nil {
 		streamWriter = c.logger.Writer()
-		if closer, ok := streamWriter.(io.Closer); ok {
-			streamClosers = append(streamClosers, closer)
-		}
 	} else if len(c.outputFile) != 0 {
 		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
 		if err != nil {
@@ -163,7 +155,7 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 		if c.taskItem != nil {
 			customWriter.Flush()
 		}
-		for _, closer := range streamClosers {
+		if closer, ok := streamWriter.(io.Closer); ok {
 			_ = closer.Close()
 		}
 		if outputFile != nil {
@@ -173,24 +165,67 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 	if err := connectPipeCommands(cmds, lastStdout, lastStderr, pipeStderr); err != nil {
 		return "", err
 	}
-
 	if err := startPipeCommands(cmds); err != nil {
 		return handleErrString(stdout.String(), stderr.String(), c.IgnoreExist1, err)
 	}
 
-	runErr := waitPipeCommands(ctx, cmds)
-	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
-		return "", buserr.New("ErrCmdTimeout")
-	}
-	if errors.Is(ctx.Err(), context.Canceled) {
-		return "", buserr.New("ErrShutDown")
-	}
+	runErr := c.pipeResultErr(ctx, waitPipeCommands(ctx, cmds))
 	if runErr != nil {
 		return handleErrString(stdout.String(), stderr.String(), c.IgnoreExist1, runErr)
 	}
 	return stdout.String(), nil
 }
 
+func (c *CommandHelper) RunPipeToFile(outputFile string, commands ...PipeCommand) (string, error) {
+	if len(commands) == 0 {
+		return "", nil
+	}
+
+	ctx, cancel, cmds := c.preparePipeCommands(commands)
+	if cancel != nil {
+		defer cancel()
+	}
+
+	file, err := os.OpenFile(outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = file.Close() }()
+
+	stderr := &lockedBuffer{limit: maxStreamOutputCapture}
+	if err := connectPipeCommands(cmds, file, stderr, stderr); err != nil {
+		return "", err
+	}
+	if err := startPipeCommands(cmds); err != nil {
+		return handleErrString("", stderr.String(), c.IgnoreExist1, err)
+	}
+
+	runErr := c.pipeResultErr(ctx, waitPipeCommands(ctx, cmds))
+	if runErr != nil {
+		return handleErrString("", stderr.String(), c.IgnoreExist1, runErr)
+	}
+	return "", nil
+}
+
+func (c *CommandHelper) preparePipeCommands(commands []PipeCommand) (context.Context, context.CancelFunc, []*exec.Cmd) {
+	ctx, cancel := c.pipeContext()
+	cmds := c.buildPipeCommands(ctx, commands)
+	if commands[0].Stdin != nil {
+		cmds[0].Stdin = commands[0].Stdin
+	}
+	return ctx, cancel, cmds
+}
+
+func (c *CommandHelper) pipeResultErr(ctx context.Context, runErr error) error {
+	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+		return buserr.New("ErrCmdTimeout")
+	}
+	if errors.Is(ctx.Err(), context.Canceled) {
+		return buserr.New("ErrShutDown")
+	}
+	return runErr
+}
+
 func (c *CommandHelper) pipeContext() (context.Context, context.CancelFunc) {
 	ctx := c.context
 	if ctx == nil {
@@ -299,29 +334,25 @@ func (c *CommandHelper) run(name string, arg ...string) (string, error) {
 
 	customWriter := &CustomWriter{taskItem: c.taskItem}
 	var stdout, stderr bytes.Buffer
-	var loggerClosers []io.Closer
+	var loggerCloser io.Closer
 	if c.taskItem != nil {
-		cmd.Stdout = customWriter
-		cmd.Stderr = customWriter
+		cmd.Stdout = io.MultiWriter(&stdout, customWriter)
+		cmd.Stderr = io.MultiWriter(&stderr, customWriter)
 	} else if c.logger != nil {
-		stdoutWriter := c.logger.Writer()
-		stderrWriter := c.logger.Writer()
-		if closer, ok := stdoutWriter.(io.Closer); ok {
-			loggerClosers = append(loggerClosers, closer)
+		streamWriter := c.logger.Writer()
+		if closer, ok := streamWriter.(io.Closer); ok {
+			loggerCloser = closer
 		}
-		if closer, ok := stderrWriter.(io.Closer); ok {
-			loggerClosers = append(loggerClosers, closer)
-		}
-		cmd.Stdout = stdoutWriter
-		cmd.Stderr = stderrWriter
+		cmd.Stdout = io.MultiWriter(&stdout, streamWriter)
+		cmd.Stderr = io.MultiWriter(&stderr, streamWriter)
 	} else if len(c.outputFile) != 0 {
-		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE, constant.FilePerm)
+		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
 		if err != nil {
 			return "", err
 		}
 		outputFile = file
-		cmd.Stdout = outputFile
-		cmd.Stderr = outputFile
+		cmd.Stdout = io.MultiWriter(&stdout, outputFile)
+		cmd.Stderr = io.MultiWriter(&stderr, outputFile)
 	} else {
 		cmd.Stdout = &stdout
 		cmd.Stderr = &stderr
@@ -336,8 +367,8 @@ func (c *CommandHelper) run(name string, arg ...string) (string, error) {
 		cmd.Stdin = c.stdin
 	}
 	defer func() {
-		for _, closer := range loggerClosers {
-			_ = closer.Close()
+		if loggerCloser != nil {
+			_ = loggerCloser.Close()
 		}
 		if outputFile != nil {
 			_ = outputFile.Close()
@@ -358,7 +389,7 @@ func (c *CommandHelper) run(name string, arg ...string) (string, error) {
 	select {
 	case err := <-done:
 		if err != nil {
-			return handleErr(stdout, stderr, c.IgnoreExist1, err)
+			return handleErr(&stdout, &stderr, c.IgnoreExist1, err)
 		}
 		return stdout.String(), nil
 	case <-contextDone(newContext):
@@ -456,11 +487,14 @@ func WithIgnoreExist1() Option {
 }
 
 type CustomWriter struct {
+	mu       sync.Mutex
 	taskItem *task.Task
 	buffer   bytes.Buffer
 }
 
 func (cw *CustomWriter) Write(p []byte) (n int, err error) {
+	cw.mu.Lock()
+	defer cw.mu.Unlock()
 	cw.buffer.Write(p)
 	lines := strings.Split(cw.buffer.String(), "\n")
 
@@ -473,13 +507,15 @@ func (cw *CustomWriter) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 func (cw *CustomWriter) Flush() {
+	cw.mu.Lock()
+	defer cw.mu.Unlock()
 	if cw.buffer.Len() > 0 {
 		cw.taskItem.Log(cw.buffer.String())
 		cw.buffer.Reset()
 	}
 }
 
-func handleErr(stdout, stderr bytes.Buffer, ignoreExist1 bool, err error) (string, error) {
+func handleErr(stdout, stderr fmt.Stringer, ignoreExist1 bool, err error) (string, error) {
 	return handleErrString(stdout.String(), stderr.String(), ignoreExist1, err)
 }
 

agent/utils/compose/compose.go

@@ -43,7 +43,7 @@ func Up(filePath string) (string, error) {
 		return "", err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "up", "-d")
 	return cmd.NewCommandMgr(cmd.WithTimeout(20*time.Minute)).RunWithStdout(base, args...)
 }
@@ -53,9 +53,9 @@ func UpWithTask(filePath string, task *task.Task, forcePull bool) error {
 		return err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "up", "-d")
-	return cmd.NewCommandMgr(cmd.WithTask(*task)).Run(base, args...)
+	return cmd.NewCommandMgr(cmd.WithTask(*task), cmd.WithTimeout(20*time.Minute)).Run(base, args...)
 }
 
 func pullComposeImages(filePath string, forcePull bool, task *task.Task) error {
@@ -139,7 +139,7 @@ func getComposeImagesByCommand(filePath string) ([]string, error) {
 		return nil, err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "config", "--format", "json", "--no-normalize")
 	stdout, err := cmd.NewCommandMgr(cmd.WithTimeout(5*time.Minute)).
 		RunWithStdout(base, args...)
@@ -180,7 +180,7 @@ func Down(filePath string) (string, error) {
 		return "", err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "down", "--remove-orphans")
 	return cmd.NewCommandMgr(cmd.WithTimeout(20*time.Minute)).RunWithStdout(base, args...)
 }
@@ -190,7 +190,7 @@ func Stop(filePath string) (string, error) {
 		return "", err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "stop")
 	return cmd.NewCommandMgr(cmd.WithTimeout(20*time.Minute)).RunWithStdout(base, args...)
 }
@@ -200,7 +200,7 @@ func Restart(filePath string) (string, error) {
 		return "", err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, "restart")
 	return cmd.NewCommandMgr(cmd.WithTimeout(20*time.Minute)).RunWithStdout(base, args...)
 }
@@ -210,7 +210,7 @@ func Operate(filePath, operation string) (string, error) {
 		return "", err
 	}
 	base, extra := getComposeBaseCmd()
-	args := append(extra, strings.Fields(loadFiles(filePath))...)
+	args := append(extra, loadFiles(filePath)...)
 	args = append(args, operation)
 	return cmd.NewCommandMgr(cmd.WithTimeout(20*time.Minute)).RunWithStdout(base, args...)
 }
@@ -221,24 +221,24 @@ func DownAndUp(filePath string) (string, error) {
 	}
 	cmdMgr := cmd.NewCommandMgr(cmd.WithTimeout(20 * time.Minute))
 	base, extra := getComposeBaseCmd()
-	argsDown := append(extra, strings.Fields(loadFiles(filePath))...)
+	argsDown := append(extra, loadFiles(filePath)...)
 	argsDown = append(argsDown, "down")
 	stdout, err := cmdMgr.RunWithStdout(base, argsDown...)
 	if err != nil {
 		return stdout, err
 	}
-	argsUp := append(extra, strings.Fields(loadFiles(filePath))...)
+	argsUp := append(extra, loadFiles(filePath)...)
 	argsUp = append(argsUp, "up", "-d")
 	stdout, err = cmdMgr.RunWithStdout(base, argsUp...)
 	return stdout, err
 }
 
-func loadFiles(filePath string) string {
+func loadFiles(filePath string) []string {
 	var fileItem []string
 	for _, item := range strings.Split(filePath, ",") {
 		if len(item) != 0 {
-			fileItem = append(fileItem, fmt.Sprintf("-f %s", item))
+			fileItem = append(fileItem, "-f", item)
 		}
 	}
-	return strings.Join(fileItem, " ")
+	return fileItem
 }

agent/utils/files/file_op.go

@@ -548,14 +548,14 @@ func (f FileOp) Cut(oldPaths []string, dst, name string, cover bool) error {
 	}
 	args = append(args, oldPaths...)
 	args = append(args, dstPath)
-	if err := cmd.NewCommandMgr().Run("mv", args...); err != nil {
+	if err := cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("mv", args...); err != nil {
 		return err
 	}
 	return nil
 }
 
 func (f FileOp) Mv(oldPath, dstPath string) error {
-	if err := cmd.NewCommandMgr().Run("mv", oldPath, dstPath); err != nil {
+	if err := cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("mv", oldPath, dstPath); err != nil {
 		return err
 	}
 	return nil
@@ -611,22 +611,22 @@ func (f FileOp) CopyAndReName(src, dst, name string, cover bool) error {
 		if name != "" && !cover {
 			dstPath = filepath.Join(dst, name)
 		}
-		return cmd.NewCommandMgr().Run("cp", "-rfp", src, dstPath)
+		return cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("cp", "-rfp", src, dstPath)
 	} else {
 		dstPath := filepath.Join(dst, name)
 		if cover {
 			dstPath = dst
 		}
-		return cmd.NewCommandMgr().Run("cp", "-fp", src, dstPath)
+		return cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("cp", "-fp", src, dstPath)
 	}
 }
 
 func (f FileOp) CopyDirWithNewName(src, dst, newName string) error {
 	if newName == "." || newName == "" {
-		return cmd.NewCommandMgr().Run("cp", "-rfp", filepath.Clean(src)+"/.", dst)
+		return cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("cp", "-rfp", filepath.Clean(src)+"/.", dst)
 	}
 	dstDir := filepath.Join(dst, newName)
-	return cmd.NewCommandMgr().Run("cp", "-rfp", src, dstDir)
+	return cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("cp", "-rfp", src, dstDir)
 }
 
 func (f FileOp) CopyDir(src, dst string) error {

agent/utils/mysql/client/local.go

@@ -245,8 +245,8 @@ func (r *Local) Backup(info BackupInfo) error {
 		args = append(args, arg)
 	}
 	args = append(args, info.Name)
-	cmdMgr := cmd.NewCommandMgr(cmd.WithOutputFile(path.Join(info.TargetDir, info.FileName)))
-	if _, err := cmdMgr.RunPipe(
+	cmdMgr := cmd.NewCommandMgr()
+	if _, err := cmdMgr.RunPipeToFile(path.Join(info.TargetDir, info.FileName),
 		cmd.PipeCommand{Name: "docker", Args: args},
 		cmd.PipeCommand{Name: "gzip", Args: []string{"-cf"}},
 	); err != nil {

agent/utils/mysql/client/remote.go

@@ -281,8 +281,8 @@ func (r *Remote) Backup(info BackupInfo) error {
 		}
 	}
 	global.LOG.Debug("docker " + strings.Join(debugArgs, " "))
-	cmdMgr := cmd.NewCommandMgr(cmd.WithOutputFile(path.Join(info.TargetDir, info.FileName)))
-	if _, err := cmdMgr.RunPipe(
+	cmdMgr := cmd.NewCommandMgr()
+	if _, err := cmdMgr.RunPipeToFile(path.Join(info.TargetDir, info.FileName),
 		cmd.PipeCommand{Name: "docker", Args: backupArgs},
 		cmd.PipeCommand{Name: "gzip", Args: []string{"-cf"}},
 	); err != nil {

agent/utils/postgresql/client/local.go

@@ -136,8 +136,8 @@ func (r *Local) Backup(info BackupInfo) error {
 	}
 	global.LOG.Infof("start to pg_dump | gzip > %s.gzip", info.TargetDir+"/"+info.FileName)
 
-	cmdMgr := cmd.NewCommandMgr(cmd.WithOutputFile(path.Join(info.TargetDir, info.FileName)))
-	if _, err := cmdMgr.RunPipe(
+	cmdMgr := cmd.NewCommandMgr()
+	if _, err := cmdMgr.RunPipeToFile(path.Join(info.TargetDir, info.FileName),
 		cmd.PipeCommand{Name: "docker", Args: []string{"exec", "-i", "-e", "PGPASSWORD=" + r.Password, r.ContainerName, "pg_dump", "-F", "c", "-U", r.Username, "-d", info.Name}},
 		cmd.PipeCommand{Name: "gzip", Args: []string{"-cf"}},
 	); err != nil {

core/app/service/logs.go

@@ -127,7 +127,7 @@ func runRemoteShellScript(url string, args ...string) error {
 	if statusCode < http.StatusOK || statusCode >= http.StatusMultipleChoices {
 		return fmt.Errorf("download script failed, status code: %d", statusCode)
 	}
-	_, err = cmd.NewCommandMgr(cmd.WithOutputFile(os.DevNull)).RunPipe(cmd.PipeCommand{
+	_, err = cmd.NewCommandMgr().RunPipeToFile(os.DevNull, cmd.PipeCommand{
 		Name:  "sh",
 		Args:  append([]string{"-s"}, args...),
 		Stdin: bytes.NewReader(script),

core/utils/cmd/cmdx.go

@@ -96,35 +96,27 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 		return "", nil
 	}
 
-	ctx, cancel := c.pipeContext()
+	ctx, cancel, cmds := c.preparePipeCommands(commands)
 	if cancel != nil {
 		defer cancel()
 	}
 
-	cmds := c.buildPipeCommands(ctx, commands)
 	customWriter := &CustomWriter{taskItem: c.taskItem}
 	var outputFile *os.File
-	limitOutputCapture := c.taskItem != nil || c.logger != nil || len(c.outputFile) != 0
 	stdout, stderr := &lockedBuffer{}, &lockedBuffer{}
+	limitOutputCapture := c.taskItem != nil || c.logger != nil || len(c.outputFile) != 0
 	if limitOutputCapture {
 		stdout.limit = maxStreamOutputCapture
 		stderr.limit = maxStreamOutputCapture
 	}
-	if commands[0].Stdin != nil {
-		cmds[0].Stdin = commands[0].Stdin
-	}
 	var pipeStderr io.Writer = stderr
 	var lastStdout io.Writer = stdout
 	var lastStderr io.Writer = stderr
 	var streamWriter io.Writer
-	var streamClosers []io.Closer
 	if c.taskItem != nil {
 		streamWriter = customWriter
 	} else if c.logger != nil {
 		streamWriter = c.logger.Writer()
-		if closer, ok := streamWriter.(io.Closer); ok {
-			streamClosers = append(streamClosers, closer)
-		}
 	} else if len(c.outputFile) != 0 {
 		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
 		if err != nil {
@@ -142,7 +134,7 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 		if c.taskItem != nil {
 			customWriter.Flush()
 		}
-		for _, closer := range streamClosers {
+		if closer, ok := streamWriter.(io.Closer); ok {
 			_ = closer.Close()
 		}
 		if outputFile != nil {
@@ -152,21 +144,67 @@ func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
 	if err := connectPipeCommands(cmds, lastStdout, lastStderr, pipeStderr); err != nil {
 		return "", err
 	}
-
 	if err := startPipeCommands(cmds); err != nil {
 		return handleErrString(stdout.String(), stderr.String(), c.IgnoreExist1, err)
 	}
 
-	runErr := waitPipeCommands(ctx, cmds)
-	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
-		return "", buserr.New("ErrCmdTimeout")
-	}
+	runErr := c.pipeResultErr(ctx, waitPipeCommands(ctx, cmds))
 	if runErr != nil {
 		return handleErrString(stdout.String(), stderr.String(), c.IgnoreExist1, runErr)
 	}
 	return stdout.String(), nil
 }
 
+func (c *CommandHelper) RunPipeToFile(outputFile string, commands ...PipeCommand) (string, error) {
+	if len(commands) == 0 {
+		return "", nil
+	}
+
+	ctx, cancel, cmds := c.preparePipeCommands(commands)
+	if cancel != nil {
+		defer cancel()
+	}
+
+	file, err := os.OpenFile(outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = file.Close() }()
+
+	stderr := &lockedBuffer{limit: maxStreamOutputCapture}
+	if err := connectPipeCommands(cmds, file, stderr, stderr); err != nil {
+		return "", err
+	}
+	if err := startPipeCommands(cmds); err != nil {
+		return handleErrString("", stderr.String(), c.IgnoreExist1, err)
+	}
+
+	runErr := c.pipeResultErr(ctx, waitPipeCommands(ctx, cmds))
+	if runErr != nil {
+		return handleErrString("", stderr.String(), c.IgnoreExist1, runErr)
+	}
+	return "", nil
+}
+
+func (c *CommandHelper) preparePipeCommands(commands []PipeCommand) (context.Context, context.CancelFunc, []*exec.Cmd) {
+	ctx, cancel := c.pipeContext()
+	cmds := c.buildPipeCommands(ctx, commands)
+	if commands[0].Stdin != nil {
+		cmds[0].Stdin = commands[0].Stdin
+	}
+	return ctx, cancel, cmds
+}
+
+func (c *CommandHelper) pipeResultErr(ctx context.Context, runErr error) error {
+	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+		return buserr.New("ErrCmdTimeout")
+	}
+	if errors.Is(ctx.Err(), context.Canceled) {
+		return buserr.New("ErrShutDown")
+	}
+	return runErr
+}
+
 func (c *CommandHelper) pipeContext() (context.Context, context.CancelFunc) {
 	ctx := c.context
 	if ctx == nil {
@@ -243,87 +281,108 @@ func waitPipeCommands(ctx context.Context, cmds []*exec.Cmd) error {
 
 func (c *CommandHelper) run(name string, arg ...string) (string, error) {
 	var cmd *exec.Cmd
-	var ctx context.Context
+	var newContext context.Context
 	var cancel context.CancelFunc
+	var outputFile *os.File
 
 	if c.timeout != 0 {
 		if c.context == nil {
-			ctx, cancel = context.WithTimeout(context.Background(), c.timeout)
+			newContext, cancel = context.WithTimeout(context.Background(), c.timeout)
 		} else {
-			ctx, cancel = context.WithTimeout(c.context, c.timeout)
+			newContext, cancel = context.WithTimeout(c.context, c.timeout)
 		}
 		defer cancel()
-		cmd = exec.CommandContext(ctx, name, arg...)
 	} else if c.context != nil {
-		ctx = c.context
-		cmd = exec.CommandContext(ctx, name, arg...)
+		newContext = c.context
+	}
+
+	if newContext != nil {
+		cmd = exec.CommandContext(newContext, name, arg...)
 	} else {
 		cmd = exec.Command(name, arg...)
 	}
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true,
+	}
 
 	customWriter := &CustomWriter{taskItem: c.taskItem}
 	var stdout, stderr bytes.Buffer
-	var loggerClosers []io.Closer
+	var loggerCloser io.Closer
 	if c.taskItem != nil {
-		cmd.Stdout = customWriter
-		cmd.Stderr = customWriter
+		cmd.Stdout = io.MultiWriter(&stdout, customWriter)
+		cmd.Stderr = io.MultiWriter(&stderr, customWriter)
 	} else if c.logger != nil {
-		stdoutWriter := c.logger.Writer()
-		stderrWriter := c.logger.Writer()
-		if closer, ok := stdoutWriter.(io.Closer); ok {
-			loggerClosers = append(loggerClosers, closer)
+		streamWriter := c.logger.Writer()
+		if closer, ok := streamWriter.(io.Closer); ok {
+			loggerCloser = closer
 		}
-		if closer, ok := stderrWriter.(io.Closer); ok {
-			loggerClosers = append(loggerClosers, closer)
-		}
-		cmd.Stdout = stdoutWriter
-		cmd.Stderr = stderrWriter
+		cmd.Stdout = io.MultiWriter(&stdout, streamWriter)
+		cmd.Stderr = io.MultiWriter(&stderr, streamWriter)
 	} else if len(c.outputFile) != 0 {
-		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE, constant.FilePerm)
+		file, err := os.OpenFile(c.outputFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, constant.FilePerm)
 		if err != nil {
 			return "", err
 		}
-		defer file.Close()
-		cmd.Stdout = file
-		cmd.Stderr = file
+		outputFile = file
+		cmd.Stdout = io.MultiWriter(&stdout, outputFile)
+		cmd.Stderr = io.MultiWriter(&stderr, outputFile)
 	} else {
 		cmd.Stdout = &stdout
 		cmd.Stderr = &stderr
 	}
-	env := os.Environ()
-	env = append(env, c.env...)
-	cmd.Env = env
+	cmd.Env = append(os.Environ(), c.env...)
 	if len(c.workDir) != 0 {
 		cmd.Dir = c.workDir
 	}
 	defer func() {
-		for _, closer := range loggerClosers {
-			_ = closer.Close()
+		if loggerCloser != nil {
+			_ = loggerCloser.Close()
+		}
+		if outputFile != nil {
+			_ = outputFile.Close()
 		}
 	}()
 
-	if c.timeout != 0 {
-		err := cmd.Run()
-		if c.taskItem != nil {
-			customWriter.Flush()
-		}
-		if ctx != nil && errors.Is(ctx.Err(), context.DeadlineExceeded) {
-			return "", buserr.New("ErrCmdTimeout")
-		}
-		if err != nil {
-			return handleErr(stdout, stderr, c.IgnoreExist1, err)
-		}
-		return stdout.String(), nil
+	if err := cmd.Start(); err != nil {
+		return "", fmt.Errorf("cmd start failed: %w", err)
+	}
+	if c.taskItem != nil {
+		defer customWriter.Flush()
 	}
 
-	err := cmd.Run()
-	if c.taskItem != nil {
-		customWriter.Flush()
+	done := make(chan error, 1)
+	go func() {
+		done <- cmd.Wait()
+	}()
+	select {
+	case err := <-done:
+		if err != nil {
+			return handleErr(&stdout, &stderr, c.IgnoreExist1, err)
+		}
+		return stdout.String(), nil
+	case <-contextDone(newContext):
+		if cmd.Process != nil && cmd.Process.Pid > 0 {
+			_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+		}
+		var err error
+		switch newContext.Err() {
+		case context.DeadlineExceeded:
+			err = buserr.New("ErrCmdTimeout")
+		case context.Canceled:
+			err = buserr.New("ErrShutDown")
+		default:
+			err = newContext.Err()
+		}
+		<-done
+		return "", err
 	}
-	if err != nil {
-		return handleErr(stdout, stderr, c.IgnoreExist1, err)
+}
+
+func contextDone(ctx context.Context) <-chan struct{} {
+	if ctx == nil {
+		return nil
 	}
-	return stdout.String(), nil
+	return ctx.Done()
 }
 
 func killStarted(cmds []*exec.Cmd) {
@@ -348,36 +407,43 @@ func WithOutputFile(outputFile string) Option {
 		s.outputFile = outputFile
 	}
 }
+
 func WithContext(ctx context.Context) Option {
 	return func(s *CommandHelper) {
 		s.context = ctx
 	}
 }
+
 func WithTimeout(timeout time.Duration) Option {
 	return func(s *CommandHelper) {
 		s.timeout = timeout
 	}
 }
+
 func WithLogger(logger *log.Logger) Option {
 	return func(s *CommandHelper) {
 		s.logger = logger
 	}
 }
+
 func WithTask(taskItem task.Task) Option {
 	return func(s *CommandHelper) {
 		s.taskItem = &taskItem
 	}
 }
+
 func WithWorkDir(workDir string) Option {
 	return func(s *CommandHelper) {
 		s.workDir = workDir
 	}
 }
+
 func WithEnv(env ...string) Option {
 	return func(s *CommandHelper) {
 		s.env = append(s.env, env...)
 	}
 }
+
 func WithIgnoreExist1() Option {
 	return func(s *CommandHelper) {
 		s.IgnoreExist1 = true
@@ -385,11 +451,14 @@ func WithIgnoreExist1() Option {
 }
 
 type CustomWriter struct {
+	mu       sync.Mutex
 	taskItem *task.Task
 	buffer   bytes.Buffer
 }
 
 func (cw *CustomWriter) Write(p []byte) (n int, err error) {
+	cw.mu.Lock()
+	defer cw.mu.Unlock()
 	cw.buffer.Write(p)
 	lines := strings.Split(cw.buffer.String(), "\n")
 
@@ -401,14 +470,17 @@ func (cw *CustomWriter) Write(p []byte) (n int, err error) {
 
 	return len(p), nil
 }
+
 func (cw *CustomWriter) Flush() {
+	cw.mu.Lock()
+	defer cw.mu.Unlock()
 	if cw.buffer.Len() > 0 {
 		cw.taskItem.Log(cw.buffer.String())
 		cw.buffer.Reset()
 	}
 }
 
-func handleErr(stdout, stderr bytes.Buffer, ignoreExist1 bool, err error) (string, error) {
+func handleErr(stdout, stderr fmt.Stringer, ignoreExist1 bool, err error) (string, error) {
 	return handleErrString(stdout.String(), stderr.String(), ignoreExist1, err)
 }