[Agent] [Executor] 完成PipelineCommand part

agent-go/executor/BasicFunction.go

@@ -1,8 +1,10 @@
 package executor
 
 import (
+	"bytes"
 	"fmt"
 	"os/exec"
+	"strings"
 )
 
 var RemoveForcePrefix = []string{"rm", "-rf"}
@@ -81,3 +83,37 @@ func BasicPrettyPrint(resultOk bool, resultLog []string) {
 		fmt.Println(s)
 	}
 }
+
+func BasicTransPipelineCommand(pipelineString string) (pipelineCommand [][]string) {
+
+	var pipelineCommandC [][]string
+	split := strings.Split(pipelineString, "|")
+	if len(split) == 0 {
+		log.WarnF("输入的pipelineString有误！ %s", pipelineString)
+		return pipelineCommandC
+	}
+
+	for _, s := range split {
+		// 去除掉无效的空行
+		singleCommand := strings.Split(s, " ")
+		var realSingleCommand []string
+
+		for i := range singleCommand {
+			if singleCommand[i] != "" {
+				realSingleCommand = append(realSingleCommand, singleCommand[i])
+			}
+		}
+		pipelineCommandC = append(pipelineCommandC, realSingleCommand)
+	}
+
+	return pipelineCommandC
+}
+
+func BasicConvertBufferToSlice(outputBuffer bytes.Buffer) (resultLog []string) {
+
+	s := outputBuffer.String()
+	split := strings.Split(s, "\n")
+
+	return split
+
+}

agent-go/executor/FunctionalExecutor.go

@@ -102,37 +102,53 @@ func FormatAllCommandExecutor(singleLineCommand []string) ([]string, error) {
 	return resultSlice, resultError
 }
 
