Method RunPipe

agent/utils/cmd/cmdx.go:115–177

(commands ...PipeCommand)

}

func (c *CommandHelper) RunPipe(commands ...PipeCommand) (string, error) {
	if len(commands) == 0 {
		return "", nil
	}

	ctx, cancel, cmds := c.preparePipeCommands(commands)
	if cancel != nil {
		defer cancel()
	}

	customWriter := &CustomWriter{taskItem: c.taskItem}
	var outputFile *os.File
	stdout, stderr := &lockedBuffer{}, &lockedBuffer{}
	limitOutputCapture := c.taskItem != nil || c.logger != nil || len(c.outputFile) != 0
	if limitOutputCapture {
		stdout.limit = maxStreamOutputCapture
		stderr.limit = maxStreamOutputCapture
	}
	var pipeStderr io.Writer = stderr
	var lastStdout io.Writer = stdout
	var lastStderr io.Writer = stderr
	var streamWriter io.Writer
	if c.taskItem != nil {
		streamWriter = customWriter
	} else if c.logger != nil {
		streamWriter = c.logger.Writer()
	} 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 {
			return "", err
		}
		outputFile = file
		lastStdout = outputFile
	}
	if streamWriter != nil {
		pipeStderr = io.MultiWriter(stderr, streamWriter)
		lastStdout = io.MultiWriter(stdout, streamWriter)
		lastStderr = io.MultiWriter(stderr, streamWriter)
	}
	defer func() {
		if c.taskItem != nil {
			customWriter.Flush()
		}
		if closer, ok := streamWriter.(io.Closer); ok {
			_ = closer.Close()
		}
		if outputFile != nil {
			_ = outputFile.Close()
		}
	}()
	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 := c.pipeResultErr(ctx, waitPipeCommands(ctx, cmds))