Update: Migrate E2E Test runner from shell to go based

Author: miladhzzzz

tests/e2e/Dockerfile.test-client

@@ -3,12 +3,14 @@ FROM golang:1.24-alpine AS builder
 WORKDIR /src
 COPY persys-scheduler ./persys-scheduler
 RUN cd /src/persys-scheduler && go build -o /out/smoke-client ./cmd/smoke-client
+COPY tests/e2e ./tests/e2e
+RUN cd /src/tests/e2e && go build -o /out/test-runner ./test-runner.go
 
 FROM alpine:3.21
-RUN apk add --no-cache ca-certificates curl jq bash
+RUN apk add --no-cache ca-certificates
 WORKDIR /app
 COPY --from=builder /out/smoke-client ./smoke-client
-COPY tests/e2e/test-suite.sh ./test-suite.sh
-RUN chmod +x ./smoke-client ./test-suite.sh
+COPY --from=builder /out/test-runner ./test-runner
+RUN chmod +x ./smoke-client ./test-runner
 
-CMD ["/bin/sh", "-c", "./test-suite.sh"]
+CMD ["./test-runner"]

tests/e2e/Makefile

@@ -1,12 +1,11 @@
-.PHONY: help test test-docker test-script test-go clean
+.PHONY: help test test-docker test-go clean
 
 help:
 	@echo "Persys Scheduler E2E"
 	@echo "===================="
 	@echo "test         - Run docker-compose based scheduler + compute-agent lifecycle suite"
 	@echo "test-docker  - Same as test"
-	@echo "test-script  - Run shell lifecycle suite against locally-running scheduler/agent"
-	@echo "test-go      - Run Go health/metrics probe against locally-running scheduler"
+	@echo "test-go      - Run Go lifecycle suite against locally-running scheduler/agent"
 	@echo "clean        - Tear down compose test stack"
 
 test: test-docker
@@ -15,10 +14,6 @@ test-docker:
 	docker compose -f docker-compose.test.yml up --build --abort-on-container-exit --exit-code-from test-client
 	docker compose -f docker-compose.test.yml down
 
-test-script:
-	chmod +x test-suite.sh
-	./test-suite.sh
-
 test-go:
 	go run test-runner.go
 

tests/e2e/README.md

@@ -8,9 +8,8 @@ The tests cover:
 
 - Scheduler startup (`/health`, `/metrics`)
 - Compute-agent startup (`/health`)
-- Scheduler gRPC control API lifecycle through `cmd/smoke-client`
-  - `RegisterNode`
-  - `Heartbeat`
+- Scheduler gRPC control API lifecycle through Go runner + `cmd/smoke-client`
+  - retries apply until scheduler can place workload
   - `ApplyWorkload` (container)
   - `GetWorkload`
   - `ListWorkloads`
@@ -32,26 +31,23 @@ This starts:
 - `etcd`
 - `compute-agent` (real runtime node service from `compute-agent/`)
 - `persys-scheduler` (with `-insecure`)
-- `test-client` (runs `test-suite.sh`)
+- `test-client` (runs `test-runner`)
 
-### Script-only (against existing local scheduler)
+### Go runner only (against existing local scheduler)
 
 ```bash
 cd tests/e2e
 SCHEDULER_METRICS_URL=http://localhost:8084 \
 SCHEDULER_GRPC_ADDR=localhost:8085 \
 AGENT_METRICS_URL=http://localhost:8080 \
-TEST_NODE_ENDPOINT=localhost:50051 \
-./test-suite.sh
+go run test-runner.go
 ```
 
 ## Environment variables
 
 - `SCHEDULER_METRICS_URL` (default `http://localhost:8084`)
 - `SCHEDULER_GRPC_ADDR` (default `localhost:8085`)
 - `AGENT_METRICS_URL` (default `http://compute-agent:8080`)
-- `TEST_NODE_ID` (default `e2e-node-1`)
-- `TEST_NODE_ENDPOINT` (default `compute-agent:50051`)
 - `TEST_WORKLOAD_ID` (default `e2e-workload-1`)
 - `RETRY_INTERVAL` (default `2`)
 - `MAX_RETRIES` (default `40`)

tests/e2e/docker-compose.test.yml

@@ -54,15 +54,13 @@ services:
       - SCHEDULER_METRICS_URL=http://persys-scheduler:8084
       - SCHEDULER_GRPC_ADDR=persys-scheduler:8085
       - AGENT_METRICS_URL=http://compute-agent:8080
-      - TEST_NODE_ID=e2e-node-1
-      - TEST_NODE_ENDPOINT=compute-agent:50051
       - TEST_WORKLOAD_ID=e2e-workload-1
     depends_on:
       - persys-scheduler
       - compute-agent
     networks:
       - persys-cloud-test
-    command: ["/bin/sh", "-c", "./test-suite.sh"]
+    command: ["./test-runner"]
 
 networks:
   persys-cloud-test:

tests/e2e/test-runner.go

