feat: add log aggregation and request correlation for Playwright tests

.github/workflows/nodejs.yml

@@ -33,6 +33,8 @@ jobs:
         version="$(bash .github/workflows/determine_docker_image_tag.sh)"
         export MINIO_ROOT_USER="minioadmin"
         export MINIO_ROOT_PASSWORD=$(openssl rand -base64 32)
+        export LOG_AGGREGATOR_URL="http://log-aggregator:8080"
+        export LOG_AGGREGATOR_ENABLED="true"
         bash k8/generate.sh $version
       env:
         WORKER_COUNT: 1
@@ -50,7 +52,9 @@ jobs:
         echo "Host: $FRONTEND_URL"
         npx wait-on "$FRONTEND_URL/service/control/health"
         kubectl wait --timeout 10m --for=condition=ready pod -l role=worker
-        ROOT_TEST_URL=$FRONTEND_URL npm run test
+        AGG_PORT=$(kubectl get svc log-aggregator -o=jsonpath='{.spec.ports[?(@.port==8080)].nodePort}')
+        LOG_AGGREGATOR_URL="http://127.0.0.1:$AGG_PORT"
+        ROOT_TEST_URL=$FRONTEND_URL LOG_AGGREGATOR_URL=$LOG_AGGREGATOR_URL npm run test
       env:
         FLAKINESS_ACCESS_TOKEN: ${{ secrets.FLAKINESS_ACCESS_TOKEN }}
     - name: Upload playwright-report
@@ -75,6 +79,7 @@ jobs:
           - file-service
           - frontend
           - control-service
+          - log-aggregator
           - squid
     steps:
       - uses: actions/checkout@v4

control-service/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.25-alpine as builder
+FROM golang:1.25-alpine AS builder
 WORKDIR /root
 COPY go.mod /root/
 COPY go.sum /root/

control-service/main.go

@@ -1,10 +1,11 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"math/rand"
 	"net/http"
 	"os"
