Merge pull request #336 from merico-dev/feat-outputs-rework

feat: outputs rework with unit tests and functional tests

Author: lfbdev

go.mod

@@ -129,7 +129,6 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.0.1 // indirect
 	github.com/patrickmn/go-cache v2.1.0+incompatible // indirect

go.sum

@@ -965,9 +965,8 @@ github.com/onsi/ginkgo v1.10.3/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/ginkgo v1.14.1/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
+github.com/onsi/ginkgo v1.16.4 h1:29JGrr5oVBm5ulCWet69zQkzWipVXIol6ygQUe/EzNc=
 github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0=
-github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
-github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/ginkgo/v2 v2.0.0 h1:CcuG/HvWNkkaqCUpJifQY8z7qEMBJya6aLPx6ftGyjQ=
 github.com/onsi/ginkgo/v2 v2.0.0/go.mod h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
 github.com/onsi/gomega v0.0.0-20151007035656-2152b45fa28a/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=

internal/pkg/pluginengine/change.go

@@ -1,28 +1,14 @@
 package pluginengine
 
 import (
-	"errors"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/merico-dev/stream/internal/pkg/configloader"
 	"github.com/merico-dev/stream/internal/pkg/statemanager"
 	"github.com/merico-dev/stream/pkg/util/log"
 )
 
