fix: add custom step output scanner with error detection (#216)

* feat: add custom step output scanner with error detection

* fix lint

* improve keywords

* add comment

* fix exec error handling bug and update error detection

* fix err handling

* fix lint

* simplify logic

* require approval for destroy stages too

---------

Co-authored-by: michaeljguarino <[email protected]>

Author: floreks

go.mod

@@ -19,7 +19,7 @@ require (
 	github.com/open-policy-agent/gatekeeper/v3 v3.15.1
 	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/pkg/errors v0.9.1
-	github.com/pluralsh/console-client-go v0.5.13
+	github.com/pluralsh/console-client-go v0.5.18
 	github.com/pluralsh/controller-reconcile-helper v0.0.4
 	github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34
 	github.com/pluralsh/polly v0.1.10
@@ -39,7 +39,6 @@ require (
 	k8s.io/apimachinery v0.29.3
 	k8s.io/cli-runtime v0.29.2
 	k8s.io/client-go v0.29.2
-	k8s.io/klog v1.0.0
 	k8s.io/klog/v2 v2.110.1
 	k8s.io/kubectl v0.29.2
 	layeh.com/gopher-luar v1.0.11

go.sum

@@ -207,7 +207,6 @@ github.com/go-ini/ini v1.67.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3I
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
-github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
@@ -527,8 +526,8 @@ github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
-github.com/pluralsh/console-client-go v0.5.13 h1:HOmkho5aaU42f6PkSb+BUFjhCJKnL5jceLZiT16HMBE=
-github.com/pluralsh/console-client-go v0.5.13/go.mod h1:eyCiLA44YbXiYyJh8303jk5JdPkt9McgCo5kBjk4lKo=
+github.com/pluralsh/console-client-go v0.5.18 h1:uwYsoGaggvi3uPZYL/+qdhvgl73sGBiuVUfQGAC/J4c=
+github.com/pluralsh/console-client-go v0.5.18/go.mod h1:eyCiLA44YbXiYyJh8303jk5JdPkt9McgCo5kBjk4lKo=
 github.com/pluralsh/controller-reconcile-helper v0.0.4 h1:1o+7qYSyoeqKFjx+WgQTxDz4Q2VMpzprJIIKShxqG0E=
 github.com/pluralsh/controller-reconcile-helper v0.0.4/go.mod h1:AfY0gtteD6veBjmB6jiRx/aR4yevEf6K0M13/pGan/s=
 github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34 h1:ab2PN+6if/Aq3/sJM0AVdy1SYuMAnq4g20VaKhTm/Bw=
@@ -1095,8 +1094,6 @@ k8s.io/client-go v0.29.2 h1:FEg85el1TeZp+/vYJM7hkDlSTFZ+c5nnK44DJ4FyoRg=
 k8s.io/client-go v0.29.2/go.mod h1:knlvFZE58VpqbQpJNbCbctTVXcd35mMyAAwBdpt4jrA=
 k8s.io/component-base v0.29.2 h1:lpiLyuvPA9yV1aQwGLENYyK7n/8t6l3nn3zAtFTJYe8=
 k8s.io/component-base v0.29.2/go.mod h1:BfB3SLrefbZXiBfbM+2H1dlat21Uewg/5qtKOl8degM=
-k8s.io/klog v1.0.0 h1:Pt+yjF5aB1xDSVbau4VsWe+dQNzA0qv1LlXdC2dF6Q8=
-k8s.io/klog v1.0.0/go.mod h1:4Bi6QPql/J/LkTDqv7R/cd3hPo4k2DG6Ptcz060Ez5I=
 k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0=
 k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/AuzbMm96cd3YHRTU83I780=

pkg/harness/controller/controller.go

@@ -100,6 +100,10 @@ func (in *stackRunController) toExecutable(ctx context.Context, step *gqlclient.
 	// ensuring they have completed all work.
 	in.wg.Add(1)
 
+	toolWriters := make([]io.WriteCloser, 0)
+	modifier := in.tool.Modifier(step.Stage)
+	args := step.Args
+	env := in.stackRun.Env()
 	consoleWriter := sink.NewConsoleWriter(
 		ctx,
 		in.consoleClient,
@@ -115,29 +119,35 @@ func (in *stackRunController) toExecutable(ctx context.Context, step *gqlclient.
 		)...,
 	)
 
-	var toolWriters []io.WriteCloser
-	modifier := in.tool.Modifier(step.Stage)
-	args := step.Args
-	env := in.stackRun.Env()
 	if modifier != nil {
 		args = modifier.Args(args)
 		env = modifier.Env(env)
 		toolWriters = modifier.WriteCloser()
 	}
 
-	return exec.NewExecutable(
-		step.Cmd,
-		append(
-			in.execOptions,
-			exec.WithDir(in.execWorkDir()),
-			exec.WithEnv(env),
-			exec.WithArgs(args),
-			exec.WithID(step.ID),
-			exec.WithOutputSinks(append(toolWriters, consoleWriter)...),
-			exec.WithHook(v1.LifecyclePreStart, in.preExecHook(step.Stage, step.ID)),
-			exec.WithHook(v1.LifecyclePostStart, in.postExecHook(step.Stage)),
-		)...,
+	// base executable options
+	options := in.execOptions
+	options = append(
+		options,
+		exec.WithDir(in.execWorkDir()),
+		exec.WithEnv(env),
+		exec.WithArgs(args),
+		exec.WithID(step.ID),
+		exec.WithOutputSinks(append(toolWriters, consoleWriter)...),
+		exec.WithHook(v1.LifecyclePreStart, in.preExecHook(step.Stage, step.ID)),
+		exec.WithHook(v1.LifecyclePostStart, in.postExecHook(step.Stage)),
+		exec.WithOutputAnalyzer(exec.NewKeywordDetector()),
 	)
+
+	// Add a custom run step output analyzer for the destroy stage to increase
+	// a chance of detecting errors during execution. On occasion executable can
+	// return exit code 0 even though there was a fatal error during execution.
+	// TODO: use destroy stage
+	// if step.Stage == gqlclient.StepStageApply {
+	//	options = append(options, exec.WithOutputAnalyzer(exec.NewKeywordDetector()))
+	//}
+
+	return exec.NewExecutable(step.Cmd, options...)
 }
 
 func (in *stackRunController) completeStackRun(status gqlclient.StackStatus, stackRunErr error) error {

pkg/harness/controller/controller_hooks.go

@@ -91,7 +91,7 @@ func (in *stackRunController) postExecHook(stage gqlclient.StepStage) v1.HookFun
 // postExecHook is a callback function started by the exec.Executable before it runs the executable.
 func (in *stackRunController) preExecHook(stage gqlclient.StepStage, id string) v1.HookFunction {
 	return func() error {
-		if stage == gqlclient.StepStageApply && in.requiresApproval() {
+		if (stage == gqlclient.StepStageApply || stage == gqlclient.StepStageDestroy) && in.requiresApproval() {
 			in.waitForApproval()
 		}
 

pkg/harness/exec/analyzer.go

@@ -0,0 +1,49 @@
+package exec
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"strings"
+)
+
+type outputAnalyzer struct {
+	stdout *bytes.Buffer
+	stderr *bytes.Buffer
+
+	heuristics []OutputAnalyzerHeuristic
+}
+
+func (in *outputAnalyzer) Stdout() io.Writer {
+	return in.stdout
+}
+
+func (in *outputAnalyzer) Stderr() io.Writer {
+	return in.stderr
+}
+
+func (in *outputAnalyzer) Detect() []error {
+	errors := make([]error, 0)
+	output := in.stdout.String()
+
+	for _, heuristic := range in.heuristics {
+		if potentialErrors := heuristic.Detect(bufio.NewScanner(strings.NewReader(output))); len(potentialErrors) > 0 {
+			errors = append(errors, potentialErrors.ToErrors()...)
+		}
+	}
+
+	if in.stderr.Len() > 0 {
+		errors = append(errors, fmt.Errorf("%s", in.stderr.String()))
+	}
+
+	return errors
+}
+
+func NewOutputAnalyzer(heuristics ...OutputAnalyzerHeuristic) OutputAnalyzer {
+	return &outputAnalyzer{
+		stdout:     bytes.NewBuffer([]byte{}),
+		stderr:     bytes.NewBuffer([]byte{}),
+		heuristics: heuristics,
+	}
+}

pkg/harness/exec/analyzer_heuristic.go

@@ -0,0 +1,63 @@
+package exec
+
+import (
+	"bufio"
+	"strings"
+
+	"github.com/pluralsh/polly/algorithms"
+)
+
+type keywordDetector struct {
+	keywords []keyword
+}
+
+type keyword struct {
+	content    string
+	ignoreCase bool
+}
+
+func (in keyword) PartOf(s string) bool {
+	if !in.ignoreCase {
+		return strings.Contains(s, in.content)
+	}
+
+	return strings.Contains(
+		strings.ToLower(s),
+		strings.ToLower(in.content),
+	)
+}
+
+// Detect implements [OutputAnalyzerHeuristic] interface.
+// TODO: we can spread actual message analysis into multiple routines to speed up the process.
+func (in *keywordDetector) Detect(input *bufio.Scanner) Errors {
+	line := 0
+	errors := make([]Error, 0)
+	for input.Scan() {
+		if !in.hasError(input.Text()) {
+			continue
+		}
+
+		errors = append(errors, Error{
+			line:    line,
+			message: input.Text(),
+		})
+	}
+
+	return errors
+}
+
+func (in *keywordDetector) hasError(message string) bool {
+	return algorithms.Index(in.keywords, func(k keyword) bool {
+		return k.PartOf(message)
+	}) >= 0
+}
+
+func NewKeywordDetector() OutputAnalyzerHeuristic {
+	return &keywordDetector{
+		keywords: []keyword{
+			{"error message: http remote state already locked", true},
+			{"error acquiring the state lock", true},
+			{"Error:", false},
+		},
+	}
+}

pkg/harness/exec/analyzer_types.go

@@ -0,0 +1,46 @@
+package exec
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+)
+
+// OutputAnalyzer captures the command output
+// and attempts to detect potential errors.
+type OutputAnalyzer interface {
+	Stdout() io.Writer
+	Stderr() io.Writer
+
+	// Detect scans the output for potential errors.
+	// It uses a custom heuristics to detect issues.
+	// It can result in a false positives.
+	//
+	// Note: Make sure that it is executed after Write
+	//		 has finished to ensure proper detection.
+	Detect() []error
+}
+
+type OutputAnalyzerHeuristic interface {
+	Detect(input *bufio.Scanner) Errors
+}
+
+type Error struct {
+	line    int
+	message string
+}
+
+func (in Error) ToError() error {
+	return fmt.Errorf("[%d] %s", in.line, in.message)
+}
+
+type Errors []Error
+
+func (in Errors) ToErrors() []error {
+	errors := make([]error, len(in))
+	for _, err := range in {
+		errors = append(errors, err.ToError())
+	}
+
+	return errors
+}

pkg/harness/exec/exec.go

@@ -2,6 +2,7 @@ package exec
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -35,6 +36,10 @@ func (in *executable) Run(ctx context.Context) error {
 	cmd.Stdout = w
 	cmd.Stderr = w
 
+	if in.outputAnalyzer != nil {
+		cmd.Stderr = io.MultiWriter(w, in.outputAnalyzer.Stderr())
+	}
+
 	// Configure environment of the executable.
 	// Root process environment is used as a base and passed in env vars
 	// are added on top of that. In case of duplicate keys, custom env
@@ -47,13 +52,17 @@ func (in *executable) Run(ctx context.Context) error {
 
 	klog.V(log.LogLevelExtended).InfoS("executing", "command", in.Command())
 	if err := cmd.Run(); err != nil {
-		if err = context.Cause(ctx); err != nil {
+		if err := context.Cause(ctx); err != nil {
 			return err
 		}
 
 		return err
 	}
 
+	if err := in.analyze(); err != nil {
+		return err
+	}
+
 	return in.runLifecycleFunction(v1.LifecyclePostStart)
 }
 
@@ -87,17 +96,22 @@ func (in *executable) ID() string {
 }
 
 func (in *executable) writer() io.Writer {
-	if len(in.outputSinks) == 0 {
-		return os.Stdout
-	}
+	writers := []io.Writer{os.Stdout}
 
-	return io.MultiWriter(
-		append(
-			algorithms.Map(in.outputSinks, func(writer io.WriteCloser) io.Writer {
+	if len(in.outputSinks) > 0 {
+		writers = append(writers, algorithms.Map(
+			in.outputSinks,
+			func(writer io.WriteCloser) io.Writer {
 				return writer
-			}),
-			os.Stdout)...,
-	)
+			})...,
+		)
+	}
+
+	if in.outputAnalyzer != nil {
+		writers = append(writers, in.outputAnalyzer.Stdout())
+	}
+
+	return io.MultiWriter(writers...)
 }
 
 func (in *executable) close(writers []io.WriteCloser) {
@@ -120,6 +134,18 @@ func (in *executable) runLifecycleFunction(lifecycle v1.Lifecycle) error {
 	return nil
 }
 
+func (in *executable) analyze() error {
+	if in.outputAnalyzer == nil {
+		return nil
+	}
+
+	if err := in.outputAnalyzer.Detect(); len(err) > 0 {
+		return errors.Join(err...)
+	}
+
+	return nil
+}
+
 func NewExecutable(command string, options ...Option) Executable {
 	result := &executable{
 		command:       command,

pkg/harness/exec/exec_options.go

@@ -50,3 +50,9 @@ func WithTimeout(timeout time.Duration) Option {
 		e.timeout = timeout
 	}
 }
+
+func WithOutputAnalyzer(heuristics ...OutputAnalyzerHeuristic) Option {
+	return func(e *executable) {
+		e.outputAnalyzer = NewOutputAnalyzer(heuristics...)
+	}
+}

pkg/harness/exec/exec_types.go

@@ -44,6 +44,9 @@ type executable struct {
 	// to the [os.Stdout].
 	outputSinks []io.WriteCloser
 
+	// outputAnalyzer
+	outputAnalyzer OutputAnalyzer
+
 	// hookFunctions ...
 	hookFunctions map[v1.Lifecycle]v1.HookFunction
 }