feat/ add log forwarding in HadesLogManager and ContinueOnError step support

.env.example

@@ -24,6 +24,10 @@ WATCH_NAMESPACE="hades"
 DELETE_ON_COMPLETE="true"
 MAX_PARALLELISM="100"
 
+### HadesLogManager Options ###
+HADESLOGMANAGER_API_PORT=8081 # Port for the Log Manager to listen on (default: 8081)
+ARTEMIS_ADAPTER_URL=http://localhost:8082/adapter/logs # URL for the Artemis Adapter to send logs to (default: http://localhost:8082/adapter/logs)
+
 ### DockerScheduler Options ###
 DOCKER_HOST="unix:///var/run/docker.sock" # Docker host to connect to (default: unix:///var/run/docker.sock)
 DOCKER_SCRIPT_EXECUTOR="/bin/bash -c" # Path to the script executor to use (default: /bin/sh)

.github/workflows/ci.yml

@@ -127,8 +127,17 @@ jobs:
       image-tag: v3.6
     secrets: inherit
 
+  build-log-manager:
+    needs: [lint, build, test]
+    if: github.actor != 'dependabot[bot]'
+    uses: ls1intum/.github/.github/workflows/build-and-push-docker-image.yml@v1.2.0
+    with:
+      image-name: ls1intum/hades/hades-log-manager
+      docker-file: ./HadesLogManager/Dockerfile
+    secrets: inherit
+
   deploy:
-    needs: [build-api, build-scheduler]
+    needs: [build-api, build-scheduler, build-log-manager]
     runs-on: ubuntu-latest
     environment: hades-test
     steps:

HadesAPI/go.sum

@@ -39,6 +39,7 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8Yc
 github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
 github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c=
+github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU=
@@ -177,11 +178,17 @@ go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzyb
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI=
 go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
 go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
+go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
 go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
+go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
 go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
+go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
 go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
 go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
 go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 golang.org/x/arch v0.27.0 h1:0WNVcR8u9yFz8j5FvdHpgwNp3FS5U4guYdzHwEiGjoU=
@@ -196,6 +203,7 @@ golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
 golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
+golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk=
 golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
 golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

HadesLogManager/Dockerfile

@@ -0,0 +1,30 @@
+# Use an official Go runtime as a parent image
+FROM golang:1.25-alpine AS builder
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# Copy go.mod and go.sum files
+COPY ./HadesLogManager/go.mod ./HadesLogManager/go.sum ./HadesLogManager/
+COPY ./shared/go.mod ./shared/go.sum ./shared/
+RUN cd HadesLogManager && go mod download
+
+# Copy the Go application source code into the container
+COPY ./HadesLogManager ./HadesLogManager
+COPY ./shared ./shared
+
+# Build the Go application
+WORKDIR /app/HadesLogManager
+RUN CGO_ENABLED=0 go build -o hades-log-manager .
+
+# Start a new stage for the minimal runtime container
+FROM gcr.io/distroless/static-debian12
+
+# Set the working directory inside the minimal runtime container
+WORKDIR /app
+
+# Copy the built binary from the builder container into the minimal runtime container
+COPY --from=builder /app/HadesLogManager/hades-log-manager . 
+
+# Run your Go application
+CMD ["/app/hades-log-manager"]

HadesLogManager/go.mod

@@ -1,4 +1,5 @@
-module HadesAdapter
+module github.com/ls1intum/hades/hadesLogManager
+replace github.com/ls1intum/hades/shared => ../shared
 
 go 1.25.0
 
