[shell-operator] Let an embedding application choose how hooks are run

pkg/executor/executor.go

@@ -51,13 +51,73 @@ func (e *StderrError) Error() string {
 	return e.Message
 }
 
+// Runner starts a prepared hook command and waits for it to finish.
+//
+// cmd carries what the hook is run with: Path, Args, Env, Dir, and the
+// writers its stdout and stderr go to. A Runner must leave the other fields
+// alone. Run returns an error when the hook could not be run or exited with a
+// non-zero code; whatever the hook wrote to stderr is turned into a
+// StderrError by the caller.
+//
+// ProcessRunner, the default, starts cmd as a child process. An application
+// embedding shell-operator may supply its own Runner, for example one that
+// hands the command to a long-lived interpreter instead of starting a fresh
+// one for every event, and fall back to ProcessRunner for the rest.
+type Runner interface {
+	Run(ctx context.Context, cmd *exec.Cmd) (*CmdUsage, error)
+}
+
+// ProcessRunner runs a hook command as a child process. The context is not
+// used to stop the process: a hook runs to completion, as it always has.
+type ProcessRunner struct{}
+
+func (ProcessRunner) Run(_ context.Context, cmd *exec.Cmd) (*CmdUsage, error) {
+	if err := startAndRegister(cmd); err != nil {
+		return nil, fmt.Errorf("cmd start: %w", err)
+	}
+	defer unregisterPID(cmd.Process.Pid)
+
+	if err := cmd.Wait(); err != nil {
+		return nil, fmt.Errorf("cmd run: %w", err)
+	}
+
+	if cmd.ProcessState == nil {
+		return nil, nil
+	}
+
+	usage := &CmdUsage{
+		Sys:  cmd.ProcessState.SystemTime(),
+		User: cmd.ProcessState.UserTime(),
+	}
+
+	// FIXME Maxrss is Unix specific.
+	sysUsage := cmd.ProcessState.SysUsage()
+	if v, ok := sysUsage.(*syscall.Rusage); ok {
+		// v.Maxrss is int32 on arm/v7
+		usage.MaxRss = int64(v.Maxrss) //nolint:unconvert
+	}
+
+	return usage, nil
+}
+
 type Executor struct {
 	cmd              *exec.Cmd
+	runner           Runner
 	logProxyHookJSON bool
 	proxyJsonKey     string
 	logger           *log.Logger
 }
 
