[shell-operator] refactoring

cmd/shell-operator/main.go

@@ -10,7 +10,7 @@ import (
 	"github.com/flant/kube-client/klogtolog"
 	"github.com/flant/shell-operator/pkg/app"
 	"github.com/flant/shell-operator/pkg/debug"
-	"github.com/flant/shell-operator/pkg/jq"
+	"github.com/flant/shell-operator/pkg/filter/jq"
 	shell_operator "github.com/flant/shell-operator/pkg/shell-operator"
 	utils_signal "github.com/flant/shell-operator/pkg/utils/signal"
 )
@@ -33,7 +33,8 @@ func main() {
 	// print version
 	kpApp.Command("version", "Show version.").Action(func(_ *kingpin.ParseContext) error {
 		fmt.Printf("%s %s\n", app.AppName, app.Version)
-		fmt.Println(jq.FilterInfo())
+		fl := jq.NewFilter(app.JqLibraryPath)
+		fmt.Println(fl.FilterInfo())
 		return nil
 	})
 

pkg/app/debug.go

@@ -10,7 +10,10 @@ var DebugUnixSocket = "/var/run/shell-operator/debug.socket"
 
 var DebugHttpServerAddr = ""
 
-var DebugKeepTmpFiles = "no"
+var (
+	DebugKeepTmpFilesVar = "no"
+	DebugKeepTmpFiles    = false
+)
 
 var DebugKubernetesAPI = false
 
@@ -27,8 +30,12 @@ func DefineDebugFlags(kpApp *kingpin.Application, cmd *kingpin.CmdClause) {
 	cmd.Flag("debug-keep-tmp-files", "set to yes to disable cleanup of temporary files").
 		Envar("DEBUG_KEEP_TMP_FILES").
 		Hidden().
-		Default(DebugKeepTmpFiles).
-		StringVar(&DebugKeepTmpFiles)
+		Default(DebugKeepTmpFilesVar).Action(func(_ *kingpin.ParseContext) error {
+		DebugKeepTmpFiles = DebugKeepTmpFilesVar == "yes"
+
+		return nil
+	}).
+		StringVar(&DebugKeepTmpFilesVar)
 
 	cmd.Flag("debug-kubernetes-api", "enable client-go debug messages").
 		Envar("DEBUG_KUBERNETES_API").

pkg/executor/executor.go

@@ -15,16 +15,9 @@ import (
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 
-	"github.com/flant/shell-operator/pkg/app"
 	utils "github.com/flant/shell-operator/pkg/utils/labels"
 )
 
-type CmdUsage struct {
-	Sys    time.Duration
-	User   time.Duration
-	MaxRss int64
-}
-
 func Run(cmd *exec.Cmd) error {
 	// TODO context: hook name, hook phase, hook binding
 	// TODO observability
@@ -33,21 +26,86 @@ func Run(cmd *exec.Cmd) error {
 	return cmd.Run()
 }
 
-func RunAndLogLines(cmd *exec.Cmd, logLabels map[string]string, logger *log.Logger) (*CmdUsage, error) {
-	// TODO observability
+type Executor struct {
+	cmd              *exec.Cmd
+	logProxyHookJSON bool
+	proxyJsonKey     string
+	logger           *log.Logger
+}
+
+func (e *Executor) WithLogProxyHookJSON(logProxyHookJSON bool) *Executor {
+	e.logProxyHookJSON = logProxyHookJSON
+
+	return e
+}
+
+func (e *Executor) WithLogProxyHookJSONKey(logProxyHookJSONKey string) *Executor {
+	if logProxyHookJSONKey == "" {
+		return e
+	}
+
+	e.proxyJsonKey = logProxyHookJSONKey
+
+	return e
+}
+
+func (e *Executor) WithLogger(logger *log.Logger) *Executor {
+	e.logger = logger
+
+	return e
+}
+
+func (e *Executor) WithCMDStdout(w io.Writer) *Executor {
+	e.cmd.Stdout = w
+
+	return e
+}
+
+func (e *Executor) WithCMDStderr(w io.Writer) *Executor {
+	e.cmd.Stderr = w
+
+	return e
+}
+
+func NewExecutor(dir string, entrypoint string, args []string, envs []string) *Executor {
+	cmd := exec.Command(entrypoint, args...)
+	cmd.Env = append(cmd.Env, envs...)
+	cmd.Dir = dir
+
+	ex := &Executor{
+		cmd:          cmd,
+		proxyJsonKey: "proxyJsonLog",
+		logger:       log.NewLogger(log.Options{}).Named("auto-executor"),
+	}
+
+	return ex
+}
+
+func (e *Executor) Output() ([]byte, error) {
+	e.logger.Debugf("Executing command '%s' in '%s' dir", strings.Join(e.cmd.Args, " "), e.cmd.Dir)
+	return e.cmd.Output()
+}
+
+type CmdUsage struct {
+	Sys    time.Duration
+	User   time.Duration
+	MaxRss int64
+}
+
+func (e *Executor) RunAndLogLines(logLabels map[string]string) (*CmdUsage, error) {
 	stdErr := bytes.NewBuffer(nil)
-	logEntry := utils.EnrichLoggerWithLabels(logger, logLabels)
+	logEntry := utils.EnrichLoggerWithLabels(e.logger, logLabels)
 	stdoutLogEntry := logEntry.With("output", "stdout")
 	stderrLogEntry := logEntry.With("output", "stderr")
 
-	logEntry.Debugf("Executing command '%s' in '%s' dir", strings.Join(cmd.Args, " "), cmd.Dir)
+	logEntry.Debugf("Executing command '%s' in '%s' dir", strings.Join(e.cmd.Args, " "), e.cmd.Dir)
 
-	plo := &proxyLogger{app.LogProxyHookJSON, stdoutLogEntry, make([]byte, 0)}
-	ple := &proxyLogger{app.LogProxyHookJSON, stderrLogEntry, make([]byte, 0)}
-	cmd.Stdout = plo
-	cmd.Stderr = io.MultiWriter(ple, stdErr)
+	plo := &proxyLogger{e.logProxyHookJSON, e.proxyJsonKey, stdoutLogEntry, make([]byte, 0)}
+	ple := &proxyLogger{e.logProxyHookJSON, e.proxyJsonKey, stderrLogEntry, make([]byte, 0)}
+	e.cmd.Stdout = plo
+	e.cmd.Stderr = io.MultiWriter(ple, stdErr)
 
-	err := cmd.Run()
+	err := e.cmd.Run()
 	if err != nil {
 		if len(stdErr.Bytes()) > 0 {
 			return nil, fmt.Errorf("%s", stdErr.String())
@@ -57,14 +115,14 @@ func RunAndLogLines(cmd *exec.Cmd, logLabels map[string]string, logger *log.Logg
 	}
 
 	var usage *CmdUsage
-	if cmd.ProcessState != nil {
+	if e.cmd.ProcessState != nil {
 		usage = &CmdUsage{
-			Sys:  cmd.ProcessState.SystemTime(),
-			User: cmd.ProcessState.UserTime(),
+			Sys:  e.cmd.ProcessState.SystemTime(),
+			User: e.cmd.ProcessState.UserTime(),
 		}
 
 		// FIXME Maxrss is Unix specific.
-		sysUsage := cmd.ProcessState.SysUsage()
+		sysUsage := e.cmd.ProcessState.SysUsage()
 		if v, ok := sysUsage.(*syscall.Rusage); ok {
 			// v.Maxrss is int32 on arm/v7
 			usage.MaxRss = int64(v.Maxrss) //nolint:unconvert
@@ -76,6 +134,7 @@ func RunAndLogLines(cmd *exec.Cmd, logLabels map[string]string, logger *log.Logg
 
 type proxyLogger struct {
 	logProxyHookJSON bool
+	proxyJsonLogKey  string
 
 	logger *log.Logger
 
@@ -116,7 +175,7 @@ func (pl *proxyLogger) Write(p []byte) (int, error) {
 		return len(p), err
 	}
 
-	logger := pl.logger.With(app.ProxyJsonLogKey, true)
+	logger := pl.logger.With(pl.proxyJsonLogKey, true)
 
 	logLineRaw, _ := json.Marshal(logMap)
 	logLine := string(logLineRaw)
@@ -176,18 +235,3 @@ func (pl *proxyLogger) writerScanner(p []byte) {
 		pl.logger.Error("reading from scanner", slog.String("error", err.Error()))
 	}
 }
-
-func Output(cmd *exec.Cmd) (output []byte, err error) {
-	// TODO context: hook name, hook phase, hook binding
-	// TODO observability
-	log.Debugf("Executing command '%s' in '%s' dir", strings.Join(cmd.Args, " "), cmd.Dir)
-	output, err = cmd.Output()
-	return
-}
-
-func MakeCommand(dir string, entrypoint string, args []string, envs []string) *exec.Cmd {
-	cmd := exec.Command(entrypoint, args...)
-	cmd.Env = append(cmd.Env, envs...)
-	cmd.Dir = dir
-	return cmd
-}

pkg/executor/executor_test.go

@@ -5,16 +5,13 @@ import (
 	"io"
 	"math/rand/v2"
 	"os"
-	"os/exec"
 	"regexp"
 	"testing"
 	"time"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	"github.com/flant/shell-operator/pkg/app"
 )
 
 func TestRunAndLogLines(t *testing.T) {
@@ -35,11 +32,11 @@ func TestRunAndLogLines(t *testing.T) {
 	logger.SetOutput(&buf)
 
 	t.Run("simple log", func(t *testing.T) {
-		app.LogProxyHookJSON = true
-
-		cmd := exec.Command("echo", `{"foo": "baz"}`)
+		ex := NewExecutor("", "echo", []string{`{"foo": "baz"}`}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
 
-		_, err := RunAndLogLines(cmd, map[string]string{"a": "b"}, logger)
+		_, err := ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
 		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","a":"b","hook":{"foo":"baz"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
@@ -48,10 +45,10 @@ func TestRunAndLogLines(t *testing.T) {
 	})
 
 	t.Run("not json log", func(t *testing.T) {
-		app.LogProxyHookJSON = false
-		cmd := exec.Command("echo", `foobar`)
+		ex := NewExecutor("", "echo", []string{"foobar"}, []string{}).
+			WithLogger(logger)
 
-		_, err := RunAndLogLines(cmd, map[string]string{"a": "b"}, logger)
+		_, err := ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
 		assert.Equal(t, buf.String(), `{"level":"info","msg":"foobar","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
@@ -67,10 +64,11 @@ func TestRunAndLogLines(t *testing.T) {
 
 		_, _ = io.WriteString(f, `{"foo": "`+randStringRunes(1024*1024)+`"}`)
 
-		app.LogProxyHookJSON = true
-		cmd := exec.Command("cat", f.Name())
+		ex := NewExecutor("", "cat", []string{f.Name()}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
 
-		_, err = RunAndLogLines(cmd, map[string]string{"a": "b"}, logger)
+		_, err = ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
 		reg := regexp.MustCompile(`{"level":"fatal","msg":"hook result","a":"b","hook":{"truncated":".*:truncated"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"`)
@@ -87,10 +85,10 @@ func TestRunAndLogLines(t *testing.T) {
 
 		_, _ = io.WriteString(f, `result `+randStringRunes(1024*1024))
 
-		app.LogProxyHookJSON = false
-		cmd := exec.Command("cat", f.Name())
+		ex := NewExecutor("", "cat", []string{f.Name()}, []string{}).
+			WithLogger(logger)
 
-		_, err = RunAndLogLines(cmd, map[string]string{"a": "b"}, logger)
+		_, err = ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
 
 		reg := regexp.MustCompile(`{"level":"info","msg":"result .*:truncated","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"`)
@@ -101,38 +99,47 @@ func TestRunAndLogLines(t *testing.T) {
 
 	t.Run("invalid json structure", func(t *testing.T) {
 		logger.SetLevel(log.LevelDebug)
-		app.LogProxyHookJSON = true
-		cmd := exec.Command("echo", `["a","b","c"]`)
-		_, err := RunAndLogLines(cmd, map[string]string{"a": "b"}, logger)
+
+		ex := NewExecutor("", "echo", []string{`["a","b","c"]`}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"a": "b"})
 		assert.NoError(t, err)
-		assert.Equal(t, buf.String(), `{"level":"debug","msg":"Executing command 'echo [\"a\",\"b\",\"c\"]' in '' dir","source":"executor/executor.go:43","a":"b","time":"2006-01-02T15:04:05Z"}`+"\n"+
-			`{"level":"debug","msg":"json log line not map[string]interface{}","source":"executor/executor.go:111","a":"b","line":["a","b","c"],"output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
-			`{"level":"info","msg":"[\"a\",\"b\",\"c\"]\n","source":"executor/executor.go:114","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
+
+		assert.Equal(t, buf.String(), `{"level":"debug","msg":"Executing command 'echo [\"a\",\"b\",\"c\"]' in '' dir","source":"executor/executor.go:101","a":"b","time":"2006-01-02T15:04:05Z"}`+"\n"+
+			`{"level":"debug","msg":"json log line not map[string]interface{}","source":"executor/executor.go:170","a":"b","line":["a","b","c"],"output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
+			`{"level":"info","msg":"[\"a\",\"b\",\"c\"]\n","source":"executor/executor.go:173","a":"b","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})
 
 	t.Run("multiline", func(t *testing.T) {
 		logger.SetLevel(log.LevelInfo)
-		app.LogProxyHookJSON = true
-		cmd := exec.Command("echo", `
+		arg := `
 {"a":"b",
 "c":"d"}
-`)
-		_, err := RunAndLogLines(cmd, map[string]string{"foor": "baar"}, logger)
+`
+		ex := NewExecutor("", "echo", []string{arg}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
 		assert.NoError(t, err)
 		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","foor":"baar","hook":{"a":"b","c":"d"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
 
 		buf.Reset()
 	})
 
 	t.Run("multiline non json", func(t *testing.T) {
-		app.LogProxyHookJSON = false
-		cmd := exec.Command("echo", `
+		arg := `
 a b
 c d
-`)
-		_, err := RunAndLogLines(cmd, map[string]string{"foor": "baar"}, logger)
+`
+		ex := NewExecutor("", "echo", []string{arg}, []string{}).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
 		assert.NoError(t, err)
 		assert.Equal(t, buf.String(), `{"level":"info","msg":"a b","foor":"baar","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n"+
 			`{"level":"info","msg":"c d","foor":"baar","output":"stdout","time":"2006-01-02T15:04:05Z"}`+"\n")
@@ -141,12 +148,15 @@ c d
 	})
 
 	t.Run("multiline json", func(t *testing.T) {
-		app.LogProxyHookJSON = true
-		cmd := exec.Command("echo", `{
+		arg := `{
 "a":"b",
 "c":"d"
-}`)
-		_, err := RunAndLogLines(cmd, map[string]string{"foor": "baar"}, logger)
+}`
+		ex := NewExecutor("", "echo", []string{arg}, []string{}).
+			WithLogProxyHookJSON(true).
+			WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(map[string]string{"foor": "baar"})
 		assert.NoError(t, err)
 		assert.Equal(t, buf.String(), `{"level":"fatal","msg":"hook result","foor":"baar","hook":{"a":"b","c":"d"},"output":"stdout","proxyJsonLog":true,"time":"2006-01-02T15:04:05Z"}`+"\n")
 

pkg/filter/filter.go

@@ -0,0 +1,6 @@
+package filter
+
+type Filter interface {
+	ApplyFilter(filterStr string, data []byte) (string, error)
+	FilterInfo() string
+}