feat: logging improvements (#683)

Signed-off-by: Ivan Dagelic <[email protected]>

internal/util/apiclient/websocket_log_reader.go

@@ -0,0 +1,93 @@
+// Copyright 2024 Daytona Platforms Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package apiclient
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/daytonaio/daytona/cmd/daytona/config"
+	"github.com/daytonaio/daytona/pkg/logs"
+	logs_view "github.com/daytonaio/daytona/pkg/views/logs"
+	"github.com/gorilla/websocket"
+)
+
+var workspaceLogsStarted bool
+
+func ReadWorkspaceLogs(activeProfile config.Profile, workspaceId string, projectNames []string, stopLogs *bool) {
+	var wg sync.WaitGroup
+	query := "follow=true"
+
+	logs_view.CalculateLongestPrefixLength(projectNames)
+
+	for index, projectName := range projectNames {
+		wg.Add(1)
+		go func(projectName string) {
+			defer wg.Done()
+
+			for {
+				// Make sure workspace logs started before showing any project logs
+				if !workspaceLogsStarted {
+					time.Sleep(250 * time.Millisecond)
+					continue
+				}
+
+				ws, _, err := GetWebsocketConn(fmt.Sprintf("/log/workspace/%s/%s", workspaceId, projectName), &activeProfile, &query)
+				// We want to retry getting the logs if it fails
+				if err != nil {
+					// TODO: return log.Trace once https://github.com/daytonaio/daytona/issues/696 is resolved
+					// log.Trace(apiclient_util.HandleErrorResponse(res, err))
+					time.Sleep(500 * time.Millisecond)
+					continue
+				}
+
+				readJSONLog(ws, stopLogs, index)
+				ws.Close()
+				break
+			}
+		}(projectName)
+	}
+
+	for {
+		ws, _, err := GetWebsocketConn(fmt.Sprintf("/log/workspace/%s", workspaceId), &activeProfile, &query)
+		// We want to retry getting the logs if it fails
+		if err != nil {
+			// TODO: return log.Trace once https://github.com/daytonaio/daytona/issues/696 is resolved
+			// log.Trace(apiclient_util.HandleErrorResponse(res, err))
+			time.Sleep(250 * time.Millisecond)
+			continue
+		}
+
+		readJSONLog(ws, stopLogs, logs_view.WORKSPACE_INDEX)
+		ws.Close()
+		break
+	}
+
+	wg.Wait()
+}
+
+func readJSONLog(ws *websocket.Conn, stopLogs *bool, index int) {
+	logEntriesChan := make(chan logs.LogEntry)
+	go logs_view.DisplayLogs(logEntriesChan, index)
+
+	for {
+		var logEntry logs.LogEntry
+		err := ws.ReadJSON(&logEntry)
+		if err != nil {
+			fmt.Println(err.Error())
+			return
+		}
+
+		logEntriesChan <- logEntry
+
+		if !workspaceLogsStarted && index == logs_view.WORKSPACE_INDEX {
+			workspaceLogsStarted = true
+		}
+
+		if *stopLogs {
+			return
+		}
+	}
+}

internal/util/log_reader.go

@@ -5,8 +5,12 @@ package util
 
 import (
 	"bufio"
+	"bytes"
 	"context"
+	"encoding/json"
 	"io"
+
+	"github.com/daytonaio/daytona/pkg/logs"
 )
 
 func ReadLog(ctx context.Context, logReader io.Reader, follow bool, c chan []byte, errChan chan error) {
@@ -32,3 +36,45 @@ func ReadLog(ctx context.Context, logReader io.Reader, follow bool, c chan []byt
 		}
 	}
 }
+
+func ReadJSONLog(ctx context.Context, logReader io.Reader, follow bool, c chan interface{}, errChan chan error) {
+	var buffer bytes.Buffer
+	reader := bufio.NewReader(logReader)
+	delimiter := []byte(logs.LogDelimiter)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			byteChunk := make([]byte, 1024)
+			n, err := reader.Read(byteChunk)
+			if err != nil {
+				if err != io.EOF {
+					errChan <- err
+				} else if !follow {
+					errChan <- io.EOF
+					return
+				}
+			}
+			buffer.Write(byteChunk[:n])
+			data := buffer.Bytes()
+
+			index := bytes.Index(data, delimiter)
+
+			if index != -1 { // if the delimiter is found, process the log entry
+
+				var logEntry logs.LogEntry
+
+				err = json.Unmarshal(data[:index], &logEntry)
+				if err != nil {
+					return
+				}
+
+				c <- logEntry
+				buffer.Reset()
+				buffer.Write(data[index+len(delimiter):]) // write remaining data to buffer
+			}
+		}
+	}
+}

pkg/api/controllers/log/websocket.go

@@ -83,6 +83,69 @@ func readLog(ginCtx *gin.Context, logReader io.Reader) {
 	}
 }
 
+func writeJSONToWs(ws *websocket.Conn, c chan interface{}, errChan chan error) {
+	for {
+		value := <-c
+		err := ws.WriteJSON(value)
+		if err != nil {
+			errChan <- err
+			break
+		}
+	}
+}
+
+func readJSONLog(ginCtx *gin.Context, logReader io.Reader) {
+	followQuery := ginCtx.Query("follow")
+	follow := followQuery == "true"
+
+	ws, err := upgrader.Upgrade(ginCtx.Writer, ginCtx.Request, nil)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	defer ws.Close()
+
+	msgChannel := make(chan interface{})
+	errChannel := make(chan error)
+	ctx, cancel := context.WithCancel(context.Background())
+
+	defer cancel()
+	go util.ReadJSONLog(ctx, logReader, follow, msgChannel, errChannel)
+	go writeJSONToWs(ws, msgChannel, errChannel)
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				err := <-errChannel
+				if err != nil {
+					if err.Error() != "EOF" {
+						log.Error(err)
+					}
+					ws.Close()
+					cancel()
+				}
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_, _, err := ws.ReadMessage()
+			if err != nil {
+				ws.Close()
+				cancel()
+				return
+			}
+		}
+	}
+}
+
 func ReadServerLog(ginCtx *gin.Context) {
 	server := server.GetInstance(nil)
 
@@ -106,7 +169,7 @@ func ReadWorkspaceLog(ginCtx *gin.Context) {
 		return
 	}
 
-	readLog(ginCtx, wsLogReader)
+	readJSONLog(ginCtx, wsLogReader)
 }
 
 func ReadProjectLog(ginCtx *gin.Context) {
@@ -121,5 +184,5 @@ func ReadProjectLog(ginCtx *gin.Context) {
 		return
 	}
 
-	readLog(ginCtx, projectLogReader)
+	readJSONLog(ginCtx, projectLogReader)
 }

pkg/builder/builder.go

@@ -9,7 +9,7 @@ import (
 	"path/filepath"
 
 	"github.com/daytonaio/daytona/pkg/gitprovider"
-	"github.com/daytonaio/daytona/pkg/logger"
+	"github.com/daytonaio/daytona/pkg/logs"
 	"github.com/daytonaio/daytona/pkg/server/containerregistries"
 	"github.com/daytonaio/daytona/pkg/workspace"
 )