-func PipelineCommandExecutor() {
+func PipelineCommandExecutor(pipelineCommand [][]string) (resultOk bool, resultLog []string) {
-
+	if len(pipelineCommand) == 0 {
-	cmd1 := exec.Command("ps", "aux")
+		return true, nil
-	cmd2 := exec.Command("grep", "apipe")
+	} else if len(pipelineCommand) == 1 {
+		log.Debug("输入的PipelineCommand长度有误！")
+		return AllCommandExecutor(pipelineCommand[0])
+	}
 
+	var c []string
+	cmd1 := exec.Command(pipelineCommand[0][0], pipelineCommand[0][1:]...)
 	var outputBuf1 bytes.Buffer
 	cmd1.Stdout = &outputBuf1
 	if err := cmd1.Start(); err != nil {
-		fmt.Printf("Error: The first command can not be startup %s\n", err)
+		sprintf := fmt.Sprintf("Error: The first command can not be startup %s", err)
-		return
+		return false, append(c, sprintf)
 	}
 	if err := cmd1.Wait(); err != nil {
-		fmt.Printf("Error: Couldn't wait for the first command: %s\n", err)
+		sprintf := fmt.Sprintf("Error: Couldn't wait for the first command: %s", err)
-		return
+		return false, append(c, sprintf)
 	}
+
+	for i := 1; i < len(pipelineCommand); i++ {
+
+		cmd2 := exec.Command(pipelineCommand[i][0], pipelineCommand[i][1:]...)
+		sprintf := fmt.Sprintf("current command is %s", pipelineCommand[i])
+		c = append(c, sprintf)
+
 		cmd2.Stdin = &outputBuf1
 		var outputBuf2 bytes.Buffer
 		cmd2.Stdout = &outputBuf2
 		if err := cmd2.Start(); err != nil {
-		fmt.Printf("Error: The second command can not be startup: %s\n", err)
+			sprintf := fmt.Sprintf("Error: The second command can not be startup: %s", err)
-		return
+			return false, append(c, sprintf)
 		}
 		if err := cmd2.Wait(); err != nil {
-		fmt.Printf("Error: Couldn't wait for the second command: %s\n", err)
+			sprintf := fmt.Sprintf("Error: Couldn't wait for the second command: %s", err)
-		return
+			return false, append(c, sprintf)
 		}
 
-	s := outputBuf2.String()
+		// change
+		outputBuf1 = outputBuf2
+	}
+
+	s := outputBuf1.String()
 	split := strings.Split(s, "\n")
 
-	BasicPrettyPrint(true, split)
+	return true, split
 }
 
 // PureResultSingleExecute 执行单行命令，忽略输出，只对执行成功与否负责

agent-go/executor/FunctionalExecutor_test.go

@@ -1,6 +1,12 @@
 package executor
 
-import "testing"
+import (
+	"bytes"
+	"fmt"
+	"os/exec"
+	"strings"
+	"testing"
+)
 
 var closeSELinux = []string{
 	"sed",
@@ -19,6 +25,29 @@ var shutdownFirewalld = []string{
 	"systemctl", "stop", "firewalld", "&&", "systemctl", "disable", "firewalld",
 }
 
+var pipelineCommandFalse = []string{
+	"ls",
+	"/etc",
+	"|",
+	"grep",
+	"passwd",
+}
+
+var pipelineCommand = [][]string{
+	{
+		"cat",
+		"/home/wdd/IdeaProjects/ProjectOctopus/agent-go/executor/script/123",
+	},
+	{
+		"grep",
+		"passwd",
+	},
+}
+
+var pipelineCommandMore = "apt-cache madison docker-ce | awk {print$3} | cut -d: -f2"
+var pipelineCommandThird = "ls /etc/ | grep passwd | head -2"
+var pipelineCommandSecond = "systemctl status docker.service | grep active | wc -w"
+
 var ifconfigCommand = []string{"ifconfig"}
 
 func TestReadTimeOutput(t *testing.T) {
@@ -28,7 +57,7 @@ func TestReadTimeOutput(t *testing.T) {
 }
 
 func TestAllCommandExecutor(t *testing.T) {
-	ok, result := AllCommandExecutor(shutdownFirewalld)
+	ok, result := AllCommandExecutor(pipelineCommandFalse)
 
 	t.Logf("执行结果为 => %#v", ok)
 	t.Logf("执行日志为 => %#v", result)
@@ -41,8 +70,50 @@ func TestPureResultSingleExecute(t *testing.T) {
 
 }
 
-func TestPipelineCommandExecutor(t *testing.T) {
+func TestPipelineCommandExecutorSingle(t *testing.T) {
 
-	PipelineCommandExecutor()
+	cmd1 := exec.Command("ls", "/etc/")
+	cmd2 := exec.Command("grep", "passwd")
+
+	var outputBuf1 bytes.Buffer
+	cmd1.Stdout = &outputBuf1
+	if err := cmd1.Start(); err != nil {
+		fmt.Printf("Error: The first command can not be startup %s\n", err)
+		return
+	}
+	if err := cmd1.Wait(); err != nil {
+		fmt.Printf("Error: Couldn't wait for the first command: %s\n", err)
+		return
+	}
+	cmd2.Stdin = &outputBuf1
+	var outputBuf2 bytes.Buffer
+	cmd2.Stdout = &outputBuf2
+	if err := cmd2.Start(); err != nil {
+		fmt.Printf("Error: The second command can not be startup: %s\n", err)
+		return
+	}
+	if err := cmd2.Wait(); err != nil {
+		fmt.Printf("Error: Couldn't wait for the second command: %s\n", err)
+		return
+	}
+
+	s := outputBuf2.String()
+	split := strings.Split(s, "\n")
+
+	BasicPrettyPrint(true, split)
 
 }
+
+func TestPipelineCommandExecutor(t *testing.T) {
+
+	//PipelineCommandExecutor(pipelineCommand)
+	pipelineStringToCommand := BasicTransPipelineCommand(pipelineCommandMore)
+
+	t.Logf("pipelineCommmand are => %#v", pipelineStringToCommand)
+
+	ok, resultLog := PipelineCommandExecutor(pipelineStringToCommand)
+
+	t.Logf("command execute ok is => %#v", ok)
+	t.Logf("command result are 下 \n")
+	BasicPrettyPrint(ok, resultLog)
+}

agent-go/executor/script/123

@@ -0,0 +1,3 @@
+passwd
+
+passwd1