fix: improve activity verification and output handling in HPC runtime service

Author: ovvesley

pkg/server/runtimes/hpc_runtime/hpc_runtime_service/hpc_runtime_service.go

@@ -3,6 +3,7 @@ package hpc_runtime_service
 import (
 	"fmt"
 	"regexp"
+	"strconv"
 	"strings"
 
 	"github.com/ovvesley/akoflow/pkg/server/config"
@@ -200,88 +201,102 @@ func (s *HPCRuntimeService) extractJobID(outputCommand string) (string, error) {
 func (s *HPCRuntimeService) VerifyActivitiesWasFinished(workflow workflow_entity.Workflow) bool {
 	config.App().Logger.Infof("WORKER: Verify activities was finished in HPCRuntime")
 
+	finishedActivities := s.getFinishedActivities(workflow)
+	if len(finishedActivities) == 0 {
+		return true
+	}
+
+	syncNeeded := false
 	for _, activity := range workflow.Spec.Activities {
-		if activity.GetRuntimeId() == s.runtimeName {
-			s.handleVerifyActivityWasFinished(activity, workflow)
+		if activity.GetRuntimeId() != s.runtimeName {
+			continue
+		}
+
+		if _, ok := finishedActivities[activity.Id]; !ok {
+			continue
+		}
+
+		if s.handleVerifyActivityWasFinished(activity) {
+			syncNeeded = true
 		}
 	}
+
+	if syncNeeded {
+		s.syncWorkflowVolumes(workflow)
+	}
+
 	return true
 }
 
-func (s *HPCRuntimeService) handleVerifyActivityWasFinished(activity workflow_activity_entity.WorkflowActivities, wf workflow_entity.Workflow) int {
-	println("Verifying activity: ", activity.Name, " with id: ", activity.Id)
-
-	wfaDatabase, _ := s.activityRepository.Find(activity.Id)
+func (s *HPCRuntimeService) getFinishedActivities(workflow workflow_entity.Workflow) map[int]struct{} {
+	finishedActivities := make(map[int]struct{})
 
-	if wfaDatabase.Status == activity_repository.StatusFinished {
-		return activity_repository.StatusFinished
+	runtimeName := s.runtimeName
+	if runtimeName == "" && len(workflow.GetRuntimeId()) > 0 {
+		runtimeName = workflow.GetRuntimeId()[0]
 	}
 
-	if wfaDatabase.Status == activity_repository.StatusCreated {
-		return activity_repository.StatusCreated
+	if runtimeName == "" {
+		return finishedActivities
 	}
 
-	runtime, err := s.runtimeRepository.GetByName(activity.GetRuntimeId())
-
+	runtime, err := s.runtimeRepository.GetByName(runtimeName)
 	if err != nil {
 		config.App().Logger.Infof("WORKER: Error getting runtime from database.")
-		return activity_repository.StatusRunning
+		return finishedActivities
 	}
 
-	if wfaDatabase.GetProcId() == "" {
-		config.App().Logger.Infof("WORKER: Activity %d has no process ID", activity.Id)
-		return activity_repository.StatusRunning
+	sharedMountPath := workflow.GetMountPath()
+	if sharedMountPath == "" {
+		sharedMountPath = runtime.GetCurrentRuntimeMetadata("MOUNT_PATH")
 	}
 
-	commandVerify := fmt.Sprintf("cat %s/akoflow_finished_%d_%d.txt", runtime.GetCurrentRuntimeMetadata("MOUNT_PATH"), wf.GetId(), activity.Id)
+	commandVerify := fmt.Sprintf("find %s -maxdepth 1 -type f -name 'akoflow_finished_%d_*.txt' -print", sharedMountPath, workflow.GetId())
+	output, err := s.connectorHPCRuntime.SetRuntime(*runtime).RunCommandWithOutputRemote(commandVerify)
+	if err != nil {
+		config.App().Logger.Infof("WORKER: Error listing finished activities for workflow %d", workflow.GetId())
+		return finishedActivities
+	}
 
-	output, _ := s.connectorHPCRuntime.SetRuntime(*runtime).RunCommandWithOutputRemote(commandVerify)
+	re := regexp.MustCompile(fmt.Sprintf(`akoflow_finished_%d_(\d+)\.txt`, workflow.GetId()))
+	for _, match := range re.FindAllStringSubmatch(output, -1) {
+		if len(match) < 2 {
+			continue
+		}
 
-	scontrolResponse, err := s.extractJobIsFinished(output)
+		activityID, err := strconv.Atoi(strings.TrimSpace(match[1]))
+		if err != nil {
+			continue
+		}
 
-	if err != nil {
-		config.App().Logger.Infof("WORKER: Error extracting job ID %s", strings.TrimSpace(wfaDatabase.GetProcId()))
-		return activity_repository.StatusRunning
+		finishedActivities[activityID] = struct{}{}
 	}
 
-	if scontrolResponse.State == "COMPLETED" {
-		config.App().Logger.Infof("WORKER: Activity %d finished", activity.Id)
-		s.syncWorkflowVolumes(wf)
-		_ = s.activityRepository.UpdateStatus(activity.Id, activity_repository.StatusFinished)
-		return activity_repository.StatusFinished
-	}
+	return finishedActivities
+}
 
-	if scontrolResponse.State == "FAILED" || scontrolResponse.State == "CANCELLED+" || scontrolResponse.State == "CANCELLED" || scontrolResponse.State == "DEADLINE" || scontrolResponse.State == "TIMEOUT" || scontrolResponse.State == "OUT_OF_MEM+" {
-		config.App().Logger.Infof("WORKER: Activity %d failed", activity.Id)
-		s.syncWorkflowVolumes(wf)
-		_ = s.activityRepository.UpdateStatus(activity.Id, activity_repository.StatusFinished)
-		return activity_repository.StatusFinished
-	}
+func (s *HPCRuntimeService) handleVerifyActivityWasFinished(activity workflow_activity_entity.WorkflowActivities) bool {
+	println("Verifying activity: ", activity.Name, " with id: ", activity.Id)
 
-	if scontrolResponse.State == "RUNNING" {
-		config.App().Logger.Infof("WORKER: Activity %d running", activity.Id)
-		_ = s.activityRepository.UpdateStatus(activity.Id, activity_repository.StatusRunning)
-		return activity_repository.StatusRunning
+	wfaDatabase, _ := s.activityRepository.Find(activity.Id)
+
+	if wfaDatabase.Status == activity_repository.StatusFinished {
+		return false
 	}
 
-	if scontrolResponse.State == "PENDING" {
-		config.App().Logger.Infof("WORKER: Activity %d pending", activity.Id)
-		_ = s.activityRepository.UpdateStatus(activity.Id, activity_repository.StatusRunning)
-		return activity_repository.StatusRunning
+	if wfaDatabase.Status == activity_repository.StatusCreated {
+		return false
 	}
 
-	return activity_repository.StatusRunning
+	if wfaDatabase.GetProcId() == "" {
+		config.App().Logger.Infof("WORKER: Activity %d has no process ID", activity.Id)
+		return false
+	}
 
-}
+	config.App().Logger.Infof("WORKER: Activity %d finished", activity.Id)
+	_ = s.activityRepository.UpdateStatus(activity.Id, activity_repository.StatusFinished)
+	return true
 
-type SaactResponse struct {
-	JobID     string `json:"JobID"`
-	JobName   string `json:"JobName"`
-	Partition string `json:"Partition"`
-	Account   string `json:"Account"`
-	AllocCPUs string `json:"AllocCPUs"`
-	State     string `json:"State"`
-	ExitCode  string `json:"ExitCode"`
 }
 
 func extractField(pattern, text string) (string, error) {
@@ -293,18 +308,6 @@ func extractField(pattern, text string) (string, error) {
 	return match[1], nil
 }
 
-func (s *HPCRuntimeService) extractJobIsFinished(output string) (SaactResponse, error) {
-	// verify if output contains "AKOFLOW_JOB_FINISHED"
-	if !strings.Contains(output, "AKOFLOW_JOB_FINISHED") {
-		return SaactResponse{}, fmt.Errorf("job not finished")
-	}
-
-	return SaactResponse{
-		State: "COMPLETED",
-	}, nil
-
-}
-
 func (s *HPCRuntimeService) HealthCheck(runtimeName string) bool {
 	config.App().Logger.Infof("WORKER: Health check HPCRuntime")
 

pkg/server/runtimes/hpc_runtime/hpc_runtime_service/make_sbatch_hpc_activity_service.go

@@ -60,11 +60,15 @@ func (m MakeSBatchHPCRuntimeActivityService) GetTemplateSbatch() string {
 func (m MakeSBatchHPCRuntimeActivityService) Handle(workflow workflow_entity.Workflow, activity workflow_activity_entity.WorkflowActivities) string {
 
 	templateSbatch := m.GetTemplateSbatch()
+	sharedMountPath := workflow.GetMountPath()
+	if sharedMountPath == "" {
+		sharedMountPath = m.runtime.GetCurrentRuntimeMetadata("MOUNT_PATH")
+	}
 
 	jobName := fmt.Sprintf("akoflow_%d_%d", workflow.GetId(), activity.GetId())
 
-	output := fmt.Sprintf("%s/akoflow_out_%d_%d.out", m.runtime.GetCurrentRuntimeMetadata("MOUNT_PATH"), workflow.GetId(), activity.GetId())
-	error := fmt.Sprintf("%s/akoflow_err_%d_%d.err", m.runtime.GetCurrentRuntimeMetadata("MOUNT_PATH"), workflow.GetId(), activity.GetId())
+	output := fmt.Sprintf("%s/akoflow_out_%d_%d.out", sharedMountPath, workflow.GetId(), activity.GetId())
+	error := fmt.Sprintf("%s/akoflow_err_%d_%d.err", sharedMountPath, workflow.GetId(), activity.GetId())
 	time := m.runtime.GetCurrentRuntimeMetadata("TIME")
 	partition := m.runtime.GetCurrentRuntimeMetadata("QUEUE")
 	ntasks := m.runtime.GetCurrentRuntimeMetadata("NTASKS")
@@ -86,7 +90,7 @@ func (m MakeSBatchHPCRuntimeActivityService) Handle(workflow workflow_entity.Wor
 	templateSbatch = strings.ReplaceAll(templateSbatch, "#MEM#", mem)
 	templateSbatch = strings.ReplaceAll(templateSbatch, "#COMMAND#", wrap)
 
-	templateSbatch += "\n\necho AKOFLOW_JOB_FINISHED > " + fmt.Sprintf("%s/akoflow_finished_%d_%d.txt", m.runtime.GetCurrentRuntimeMetadata("MOUNT_PATH"), workflow.GetId(), activity.GetId())
+	templateSbatch += "\n\necho AKOFLOW_JOB_FINISHED > " + fmt.Sprintf("%s/akoflow_finished_%d_%d.txt", sharedMountPath, workflow.GetId(), activity.GetId())
 
 	templateSbatchBase64 := base64.StdEncoding.EncodeToString([]byte(templateSbatch))
 

pkg/server/runtimes/singularity_runtime/singularity_runtime_service/make_singularity_activity_service.go

@@ -65,10 +65,8 @@ func (s *MakeSingularityActivityService) MakeContainerCommandActivityToHPC(wf wo
 	commandBase64 := base64.StdEncoding.EncodeToString([]byte(command))
 	commandFinal := "echo " + commandBase64 + " | base64 -d | sh"
 
-	entryPoint := fmt.Sprintf("singularity exec --bind %s:%s --pwd %s %s bash -c '%s'",
+	entryPoint := fmt.Sprintf("singularity exec --bind %s:/akoflow-wfa-shared --pwd /akoflow-wfa-shared %s bash -c '%s'",
 		mountPath,
-		mountPath,
-		"/akoflow-wfa-shared ",
 		imageSifPath,
 		commandFinal,
 	)