Port conflict management

Author: ivov

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/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))