feat: Manage lifecycle for multiple runner types (#75)

Author: ivov

cmd/launcher/main.go

@@ -4,7 +4,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
-
+	"sync"
 	"task-runner-launcher/internal/commands"
 	"task-runner-launcher/internal/config"
 	"task-runner-launcher/internal/errorreporting"
@@ -16,19 +16,19 @@ import (
 
 func main() {
 	flag.Usage = func() {
-		fmt.Printf("Usage: %s [runner-type]\n", os.Args[0])
+		fmt.Printf("Usage: %s [runner-type(s)]\n", os.Args[0])
 		flag.PrintDefaults()
 	}
 
 	if len(os.Args) < 2 {
-		os.Stderr.WriteString("Missing runner-type argument")
+		os.Stderr.WriteString("Missing runner-type argument(s)\n")
 		flag.Usage()
 		os.Exit(1)
 	}
 
-	runnerType := os.Args[1]
+	runnerTypes := os.Args[1:]
 
-	launcherConfig, err := config.LoadLauncherConfig([]string{runnerType}, envconfig.OsLookuper())
+	launcherConfig, err := config.LoadLauncherConfig(runnerTypes, envconfig.OsLookuper())
 	if err != nil {
 		logs.Errorf("Failed to load config: %v", err)
 		os.Exit(1)
@@ -39,13 +39,23 @@ func main() {
 
 	http.InitHealthCheckServer(launcherConfig.BaseConfig.HealthCheckServerPort)
 
-	logLevel := logs.ParseLevel(launcherConfig.BaseConfig.LogLevel)
-	logPrefix := logs.GetLauncherPrefix(runnerType)
-	logger := logs.NewLogger(logLevel, logPrefix)
+	var wg sync.WaitGroup
+
+	for _, runnerType := range runnerTypes {
+		wg.Add(1)
+		go func(rt string) {
+			defer wg.Done()
 
-	cmd := commands.NewLaunchCommand(logger)
+			logLevel := logs.ParseLevel(launcherConfig.BaseConfig.LogLevel)
+			logPrefix := logs.GetLauncherPrefix(runnerType)
+			logger := logs.NewLogger(logLevel, logPrefix)
 
-	if err := cmd.Execute(launcherConfig, runnerType); err != nil {
-		logs.Errorf("Failed to execute `launch` command: %s", err)
+			cmd := commands.NewLaunchCommand(logger)
+			if err := cmd.Execute(launcherConfig, rt); err != nil {
+				logger.Errorf("Failed to execute `launch` command: %v", err)
+			}
+		}(runnerType)
 	}
+
+	wg.Wait()
 }

docs/lifecycle.md

@@ -2,7 +2,9 @@
 
 ## Summary
 
-The purpose of the launcher is to minimize resource use by launching a runner on demand. To do so, the launcher impersonates a task runner until a task is ready for pickup, and then it launches a runner to run the task, and after the runner has automatically shut down, the launcher will re-launch a runner to pick up the next task.
+The purpose of the launcher is to minimize resource use by launching a runner on demand. 
+
+To do so, the launcher impersonates a task runner until a task is ready for pickup, then it launches a runner to run the task, and after the runner has automatically shut down, the launcher will re-launch a runner once the next task is ready for pickup. The launcher follows this cycle independently for every runner type (i.e., language) configured to run.
 
 ## Step by step
 

docs/setup.md

@@ -20,6 +20,8 @@ To set up the launcher:
 - The launcher exposes a health check endpoint at `/healthz` on port `5680`, configurable via `N8N_RUNNERS_LAUNCHER_HEALTH_CHECK_PORT`.
 - The task broker exposes a health check endpoint at `/healthz` on port `5679`, configurable via `N8N_RUNNERS_BROKER_PORT`.
 
+<br>
+
 ```mermaid
 sequenceDiagram
     participant k8s
@@ -56,6 +58,7 @@ Example config file at `/etc/n8n-task-runners.json`:
         "--disable-proto=delete",
         "/usr/local/lib/node_modules/n8n/node_modules/@n8n/task-runner/dist/start.js"
       ],
+      "health-check-server-port": "5681",
       "allowed-env": ["PATH", "GENERIC_TIMEZONE"],
       "env-overrides": {
         "N8N_RUNNERS_TASK_TIMEOUT": "80",
@@ -73,6 +76,7 @@ Example config file at `/etc/n8n-task-runners.json`:
       "args": [
         "/usr/local/lib/python3.13/site-packages/n8n/task-runner-python/main.py"
       ],
+      "health-check-server-port": "5682", 
       "allowed-env": ["PATH", "GENERIC_TIMEZONE"],
       "env-overrides": {
         "N8N_RUNNERS_TASK_TIMEOUT": "30",
@@ -90,6 +94,7 @@ Example config file at `/etc/n8n-task-runners.json`:
 | `workdir`       | Path where the task runner's `command` will run.                                                                                          |
 | `command`       | Command to start the task runner.                                                                                       |
 | `args`          | Args and flags to use with `command`.                                                                                           |
+| `health-check-server-port` | Port for the runner's health check server. When a single runner is configured, this is optional and defaults to `5681`. When multiple runners are configured, this is required and must be unique per runner.
 | `allowed-env`   | Env vars that the launcher will pass through from its own environment to the runner. See [environment](environment.md). |
 | `env-overrides` | Env vars that the launcher will set directly on the runner. See [environment](environment.md).                          |
 
@@ -104,8 +109,9 @@ The launcher can pass env vars to task runners in two ways, as specified in the
 | `allowed-env` | Env vars filtered from the launcher's own environment | Passing env vars common to all runner types |
 | `env-overrides` | Env vars set by the launcher directly on the runner, with precedence over `allowed-env` | Passing env vars specific to a single runner type |
 
-Exceptionally, these three env vars cannot be disallowed or overridden:
+Exceptionally, these four env vars cannot be disallowed or overridden:
 
 - `N8N_RUNNERS_TASK_BROKER_URI`
 - `N8N_RUNNERS_GRANT_TOKEN`
-- `N8N_RUNNERS_HEALTH_CHECK_SERVER_ENABLED=true`
+- `N8N_RUNNERS_HEALTH_CHECK_SERVER_ENABLED=true`
+- `N8N_RUNNERS_HEALTH_CHECK_SERVER_PORT`

internal/commands/launch.go

@@ -45,7 +45,7 @@ func (c *LaunchCommand) Execute(launcherConfig *config.LauncherConfig, runnerTyp
 	// 2. prepare env vars to pass to runner
 
 	runnerEnv := env.PrepareRunnerEnv(baseConfig, runnerConfig, c.logger)
-	runnerServerURI := fmt.Sprintf("http://%s:%s", baseConfig.RunnerHealthCheckServerHost, baseConfig.RunnerHealthCheckServerPort)
+	runnerServerURI := fmt.Sprintf("http://%s:%s", baseConfig.RunnerHealthCheckServerHost, runnerConfig.HealthCheckServerPort)
 
 	for {
 		// 3. check until task broker is ready

internal/config/config.go

@@ -49,12 +49,9 @@ type BaseConfig struct {
 	// HealthCheckServerPort is the port for the launcher's health check server.
 	HealthCheckServerPort string `env:"N8N_RUNNERS_LAUNCHER_HEALTH_CHECK_PORT, default=5680"`
 
-	// RunnerHealthCheckServerHost is the host for the runner's health check server.
+	// RunnerHealthCheckServerHost is the host for all runners' health check servers.
 	RunnerHealthCheckServerHost string `env:"N8N_RUNNERS_HEALTH_CHECK_SERVER_HOST, default=127.0.0.1"`
 
-	// RunnerHealthCheckServerPort is the port for the runner's health check server.
-	RunnerHealthCheckServerPort string `env:"N8N_RUNNERS_HEALTH_CHECK_SERVER_PORT, default=5681"`
-
 	// Sentry is the Sentry config for the launcher, a subset of what is defined in:
 	// https://docs.sentry.io/platforms/go/configuration/options/
 	Sentry *SentryConfig
@@ -82,6 +79,11 @@ type RunnerConfig struct {
 	// Arguments for command, currently path to runner entrypoint.
 	Args []string `json:"args"`
 
+	// Port for the runner's health check server.
+	// When a single runner is configured, this is optional and defaults to 5681.
+	// When multiple runners are configured, this is required and must be unique per runner.
+	HealthCheckServerPort string `json:"health-check-server-port,omitempty"`
+
 	// Env vars for the launcher to pass from its own environment to the runner.
 	AllowedEnv []string `json:"allowed-env"`
 
@@ -180,6 +182,24 @@ func readLauncherConfigFile(runnerTypes []string) (map[string]*RunnerConfig, err
 		}
 	}
 
+	if len(runnerConfigs) == 1 {
+		for _, config := range runnerConfigs {
+			if config.HealthCheckServerPort == "" {
+				config.HealthCheckServerPort = "5681"
+			}
+		}
+	} else {
+		for runnerType, config := range runnerConfigs {
+			if config.HealthCheckServerPort == "" {
+				return nil, fmt.Errorf("runner %s: health-check-server-port is required with multiple runners", runnerType)
+			}
+		}
+	}
+
+	if err := validateRunnerPorts(runnerConfigs); err != nil {
+		return nil, err
+	}
+
 	if taskRunnersNum == 1 {
 		logs.Debug("Loaded config file with a single runner config")
 	} else {
@@ -188,3 +208,33 @@ func readLauncherConfigFile(runnerTypes []string) (map[string]*RunnerConfig, err
 
 	return runnerConfigs, nil
 }
+
+func validateRunnerPorts(runnerConfigs map[string]*RunnerConfig) error {
+	reservedPorts := map[string]string{
+		"5678": "n8n main server",
+		"5679": "n8n broker server",
+		"5680": "launcher health check server",
+	}
+
+	usedPorts := make(map[string]string)
+
+	for runnerType, config := range runnerConfigs {
+		port := config.HealthCheckServerPort
+
+		if port, err := strconv.Atoi(port); err != nil || port <= 0 || port >= 65536 {
+			return fmt.Errorf("runner %s: health-check-server-port must be a valid port number", runnerType)
+		}
+
+		if service, exists := reservedPorts[port]; exists {
+			return fmt.Errorf("runner %s: health-check-server-port %s conflicts with %s", runnerType, port, service)
+		}
+
+		if existingRunner, exists := usedPorts[port]; exists {
+			return fmt.Errorf("runners %s and %s cannot use the same health-check-server-port %s", existingRunner, runnerType, port)
+		}
+
+		usedPorts[port] = runnerType
+	}
+
+	return nil
+}

internal/config/config_test.go

@@ -143,3 +143,127 @@ func TestConfigFileErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateRunnerPorts(t *testing.T) {
+	tests := []struct {
+		name          string
+		runnerConfigs map[string]*RunnerConfig
+		expectedError string
+	}{
+		{
+			name: "valid unique ports",
+			runnerConfigs: map[string]*RunnerConfig{
+				"javascript": {HealthCheckServerPort: "5681"},
+				"python":     {HealthCheckServerPort: "5682"},
+			},
+			expectedError: "",
+		},
+		{
+			name: "duplicate ports",
+			runnerConfigs: map[string]*RunnerConfig{
+				"javascript": {HealthCheckServerPort: "5681"},
+				"python":     {HealthCheckServerPort: "5681"},
+			},
+			expectedError: "cannot use the same health-check-server-port",
+		},
+		{
+			name: "reserved port conflict",
+			runnerConfigs: map[string]*RunnerConfig{
+				"javascript": {HealthCheckServerPort: "5679"},
+			},
+			expectedError: "conflicts with n8n broker server",
+		},
+		{
+			name: "invalid port number",
+			runnerConfigs: map[string]*RunnerConfig{
+				"javascript": {HealthCheckServerPort: "not-a-port"},
+			},
+			expectedError: "must be a valid port number",
+		},
+		{
+			name: "port out of range",
+			runnerConfigs: map[string]*RunnerConfig{
+				"javascript": {HealthCheckServerPort: "70000"},
+			},
+			expectedError: "must be a valid port number",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateRunnerPorts(tt.runnerConfigs)
+			if tt.expectedError == "" {
+				assert.NoError(t, err)
+			} else {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), tt.expectedError)
+			}
+		})
+	}
+}
+
+func TestBackwardsCompatibilityPortDefaults(t *testing.T) {
+	tests := []struct {
+		name          string
+		configContent string
+		runnerTypes   []string
+		expectError   bool
+		expectedPorts map[string]string
+	}{
+		{
+			name: "single runner gets default port",
+			configContent: `{
+				"task-runners": [{
+					"runner-type": "javascript",
+					"workdir": "/test",
+					"command": "node",
+					"args": ["test.js"]
+				}]
+			}`,
+			runnerTypes: []string{"javascript"},
+			expectedPorts: map[string]string{
+				"javascript": "5681",
+			},
+		},
+		{
+			name: "multiple runners require explicit ports",
+			configContent: `{
+				"task-runners": [
+					{
+						"runner-type": "javascript",
+						"workdir": "/test",
+						"command": "node",
+						"args": ["test.js"]
+					},
+					{
+						"runner-type": "python", 
+						"workdir": "/test",
+						"command": "python",
+						"args": ["test.py"]
+					}
+				]
+			}`,
+			runnerTypes: []string{"javascript", "python"},
+			expectError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			configPath = filepath.Join(t.TempDir(), "test-config.json")
+			err := os.WriteFile(configPath, []byte(tt.configContent), 0600)
+			require.NoError(t, err)
+
+			configs, err := readLauncherConfigFile(tt.runnerTypes)
+
+			if tt.expectError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				for runnerType, expectedPort := range tt.expectedPorts {
+					assert.Equal(t, expectedPort, configs[runnerType].HealthCheckServerPort)
+				}
+			}
+		})
+	}
+}

internal/env/env.go

@@ -22,6 +22,9 @@ const (
 	// EnvVarHealthCheckServerEnabled is the env var to enable the runner's health check server.
 	EnvVarHealthCheckServerEnabled = "N8N_RUNNERS_HEALTH_CHECK_SERVER_ENABLED"
 
+	// EnvVarHealthCheckServerPort is the env var for the runner's health check server port.
+	EnvVarHealthCheckServerPort = "N8N_RUNNERS_HEALTH_CHECK_SERVER_PORT"
+
 	// EnvVarAutoShutdownTimeout is the env var for how long (in seconds) a runner
 	// may be idle for before exit.
 	EnvVarAutoShutdownTimeout = "N8N_RUNNERS_AUTO_SHUTDOWN_TIMEOUT"
@@ -103,6 +106,7 @@ var requiredRuntimeEnvVars = []string{
 	EnvVarTaskBrokerURI,
 	EnvVarHealthCheckServerEnabled,
 	EnvVarGrantToken,
+	EnvVarHealthCheckServerPort,
 }
 
 // PrepareRunnerEnv prepares the environment variables to pass to the runner.
@@ -120,6 +124,7 @@ func PrepareRunnerEnv(baseConfig *config.BaseConfig, runnerConfig *config.Runner
 	}
 	runnerEnv = append(runnerEnv, fmt.Sprintf("%s=%s", EnvVarTaskBrokerURI, baseConfig.TaskBrokerURI))
 	runnerEnv = append(runnerEnv, fmt.Sprintf("%s=true", EnvVarHealthCheckServerEnabled))
+	runnerEnv = append(runnerEnv, fmt.Sprintf("%s=%s", EnvVarHealthCheckServerPort, runnerConfig.HealthCheckServerPort))
 
 	// TODO: The next two lines are legacy behavior to remove after deprecation period.
 	runnerEnv = append(runnerEnv, fmt.Sprintf("%s=%s", EnvVarAutoShutdownTimeout, baseConfig.AutoShutdownTimeout))