+// WithRunner sets how RunAndLogLines runs the command. A nil runner keeps
+// ProcessRunner. Output, used for the --config call, always starts a process.
+func (e *Executor) WithRunner(runner Runner) *Executor {
+	if runner != nil {
+		e.runner = runner
+	}
+
+	return e
+}
+
 func (e *Executor) WithLogProxyHookJSON(logProxyHookJSON bool) *Executor {
 	e.logProxyHookJSON = logProxyHookJSON
 
@@ -111,6 +171,7 @@ func NewExecutor(dir string, entrypoint string, args []string, envs []string) *E
 
 	ex := &Executor{
 		cmd:          cmd,
+		runner:       ProcessRunner{},
 		proxyJsonKey: "proxyJsonLog",
 		logger:       log.NewLogger().Named("auto-executor"),
 	}
@@ -190,33 +251,13 @@ func (e *Executor) RunAndLogLines(ctx context.Context, logLabels map[string]stri
 	e.cmd.Stdout = plo
 	e.cmd.Stderr = io.MultiWriter(ple, stdErr)
 
-	if err := startAndRegister(e.cmd); err != nil {
-		return nil, fmt.Errorf("cmd start: %w", err)
-	}
-	defer unregisterPID(e.cmd.Process.Pid)
-
-	err := e.cmd.Wait()
+	usage, err := e.runner.Run(ctx, e.cmd)
 	if err != nil {
 		if len(stdErr.Bytes()) > 0 {
 			return nil, &StderrError{Message: stdErr.String()}
 		}
 
-		return nil, fmt.Errorf("cmd run: %w", err)
-	}
-
-	var usage *CmdUsage
-	if e.cmd.ProcessState != nil {
-		usage = &CmdUsage{
-			Sys:  e.cmd.ProcessState.SystemTime(),
-			User: e.cmd.ProcessState.UserTime(),
-		}
-
-		// FIXME Maxrss is Unix specific.
-		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
-		}
+		return nil, err
 	}
 
 	return usage, nil

pkg/executor/executor_test.go

@@ -470,3 +470,77 @@ func TestGlobalRegistry_RunAndLogLines_FailedStart(t *testing.T) {
 	_, err := ex.RunAndLogLines(context.Background(), map[string]string{})
 	assert.Error(t, err)
 }
+
+// fakeRunner stands in for a Runner that does not start a process: it writes
+// what a hook would have printed and reports the configured result.
+type fakeRunner struct {
+	stdout, stderr string
+	usage          *CmdUsage
+	err            error
+
+	gotCmd *exec.Cmd
+}
+
+func (r *fakeRunner) Run(_ context.Context, cmd *exec.Cmd) (*CmdUsage, error) {
+	r.gotCmd = cmd
+	_, _ = io.WriteString(cmd.Stdout, r.stdout)
+	_, _ = io.WriteString(cmd.Stderr, r.stderr)
+
+	return r.usage, r.err
+}
+
+func TestRunAndLogLines_WithRunner(t *testing.T) {
+	logger := log.NewLogger()
+	logger.SetLevel(log.LevelInfo)
+
+	var buf bytes.Buffer
+	logger.SetOutput(&buf)
+
+	t.Run("runner gets the command and its output is logged", func(t *testing.T) {
+		buf.Reset()
+		runner := &fakeRunner{stdout: "from runner\n", usage: &CmdUsage{User: time.Second}}
+
+		ex := NewExecutor("/hooks", "/hooks/hook.py", []string{"arg"}, []string{"FOO=bar"}).
+			WithRunner(runner).
+			WithLogger(logger)
+
+		usage, err := ex.RunAndLogLines(context.Background(), map[string]string{})
+		require.NoError(t, err)
+
+		assert.Equal(t, &CmdUsage{User: time.Second}, usage)
+		assert.Equal(t, []string{"/hooks/hook.py", "arg"}, runner.gotCmd.Args)
+		assert.Equal(t, "/hooks", runner.gotCmd.Dir)
+		assert.Contains(t, runner.gotCmd.Env, "FOO=bar")
+		assert.Contains(t, buf.String(), `"msg":"from runner"`)
+	})
+
+	t.Run("failure with stderr is a StderrError", func(t *testing.T) {
+		runner := &fakeRunner{stderr: "boom", err: fmt.Errorf("exit status 1")}
+
+		ex := NewExecutor("", "hook", nil, nil).WithRunner(runner).WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(context.Background(), map[string]string{})
+
+		var stderrErr *StderrError
+		require.ErrorAs(t, err, &stderrErr)
+		assert.Equal(t, "boom", stderrErr.Message)
+	})
+
+	t.Run("failure without stderr keeps the runner error", func(t *testing.T) {
+		runnerErr := fmt.Errorf("runner is gone")
+		runner := &fakeRunner{err: runnerErr}
+
+		ex := NewExecutor("", "hook", nil, nil).WithRunner(runner).WithLogger(logger)
+
+		_, err := ex.RunAndLogLines(context.Background(), map[string]string{})
+		assert.ErrorIs(t, err, runnerErr)
+	})
+
+	t.Run("nil runner keeps starting a process", func(t *testing.T) {
+		ex := NewExecutor("", "sh", []string{"-c", "exit 0"}, nil).WithRunner(nil).WithLogger(logger)
+
+		usage, err := ex.RunAndLogLines(context.Background(), map[string]string{})
+		require.NoError(t, err)
+		assert.NotNil(t, usage)
+	})
+}

pkg/hook/hook.go

@@ -53,6 +53,9 @@ type Hook struct {
 
 	Logger *log.Logger
 
+	// runner runs the hook for an event. Nil means executor.ProcessRunner.
+	runner executor.Runner
+
 	// ioProvider prepares and cleans up temporary files for hook execution.
 	// Defaults to hookIOProvider. Tests may inject a stub.
 	ioProvider IOProvider
@@ -76,6 +79,12 @@ func NewHook(name, path string, keepTemporaryHookFiles bool, logProxyHookJSON bo
 	return h
 }
 
+// WithRunner sets how the hook is run for an event. The --config call made
+// while loading the hook always starts a process.
+func (h *Hook) WithRunner(runner executor.Runner) {
+	h.runner = runner
+}
+
 func (h *Hook) WithTmpDir(dir string) {
 	h.TmpDir = dir
 }
@@ -131,6 +140,7 @@ func (h *Hook) Run(ctx context.Context, _ htypes.BindingType, context []bctx.Bin
 		h.Path,
 		[]string{},
 		env.Envs()).
+		WithRunner(h.runner).
 		WithLogProxyHookJSON(h.LogProxyHookJSON).
 		WithLogProxyHookJSONKey(h.LogProxyHookJSONKey).
 		WithLogger(h.Logger.Named("executor"))

pkg/hook/hook_manager.go

@@ -45,6 +45,7 @@ type Manager struct {
 	keepTemporaryHookFiles bool
 	logProxyHookJSON       bool
 	logProxyHookJSONKey    string
+	hookRunner             executor.Runner
 
 	// indexMu protects hookNamesInOrder, hooksByName and hooksInOrder so that
 	// Init() (which fully rebuilds the indices) does not race with concurrent
@@ -82,6 +83,10 @@ type ManagerConfig struct {
 	LogProxyHookJSON       bool
 	LogProxyHookJSONKey    string
 
+	// HookRunner runs hooks for events. When nil, every hook is started as a
+	// child process (executor.ProcessRunner).
+	HookRunner executor.Runner
+
 	Logger *log.Logger
 }
 
@@ -107,6 +112,7 @@ func NewHookManager(config *ManagerConfig) *Manager {
 		keepTemporaryHookFiles: config.KeepTemporaryHookFiles,
 		logProxyHookJSON:       config.LogProxyHookJSON,
 		logProxyHookJSONKey:    config.LogProxyHookJSONKey,
+		hookRunner:             config.HookRunner,
 
 		logger: config.Logger,
 	}
@@ -270,6 +276,7 @@ func (hm *Manager) wireHookController(hook *Hook) {
 	hookCtrl.InitAdmissionBindings(hook.GetConfig().KubernetesValidating, hook.GetConfig().KubernetesMutating, hm.admissionWebhookManager)
 	hook.WithHookController(hookCtrl)
 	hook.WithTmpDir(hm.TempDir())
+	hook.WithRunner(hm.hookRunner)
 }
 
 func (hm *Manager) execCommandOutput(hookName string, dir string, entrypoint string, envs []string, args []string) ([]byte, error) {

pkg/hook/hook_manager_test.go

@@ -2,13 +2,15 @@ package hook
 
 import (
 	"context"
+	"os/exec"
 	"path/filepath"
 	"strings"
 	"testing"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 	. "github.com/onsi/gomega"
 
+	"github.com/flant/shell-operator/pkg/executor"
 	"github.com/flant/shell-operator/pkg/hook/controller"
 	"github.com/flant/shell-operator/pkg/hook/types"
 	"github.com/flant/shell-operator/pkg/webhook/admission"
@@ -448,3 +450,36 @@ func Test_ManagerConfig_RenamedFields_CompileCheck(t *testing.T) {
 		t.Fatal("expected nil ScheduleManager")
 	}
 }
+
+type recordingRunner struct {
+	paths []string
+	envs  [][]string
+}
+
+func (r *recordingRunner) Run(_ context.Context, cmd *exec.Cmd) (*executor.CmdUsage, error) {
+	r.paths = append(r.paths, cmd.Path)
+	r.envs = append(r.envs, cmd.Env)
+
+	return &executor.CmdUsage{}, nil
+}
+
+func Test_ManagerConfig_HookRunner_RunsHooks(t *testing.T) {
+	g := NewWithT(t)
+
+	hm := newHookManager(t, "testdata/hook_manager")
+	runner := &recordingRunner{}
+	hm.hookRunner = runner
+
+	// --config is still fetched by starting the hook as a process.
+	g.Expect(hm.Init()).To(Succeed())
+	g.Expect(runner.paths).To(BeEmpty())
+
+	h := hm.GetHook("configMapHooks/hook.sh")
+	g.Expect(h).NotTo(BeNil())
+
+	_, err := h.Run(context.Background(), types.OnKubernetesEvent, nil, map[string]string{})
+	g.Expect(err).NotTo(HaveOccurred())
+
+	g.Expect(runner.paths).To(Equal([]string{h.Path}))
+	g.Expect(runner.envs[0]).To(ContainElement(HavePrefix("BINDING_CONTEXT_PATH=")))
+}

pkg/shell-operator/bootstrap.go

@@ -318,6 +318,7 @@ func (op *ShellOperator) setupHookManagers(cfg *app.Config, hooksDir string, tem
 		KeepTemporaryHookFiles:   cfg.Debug.KeepTempFiles,
 		LogProxyHookJSON:         cfg.Log.ProxyHookJSON,
 		LogProxyHookJSONKey:      app.ProxyJsonLogKey,
+		HookRunner:               op.hookRunner,
 		Logger:                   op.logger.Named("hook-manager"),
 	}
 	op.HookManager = hook.NewHookManager(hookCfg)

pkg/shell-operator/operator.go

@@ -29,6 +29,7 @@ import (
 
 	klient "github.com/flant/kube-client/client"
 	pkg "github.com/flant/shell-operator/pkg"
+	"github.com/flant/shell-operator/pkg/executor"
 	"github.com/flant/shell-operator/pkg/hook"
 	bindingcontext "github.com/flant/shell-operator/pkg/hook/binding_context"
 	"github.com/flant/shell-operator/pkg/hook/controller"
@@ -96,6 +97,9 @@ type ShellOperator struct {
 	// as a field so extenders can substitute a custom factory if needed.
 	taskFactory HookTaskFactory
 
+	// hookRunner runs hooks for events; nil starts each hook as a process.
+	hookRunner executor.Runner
+
 	// MetricNames is a per-instance snapshot of resolved metric names.
 	// Library callers may consult it to avoid relying on the package-level
 	// metrics globals. NewShellOperator populates it from cfg's prefix; the
@@ -138,6 +142,14 @@ func WithMetricStorage(ms metricsstorage.Storage) Option {
 	}
 }
 
+// WithHookRunner sets how hooks are run for events. By default every hook is
+// started as a child process; see executor.Runner.
+func WithHookRunner(runner executor.Runner) Option {
+	return func(operator *ShellOperator) {
+		operator.hookRunner = runner
+	}
+}
+
 // WithHookMetricStorage overrides the default per-instance HookMetricStorage.
 func WithHookMetricStorage(ms metricsstorage.Storage) Option {
 	return func(operator *ShellOperator) {