feature(systemd): added the ability to start a service with the predefined config

Author: pzindyaev

.gitignore

@@ -1,2 +1,3 @@
 # Added by goreleaser init:
 dist/
+cmd/ad-runtime-utils/ad-runtime-utils

README.MD

@@ -91,4 +91,15 @@ Default runtimes:
 Service <service-name>:
   <runtime>: <path or "error: <message>">
   …
-```
+```
+
+### 4. Starting a Service
+
+When --start and/or --supervise flags are provided --service is a required argument.
+
+- It starts the executable `services.<service-name>.executable` with the specified arguments from `services.<service-name>.executable_args`.
+
+- If the `--supervise` flag is provided, it will run the health checks, defined in `services.<service-name>.health_checks` on service start to make sure the service is operational. If any of the checks fail the service will be stopped. `Type=notify` should be used in the systemd unit(see example in `examples/systemd` directory).
+
+- While using `--supervise` and health checks, make sure that systemd service has enough `TimeoutStartSec`. ideally should be a combined timeout of all health checks.
+

cmd/ad-runtime-utils/app.go

@@ -4,10 +4,13 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"os"
 	"strings"
 
 	"github.com/arenadata/ad-runtime-utils/internal/config"
 	"github.com/arenadata/ad-runtime-utils/internal/detect"
+	"github.com/arenadata/ad-runtime-utils/internal/exec"
+	"github.com/coreos/go-systemd/v22/daemon"
 )
 
 // exit codes.
@@ -27,6 +30,8 @@ func Run(args []string, stdout, stderr io.Writer) int {
 	listAll := fs.Bool("list", false, "List all detected runtimes (default + services)")
 	fs.BoolVar(listAll, "l", false, "shorthand for --list")
 	printCACerts := fs.Bool("print-cacerts", false, "When used with --runtime=java, prints the cacerts path and exits")
+	start := fs.Bool("start", false, "Start the service. Use with simple/exec services")
+	supervise := fs.Bool("supervise", false, "Supervise the service. Use with norify services")
 
 	if err := fs.Parse(args); err != nil {
 		return exitParseError
@@ -76,6 +81,15 @@ func Run(args []string, stdout, stderr io.Writer) int {
 	}
 
 	envName := detectEnvName(cfg, *service, *runtime)
+
+	if *start {
+		if err = startService(*service, envName, path, *cfg, *supervise); err != nil {
+			fmt.Fprintf(stderr, "start service failed: %v\n", err)
+			return exitUserError
+		}
+		return exitOK
+	}
+
 	fmt.Fprintf(stdout, "export %s=%s\n", envName, path)
 	return exitOK
 }
@@ -124,3 +138,52 @@ func runList(cfg *config.Config, stdout, stderr io.Writer) int {
 	}
 	return exitOK
 }
+
+func startService(service string, envName string, envPath string, cfg config.Config, supervise bool) error {
+	srvConfig, ok := cfg.Services[service]
+	if !ok {
+		return fmt.Errorf("service %s not found in config", service)
+	}
+	// Append the env for the runtime (eg. JAVA_HOME)
+	if srvConfig.EnvVars == nil {
+		srvConfig.EnvVars = make(map[string]string)
+	}
+	srvConfig.EnvVars[envName] = envPath
+	if !supervise {
+		return exec.RunExecutable(srvConfig.Executable, srvConfig.ExecutableArgs, srvConfig.EnvVars)
+	}
+	process, err := exec.RunExecutableAsync(srvConfig.Executable, srvConfig.ExecutableArgs, srvConfig.EnvVars)
+	if err != nil {
+		return err
+	}
+	// Run the health checks
+	for _, checkCfg := range srvConfig.HealthChecks {
+		switch checkCfg.Type {
+		case exec.PortHealthCheckType:
+			portheck := exec.PortHealthCheck{
+				PID:    process.Process.Pid,
+				Config: checkCfg,
+			}
+			if err = portheck.Check(); err != nil {
+				if err = process.Process.Signal(os.Interrupt); err != nil {
+					return fmt.Errorf("failed to send interrupt signal to process: %w", err)
+				}
+				return fmt.Errorf("health check failed: %w", err)
+			}
+		default:
+			if err = process.Process.Signal(os.Interrupt); err != nil {
+				return fmt.Errorf("failed to send interrupt signal to process: %w", err)
+			}
+			return fmt.Errorf("unknown health check type: %s", checkCfg.Type)
+		}
+	}
+	// Notify systemd daemon that service has started
+	if _, err = daemon.SdNotify(false, daemon.SdNotifyReady); err != nil {
+		fmt.Fprintf(os.Stderr, "systemd notification failed: %v\n", err)
+	}
+	// TODO: Replace this with an actual supervisor loop
+	if err = process.Wait(); err != nil {
+		return err
+	}
+	return nil
+}

configs/config.yaml

@@ -91,4 +91,4 @@ services:
   FLINK2:
       path: /etc/flink2/conf/flink2-java.yaml
   OZONE:
