fix for incorrect error stopping the pipeline

Author: saul-data

app/workers/runtask/runtask.go

@@ -87,7 +87,10 @@ func worker(ctx context.Context, msg modelmain.WorkerTaskSend) {
 			log.Println(errl.Error())
 		}
 
-		WSLogError("Lock for run and node exists:"+errl.Error(), msg, TaskUpdate)
+		log.Println("Lock for run and node exists:", msg.RunID, msg.NodeID)
+
+		// == NB: this should be a silent fail and continue, the below will fail the entire graph
+		SilentWSLogError("Lock for run and node exists:"+errl.Error(), msg)
 
 		return
 	}
@@ -96,14 +99,14 @@ func worker(ctx context.Context, msg modelmain.WorkerTaskSend) {
 	var lockCheck modelmain.WorkerTasks
 	err2 := database.DBConn.Select("task_id", "status").Where("task_id = ? and environment_id= ?", msg.TaskID, msg.EnvironmentID).First(&lockCheck).Error
 	if err2 != nil {
-		log.Println(err2.Error())
-		WSLogError("Task already running:"+err2.Error(), msg, TaskUpdate)
+		log.Println("Task already running", err2.Error())
+		SilentWSLogError("Task already running:"+err2.Error(), msg)
 		return
 	}
 
 	if lockCheck.Status != "Queue" {
 		log.Println("Skipping not in queue", msg.RunID, msg.NodeID)
-		WSLogError("Skipping not in queue - runid:"+msg.RunID+" - node:"+msg.NodeID, msg, TaskUpdate)
+		SilentWSLogError("Skipping not in queue - runid:"+msg.RunID+" - node:"+msg.NodeID, msg)
 		return
 	}
 
@@ -114,7 +117,7 @@ func worker(ctx context.Context, msg modelmain.WorkerTaskSend) {
 	err2 = database.DBConn.Select("run_id", "status").Where("run_id = ?", msg.RunID).First(&pipelineCheck).Error
 	if err2 != nil {
 		log.Println(err2.Error())
-		WSLogError("Skipping not in queue - runid:"+msg.RunID+" - node:"+msg.NodeID, msg, TaskUpdate)
+		WSLogError("Pipeline marked as failed - runid:"+msg.RunID+" - node:"+msg.NodeID, msg, TaskUpdate)
 		return
 	}
 

app/workers/runtask/ws_silent_log_error.go

@@ -0,0 +1,28 @@
+package runtask
+
+import (
+	"time"
+
+	"github.com/dataplane-app/dataplane/app/mainapp/database"
+	"github.com/dataplane-app/dataplane/app/mainapp/database/models"
+	"github.com/dataplane-app/dataplane/app/workers/mqworker"
+	"github.com/google/uuid"
+)
+
+/* Return errors to the logging console without killing the pipeline (letting it complete) */
+func SilentWSLogError(logline string, msg models.WorkerTaskSend) {
+
+	uidstring := uuid.NewString()
+	sendmsg := models.LogsSend{
+		CreatedAt:     time.Now().UTC(),
+		UID:           uidstring,
+		Log:           logline,
+		LogType:       "error",
+		EnvironmentID: msg.EnvironmentID,
+		RunID:         msg.RunID,
+	}
+
+	mqworker.MsgSend("workerlogs."+msg.EnvironmentID+"."+msg.RunID+"."+msg.NodeID, sendmsg)
+	database.DBConn.Create(&sendmsg)
+
+}