Added support for Shell runtime

cli/cli.go

@@ -4,6 +4,7 @@ import (
 	"os"
 
 	"github.com/runabol/tork/internal/logging"
+	"github.com/runabol/tork/internal/reexec"
 	ucli "github.com/urfave/cli/v2"
 )
 
@@ -25,6 +26,9 @@ func New() *CLI {
 }
 
 func (c *CLI) Run() error {
+	if reexec.Init() {
+		return nil
+	}
 	return c.app.Run(os.Args)
 }
 

conf/conf.go

@@ -55,7 +55,7 @@ func LoadConfig() error {
 	if userConfig != "" {
 		return errors.Errorf(errMsg)
 	} else {
-		logger.Warn().Msg(errMsg)
+		logger.Debug().Msg(errMsg)
 	}
 	return nil
 }

configs/sample.config.toml

@@ -83,3 +83,11 @@ denylist = []   # supports wildcards (*)
 
 [mounts.temp]
 dir = "/tmp"
+
+[runtime]
+type = "docker" # docker | shell
+
+[runtime.shell]
+cmd = ["bash", "-c"]
+uid = "1000"         # requires running the Tork process as root
+gid = "1000"         # requires running the Tork process as root

engine/worker.go

@@ -4,12 +4,15 @@ import (
 	"github.com/pkg/errors"
 	"github.com/runabol/tork/conf"
 	"github.com/runabol/tork/internal/runtime"
+	"github.com/runabol/tork/internal/runtime/docker"
+	"github.com/runabol/tork/internal/runtime/shell"
 	"github.com/runabol/tork/internal/worker"
 	"github.com/runabol/tork/mount"
 )
 
 func (e *Engine) initWorker() error {
-	rt, err := runtime.NewDockerRuntime()
+	// init the runtime
+	rt, err := initRuntime()
 	if err != nil {
 		return err
 	}
@@ -50,3 +53,19 @@ func (e *Engine) initWorker() error {
 	e.worker = w
 	return nil
 }
+
+func initRuntime() (runtime.Runtime, error) {
+	runtimeType := conf.StringDefault("runtime.type", "docker")
+	switch runtimeType {
+	case "docker":
+		return docker.NewDockerRuntime()
+	case "shell":
+		return shell.NewShellRuntime(shell.Config{
+			CMD: conf.Strings("runtime.shell.cmd"),
+			UID: conf.StringDefault("runtime.shell.uid", shell.DEFAULT_UID),
+			GID: conf.StringDefault("runtime.shell.gid", shell.DEFAULT_GID),
+		}), nil
+	default:
+		return nil, errors.Errorf("unknown runtime type: %s", runtimeType)
+	}
+}

go.mod

@@ -25,8 +25,10 @@ require (
 	github.com/stretchr/testify v1.8.4
 	github.com/urfave/cli/v2 v2.25.7
 	golang.org/x/exp v0.0.0-20230807204917-050eac23e9de
+	golang.org/x/sys v0.12.0
 	golang.org/x/time v0.3.0
 	gopkg.in/yaml.v3 v3.0.1
+	gotest.tools/v3 v3.5.1
 )
 
 require (
@@ -42,6 +44,7 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/labstack/gommon v0.4.0 // indirect
 	github.com/leodido/go-urn v1.2.4 // indirect
@@ -69,8 +72,6 @@ require (
 	golang.org/x/crypto v0.11.0 // indirect
 	golang.org/x/mod v0.11.0 // indirect
 	golang.org/x/net v0.12.0 // indirect
-	golang.org/x/sys v0.12.0 // indirect
 	golang.org/x/text v0.11.0 // indirect
 	golang.org/x/tools v0.6.0 // indirect
-	gotest.tools/v3 v3.5.1 // indirect
 )

input/validate.go

@@ -83,7 +83,6 @@ func validateQueue(fl validator.FieldLevel) bool {
 
 func taskInputValidation(sl validator.StructLevel) {
 	taskTypeValidation(sl)
-	regularTaskValidation(sl)
 	compositeTaskValidation(sl)
 }
 
@@ -149,13 +148,3 @@ func compositeTaskValidation(sl validator.StructLevel) {
 		sl.ReportError(t.Timeout, "timeout", "Timeout", "invalidcompositetask", "")
 	}
 }
-
-func regularTaskValidation(sl validator.StructLevel) {
-	t := sl.Current().Interface().(Task)
-	if t.Parallel != nil || t.Each != nil || t.SubJob != nil {
-		return
-	}
-	if t.Image == "" {
-		sl.ReportError(t.Image, "image", "Image", "required", "")
-	}
-}

input/validate_test.go

@@ -118,6 +118,19 @@ func TestValidateJobTaskNoName(t *testing.T) {
 	assert.Error(t, err)
 }
 
+func TestValidateJobTaskNoImage(t *testing.T) {
+	j := Job{
+		Name: "test job",
+		Tasks: []Task{
+			{
+				Name: "some task",
+			},
+		},
+	}
+	err := j.Validate()
+	assert.NoError(t, err)
+}
+
 func TestValidateJobTaskRetry(t *testing.T) {
 	j := Job{
 		Name: "test job",

internal/cache/cache.go

@@ -113,6 +113,7 @@ func (c *Cache[V]) Modify(k string, m func(x V) (V, error)) error {
 	v := item.Object
 	v, err := m(v)
 	if err != nil {
+		item.mu.Unlock()
 		return err
 	}
 	item.Object = v

internal/cache/cache_test.go

@@ -255,6 +255,10 @@ func TestModify(t *testing.T) {
 		return 0, errors.New("something bad happened")
 	})
 	assert.Error(t, err)
+	err = tc.Modify("number", func(x int) (int, error) {
+		return 17, nil
+	})
+	assert.NoError(t, err)
 }
 
 func TestModifyObjectConcurrently(t *testing.T) {

internal/coordinator/coordinator_test.go

@@ -15,8 +15,7 @@ import (
 	"github.com/runabol/tork/mount"
 	"github.com/runabol/tork/mq"
 
-	"github.com/runabol/tork/internal/runtime"
-
+	"github.com/runabol/tork/internal/runtime/docker"
 	"github.com/runabol/tork/internal/uuid"
 	"github.com/runabol/tork/internal/worker"
 	"github.com/stretchr/testify/assert"
@@ -305,7 +304,7 @@ func doRunJob(t *testing.T, filename string) *tork.Job {
 		assert.NoError(t, c.Stop())
 	}()
 
-	rt, err := runtime.NewDockerRuntime()
+	rt, err := docker.NewDockerRuntime()
 	assert.NoError(t, err)
 
 	mounter, err := mount.NewVolumeMounter()

internal/reexec/command_linux.go

@@ -0,0 +1,35 @@
+//go:build linux
+
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"os/exec"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+// Self returns the path to the current process's binary.
+// Returns "/proc/self/exe".
+func Self() string {
+	return "/proc/self/exe"
+}
+
+// Command returns *exec.Cmd which has Path as current binary. Also it setting
+// SysProcAttr.Pdeathsig to SIGTERM.
+// This will use the in-memory version (/proc/self/exe) of the current binary,
+// it is thus safe to delete or replace the on-disk binary (os.Args[0]).
+//
+// As SysProcAttr.Pdeathsig is set, the signal will be sent to the process when
+// the OS thread which created the process dies. It is the caller's
+// responsibility to ensure that the creating thread is not terminated
+// prematurely. See https://go.dev/issue/27505 for more details.
+func Command(args ...string) *exec.Cmd {
+	return &exec.Cmd{
+		Path: Self(),
+		Args: args,
+		SysProcAttr: &syscall.SysProcAttr{
+			Pdeathsig: unix.SIGTERM,
+		},
+	}
+}

internal/reexec/command_unix.go

@@ -0,0 +1,23 @@
+//go:build freebsd || darwin
+
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"os/exec"
+)
+
+// Self returns the path to the current process's binary.
+// Uses os.Args[0].
+func Self() string {
+	return naiveSelf()
+}
+
+// Command returns *exec.Cmd which has Path as current binary.
+// For example if current binary is "docker" at "/usr/bin/", then cmd.Path will
+// be set to "/usr/bin/docker".
+func Command(args ...string) *exec.Cmd {
+	return &exec.Cmd{
+		Path: Self(),
+		Args: args,
+	}
+}

internal/reexec/command_unsupported.go

@@ -0,0 +1,17 @@
+//go:build !linux && !windows && !freebsd && !darwin
+// +build !linux,!windows,!freebsd,!darwin
+
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"os/exec"
+)
+
+func Self() string {
+	return ""
+}
+
+// Command is unsupported on operating systems apart from Linux, Windows, and Darwin.
+func Command(args ...string) *exec.Cmd {
+	return nil
+}

internal/reexec/command_windows.go

@@ -0,0 +1,21 @@
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"os/exec"
+)
+
+// Self returns the path to the current process's binary.
+// Uses os.Args[0].
+func Self() string {
+	return naiveSelf()
+}
+
+// Command returns *exec.Cmd which has Path as current binary.
+// For example if current binary is "docker.exe" at "C:\", then cmd.Path will
+// be set to "C:\docker.exe".
+func Command(args ...string) *exec.Cmd {
+	return &exec.Cmd{
+		Path: Self(),
+		Args: args,
+	}
+}

internal/reexec/reexec_test.go

@@ -0,0 +1,52 @@
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+func init() {
+	Register("reexec", func() {
+		panic("Return Error")
+	})
+	Init()
+}
+
+func TestRegister(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			assert.Equal(t, `reexec func already registered under name "reexec"`, r)
+		}
+	}()
+	Register("reexec", func() {})
+}
+
+func TestCommand(t *testing.T) {
+	cmd := Command("reexec")
+	w, err := cmd.StdinPipe()
+	assert.NilError(t, err, "Error on pipe creation: %v", err)
+	defer w.Close()
+
+	err = cmd.Start()
+	assert.NilError(t, err, "Error on re-exec cmd: %v", err)
+	err = cmd.Wait()
+	assert.Error(t, err, "exit status 2")
+}
+
+func TestNaiveSelf(t *testing.T) {
+	if os.Getenv("TEST_CHECK") == "1" {
+		os.Exit(2)
+	}
+	cmd := exec.Command(naiveSelf(), "-test.run=TestNaiveSelf")
+	cmd.Env = append(os.Environ(), "TEST_CHECK=1")
+	err := cmd.Start()
+	assert.NilError(t, err, "Unable to start command")
+	err = cmd.Wait()
+	assert.Error(t, err, "exit status 2")
+
+	os.Args[0] = "mkdir"
+	assert.Check(t, naiveSelf() != os.Args[0])
+}

internal/reexec/rexec.go

@@ -0,0 +1,51 @@
+// Package reexec facilitates the busybox style reexec of the docker binary that
+// we require because of the forking limitations of using Go.  Handlers can be
+// registered with a name and the argv 0 of the exec of the binary will be used
+// to find and execute custom init paths.
+package reexec // import "github.com/docker/docker/pkg/reexec"
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+)
+
+var registeredInitializers = make(map[string]func())
+
+// Register adds an initialization func under the specified name
+func Register(name string, initializer func()) {
+	if _, exists := registeredInitializers[name]; exists {
+		panic(fmt.Sprintf("reexec func already registered under name %q", name))
+	}
+
+	registeredInitializers[name] = initializer
+}
+
+// Init is called as the first part of the exec process and returns true if an
+// initialization function was called.
+func Init() bool {
+	initializer, exists := registeredInitializers[os.Args[0]]
+	if exists {
+		initializer()
+
+		return true
+	}
+	return false
+}
+
+func naiveSelf() string {
+	name := os.Args[0]
+	if filepath.Base(name) == name {
+		if lp, err := exec.LookPath(name); err == nil {
+			return lp
+		}
+	}
+	// handle conversion of relative paths to absolute
+	if absName, err := filepath.Abs(name); err == nil {
+		return absName
+	}
+	// if we couldn't get absolute name, return original
+	// (NOTE: Go only errors on Abs() if os.Getwd fails)
+	return name
+}

internal/runtime/docker.go → internal/runtime/docker/docker.go

@@ -1,4 +1,4 @@
-package runtime
+package docker
 
 import (
 	"archive/tar"