-      path: /etc/ozone/conf/ozone-java.yaml
+      path: /etc/ozone/conf/ozone-java.yaml

examples/config/kafka.yaml

@@ -0,0 +1,17 @@
+executable: bin/kafka-server-start.sh
+executable_args:
+  - config/server.properties
+runtimes:
+  java:
+    version: "21"
+health_checks:
+  - type: port
+    params:
+      port: 9092
+      timeout: 20
+      protocol: tcp
+  - type: port
+    params:
+      port: 9093
+      timeout: 20
+      protocol: tcp

examples/systemd/kafka.service

@@ -0,0 +1,10 @@
+[Unit]
+Description=Test Sleep
+
+[Service]
+Type=notify
+ExecStart=bin/ad-runtime-utils --config configs/config.yaml --service kafka --runtime java --start --supervise
+TimeoutStartSec=120
+
+[Install]
+WantedBy=multi-user.target

go.mod

@@ -2,4 +2,7 @@ module github.com/arenadata/ad-runtime-utils
 
 go 1.24.4
 
-require github.com/goccy/go-yaml v1.18.0
+require (
+	github.com/coreos/go-systemd/v22 v22.6.0
+	github.com/goccy/go-yaml v1.18.0
+)

go.sum

@@ -1,2 +1,4 @@
+github.com/coreos/go-systemd/v22 v22.6.0 h1:aGVa/v8B7hpb0TKl0MWoAavPDmHvobFe5R5zn0bCJWo=
+github.com/coreos/go-systemd/v22 v22.6.0/go.mod h1:iG+pp635Fo7ZmV/j14KUcmEyWF+0X7Lua8rrTWzYgWU=
 github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw=
 github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=

internal/config/config.go

@@ -1,8 +1,10 @@
 package config
 
 import (
+	"bufio"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/goccy/go-yaml"
 )
@@ -14,9 +16,19 @@ type RuntimeSetting struct {
 	Paths        []string `yaml:"paths,omitempty"`
 }
 