@@ -1,44 +1,179 @@
 package main
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"net/http"
 	"os"
+	"os/exec"
+	"strconv"
 	"strings"
 	"time"
 )
 
 func main() {
-	metricsURL := getenvDefault("SCHEDULER_METRICS_URL", "http://localhost:8084")
+	cfg := loadConfig()
 	client := &http.Client{Timeout: 10 * time.Second}
 
-	if err := waitFor(metricsURL+"/health", client, 30, 2*time.Second); err != nil {
-		panic(fmt.Sprintf("health check failed: %v", err))
-	}
-	if err := waitFor(metricsURL+"/metrics", client, 30, 2*time.Second); err != nil {
-		panic(fmt.Sprintf("metrics endpoint failed: %v", err))
+	must(waitForHTTP(cfg.schedulerMetricsURL+"/health", client, cfg.maxRetries, cfg.retryInterval))
+	must(waitForHTTP(cfg.schedulerMetricsURL+"/metrics", client, cfg.maxRetries, cfg.retryInterval))
+	must(waitForHTTP(cfg.agentMetricsURL+"/health", client, cfg.maxRetries, cfg.retryInterval))
+
+	logStep("apply workload")
+	must(applyWorkloadWithRetry(cfg))
+
+	var workload getWorkloadResponse
+	must(poll(cfg.maxRetries, cfg.retryInterval, func() (bool, error) {
+		out, err := runSmokeCapture(cfg.schedulerGRPCAddr, "-op", "get-workload", "-workload-id", cfg.testWorkloadID)
+		if err != nil {
+			return false, nil
+		}
+		if err := unmarshalSmokeJSON(out, &workload); err != nil {
+			return false, nil
+		}
+		s := strings.ToLower(strings.TrimSpace(workload.Workload.Status))
+		return s == "running" || s == "pending" || s == "unknown", nil
+	}))
+
+	listOut, err := runSmokeCapture(cfg.schedulerGRPCAddr, "-op", "list-workloads")
+	must(err)
+	var listResp listWorkloadsResponse
+	must(unmarshalSmokeJSON(listOut, &listResp))
+	if !containsWorkload(listResp.Workloads, cfg.testWorkloadID) {
+		failf("workload %s missing from list-workloads", cfg.testWorkloadID)
 	}
 
-	body, err := getBody(metricsURL+"/metrics", client)
+	summaryOut, err := runSmokeCapture(cfg.schedulerGRPCAddr, "-op", "cluster-summary")
 	if err != nil {
-		panic(fmt.Sprintf("metrics fetch failed: %v", err))
+		failf("cluster-summary failed: %v", err)
+	}
+	var summary clusterSummaryResponse
+	must(unmarshalSmokeJSON(summaryOut, &summary))
+	if summary.TotalWorkloads < 1 {
+		failf("cluster summary reported total_workloads=%d, want >=1", summary.TotalWorkloads)
 	}
 
+	logStep("delete workload")
+	must(runSmoke(cfg.schedulerGRPCAddr, "-op", "delete-workload", "-workload-id", cfg.testWorkloadID))
+	must(poll(cfg.maxRetries, cfg.retryInterval, func() (bool, error) {
+		out, err := runSmokeCapture(cfg.schedulerGRPCAddr, "-op", "get-workload", "-workload-id", cfg.testWorkloadID)
+		if err != nil {
+			return true, nil // NotFound is acceptable terminal state.
+		}
+		var r getWorkloadResponse
+		if err := unmarshalSmokeJSON(out, &r); err != nil {
+			return false, nil
+		}
+		s := strings.ToLower(strings.TrimSpace(r.Workload.Status))
+		return s == "deleting" || s == "deleted", nil
+	}))
+
+	body, err := getBody(cfg.schedulerMetricsURL+"/metrics", client)
+	must(err)
 	required := []string{
-		"go_gc_duration_seconds",
-		"process_cpu_seconds_total",
+		"persys_scheduler_grpc_server_requests_total",
+		"persys_scheduler_grpc_server_request_duration_seconds",
+		"persys_scheduler_agent_rpc_requests_total",
+		"persys_scheduler_agent_rpc_duration_seconds",
+		"persys_scheduler_nodes_status",
+		"persys_scheduler_workloads_status",
+		"persys_scheduler_reconciliation_results_total",
 	}
 	for _, metric := range required {
 		if !strings.Contains(body, metric) {
-			panic(fmt.Sprintf("expected metric %q not found", metric))
+			failf("expected metric %q not found", metric)
 		}
 	}
 
-	fmt.Println("✅ Go-based E2E probe passed")
+	fmt.Println("E2E scheduler + compute-agent suite passed")
+}
+
+type config struct {
+	schedulerMetricsURL string
+	schedulerGRPCAddr   string
+	agentMetricsURL     string
+	testWorkloadID      string
+	retryInterval       time.Duration
+	maxRetries          int
+}
+
+type getWorkloadResponse struct {
+	Workload workloadView `json:"workload"`
+}
+
+type listWorkloadsResponse struct {
+	Workloads []workloadView `json:"workloads"`
+}
+
+type workloadView struct {
+	WorkloadID string `json:"workloadId"`
+	Status     string `json:"status"`
+}
+
+type clusterSummaryResponse struct {
+	TotalWorkloads int32 `json:"totalWorkloads"`
+}
+
+func loadConfig() config {
+	return config{
+		schedulerMetricsURL: getenvDefault("SCHEDULER_METRICS_URL", "http://localhost:8084"),
+		schedulerGRPCAddr:   getenvDefault("SCHEDULER_GRPC_ADDR", "localhost:8085"),
+		agentMetricsURL:     getenvDefault("AGENT_METRICS_URL", "http://compute-agent:8080"),
+		testWorkloadID:      getenvDefault("TEST_WORKLOAD_ID", "e2e-workload-1"),
+		retryInterval:       getenvDurationDefault("RETRY_INTERVAL", 2*time.Second),
+		maxRetries:          getenvIntDefault("MAX_RETRIES", 40),
+	}
+}
+
+func applyWorkloadWithRetry(cfg config) error {
+	return poll(cfg.maxRetries, cfg.retryInterval, func() (bool, error) {
+		err := runSmoke(cfg.schedulerGRPCAddr,
+			"-op", "apply-container",
+			"-workload-id", cfg.testWorkloadID,
+			"-container-image", "busybox:latest",
+			"-container-cmd", "sh,-c,sleep 60",
+			"-w-cpu", "100",
+			"-w-mem", "128",
+			"-w-disk", "1",
+		)
+		if err != nil {
+			msg := strings.ToLower(err.Error())
+			if strings.Contains(msg, "no suitable node") || strings.Contains(msg, "cannot place workload") {
+				return false, nil
+			}
+			return false, err
+		}
+		return true, nil
+	})
+}
+
+func runSmoke(schedulerAddr string, args ...string) error {
+	_, err := runSmokeCapture(schedulerAddr, args...)
+	return err
 }
 
-func waitFor(url string, client *http.Client, retries int, interval time.Duration) error {
+func runSmokeCapture(schedulerAddr string, args ...string) (string, error) {
+	cmdArgs := append([]string{"-scheduler", schedulerAddr}, args...)
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "./smoke-client", cmdArgs...)
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	err := cmd.Run()
+	out := strings.TrimSpace(stdout.String() + "\n" + stderr.String())
+	if err != nil {
+		return out, fmt.Errorf("smoke-client %v failed: %w\n%s", args, err, out)
+	}
+	return out, nil
+}
+
+func waitForHTTP(url string, client *http.Client, retries int, interval time.Duration) error {
 	for i := 0; i < retries; i++ {
 		resp, err := client.Get(url)
 		if err == nil && resp.StatusCode == http.StatusOK {
@@ -69,9 +204,89 @@ func getBody(url string, client *http.Client) (string, error) {
 	return string(b), nil
 }
 
+func poll(retries int, interval time.Duration, f func() (bool, error)) error {
+	for i := 0; i < retries; i++ {
+		ok, err := f()
+		if err != nil {
+			return err
+		}
+		if ok {
+			return nil
+		}
+		time.Sleep(interval)
+	}
+	return fmt.Errorf("timeout while polling condition")
+}
+
+func unmarshalSmokeJSON(out string, target interface{}) error {
+	raw, err := extractJSONObject(out)
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal([]byte(raw), target)
+}
+
+func extractJSONObject(s string) (string, error) {
+	start := strings.Index(s, "{")
+	end := strings.LastIndex(s, "}")
+	if start == -1 || end == -1 || end <= start {
+		return "", fmt.Errorf("no JSON object found in output: %s", s)
+	}
+	return s[start : end+1], nil
+}
+
+func containsWorkload(workloads []workloadView, workloadID string) bool {
+	for _, w := range workloads {
+		if strings.TrimSpace(w.WorkloadID) == workloadID {
+			return true
+		}
+	}
+	return false
+}
+
+func getenvIntDefault(key string, fallback int) int {
+	v := strings.TrimSpace(os.Getenv(key))
+	if v == "" {
+		return fallback
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return fallback
+	}
+	return n
+}
+
+func getenvDurationDefault(key string, fallback time.Duration) time.Duration {
+	v := strings.TrimSpace(os.Getenv(key))
+	if v == "" {
+		return fallback
+	}
+	if d, err := time.ParseDuration(v); err == nil {
+		return d
+	}
+	if s, err := strconv.Atoi(v); err == nil {
+		return time.Duration(s) * time.Second
+	}
+	return fallback
+}
+
 func getenvDefault(key, fallback string) string {
 	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
 		return v
 	}
 	return fallback
 }
+
+func logStep(msg string) {
+	log.Printf("==> %s", msg)
+}
+
+func must(err error) {
+	if err != nil {
+		failf("%v", err)
+	}
+}
+
+func failf(format string, args ...interface{}) {
+	log.Fatalf(format, args...)
+}