@@ -42,7 +43,7 @@ require (
 
 require (
 	github.com/klauspost/compress v1.18.6 // indirect
-	github.com/ls1intum/hades/shared v0.0.0-20260505155304-a178a43cb5c2
+	github.com/ls1intum/hades/shared v0.0.0-20260528212350-4905a32aa556
 	github.com/nats-io/nkeys v0.4.15 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	golang.org/x/crypto v0.52.0 // indirect

HadesLogManager/go.sum

@@ -45,8 +45,8 @@ github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzh
 github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
 github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ=
 github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI=
-github.com/ls1intum/hades/shared v0.0.0-20260505155304-a178a43cb5c2 h1:vrqgmPmtTBRBBZEAEZMKqcxQK5OFmM4IE7+nDUOKd2w=
-github.com/ls1intum/hades/shared v0.0.0-20260505155304-a178a43cb5c2/go.mod h1:qB8+4591QBf959Hd6DGmE0ATXsF0SsR1UDTGLKCbC1M=
+github.com/ls1intum/hades/shared v0.0.0-20260528212350-4905a32aa556 h1:kfdGL7z/A4o7rr3Vi8fUkqAC+i6am/IUMkBjNnPX5Jg=
+github.com/ls1intum/hades/shared v0.0.0-20260528212350-4905a32aa556/go.mod h1:rB2Xc7vj0gHoGp8Jq91fE9rCakAfBFqNqlJfWwcQ72A=
 github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4=
 github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -88,6 +88,7 @@ github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2W
 go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8=
 go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
 go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
 golang.org/x/arch v0.27.0 h1:0WNVcR8u9yFz8j5FvdHpgwNp3FS5U4guYdzHwEiGjoU=
 golang.org/x/arch v0.27.0/go.mod h1:0X+GdSIP+kL5wPmpK7sdkEVTt2XoYP0cSjQSbZBwOi8=
 golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=

HadesLogManager/log_subscriber.go

@@ -32,6 +32,7 @@ type DynamicLogManager struct {
 type watcherState struct {
 	ctx    context.Context
 	cancel context.CancelFunc
+	wg     *sync.WaitGroup
 }
 
 // NewDynamicLogManager creates a new DynamicLogManager instance with the provided dependencies.
@@ -164,23 +165,28 @@ func (dlm *DynamicLogManager) cleanupSubscriptions(subs []*nats.Subscription) {
 func (dlm *DynamicLogManager) startWatchingJobLogs(ctx context.Context, jobID string) {
 	// Create new context for this job outside the lock
 	jobCtx, cancel := context.WithCancel(ctx)
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
 
 	// Minimize critical section - only lock for map operations
 	dlm.mu.Lock()
 	oldWatcher, exists := dlm.watchers[jobID]
 	dlm.watchers[jobID] = watcherState{
 		ctx:    jobCtx,
 		cancel: cancel,
+		wg:     wg,
 	}
 	dlm.mu.Unlock()
 
 	// Cancel old watcher outside the lock to avoid potential deadlock
 	if exists {
 		oldWatcher.cancel()
+		oldWatcher.wg.Wait()
 	}
 
 	// Start watching logs for this job
 	go func() {
+		defer wg.Done()
 		defer func() {
 			// Use a more efficient cleanup check
 			dlm.mu.Lock()
@@ -218,13 +224,27 @@ func (dlm *DynamicLogManager) startWatchingJobLogs(ctx context.Context, jobID st
 //   - jobID: Unique identifier of the job to stop watching logs for
 func (dlm *DynamicLogManager) stopWatchingJobLogs(jobID string) {
 	dlm.mu.Lock()
-	defer dlm.mu.Unlock()
-
-	if watcher, exists := dlm.watchers[jobID]; exists {
+	watcher, exists := dlm.watchers[jobID]
+	if exists {
 		delete(dlm.watchers, jobID)
+	}
+	dlm.mu.Unlock()
+
+	if exists {
 		slog.Info("Stopping log watch", "job_id", jobID)
 		watcher.cancel()
+		watcher.wg.Wait() // Wait outside the lock
 
 		dlm.logAggregator.MarkJobCompleted(jobID)
+		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("Panic while sending job logs", "job_id", jobID, "panic", r)
+				}
+			}()
+			if err := dlm.logAggregator.SendJobLogs(jobID); err != nil {
+				slog.Error("Failed to send job logs", "job_id", jobID, "error", err)
+			}
+		}()
 	}
 }

HadesLogManager/main.go

@@ -25,7 +25,7 @@ const (
 // HadesLogManagerConfig holds the configuration for the log manager
 type HadesLogManagerConfig struct {
 	NatsConfig hadesnats.ConnectionConfig
-	APIPort    string `env:"API_PORT" envDefault:"8081"`
+	APIPort    string `env:"HADESLOGMANAGER_API_PORT" envDefault:"8081"`
 }
 
 func main() {
@@ -80,12 +80,7 @@ func connectNATS(config hadesnats.ConnectionConfig) (*nats.Conn, error) {
 }
 
 // runWithGracefulShutdown starts services and handles graceful shutdown
-func runWithGracefulShutdown(
-	ctx context.Context,
-	cancel context.CancelFunc,
-	cfg HadesLogManagerConfig,
-	dynamicManager buildlogs.LogManager,
-	logAggregator buildlogs.LogAggregator,
+func runWithGracefulShutdown(ctx context.Context, cancel context.CancelFunc, cfg HadesLogManagerConfig, dynamicManager buildlogs.LogManager, logAggregator buildlogs.LogAggregator,
 ) error {
 	var wg sync.WaitGroup
 	errChan := make(chan error, 2)
@@ -128,12 +123,7 @@ func runWithGracefulShutdown(
 }
 
 // waitForShutdown waits for OS signal or error and performs graceful shutdown
-func waitForShutdown(
-	ctx context.Context,
-	cancel context.CancelFunc,
-	server *http.Server,
-	wg *sync.WaitGroup,
-	errChan chan error,
+func waitForShutdown(ctx context.Context, cancel context.CancelFunc, server *http.Server, wg *sync.WaitGroup, errChan chan error,
 ) error {
 	sigChan := make(chan os.Signal, 1)
 	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
@@ -172,6 +162,13 @@ func waitForShutdown(
 	return shutdownErr
 }
 
+// setupAPIRoute creates and configures the Gin router with all log manager endpoints.
+// It registers the following routes:
+//
+//   - GET /jobs/:jobId/logs   — returns all aggregated log entries for the given job ID. (Used for testing purposes)
+//   - GET /jobs/:jobId/status — returns the current build status for the given job ID, or 404 if the job is not found.
+//   - GET /jobs               — returns a list of all known job IDs (active and completed).
+//   - GET /health             — liveness probe returning a static OK response.
 func setupAPIRoute(aggregator buildlogs.LogAggregator) *gin.Engine {
 	r := gin.Default()
 	jobs := r.Group("/jobs")