add tail handler in log controller

internal/api/controller/logs/tail.go

@@ -0,0 +1,46 @@
+// Copyright 2022 Harness Inc. All rights reserved.
+// Use of this source code is governed by the Polyform Free Trial License
+// that can be found in the LICENSE.md file for this repository.
+
+package logs
+
+import (
+	"context"
+	"fmt"
+
+	apiauth "github.com/harness/gitness/internal/api/auth"
+	"github.com/harness/gitness/internal/auth"
+	"github.com/harness/gitness/livelog"
+	"github.com/harness/gitness/types/enum"
+)
+
+func (c *Controller) Tail(
+	ctx context.Context,
+	session *auth.Session,
+	spaceRef string,
+	pipelineUID string,
+	executionNum int64,
+	stageNum int64,
+	stepNum int64,
+) (<-chan *livelog.Line, <-chan error, error) {
+	space, err := c.spaceStore.FindByRef(ctx, spaceRef)
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not find parent space: %w", err)
+	}
+
+	pipeline, err := c.pipelineStore.FindByUID(ctx, space.ID, pipelineUID)
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not find pipeline: %w", err)
+	}
+
+	err = apiauth.CheckPipeline(ctx, c.authorizer, session, space.Path, pipeline.UID, enum.PermissionPipelineView)
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not authorize: %w", err)
+	}
+
+	// TODO: Figure out step ID by querying stages and steps tables
+	var id int64
+
+	linec, errc := c.logStream.Tail(ctx, id)
+	return linec, errc, nil
+}

internal/api/handler/logs/find.go

@@ -2,7 +2,7 @@
 // Use of this source code is governed by the Polyform Free Trial License
 // that can be found in the LICENSE.md file for this repository.
 
-package execution
+package logs
 
 import (
 	"io"

internal/api/handler/logs/tail.go

@@ -0,0 +1,116 @@
+// Copyright 2022 Harness Inc. All rights reserved.
+// Use of this source code is governed by the Polyform Free Trial License
+// that can be found in the LICENSE.md file for this repository.
+
+package logs
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/harness/gitness/internal/api/controller/logs"
+	"github.com/harness/gitness/internal/api/render"
+	"github.com/harness/gitness/internal/api/request"
+	"github.com/harness/gitness/internal/paths"
+)
+
+var (
+	pingInterval = 30 * time.Second
+	tailMaxTime  = 1 * time.Hour
+)
+
+func HandleTail(logCtrl *logs.Controller) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		session, _ := request.AuthSessionFrom(ctx)
+		pipelineRef, err := request.GetPipelineRefFromPath(r)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+		executionNum, err := request.GetExecutionNumberFromPath(r)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+		stageNum, err := request.GetStageNumberFromPath(r)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+		stepNum, err := request.GetStepNumberFromPath(r)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+		spaceRef, pipelineUID, err := paths.DisectLeaf(pipelineRef)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+
+		f, ok := w.(http.Flusher)
+		if !ok {
+			return
+		}
+
+		io.WriteString(w, ": ping\n\n")
+		f.Flush()
+
+		linec, errc, err := logCtrl.Tail(
+			ctx, session, spaceRef, pipelineUID,
+			executionNum, stageNum, stepNum)
+		if err != nil {
+			render.TranslatedUserError(w, err)
+			return
+		}
+
+		// could not get error channel
+		if errc == nil {
+			io.WriteString(w, "event: error\ndata: eof\n\n")
+			return
+		}
+
+		h := w.Header()
+		h.Set("Content-Type", "text/event-stream")
+		h.Set("Cache-Control", "no-cache")
+		h.Set("Connection", "keep-alive")
+		h.Set("X-Accel-Buffering", "no")
+		h.Set("Access-Control-Allow-Origin", "*")
+
+		ctx, cancel := context.WithCancel(r.Context())
+		defer cancel()
+
+		enc := json.NewEncoder(w)
+
+		tailMaxTimeTimer := time.After(tailMaxTime)
+		msgDelayTimer := time.NewTimer(pingInterval) // if time b/w messages takes longer, send a ping
+		defer msgDelayTimer.Stop()
+	L:
+		for {
+			msgDelayTimer.Reset(pingInterval)
+			select {
+			case <-ctx.Done():
+				break L
+			case <-errc:
+				break L
+			case <-tailMaxTimeTimer:
+				break L
+			case <-msgDelayTimer.C:
+				io.WriteString(w, ": ping\n\n")
+				f.Flush()
+			case line := <-linec:
+				io.WriteString(w, "data: ")
+				enc.Encode(line)
+				io.WriteString(w, "\n\n")
+				f.Flush()
+			}
+		}
+
+		io.WriteString(w, "event: error\ndata: eof\n\n")
+		f.Flush()
+	}
+}