@@ -15,12 +16,14 @@ import (
 
 	"github.com/labstack/echo/v4"
 	"github.com/mxschmitt/try-playwright/internal/echoutils"
+	"github.com/mxschmitt/try-playwright/internal/logagg"
 	"github.com/mxschmitt/try-playwright/internal/workertypes"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/getsentry/sentry-go"
 	sentryecho "github.com/getsentry/sentry-go/echo"
 
+	"github.com/google/uuid"
 	"github.com/streadway/amqp"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"k8s.io/client-go/kubernetes"
@@ -36,8 +39,11 @@ const (
 
 func init() {
 	rand.Seed(time.Now().UTC().UnixNano())
-	log.SetFormatter(&log.TextFormatter{
-		TimestampFormat: time.StampMilli,
+	log.SetFormatter(&log.JSONFormatter{
+		TimestampFormat: time.RFC3339Nano,
+		FieldMap: log.FieldMap{
+			log.FieldKeyMsg: "message",
+		},
 	})
 }
 
@@ -133,44 +139,78 @@ func getTurnstileIP(c echo.Context) string {
 	return c.RealIP()
 }
 
+func respondError(c echo.Context, status int, requestID, testID string, logBuffer *bytes.Buffer, msg string) error {
+	return c.JSON(status, echo.Map{
+		"error":     msg,
+		"requestId": requestID,
+		"testId":    testID,
+		"logs": echo.Map{
+			"control": logBuffer.String(),
+		},
+	})
+}
+
 func (s *server) handleRun(c echo.Context) error {
+	requestID := uuid.New().String()
+	testID := c.Request().Header.Get("X-Test-ID")
+	if testID == "" {
+		testID = requestID
+	}
+	c.Set("requestId", requestID)
+	c.Set("testId", testID)
+	c.Response().Header().Set("X-Request-ID", requestID)
+	c.Response().Header().Set("X-Test-ID", testID)
+	logBuffer := &bytes.Buffer{}
+	defer logagg.DeferPost("control", &testID, &requestID, logBuffer)
+	requestScopedLogger := log.New()
+	requestScopedLogger.SetFormatter(log.StandardLogger().Formatter)
+	requestScopedLogger.SetLevel(log.GetLevel())
+	requestScopedLogger.SetOutput(io.MultiWriter(os.Stdout, logBuffer))
+	logger := requestScopedLogger.WithFields(log.Fields{
+		"request-id": requestID,
+		"testId":     testID,
+		"service":    "control",
+	})
+	logger.Logger.AddHook(logagg.NewHook())
+
 	var req *workertypes.WorkerRequestPayload
 	if err := c.Bind(&req); err != nil {
-		return c.JSON(http.StatusBadRequest, echo.Map{
-			"error": "could not decode request body",
-		})
+		return respondError(c, http.StatusBadRequest, requestID, testID, logBuffer, "could not decode request body")
+	}
+	req.RequestID = requestID
+	if req.TestID == "" {
+		req.TestID = testID
 	}
 	if !req.Language.IsValid() {
-		return c.JSON(http.StatusBadRequest, echo.Map{
-			"error": "could not recognize language",
-		})
+		return respondError(c, http.StatusBadRequest, requestID, testID, logBuffer, "could not recognize language")
 	}
 
-	log.Printf("Validating turnstile")
+	logger.Printf("Validating turnstile")
 	if err := ValidateTurnstile(c.Request().Context(), req.Token, getTurnstileIP(c), os.Getenv("TURNSTILE_SECRET_KEY")); err != nil {
-		log.Printf("Could not validate turnstile: %v", err)
-		return c.JSON(http.StatusUnauthorized, echo.Map{
-			"error": err.Error(),
-		})
+		logger.Printf("Could not validate turnstile: %v", err)
+		return respondError(c, http.StatusUnauthorized, requestID, testID, logBuffer, err.Error())
 	}
-	log.Printf("Validated turnstile successfully")
-	log.Printf("Obtaining worker")
+	logger = logger.WithField("request-id", requestID)
+	logger.Printf("Validated turnstile successfully")
+	logger.Printf("Obtaining worker")
 	var worker *Worker
 	select {
 	case worker = <-s.workers[req.Language].GetCh():
 	case <-time.After(WORKER_TIMEOUT * time.Second):
-		log.Println("Got Worker timeout, was not able to get a worker!")
-		return c.JSON(http.StatusServiceUnavailable, echo.Map{
-			"error": "Timeout in getting a worker!",
-		})
+		logger.Println("Got Worker timeout, was not able to get a worker!")
+		return respondError(c, http.StatusServiceUnavailable, requestID, testID, logBuffer, "Timeout in getting a worker!")
 	}
 
-	logger := log.WithField("worker-id", worker.id)
+	logger = logger.WithFields(log.Fields{
+		"worker-id": worker.id,
+		"testId":    testID,
+	})
 	logger.Infof("Received code: '%s'", req.Code)
 	logger.Info("Obtained worker successfully")
 	logger.Info("Publishing job")
-	if err := worker.Publish(req.Code); err != nil {
-		return fmt.Errorf("could not create new worker job: %w", err)
+	if err := worker.Publish(req.Code, req.RequestID, req.TestID); err != nil {
+		logger.Errorf("could not create new worker job: %v", err)
+		return respondError(c, http.StatusInternalServerError, requestID, testID, logBuffer, "could not create new worker job")
 	}
 	logger.Println("Published message")
 
@@ -205,13 +245,22 @@ func (s *server) handleRun(c echo.Context) error {
 
 	if timeout {
 		return c.JSON(http.StatusServiceUnavailable, echo.Map{
-			"error": "Execution timeout!",
+			"error":     "Execution timeout!",
+			"requestId": requestID,
+			"testId":    testID,
+			"logs": echo.Map{
+				"control": logBuffer.String(),
+			},
 		})
 	}
 
+	payload.RequestID = requestID
+	payload.TestID = testID
 	if !payload.Success {
 		return c.JSON(http.StatusBadRequest, payload)
 	}
+	c.Response().Header().Set("X-Request-ID", requestID)
+	c.Response().Header().Set("X-Test-ID", testID)
 	return c.JSON(http.StatusOK, payload)
 }
 
@@ -228,7 +277,7 @@ func (s *server) handleShareGet(c echo.Context) error {
 }
 
 func (s *server) handleShareCreate(c echo.Context) error {
-	code, err := ioutil.ReadAll(http.MaxBytesReader(c.Response().Writer, c.Request().Body, 1<<20))
+	code, err := io.ReadAll(http.MaxBytesReader(c.Response().Writer, c.Request().Body, 1<<20))
 	if err != nil {
 		return fmt.Errorf("could not read request body: %w", err)
 	}

control-service/workers.go

@@ -191,6 +191,14 @@ func (w *Worker) createPod() error {
 							Name:  "FILE_SERVICE_URL",
 							Value: "http://file:8080",
 						},
+						{
+							Name:  "LOG_AGGREGATOR_URL",
+							Value: os.Getenv("LOG_AGGREGATOR_URL"),
+						},
+						{
+							Name:  "LOG_AGGREGATOR_ENABLED",
+							Value: os.Getenv("LOG_AGGREGATOR_ENABLED"),
+						},
 					},
 					Resources: v1.ResourceRequirements{
 						Limits: v1.ResourceList{
@@ -219,9 +227,11 @@ func determineWorkerImageName(language workertypes.WorkerLanguage) string {
 	return fmt.Sprintf("ghcr.io/mxschmitt/try-playwright/worker-%s:%s", language, tag)
 }
 
-func (w *Worker) Publish(code string) error {
-	msgBody, err := json.Marshal(map[string]string{
-		"code": code,
+func (w *Worker) Publish(code string, requestID string, testID string) error {
+	msgBody, err := json.Marshal(&workertypes.WorkerRequestPayload{
+		Code:      code,
+		RequestID: requestID,
+		TestID:    testID,
 	})
 	if err != nil {
 		return fmt.Errorf("could not marshal json: %v", err)

e2e/tests/api.spec.ts

@@ -1,19 +1,48 @@
 import { expect, test as base, APIResponse } from '@playwright/test';
 
+async function attachAggregatorLogs(testId?: string) {
+  const testInfo = test.info();
+  const effectiveTestId = testId || testInfo.testId;
+  const base = (process.env.LOG_AGGREGATOR_URL || '').replace(/\/$/, '');
+  if (!base) return;
+  try {
+    const res = await fetch(`${base}/logs/${encodeURIComponent(effectiveTestId)}`);
+    if (!res.ok) return;
+    const body = await res.text();
+    if (body.trim().length === 0) return;
+    await testInfo.attach(`logs-${effectiveTestId}`, {
+      body,
+      contentType: 'text/plain',
+    });
+  } catch {
+    // best-effort; ignore
+  }
+}
+
 type TestFixtures = {
   executeCode: (code: string, language: string) => Promise<APIResponse>
 };
 
 const test = base.extend<TestFixtures>({
   executeCode: async ({ request }, use) => {
     await use(async (code: string, language: string) => {
-      return await request.post('/service/control/run', {
+      const requestId = test.info().testId; // align requestId with Playwright testId for log correlation
+      const testId = test.info().testId;
+      const resp = await request.post('/service/control/run', {
+        headers: {
+          'X-Request-ID': requestId,
+          'X-Test-ID': testId,
+        },
         data: {
           code,
-          language
+          language,
+          requestId,
+          testId,
         },
         timeout: 30 * 1000,
-      })
+      });
+      await attachAggregatorLogs(testId);
+      return resp;
     });
   },
 });

e2e/tests/visual.spec.ts

@@ -1,14 +1,41 @@
 import { expect, test as base, Page } from '@playwright/test';
 
+async function attachAggregatorLogs(testId?: string) {
+  const testInfo = test.info();
+  const effectiveTestId = testId || testInfo.testId;
+  const base = (process.env.LOG_AGGREGATOR_URL || '').replace(/\/$/, '');
+  if (!base) return;
+  try {
+    const res = await fetch(`${base}/logs/${encodeURIComponent(effectiveTestId)}`);
+    if (!res.ok) return;
+    const body = await res.text();
+    if (body.trim().length === 0) return;
+    await testInfo.attach(`logs-${effectiveTestId}`, {
+      body,
+      contentType: 'text/plain',
+    });
+  } catch {
+    // best-effort; ignore
+  }
+}
+
 class TryPlaywrightPage {
   constructor(private readonly page: Page) { }
   async executeExample(nth: number): Promise<void> {
     await this.page.goto('/?l=javascript');
     await this.page.locator(`.rs-panel-group > .rs-panel:nth-child(${nth})`).click();
+    const responsePromise = this.page.waitForResponse("**/service/control/run");
     await Promise.all([
-      this.page.waitForResponse("**/service/control/run"),
+      responsePromise,
       this.page.getByRole('button', { name: 'Run' }).click(),
-    ])
+    ]);
+    const resp = await responsePromise;
+    try {
+      const payload = await resp.json();
+      await attachAggregatorLogs(payload?.testId);
+    } catch (_) {
+      // ignore: best-effort log attachment
+    }
   }
   async getConsoleLines(): Promise<string[]> {
     let consoleLines = this.page.locator(".rs-panel-body code")

file-service/Dockerfile

@@ -1,11 +1,11 @@
-FROM golang:1.25-alpine as builder
+FROM golang:1.25-alpine AS builder
 WORKDIR /root
 COPY go.mod /root/
 COPY go.sum /root/
 RUN go mod download
 
 COPY file-service/* /root/
-COPY internal/echoutils /root/internal/echoutils
+COPY internal/ /root/internal/
 RUN CGO_ENABLED=0 GOOS=linux go build -o /app main.go
 
 FROM alpine:latest