KubeExec: parse logs and return output (#4066)

* KubeExec: parse logs and return output

* Incorporated review comments

pkg/function/kube_exec.go

@@ -5,10 +5,12 @@ import (
 	"regexp"
 	"strings"
 
-	log "github.com/sirupsen/logrus"
+	"github.com/pkg/errors"
 
 	kanister "github.com/kanisterio/kanister/pkg"
+	"github.com/kanisterio/kanister/pkg/format"
 	"github.com/kanisterio/kanister/pkg/kube"
+	"github.com/kanisterio/kanister/pkg/output"
 	"github.com/kanisterio/kanister/pkg/param"
 )
 
@@ -33,6 +35,30 @@ func (*kubeExecFunc) Name() string {
 	return "KubeExec"
 }
 
+func parseLogAndCreateOutput(out string) (map[string]interface{}, error) {
+	if out == "" {
+		return nil, nil
+	}
+	var op map[string]interface{}
+	logs := regexp.MustCompile("[\n]").Split(out, -1)
+	for _, l := range logs {
+		// Log should contain "###Phase-output###:" string
+		if strings.Contains(l, output.PhaseOpString) {
+			if op == nil {
+				op = make(map[string]interface{})
+			}
+			pattern := regexp.MustCompile(`###Phase-output###:(.*?)*$`)
+			match := pattern.FindAllStringSubmatch(l, 1)
+			opObj, err := output.UnmarshalOutput(match[0][1])
+			if err != nil {
+				return nil, err
+			}
+			op[opObj.Key] = opObj.Value
+		}
+	}
+	return op, nil
+}
+
 func (kef *kubeExecFunc) Exec(ctx context.Context, tp param.TemplateParams, args map[string]interface{}) (map[string]interface{}, error) {
 	cli, err := kube.NewClient()
 	if err != nil {
@@ -54,23 +80,14 @@ func (kef *kubeExecFunc) Exec(ctx context.Context, tp param.TemplateParams, args
 	}
 
 	stdout, stderr, err := kube.Exec(cli, namespace, pod, container, cmd)
-	if stdout != "" {
-		logs := regexp.MustCompile("[\r\n]").Split(stdout, -1)
-		for _, stdoutLog := range logs {
-			if strings.TrimSpace(stdoutLog) != "" {
-				log.Info(stdoutLog)
-			}
-		}
-	}
-	if stderr != "" {
-		logs := regexp.MustCompile("[\r\n]").Split(stderr, -1)
-		for _, stderrLog := range logs {
-			if strings.TrimSpace(stderrLog) != "" {
-				log.Info(stderrLog)
-			}
-		}
+	format.Log(pod, container, stdout)
+	format.Log(pod, container, stderr)
+	if err != nil {
+		return nil, err
 	}
-	return nil, err
+
+	out, err := parseLogAndCreateOutput(stdout)
+	return out, errors.Wrap(err, "Failed to generate output")
 }
 
 func (*kubeExecFunc) RequiredArgs() []string {

pkg/function/kube_exec_test.go

@@ -149,3 +149,27 @@ func (s *KubeExecTest) TestKubeExec(c *C) {
 		c.Assert(err, IsNil)
 	}
 }
+
+func (s *KubeExecTest) TestParseLogAndCreateOutput(c *C) {
+	for _, tc := range []struct {
+		log        string
+		expected   map[string]interface{}
+		errChecker Checker
+		outChecker Checker
+	}{
+		{"###Phase-output###: {\"key\":\"version\",\"value\":\"0.12.0\"}", map[string]interface{}{"version": "0.12.0"}, IsNil, NotNil},
+		{"###Phase-output###: {\"key\":\"version\",\"value\":\"0.12.0\"}\n###Phase-output###: {\"key\":\"path\",\"value\":\"/backup/path\"}",
+			map[string]interface{}{"version": "0.12.0", "path": "/backup/path"}, IsNil, NotNil},
+		{"Random message ###Phase-output###: {\"key\":\"version\",\"value\":\"0.12.0\"}", map[string]interface{}{"version": "0.12.0"}, IsNil, NotNil},
+		{"Random message with newline \n###Phase-output###: {\"key\":\"version\",\"value\":\"0.12.0\"}", map[string]interface{}{"version": "0.12.0"}, IsNil, NotNil},
+		{"###Phase-output###: Invalid message", nil, NotNil, IsNil},
+		{"Random message", nil, IsNil, IsNil},
+	} {
+		out, err := parseLogAndCreateOutput(tc.log)
+		c.Check(err, tc.errChecker)
+		c.Check(out, tc.outChecker)
+		if out != nil {
+			c.Check(out, DeepEquals, tc.expected)
+		}
+	}
+}

pkg/output/output.go

@@ -9,7 +9,7 @@ import (
 )
 
 const (
-	phaseOpString = "###Phase-output###:"
+	PhaseOpString = "###Phase-output###:"
 )
 
 type Output struct {
@@ -24,11 +24,18 @@ func marshalOutput(key, value string) (string, error) {
 	}
 	outString, err := json.Marshal(out)
 	if err != nil {
-		return "", errors.Wrapf(err, "Failed to marshall key-value pair")
+		return "", errors.Wrap(err, "Failed to marshal key-value pair")
 	}
 	return string(outString), nil
 }
 
+// UnmarshalOutput unmarshals output json into Output struct
+func UnmarshalOutput(opString string) (*Output, error) {
+	p := &Output{}
+	err := json.Unmarshal([]byte(opString), p)
+	return p, errors.Wrap(err, "Failed to unmarshal key-value pair")
+}
+
 // ValidateKey validates the key argument
 func ValidateKey(key string) error {
 	// key should be non-empty
@@ -49,6 +56,6 @@ func PrintOutput(key, value string) error {
 	if err != nil {
 		return err
 	}
-	fmt.Println(phaseOpString, outString)
+	fmt.Println(PhaseOpString, outString)
 	return nil
 }