Merge branch 'develop' into chore/clean-codes

Author: fiftin

.github/workflows/release.yml

@@ -164,6 +164,8 @@ jobs:
         with:
           builder: ${{ steps.buildx.outputs.name }}
           context: .
+          build-args: |
+            ANSIBLE_VERSION=9.4.0
           file: deployment/docker/runner/Dockerfile
           platforms: linux/amd64,linux/arm64 #,linux/arm/v6
           push: ${{ github.event_name != 'pull_request' }}

api/sockets/handler.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/gorilla/context"
 	"github.com/gorilla/websocket"
-	"github.com/semaphoreui/semaphore/util"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -33,6 +32,10 @@ const (
 
 	// Maximum message size allowed from peer.
 	maxMessageSize = 512
+
+	// Maximum size of the connection.send channel.
+	// When the channel is full, the hub closes it (see method hub.run).
+	connectionChannelSize = 256
 )
 
 type connection struct {
@@ -41,6 +44,25 @@ type connection struct {
 	userID int
 }
 
+func (c *connection) log(level log.Level, err error, msg string) {
+	log.WithError(err).WithFields(log.Fields{
+		"context": "websocket",
+		"user_id": c.userID,
+	}).Log(level, msg)
+}
+
+func (c *connection) logError(err error, msg string) {
+	c.log(log.ErrorLevel, err, msg)
+}
+
+func (c *connection) logWarn(err error, msg string) {
+	c.log(log.DebugLevel, err, msg)
+}
+
+func (c *connection) logDebug(err error, msg string) {
+	c.log(log.DebugLevel, err, msg)
+}
+
 // readPump pumps messages from the websocket connection to the hub.
 func (c *connection) readPump() {
 	defer func() {
@@ -50,11 +72,17 @@ func (c *connection) readPump() {
 
 	c.ws.SetReadLimit(maxMessageSize)
 
-	util.LogErrorF(c.ws.SetReadDeadline(tz.Now().Add(pongWait)), log.Fields{"error": "Cannot set read deadline"})
+	if err := c.ws.SetReadDeadline(tz.Now().Add(pongWait)); err != nil {
+		c.logWarn(err, "Failed to set read deadline")
+	}
 
 	c.ws.SetPongHandler(func(string) error {
-		err := c.ws.SetReadDeadline(tz.Now().Add(pongWait))
-		util.LogErrorF(err, log.Fields{"error": "Cannot set read deadline"})
+		deadline := tz.Now().Add(pongWait)
+
+		if err := c.ws.SetReadDeadline(deadline); err != nil {
+			c.logWarn(err, "Failed to set read deadline")
+		}
+
 		return nil
 	})
 
@@ -64,7 +92,7 @@ func (c *connection) readPump() {
 
 		if err != nil {
 			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
-				util.LogError(err)
+				c.logDebug(err, "Failed to read message from client")
 			}
 			break
 		}
@@ -74,9 +102,11 @@ func (c *connection) readPump() {
 // write writes a message with the given message type and payload.
 func (c *connection) write(mt int, payload []byte) error {
 
-	err := c.ws.SetWriteDeadline(tz.Now().Add(writeWait))
+	deadline := tz.Now().Add(writeWait)
 
-	util.LogErrorF(err, log.Fields{"error": "Cannot set write deadline"})
+	if err := c.ws.SetWriteDeadline(deadline); err != nil {
+		c.logWarn(err, "Cannot set write deadline")
+	}
 
 	return c.ws.WriteMessage(mt, payload)
 }
@@ -93,28 +123,23 @@ func (c *connection) writePump() {
 	for {
 		select {
 		case message, ok := <-c.send:
+
 			if !ok {
 				if err := c.write(websocket.CloseMessage, []byte{}); err != nil {
-					log.WithError(err).WithFields(log.Fields{
-						"context": "websocket",
-						"user_id": c.userID,
-					}).Debug("Cannot send close message")
+					c.logDebug(err, "Failed to write close message to client")
 				}
 				return
 			}
+
 			if err := c.write(websocket.TextMessage, message); err != nil {
-				log.WithError(err).WithFields(log.Fields{
-					"context": "websocket",
-					"user_id": c.userID,
-				}).Debug("Cannot send text message")
+				c.logDebug(err, "Failed to write message to client")
 				return
 			}
+
 		case <-ticker.C:
+
 			if err := c.write(websocket.PingMessage, []byte{}); err != nil {
-				log.WithError(err).WithFields(log.Fields{
-					"context": "websocket",
-					"user_id": c.userID,
-				}).Debug("Cannot send ping message")
+				c.logDebug(err, "Failed to write ping message to client")
 				return
 			}
 		}
@@ -131,13 +156,18 @@ func Handler(w http.ResponseWriter, r *http.Request) {
 	user := usr.(*db.User)
 	ws, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
-		log.Error(err)
+
+		log.WithError(err).WithFields(log.Fields{
+			"context": "websocket",
+			"user_id": user.ID,
+		}).Error("Failed to upgrade connection to websocket")
+
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 
 	c := &connection{
-		send:   make(chan []byte, 256),
+		send:   make(chan []byte, connectionChannelSize),
 		ws:     ws,
 		userID: user.ID,
 	}

api/sockets/pool.go

@@ -1,9 +1,11 @@
 package sockets
 
+import log "github.com/sirupsen/logrus"
+
 // hub maintains the set of active connections and broadcasts messages to the
 // connections.
 type hub struct {
-	// Registered connections.
+	// Registered websocket connections.
 	connections map[*connection]bool
 
 	// Inbound messages from the connections.
@@ -28,7 +30,6 @@ var h = hub{
 	connections: make(map[*connection]bool),
 }
 
-// nolint: gocyclo
 func (h *hub) run() {
 	for {
 		select {
@@ -40,16 +41,23 @@ func (h *hub) run() {
 				close(c.send)
 			}
 		case m := <-h.broadcast:
-			for c := range h.connections {
-				if m.userID > 0 && m.userID != c.userID {
+			for conn := range h.connections {
+				if m.userID > 0 && m.userID != conn.userID {
 					continue
 				}
 
 				select {
-				case c.send <- m.msg:
+				case conn.send <- m.msg:
 				default:
-					close(c.send)
-					delete(h.connections, c)
+
+					log.WithFields(log.Fields{
+						"context": "websocket",
+						"user_id": conn.userID,
+					}).Error("Connection send channel is full, connection closing")
+
+					close(conn.send)
+					delete(h.connections, conn)
+					_ = conn.ws.Close() // Close the WebSocket connection first
 				}
 			}
 		}

cli/cmd/root.go

@@ -36,16 +36,22 @@ Complete documentation is available at https://semaphoreui.com.`,
 		_ = cmd.Help()
 		os.Exit(0)
 	},
+
 	PersistentPreRun: func(cmd *cobra.Command, args []string) {
-		if persistentFlags.logLevel == "" {
+		str := persistentFlags.logLevel
+		if str == "" {
+			str = os.Getenv("SEMAPHORE_LOG_LEVEL")
+		}
+		if str == "" {
 			return
 		}
 
-		lvl, err := log.ParseLevel(persistentFlags.logLevel)
+		lvl, err := log.ParseLevel(str)
 		if err != nil {
 			log.Panic(err)
 		}
 
+		fmt.Println("Log level set to", lvl)
 		log.SetLevel(lvl)
 	},
 }

services/runners/running_job.go

@@ -96,17 +96,37 @@ func (p *runningJob) logPipe(reader io.Reader) {
 	defer p.logWG.Done()
 
 	scanner := bufio.NewScanner(reader)
+	const maxCapacity = 10 * 1024 * 1024 // 10 MB
+	buf := make([]byte, maxCapacity)
+	scanner.Buffer(buf, maxCapacity)
 
 	for scanner.Scan() {
 		line := scanner.Text()
 		p.Log(line)
 	}
 
-	if scanner.Err() != nil && scanner.Err().Error() != "EOF" {
-		//don't panic on these errors, sometimes it throws not dangerous "read |0: file already closed" error
-		log.WithError(scanner.Err()).WithFields(log.Fields{
-			"context": "task_log",
+	err := scanner.Err()
+
+	if err != nil {
+		msg := "Failed to read TaskRunner output"
+
+		switch err.Error() {
+		case "EOF",
+			"os: process already finished",
+			"read |0: file already closed":
+			return // it is ok
+		case "bufio.Scanner: token too long":
+			msg = "TaskRunner output exceeds the maximum allowed size of 10MB"
+			break
+		}
+
+		p.job.Kill() // kill the job because stdout cannot be read.
+
+		log.WithError(err).WithFields(log.Fields{
 			"task_id": p.job.Task.ID,
-		}).Debug("failed to read log")
+			"context": "task_logger",
+		}).Error(msg)
+
+		p.Log("Fatal error: " + msg)
 	}
 }

services/tasks/TaskRunner.go

@@ -192,7 +192,11 @@ func (t *TaskRunner) run() {
 		if t.job.IsKilled() {
 			t.SetStatus(task_logger.TaskStoppedStatus)
 		} else {
-			log.WithError(err).Warn("Failed to run task")
+			log.WithError(err).WithFields(log.Fields{
+				"task_id":     t.Task.ID,
+				"context":     "task_runner",
+				"task_status": t.Task.Status,
+			}).Warn("Failed to run task")
 			t.Log("Failed to run task: " + err.Error())
 			t.SetStatus(task_logger.TaskFailStatus)
 		}

services/tasks/TaskRunner_logging.go

@@ -153,6 +153,9 @@ func (t *TaskRunner) logPipe(reader io.Reader) {
 	}()
 
 	scanner := bufio.NewScanner(reader)
+	const maxCapacity = 10 * 1024 * 1024 // 10 MB
+	buf := make([]byte, maxCapacity)
+	scanner.Buffer(buf, maxCapacity)
 
 	for scanner.Scan() {
 		line := scanner.Text()
@@ -161,7 +164,28 @@ func (t *TaskRunner) logPipe(reader io.Reader) {
 
 	close(linesCh)
 
-	if scanner.Err() != nil && scanner.Err().Error() != "EOF" {
-		util.LogDebugF(scanner.Err(), log.Fields{"error": "Failed to read TaskRunner output"})
+	err := scanner.Err()
+
+	if err != nil {
+		msg := "Failed to read TaskRunner output"
+
+		switch err.Error() {
+		case "EOF",
+			"os: process already finished",
+			"read |0: file already closed":
+			return // it is ok
+		case "bufio.Scanner: token too long":
+			msg = "TaskRunner output exceeds the maximum allowed size of 10MB"
+			break
+		}
+
+		t.kill() // kill the job because stdout cannot be read.
+
+		log.WithError(err).WithFields(log.Fields{
+			"task_id": t.Task.ID,
+			"context": "task_logger",
+		}).Error(msg)
+
+		t.Log("Fatal error: " + msg)
 	}
 }