feat: add real-time log streaming endpoint (#55)

Author: w-h-a

cmd/container_logs.go

@@ -0,0 +1,16 @@
+package cmd
+
+import (
+	"net/http"
+
+	"github.com/urfave/cli/v2"
+	containerlogs "github.com/w-h-a/workflow/internal/container_logs"
+)
+
+func StreamContainerLogs(ctx *cli.Context) error {
+	taskID := ctx.String("task_id")
+
+	containerlogs.RunStreamerClient(&http.Client{}, taskID)
+
+	return nil
+}

cmd/engine.go

@@ -19,6 +19,7 @@ import (
 	"github.com/w-h-a/workflow/internal/engine/clients/runner/docker"
 	"github.com/w-h-a/workflow/internal/engine/config"
 	"github.com/w-h-a/workflow/internal/engine/services/coordinator"
+	"github.com/w-h-a/workflow/internal/engine/services/streamer"
 	"github.com/w-h-a/workflow/internal/engine/services/worker"
 	"go.opentelemetry.io/contrib/bridges/otelslog"
 	"go.opentelemetry.io/contrib/instrumentation/host"
@@ -141,6 +142,17 @@ func StartEngine(ctx *cli.Context) error {
 	stopChannels := map[string]chan struct{}{}
 	numServices := 0
 
+	// streamer
+	var streamerServer serverv2.Server
+	var s *streamer.Service
+	if config.Mode() == "standalone" || config.Mode() == "streamer" {
+		streamerServer, s = engine.NewStreamer(
+			brokerClient,
+		)
+
+		numServices += 2
+	}
+
 	// worker
 	var workerServer serverv2.Server
 	var w *worker.Service
@@ -172,6 +184,26 @@ func StartEngine(ctx *cli.Context) error {
 	// error chan
 	ch := make(chan error, numServices)
 
+	// start streamer
+	if s != nil {
+		stop := make(chan struct{})
+		stopChannels["streamer"] = stop
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			slog.InfoContext(ctx.Context, "starting streamer")
+			ch <- s.Start(stop)
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			slog.InfoContext(ctx.Context, "starting streamer http server", "address", config.StreamerHttp())
+			ch <- streamerServer.Start()
+		}()
+	}
+
 	// start worker
 	if w != nil {
 		stop := make(chan struct{})
@@ -222,6 +254,10 @@ func StartEngine(ctx *cli.Context) error {
 	// graceful shutdown
 	slog.InfoContext(ctx.Context, "stopping...")
 
+	if stop, ok := stopChannels["streamer"]; ok {
+		close(stop)
+	}
+
 	if stop, ok := stopChannels["worker"]; ok {
 		close(stop)
 	}

docker-compose.yaml

@@ -58,6 +58,31 @@ services:
       retries: 5
       start_period: 30s
 
+  streamer:
+    container_name: streamer
+    build: .
+    command: engine
+    ports:
+      - "4002:4002"
+    environment:
+      - ENV=dev
+      - NAME=streamer
+      - VERSION=0.1.0-alpha.0
+      - MODE=streamer
+      - HTTP_ADDRESS=:4002
+      - TRACES_ADDRESS=jaeger:4318
+      - METRICS_ADDRESS=prometheus:9090
+      - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=http://prometheus:9090/api/v1/otlp/v1/metrics
+      - BROKER=rabbit
+      - BROKER_LOCATION=amqp://guest:guest@rabbitmq:5672
+    depends_on:
+      jaeger:
+        condition: service_healthy
+      prometheus:
+        condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
+
   worker:
     container_name: worker
     build: .

internal/container_logs/container_logs.go

@@ -0,0 +1,36 @@
+package containerlogs
+
+import (
+	"bufio"
+	"fmt"
+	"net/http"
+)
+
+func RunStreamerClient(client *http.Client, id string) {
+	url := fmt.Sprintf("http://localhost:4002/logs/%s", id)
+
+	resp, err := client.Get(url)
+	if err != nil {
+		fmt.Println("Error making request:", err)
+		return
+	}
+
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		fmt.Printf("Server returned non-200 status: %s\n", resp.Status)
+		return
+	}
+
+	scanner := bufio.NewScanner(resp.Body)
+
+	for scanner.Scan() {
+		fmt.Println(scanner.Text())
+	}
+
+	if err := scanner.Err(); err != nil {
+		fmt.Println("Error reading from stream:", err)
+	}
+
+	fmt.Println("Connection closed.")
+}

internal/engine/config/config.go

@@ -26,6 +26,7 @@ type config struct {
 	mode                string
 	coordinatorHttp     string
 	workerHttp          string
+	streamerHttp        string
 	workerQueues        map[string]int
 	tracesAddress       string
 	metricsAddress      string
@@ -50,6 +51,7 @@ func New() {
 			mode:                "standalone",
 			coordinatorHttp:     ":4000",
 			workerHttp:          ":4001",
+			streamerHttp:        ":4002",
 			workerQueues:        map[string]int{string(task.Scheduled): 1, string(task.Cancelled): 1},
 			tracesAddress:       "localhost:4318",
 			metricsAddress:      "localhost:4318",
@@ -93,6 +95,9 @@ func New() {
 			if instance.mode == "worker" {
 				instance.workerHttp = httpAddress
 			}
+			if instance.mode == "streamer" {
+				instance.streamerHttp = httpAddress
+			}
 		}
 
 		qs := os.Getenv("QUEUES")
@@ -247,6 +252,14 @@ func WorkerHttp() string {
 	return instance.workerHttp
 }
 
+func StreamerHttp() string {
+	if instance == nil {
+		panic("cfg is nil")
+	}
+
+	return instance.streamerHttp
+}
+
 func WorkerQueues() map[string]int {
 	if instance == nil {
 		panic("cfg is nil")

internal/engine/engine.go

@@ -12,7 +12,9 @@ import (
 	"github.com/w-h-a/workflow/internal/engine/config"
 	httphandlers "github.com/w-h-a/workflow/internal/engine/handlers/http"
 	"github.com/w-h-a/workflow/internal/engine/services/coordinator"
+	"github.com/w-h-a/workflow/internal/engine/services/streamer"
 	"github.com/w-h-a/workflow/internal/engine/services/worker"
+	"github.com/w-h-a/workflow/internal/log"
 	"github.com/w-h-a/workflow/internal/task"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	"go.opentelemetry.io/otel"
@@ -122,3 +124,53 @@ func NewWorker(
 
 	return httpServer, workerService
 }
+
+func NewStreamer(
+	brokerClient broker.Broker,
+) (serverv2.Server, *streamer.Service) {
+	// services
+	streamerService := streamer.New(
+		brokerClient,
+		map[string]int{
+			string(log.Queue): 1,
+		},
+	)
+
+	// base server options
+	opts := []serverv2.ServerOption{
+		serverv2.ServerWithNamespace(config.Env()),
+		serverv2.ServerWithName(config.Name()),
+		serverv2.ServerWithVersion(config.Version()),
+	}
+
+	// create http router
+	router := mux.NewRouter()
+
+	httpStatus := httphandlers.NewStatusHandler(streamerService)
+	router.Methods(http.MethodGet).Path("/status").HandlerFunc(httpStatus.GetStatus)
+
+	httpLogs := httphandlers.NewLogsHandler(streamerService)
+	router.Methods(http.MethodGet).Path("/logs/{id}").HandlerFunc(httpLogs.StreamLogs)
+
+	// create http server
+	httpOpts := []serverv2.ServerOption{
+		serverv2.ServerWithAddress(config.StreamerHttp()),
+	}
+
+	httpOpts = append(httpOpts, opts...)
+
+	httpServer := httpserver.NewServer(httpOpts...)
+
+	handler := otelhttp.NewHandler(
+		router,
+		"",
+		otelhttp.WithSpanNameFormatter(func(operation string, r *http.Request) string { return r.URL.Path }),
+		otelhttp.WithTracerProvider(otel.GetTracerProvider()),
+		otelhttp.WithPropagators(otel.GetTextMapPropagator()),
+		otelhttp.WithFilter(func(r *http.Request) bool { return r.URL.Path != "/status" }),
+	)
+
+	httpServer.Handle(handler)
+
+	return httpServer, streamerService
+}

internal/engine/handlers/http/log.go

@@ -0,0 +1,64 @@
+package http
+
+import (
+	"net/http"
+
+	"github.com/w-h-a/workflow/internal/engine/services/streamer"
+)
+
+type Logs struct {
+	parser   *Parser
+	streamer *streamer.Service
+}
+
+func (l *Logs) StreamLogs(w http.ResponseWriter, r *http.Request) {
+	ctx := reqToCtx(r)
+
+	taskId, err := l.parser.ParseTaskId(ctx, r)
+	if err != nil {
+		wrtRsp(w, http.StatusBadRequest, map[string]any{"error": err.Error()})
+		return
+	}
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.Header().Set("Connection", "keep-alive")
+
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "streaming not supported", http.StatusInternalServerError)
+		return
+	}
+
+	logStream, err := l.streamer.StreamLogs(ctx, taskId)
+	if err != nil {
+		http.Error(w, "failed to stream logs", http.StatusInternalServerError)
+		return
+	}
+
+	defer close(logStream.Stop)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case logLine, ok := <-logStream.Logs:
+			if !ok {
+				return
+			}
+
+			data := []byte("data: " + logLine + "\n\n")
+			if _, err := w.Write(data); err != nil {
+				return
+			}
+			flusher.Flush()
+		}
+	}
+}
+
+func NewLogsHandler(streamerService *streamer.Service) *Logs {
+	return &Logs{
+		parser:   &Parser{},
+		streamer: streamerService,
+	}
+}