Merge branch 'main' into manage-lifecycle-for-multiple-runner-types

Author: ivov

Makefile

@@ -1,15 +1,10 @@
-.PHONY: audit build lint lintfix fmt fmt-check run test test-verbose test-coverage
-
-check: lint
-	go fmt ./...
-	go vet ./...
-
 build:
 	go build -o bin cmd/launcher/main.go
 	@echo "Binary built at: $(shell pwd)/bin/main"
 
-lint:
-	golangci-lint run
+check: lint
+	go fmt ./...
+	go vet ./...
 
 lintfix:
 	golangci-lint run --fix
@@ -23,9 +18,15 @@ fmt-check:
 		exit 1; \
 	fi
 
+lint:
+	golangci-lint run
+
 run: build
 	./bin/main javascript
 
+run-all: build
+	./bin/main javascript python
+
 test:
 	go test -race ./...
 
@@ -36,3 +37,5 @@ test-coverage:
 	go test -race -coverprofile=coverage.out ./...
 	go tool cover -html=coverage.out -o coverage.html
 	open coverage.html
+
+.PHONY: build check lint lintfix fmt fmt-check run run-all test test-verbose test-coverage

internal/commands/launch.go

@@ -45,6 +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)
 
 	for {
 		// 3. check until task broker is ready
@@ -110,7 +111,7 @@ func (c *LaunchCommand) Execute(launcherConfig *config.LauncherConfig, runnerTyp
 			return fmt.Errorf("failed to start runner process: %w", err)
 		}
 
-		go http.ManageRunnerHealth(ctx, cmd, env.RunnerServerURI, &wg, c.logger)
+		go http.ManageRunnerHealth(ctx, cmd, runnerServerURI, &wg, c.logger)
 
 		err = cmd.Wait()
 		if err != nil && err.Error() == "signal: killed" {

internal/config/config.go

@@ -49,6 +49,12 @@ 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 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

internal/env/env.go

@@ -31,12 +31,6 @@ const (
 	EnvVarTaskTimeout = "N8N_RUNNERS_TASK_TIMEOUT"
 )
 
-const (
-	// URI of the runner. Used for monitoring the runner's health
-	// BUG: Harcoded value makes N8N_RUNNERS_HEALTH_CHECK_SERVER_HOST and N8N_RUNNERS_HEALTH_CHECK_SERVER_PORT non-configurable.
-	RunnerServerURI = "http://127.0.0.1:5681"
-)
-
 // partitionByAllowlist divides the current env vars into those included in and
 // excluded from the allowlist.
 func partitionByAllowlist(allowlist []string) (included, excluded []string) {

internal/http/healthcheck_server.go

@@ -19,8 +19,8 @@ const (
 // exposing `/healthz` at the given port, running in a goroutine.
 func InitHealthCheckServer(port string) {
 	srv := newHealthCheckServer(port)
+	logs.Infof("Starting launcher's health check server at port %s", port)
 	go func() {
-		logs.Infof("Starting launcher's health check server at port %s", port)
 		if err := srv.ListenAndServe(); err != nil {
 			errMsg := "Health check server failed to start"
 			if opErr, ok := err.(*net.OpError); ok && opErr.Op == "listen" {

internal/logs/logger.go

@@ -116,7 +116,7 @@ func (l *Logger) Infof(msg string, xs ...interface{}) {
 
 func (l *Logger) Warn(msg string) {
 	if l.level <= WarnLevel {
-		l.warn.Printf("%sWARN %s%s%s", ColorYellow, l.prefix, msg, ColorReset)
+		l.warn.Printf("%sWARN  %s%s%s", ColorYellow, l.prefix, msg, ColorReset)
 	}
 }
 

internal/logs/logger_test.go

@@ -130,22 +130,22 @@ func TestWarnLogs(t *testing.T) {
 			level:          DebugLevel,
 			logFunc:        Warn,
 			message:        "test warn message",
-			expectedOutput: "WARN test warn message",
+			expectedOutput: "WARN  test warn message",
 			shouldLog:      true,
 		},
 		{
 			name:           "warn level logs warn message",
 			level:          WarnLevel,
 			logFunc:        Warn,
 			message:        "test warn message",
-			expectedOutput: "WARN test warn message",
+			expectedOutput: "WARN  test warn message",
 			shouldLog:      true,
 		},
 		{
 			name:           "error level does not log warn message",
 			level:          ErrorLevel,
 			logFunc:        Warn,
-			message:        "test warn message",
+			message:        "test  warn message",
 			expectedOutput: "",
 			shouldLog:      false,
 		},

internal/logs/runner_writers.go

@@ -26,8 +26,28 @@ func NewRunnerWriter(w io.Writer, prefix string, level string, color string) *Ru
 	}
 }
 
+var (
+	shouldLogDebug  bool
+	logLevelChecked bool
+)
+
+func initLogLevel() {
+	if !logLevelChecked {
+		logLevel := strings.ToUpper(os.Getenv("N8N_RUNNERS_LAUNCHER_LOG_LEVEL"))
+		shouldLogDebug = logLevel == "DEBUG" || logLevel == ""
+		logLevelChecked = true
+	}
+}
+
 // Write implements `io.Writer` and adds color, timestamp, level and a prefix to each line.
 func (w *RunnerWriter) Write(p []byte) (n int, err error) {
+	if w.level == "DEBUG" {
+		initLogLevel()
+		if !shouldLogDebug {
+			return len(p), nil
+		}
+	}
+
 	scanner := bufio.NewScanner(strings.NewReader(string(p)))
 
 	for scanner.Scan() {