-const REF_PREFIX = "${{"
-const REF_SUFFIX = "}}"
-
-// eg. ${{name.kind.outputs.key}},name setgment number is 0
-const NAME_SEGMENT_NUM = 0
-
-// eg. ${{name.kind.outputs.key}},kind setgment number is 1
-const KIND_SEGMENT_NUM = 1
-
-// eg. ${{name.kind.outputs.key}},key setgment number is 3
-const REF_SEGMENT_NUM = 3
-
 // Change is a wrapper with a single Tool and its Action should be execute.
 type Change struct {
 	Tool        *configloader.Tool
@@ -111,12 +97,19 @@ func execute(smgr statemanager.Manager, changes []*Change) map[string]error {
 		var returnValue map[string]interface{}
 
 		log.Debugf("Tool's raw changes are: %s.", c.Tool.Options)
-		// fill ref inputs
-		err = fillRefValueWithOutputs(smgr, c.Tool.Options)
-		if err != nil {
+
+		errs := HandleOutputsReferences(smgr, c.Tool.Options)
+		if len(errs) != 0 {
 			succeeded = false
+
+			for _, e := range errs {
+				log.Errorf("Error: %s.", e)
+			}
+			log.Errorf("The outputs reference in tool %s (%s) can't be resolved. Please double check your config.", c.Tool.Name, c.Tool.Plugin.Kind)
+
+			// not executing this change since its input isn't valid
+			continue
 		}
-		log.Debugf("Tool's changes with filled inputs: %s.", c.Tool.Options)
 
 		switch c.ActionName {
 		case statemanager.ActionCreate:
@@ -192,56 +185,3 @@ func handleResult(smgr statemanager.Manager, change *Change) error {
 	log.Successf("Plugin %s(%s) %s done.", change.Tool.Name, change.Tool.Plugin.Kind, change.ActionName)
 	return nil
 }
-
-// fillRefValueWithOutputs fill inputs from state
-func fillRefValueWithOutputs(smgr statemanager.Manager, options map[string]interface{}) error {
-	for key, value := range options {
-		log.Debugf("Key: %s,  Value: %s.", key, value)
-		// judge whether the value is a string
-		if inst, ok := value.(string); ok {
-			// judge whether the format is ${{xxx}}
-			if isValidRefFormat(inst) {
-				ref := getRefFormatString(inst)
-				log.Debug("Ref inputs: ", ref)
-				refParam := strings.Split(ref, ".")
-				if len(refParam) <= 3 {
-					return errors.New("incorrect output reference: " + ref)
-				}
-
-				outputs, err := smgr.GetOutputs(statemanager.GenStateKey(refParam[NAME_SEGMENT_NUM], refParam[KIND_SEGMENT_NUM]))
-				if err != nil {
-					return err
-				}
-				log.Debug("Ref outputs: ", outputs)
-
-				if outs, ok := outputs.(map[string]interface{}); ok {
-					log.Debug("Ref outs: ", outs)
-					log.Debug("Ref param: ", refParam[REF_SEGMENT_NUM])
-					if value == nil {
-						return errors.New("ref input value is null: " + refParam[REF_SEGMENT_NUM])
-					}
-					if options[key], ok = outs[refParam[REF_SEGMENT_NUM]]; !ok {
-						return fmt.Errorf("can not find %s in dependency outputs", refParam[REF_SEGMENT_NUM])
-					}
-				}
-			}
-		}
-		// judge wheter the format is map[string]interface{}, if true, then recursion
-		if _, ok := value.(map[string]interface{}); ok {
-			if err := fillRefValueWithOutputs(smgr, value.(map[string]interface{})); err != nil {
-				return err
-			}
-		}
-	}
-	return nil
-}
-
-// isValidRefFormat if the format is ${{abc}}
-func isValidRefFormat(ref string) bool {
-	return strings.HasPrefix(ref, REF_PREFIX) && strings.HasSuffix(ref, REF_SUFFIX)
-}
-
-// getRefFormatString get abc from ${{abc}} or ${{ abc }}
-func getRefFormatString(rawFormatString string) string {
-	return strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(rawFormatString, REF_PREFIX), REF_SUFFIX))
-}

internal/pkg/pluginengine/change_helper.go

@@ -82,10 +82,12 @@ func changesForApply(smgr statemanager.Manager, cfg *configloader.Config) ([]*Ch
 				description := fmt.Sprintf("Tool %s (%s) found in config but doesn't exist in the state, will be created.", tool.Name, tool.Plugin.Kind)
 				changes = append(changes, generateCreateAction(&tool, description))
 			} else {
-				if err := renderRefByDependency(&tool, cfg.Tools, smgr); err != nil {
-					return nil, err
-				}
 				// tool found in the state
+
+				// first, handle possible "outputs" referneces in the tool's config
+				// ignoring errors, since at this stage we are calculating changes, and the dependency might not have its output in the state yet
+				_ = HandleOutputsReferences(smgr, tool.Options)
+
 				if drifted(tool.Options, state.Options) {
 					// tool's config differs from State's, Update
 					description := fmt.Sprintf("Tool %s (%s) config drifted from the state, will be updated.", tool.Name, tool.Plugin.Kind)
@@ -169,33 +171,3 @@ func changesForForceDelete(smgr statemanager.Manager, cfg *configloader.Config)
 	}
 	return changes
 }
-
-// renderRefByDependency
-// 1. if dependency plugin changed, do not fill ${{***}} with ref value;
-// 2. if dependency plugin did not change, fill ${{***}} with ref value;
-func renderRefByDependency(tool *configloader.Tool, tools []configloader.Tool, smgr statemanager.Manager) error {
-	if len(tool.DependsOn) > 0 {
-		for _, dependency := range tool.DependsOn {
-			dependencyChange := false
-			for _, c := range tools {
-				log.Debugf("====== Name: %s kind: %s dependency: %s  =====", c.Name, c.Plugin.Kind, dependency)
-				if fmt.Sprintf("%s%s%s", c.Name, ".", c.Plugin.Kind) == dependency {
-					state := smgr.GetState(statemanager.StateKeyGenerateFunc(&c))
-					if drifted(c.Options, state.Options) {
-						dependencyChange = true
-					}
-				}
-			}
-			if !dependencyChange {
-				// fill ref inputs,
-				if err := fillRefValueWithOutputs(smgr, tool.Options); err != nil {
-					return err
-				}
-				log.Infof("Dependency plugin no changes, ref inputs will be filled: %s.", tool.Options)
-			} else {
-				log.Infof("Do not fill ref inputs now, they will be filled when dependency plugin complete: %s.", tool.Options)
-			}
-		}
-	}
-	return nil
-}