Feat/persisting generated log files and updating SR cancellation endpoint

README.md

@@ -17,10 +17,10 @@ Run the following command to seed the database wth sample data
 ```bash
 docker compose --profile be-seed -p flowforge up --build
 ```
-You may specify the following arguments in front of the above command to seed your own files:
-create_users -> specifies whether to create user in auth0, default is "false"
-users -> specify user csv file name, default is "user.csv"
-orgs -> specify org csv file name, default is "org.csv"
+You may specify the following arguments in front of the above command to seed your own files:<br>
+create_users -> specifies whether to create user in auth0, default is "false"<br>
+users -> specify user csv file name, default is "user.csv"<br>
+orgs -> specify org csv file name, default is "org.csv"<br>
 memberships -> specify membership csv file name, default is "membership.csv"
 
 Example:
@@ -44,8 +44,9 @@ docker compose --profile be -p flowforge up --build
 Run the following command to delete container
 
 ```bash
-docker compose -p flowforge down -v --rmi local 
+docker compose -p flowforge down --rmi local 
 ```
+**Note: To remove all volumes associated with the service, add -v flag after down**
 
 ## Setup Frontend Development
 

backend/database/mongo_seed/main.go

@@ -4,11 +4,9 @@ import (
 	"os"
 	"time"
 
-	"github.com/gookit/event"
 	"github.com/joshtyf/flowforge/src/database"
 	"github.com/joshtyf/flowforge/src/database/client"
 	"github.com/joshtyf/flowforge/src/database/models"
-	"github.com/joshtyf/flowforge/src/events"
 	"github.com/joshtyf/flowforge/src/logger"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 )
@@ -601,13 +599,4 @@ func main() {
 			panic("Inserted ID is not an ObjectID")
 		}
 	}
-
-	// start SRs
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest1))
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest7))
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest3))
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest6))
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest9))
-	event.FireAsync(events.NewNewServiceRequestEvent(&serviceRequest12))
-
 }

backend/src/database/models/pipeline.go

@@ -37,6 +37,19 @@ const (
 	WaitForApprovalStep PipelineStepType = "WAIT_FOR_APPROVAL"
 )
 
+var cancellableStepTypes = []PipelineStepType{
+	WaitForApprovalStep,
+}
+
+func IsCancellablePipelineStepType(stepType PipelineStepType) bool {
+	for _, cancellableStepType := range cancellableStepTypes {
+		if stepType == cancellableStepType {
+			return true
+		}
+	}
+	return false
+}
+
 var allPipelineStepTypes = []PipelineStepType{APIStep, WaitForApprovalStep}
 
 func IsValidPipelineStepType(stepType PipelineStepType) bool {

backend/src/execute/step_exec_mgr.go

@@ -141,7 +141,6 @@ func (srm *ExecutionManager) execute(serviceRequest *models.ServiceRequestModel,
 	}
 
 	// Create a log file for the current step
-	// TODO: file is not persisted in Docker container
 	f, err := os.OpenFile(
 		logger.CreateExecutorLogFilePath(serviceRequest.Id.Hex(), step.StepName),
 		os.O_RDWR|os.O_CREATE|os.O_APPEND,
@@ -220,6 +219,17 @@ func (srm *ExecutionManager) handleCompletedStepEvent(e event.Event) error {
 		return nil
 	}
 
+	// Check if SR has been cancelled
+	sr, err := database.NewServiceRequest(srm.mongoClient).GetById(serviceRequest.Id.Hex())
+	if err != nil {
+		srm.logger.Error(fmt.Sprintf("error encounter while verifying sr status: %s", err))
+		return err
+	}
+	if sr.Status == models.CANCELLED {
+		srm.logger.Info(fmt.Sprintf("service request %s has been cancelled. Will not proceed to execute next step", sr.Id.Hex()))
+		return nil
+	}
+
 	// Set the current executor to the next executor
 	nextStep := pipeline.GetPipelineStep(completedStepModel.NextStepName)
 	if nextStep == nil {

backend/src/server/handlers.go

@@ -56,7 +56,7 @@ func (s *ServerHandler) registerRoutes(r *mux.Router) {
 	r.Handle("/api/service_request/{requestId}", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgMember(s.psqlClient, handleGetServiceRequest(s.logger, s.mongoClient, s.psqlClient), s.logger), s.logger), s.logger)).Methods("GET")
 	r.Handle("/api/service_request", isAuthenticated(getOrgIdFromRequestBody(isOrgMember(s.psqlClient, handleCreateServiceRequest(s.logger, s.mongoClient, s.psqlClient), s.logger), s.logger), s.logger)).Methods("POST").Headers("Content-Type", "application/json")
 	r.Handle("/api/service_request/{requestId}", isAuthenticated(getOrgIdFromRequestBody(isOrgMember(s.psqlClient, handleUpdateServiceRequest(s.logger, s.mongoClient), s.logger), s.logger), s.logger)).Methods("PATCH").Headers("Content-Type", "application/json")
-	r.Handle("/api/service_request/{requestId}/cancel", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgMember(s.psqlClient, handleCancelStartedServiceRequest(s.logger, s.mongoClient), s.logger), s.logger), s.logger)).Methods("PUT")
+	r.Handle("/api/service_request/{requestId}/cancel", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgMember(s.psqlClient, handleCancelStartedServiceRequest(s.logger, s.mongoClient, s.psqlClient), s.logger), s.logger), s.logger)).Methods("PUT")
 	r.Handle("/api/service_request/{requestId}/start", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgMember(s.psqlClient, handleStartServiceRequest(s.logger, s.mongoClient), s.logger), s.logger), s.logger)).Methods("PUT")
 	r.Handle("/api/service_request/{requestId}/approve", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgAdmin(s.psqlClient, handleApproveServiceRequest(s.logger, s.mongoClient, s.psqlClient), s.logger), s.logger), s.logger)).Methods("PUT")
 	r.Handle("/api/service_request/{requestId}/reject", isAuthenticated(getOrgIdUsingSrId(s.mongoClient, isOrgAdmin(s.psqlClient, handleRejectServiceRequest(s.logger, s.mongoClient, s.psqlClient), s.logger), s.logger), s.logger)).Methods("PUT")