@@ -30,7 +30,7 @@ type BuilderConfig struct {
 	// Namespace to be used when tagging and pushing the build image
 	BuildImageNamespace             string
 	BasePath                        string
-	LoggerFactory                   logger.LoggerFactory
+	LoggerFactory                   logs.LoggerFactory
 	DefaultProjectImage             string
 	DefaultProjectUser              string
 	DefaultProjectPostStartCommands []string
@@ -56,7 +56,7 @@ type Builder struct {
 	buildImageNamespace             string
 	serverConfigFolder              string
 	basePath                        string
-	loggerFactory                   logger.LoggerFactory
+	loggerFactory                   logs.LoggerFactory
 	defaultProjectImage             string
 	defaultProjectUser              string
 	defaultProjectPostStartCommands []string

pkg/builder/devcontainer.go

@@ -20,6 +20,7 @@ import (
 	"github.com/daytonaio/daytona/pkg/builder/devcontainer"
 	"github.com/daytonaio/daytona/pkg/containerregistry"
 	"github.com/daytonaio/daytona/pkg/docker"
+	"github.com/daytonaio/daytona/pkg/logs"
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/mount"
@@ -92,7 +93,7 @@ func (b *DevcontainerBuilder) CleanUp() error {
 }
 
 func (b *DevcontainerBuilder) Publish() error {
-	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name)
+	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name, logs.LogSourceBuilder)
 	defer projectLogger.Close()
 
 	cliBuilder, err := b.getBuilderDockerClient()
@@ -113,7 +114,7 @@ func (b *DevcontainerBuilder) Publish() error {
 }
 
 func (b *DevcontainerBuilder) buildDevcontainer() error {
-	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name)
+	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name, logs.LogSourceBuilder)
 	defer projectLogger.Close()
 
 	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
@@ -204,7 +205,7 @@ func (b *DevcontainerBuilder) buildDevcontainer() error {
 }
 
 func (b *DevcontainerBuilder) readConfiguration() error {
-	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name)
+	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name, logs.LogSourceBuilder)
 	defer projectLogger.Close()
 
 	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
@@ -278,7 +279,7 @@ func (b *DevcontainerBuilder) readConfiguration() error {
 func (b *DevcontainerBuilder) startContainer() error {
 	ctx := context.Background()
 
-	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name)
+	projectLogger := b.loggerFactory.CreateProjectLogger(b.project.WorkspaceId, b.project.Name, logs.LogSourceBuilder)
 	defer projectLogger.Close()
 
 	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())

pkg/builder/factory.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/daytonaio/daytona/pkg/git"
 	"github.com/daytonaio/daytona/pkg/gitprovider"
-	"github.com/daytonaio/daytona/pkg/logger"
+	"github.com/daytonaio/daytona/pkg/logs"
 	"github.com/daytonaio/daytona/pkg/ports"
 	"github.com/daytonaio/daytona/pkg/server/containerregistries"
 	"github.com/daytonaio/daytona/pkg/workspace"
@@ -30,7 +30,7 @@ type BuilderFactory struct {
 	containerRegistryServer         string
 	buildImageNamespace             string
 	basePath                        string
-	loggerFactory                   logger.LoggerFactory
+	loggerFactory                   logs.LoggerFactory
 	image                           string
 	containerRegistryService        containerregistries.IContainerRegistryService
 	defaultProjectImage             string
@@ -68,7 +68,7 @@ func (f *BuilderFactory) Create(p workspace.Project, gpc *gitprovider.GitProvide
 		return nil, err
 	}
 
-	projectLogger := f.loggerFactory.CreateProjectLogger(p.WorkspaceId, p.Name)
+	projectLogger := f.loggerFactory.CreateProjectLogger(p.WorkspaceId, p.Name, logs.LogSourceBuilder)
 	defer projectLogger.Close()
 
 	gitservice := git.Service{

pkg/cmd/server/serve.go

@@ -17,7 +17,7 @@ import (
 	"github.com/daytonaio/daytona/pkg/apikey"
 	"github.com/daytonaio/daytona/pkg/builder"
 	"github.com/daytonaio/daytona/pkg/db"
-	"github.com/daytonaio/daytona/pkg/logger"
+	"github.com/daytonaio/daytona/pkg/logs"
 	"github.com/daytonaio/daytona/pkg/provider/manager"
 	"github.com/daytonaio/daytona/pkg/provisioner"
 	"github.com/daytonaio/daytona/pkg/server"
@@ -63,7 +63,7 @@ var ServeCmd = &cobra.Command{
 		if err != nil {
 			log.Fatal(err)
 		}
-		loggerFactory := logger.NewLoggerFactory(logsDir)
+		loggerFactory := logs.NewLoggerFactory(logsDir)
 
 		dbPath, err := getDbPath()
 		if err != nil {