+type HealthCheckConfig struct {
+	Type   string         `yaml:"type"`
+	Params map[string]any `yaml:"params,omitempty"`
+}
+
 type ServiceConfig struct {
-	Runtimes map[string]RuntimeSetting `yaml:"runtimes,omitempty"`
-	Path     string                    `yaml:"path,omitempty"`
+	Runtimes       map[string]RuntimeSetting `yaml:"runtimes,omitempty"`
+	Path           string                    `yaml:"path,omitempty"`
+	Executable     string                    `yaml:"executable,omitempty"`
+	ExecutableArgs []string                  `yaml:"executable_args,omitempty"`
+	EnvVars        map[string]string         `yaml:"env_vars,omitempty"`
+	EnvVarsFile    string                    `yaml:"env_vars_file,omitempty"`
+	HealthChecks   []HealthCheckConfig       `yaml:"health_checks,omitempty"`
 }
 
 type Config struct {
@@ -31,6 +43,30 @@ type Config struct {
 	Services map[string]ServiceConfig `yaml:"services"`
 }
 
+func (h *HealthCheckConfig) ParamToInt(name string) (int, error) {
+	val, found := h.Params[name]
+	if !found {
+		return 0, fmt.Errorf("health check param %q not found", name)
+	}
+	intVal, ok := val.(int)
+	if !ok {
+		return 0, fmt.Errorf("health check param %q is not an integer", name)
+	}
+	return intVal, nil
+}
+
+func (h *HealthCheckConfig) ParamToString(name string) (string, error) {
+	val, found := h.Params[name]
+	if !found {
+		return "", fmt.Errorf("health check param %q not found", name)
+	}
+	strVal, ok := val.(string)
+	if !ok {
+		return "", fmt.Errorf("health check param %q is not a string", name)
+	}
+	return strVal, nil
+}
+
 func Load(path string) (*Config, error) {
 	data, readErr := os.ReadFile(path)
 	if readErr != nil {
@@ -44,37 +80,97 @@ func Load(path string) (*Config, error) {
 	}
 
 	for name, svc := range cfg.Services {
-		if svc.Path == "" {
-			continue
-		}
-
-		extData, readExtErr := os.ReadFile(svc.Path)
-		if readExtErr != nil {
-			if os.IsNotExist(readExtErr) {
-				continue
+		// Load Service config from file if specified
+		if svc.Path != "" {
+			if extCfg, fullCfgErr := parseExternalConfig(svc.Path); fullCfgErr == nil {
+				if extSvc, found := extCfg.Services[name]; found {
+					cfg.Services[name] = extSvc
+					continue
+				}
 			}
-			return nil, fmt.Errorf("read service config %q: %w", svc.Path, readExtErr)
-		}
 
-		var extCfg Config
-		fullCfgErr := yaml.UnmarshalWithOptions(extData, &extCfg, yaml.Strict())
-		if fullCfgErr == nil {
-			if extSvc, found := extCfg.Services[name]; found && len(extSvc.Runtimes) > 0 {
-				svc.Runtimes = extSvc.Runtimes
-				cfg.Services[name] = svc
-				continue
+			// Load Service from external file if specified
+			ext, err := parseExternalServiceConfig(svc.Path)
+			if err != nil {
+				return nil, fmt.Errorf("parse external service config %q: %w", svc.Path, err)
 			}
+			// Replace ServiceConfig with the loaded one
+			svc = ext
+			// Keep the path to the Service config file for future references
+			svc.Path = path
 		}
 
-		var ext ServiceConfig
-		fallbackErr := yaml.UnmarshalWithOptions(extData, &ext, yaml.Strict())
-		if fallbackErr != nil {
-			return nil, fmt.Errorf("parse service config %q: %w", svc.Path, fallbackErr)
-		}
-
-		svc.Runtimes = ext.Runtimes
 		cfg.Services[name] = svc
 	}
 
 	return &cfg, nil
 }
+
+func parseExternalConfig(path string) (Config, error) {
+	var cfg Config
+
+	extData, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return cfg, nil
+		}
+		return cfg, fmt.Errorf("read service config %q: %w", path, err)
+	}
+
+	if err = yaml.UnmarshalWithOptions(extData, &cfg, yaml.Strict()); err == nil {
+		return cfg, nil
+	}
+	return cfg, nil
+}
+
+func parseExternalServiceConfig(path string) (ServiceConfig, error) {
+	var cfg ServiceConfig
+	extData, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return cfg, nil
+		}
+		return cfg, fmt.Errorf("read service config %q: %w", path, err)
+	}
+	if err = yaml.UnmarshalWithOptions(extData, &cfg, yaml.Strict()); err != nil {
+		return cfg, fmt.Errorf("parse service config %q: %w", path, err)
+	}
+	// Load EnvVars from file if specified in the external config
+	if cfg.EnvVarsFile != "" {
+		err = loadEnvVarsFile(&cfg)
+		if err != nil {
+			return cfg, fmt.Errorf("load env vars from file %q: %w", cfg.EnvVarsFile, err)
+		}
+	}
+	return cfg, nil
+}
+
+func loadEnvVarsFile(svc *ServiceConfig) error {
+	envVarsFile, err := os.Open(svc.EnvVarsFile)
+	if err != nil {
+		return err
+	}
+	defer envVarsFile.Close()
+	if svc.EnvVars == nil {
+		svc.EnvVars = make(map[string]string)
+	}
+	scanner := bufio.NewScanner(envVarsFile)
+	for scanner.Scan() {
+		line := scanner.Text()
+		// Skip comments and empty lines
+		if strings.HasPrefix(line, "#") || line == "" {
+			continue
+		}
+		// Split on the first '=' to get the key-value pair
+		splitSubstrings := 2
+		parts := strings.SplitN(line, "=", splitSubstrings)
+		if len(parts) != splitSubstrings {
+			continue
+		}
+		// Add new env var only if it doesn't already exist
+		if _, found := svc.EnvVars[parts[0]]; !found {
+			svc.EnvVars[parts[0]] = strings.Trim(parts[1], "\"")
+		}
+	}
+	return nil
+}

internal/config/config_test.go

@@ -158,3 +158,42 @@ services:
 		t.Fatal("expected error parsing external service config")
 	}
 }
+func TestLoad_ServiceEnvVarsFile(t *testing.T) {
+	tmpDir := t.TempDir()
+	envFilePath := filepath.Join(tmpDir, "env.sh")
+	svcContent := []byte(`
+env_vars_file: "` + envFilePath + `"
+`)
+	svcCfgFile := filepath.Join(tmpDir, "trino.yaml")
+	if err := os.WriteFile(svcCfgFile, svcContent, 0o644); err != nil {
+		t.Fatalf("write external service config: %v", err)
+	}
+	envContent := []byte(`
+# some comment
+ENV_TEST="test_value"
+
+`)
+	if err := os.WriteFile(envFilePath, envContent, 0o644); err != nil {
+		t.Fatalf("write environment variables file: %v", err)
+	}
+	mainContent := []byte(`
+services:
+  svc1:
+    path: "` + svcCfgFile + `"
+`)
+	mainFilePath := filepath.Join(tmpDir, "config.yaml")
+	if err := os.WriteFile(mainFilePath, mainContent, 0o644); err != nil {
+		t.Fatalf("write main config: %v", err)
+	}
+	cfg, err := Load(mainFilePath)
+	if err != nil {
+		t.Fatalf("Load() failed: %v", err)
+	}
+	envValue, found := cfg.Services["svc1"].EnvVars["ENV_TEST"]
+	if !found {
+		t.Fatalf("env_vars_file not loaded correctly. config: %q", cfg)
+	}
+	if envValue != "test_value" {
+		t.Fatalf("env_vars_file loaded incorrectly, got '%s', expected 'test_value'", envValue)
+	}
+}