@@ -287,7 +287,7 @@ func handleCreateServiceRequest(logger logger.ServerLogger, mongoClient *mongo.C
 	})
 }
 
-func handleCancelStartedServiceRequest(logger logger.ServerLogger, client *mongo.Client) http.Handler {
+func handleCancelStartedServiceRequest(logger logger.ServerLogger, client *mongo.Client, psqlClient *sql.DB) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		vars := mux.Vars(r)
 		requestId := vars["requestId"]
@@ -298,20 +298,19 @@ func handleCancelStartedServiceRequest(logger logger.ServerLogger, client *mongo
 			return
 		}
 		status := sr.Status
-		if status != models.PENDING && status != models.RUNNING {
+		if status == models.COMPLETED {
 			logger.Error(fmt.Sprintf("failed to %s service request %s: %s", "cancel", requestId, "execution has been completed"))
 			encode(w, r, http.StatusBadRequest, newHandlerError(ErrServiceRequestAlreadyCompleted, http.StatusBadRequest))
 			return
 		}
 
-		if status == models.NOT_STARTED {
-			logger.Error(fmt.Sprintf("failed to %s service request %s: %s", "cancel", requestId, "execution has not been started"))
-			encode(w, r, http.StatusBadRequest, newHandlerError(ErrServiceRequestNotStarted, http.StatusBadRequest))
+		sre, err := database.NewServiceRequestEvent(psqlClient).GetLatestStepEvent(requestId)
+		if err != nil {
+			logger.Error(fmt.Sprintf("error encountered while handling API request: %s", err))
+			encode(w, r, http.StatusInternalServerError, newHandlerError(ErrInternalServerError, http.StatusInternalServerError))
 			return
 		}
 
-		// TODO: implement cancellation of sr
-
 		err = database.NewServiceRequest(client).UpdateStatus(requestId, models.CANCELLED)
 
 		// TODO: discuss how to handle this error
@@ -320,6 +319,24 @@ func handleCancelStartedServiceRequest(logger logger.ServerLogger, client *mongo
 			encode(w, r, http.StatusInternalServerError, newHandlerError(ErrInternalServerError, http.StatusInternalServerError))
 			return
 		}
+
+		if models.IsCancellablePipelineStepType(sre.StepType) {
+			userId := r.Context().Value(jwtmiddleware.ContextKey{}).(*validator.ValidatedClaims).RegisteredClaims.Subject
+			// Log step cancelled event
+			serviceRequestEvent := database.NewServiceRequestEvent(psqlClient)
+			err = serviceRequestEvent.Create(&models.ServiceRequestEventModel{
+				EventType:        models.STEP_COMPLETED,
+				ServiceRequestId: sr.Id.Hex(),
+				StepName:         sre.StepName,
+				CreatedBy:        userId,
+				StepType:         sre.StepType,
+			})
+			if err != nil {
+				encode(w, r, http.StatusInternalServerError, newHandlerError(ErrInternalServerError, http.StatusInternalServerError))
+				return
+			}
+		}
+
 		encode[any](w, r, http.StatusOK, nil)
 	})
 }

compose.yaml

@@ -20,6 +20,8 @@ services:
       - be
     build:
       context: backend
+    volumes:
+      - flowforge_executor_logs:/backend/executor_logs
     environment:
       - ENV=${ENV}
       - MONGO_URI=${MONGO_URI}
@@ -123,3 +125,6 @@ services:
 
     environment:
       - MONGO_URI=${MONGO_URI}
+
+volumes:
+  flowforge_executor_logs: