Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

use last instead of first bacalhau execution #913

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

use last instead of first bacalhau execution #913

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 4 additions & 2 deletions gateway/utils/queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,8 @@ func processJob(jobID uint, db *gorm.DB) error {
} else if bacalhau.JobCompleted(bacalhauJob) {
fmt.Printf("Job %v , %v completed\n", job.ID, job.BacalhauJobID)
if len(bacalhauJob.State.Executions) > 0 {
return completeJobAndAddOutputFiles(&job, models.JobStateCompleted, bacalhauJob.State.Executions[0].PublishedResult.CID, db)
lastExecutionIndex := len(bacalhauJob.State.Executions) - 1
return completeJobAndAddOutputFiles(&job, models.JobStateCompleted, bacalhauJob.State.Executions[lastExecutionIndex].PublishedResult.CID, db)
}
return setJobStatus(&job, models.JobStateFailed, fmt.Sprintf("Output execution data lost for %v", job.BacalhauJobID), db)
} else if bacalhau.JobCancelled(bacalhauJob) {
Expand Down Expand Up @@ -220,7 +221,8 @@ func checkRunningJob(jobID uint, db *gorm.DB) error {
} else if bacalhau.JobCompleted(bacalhauJob) {
fmt.Printf("Job %v , %v completed, updating status and adding output files\n", job.ID, job.BacalhauJobID)
if len(bacalhauJob.State.Executions) > 0 {
return completeJobAndAddOutputFiles(&job, models.JobStateCompleted, bacalhauJob.State.Executions[0].PublishedResult.CID, db)
lastExecutionIndex := len(bacalhauJob.State.Executions) - 1
return completeJobAndAddOutputFiles(&job, models.JobStateCompleted, bacalhauJob.State.Executions[lastExecutionIndex].PublishedResult.CID, db)
}
return setJobStatus(&job, models.JobStateFailed, fmt.Sprintf("Output execution data lost for %v", job.BacalhauJobID), db)
} else {
Expand Down
18 changes: 11 additions & 7 deletions internal/bacalhau/bacalhau.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,14 +239,15 @@ func JobFailedWithCapacityError(job *model.JobWithInfo) bool {
capacityErrorMessages := []string{"not enough capacity", "not enough nodes", "does not have capacity"}
falseCapacityMessages := []string{"Could not inspect image", "node does not support the available image platforms"}
if len(job.State.Executions) > 0 {
fmt.Printf("Checking for capacity error, got error: %v\n", job.State.Executions[0].Status)
lastExecutionIndex := len(job.State.Executions) - 1
fmt.Printf("Checking for capacity error, got error: %v\n", job.State.Executions[lastExecutionIndex].Status)
for _, errorMsg := range falseCapacityMessages {
if job.State.State == model.JobStateError && strings.Contains(job.State.Executions[0].Status, errorMsg) {
if job.State.State == model.JobStateError && strings.Contains(job.State.Executions[lastExecutionIndex].Status, errorMsg) {
return false
}
}
for _, errorMsg := range capacityErrorMessages {
if job.State.State == model.JobStateError && strings.Contains(job.State.Executions[0].Status, errorMsg) {
if job.State.State == model.JobStateError && strings.Contains(job.State.Executions[lastExecutionIndex].Status, errorMsg) {
return true
}
}
Expand All @@ -257,16 +258,18 @@ func JobFailedWithCapacityError(job *model.JobWithInfo) bool {
func JobFailedWithBidRejectedError(job *model.JobWithInfo) bool {
capacityErrorMsg := "bid rejected"
if len(job.State.Executions) > 0 {
fmt.Printf("Checking for capacity error, got error: %v\n", job.State.Executions[0].Status)
return job.State.State == model.JobStateError && strings.Contains(job.State.Executions[0].Status, capacityErrorMsg)
lastExecutionIndex := len(job.State.Executions) - 1
fmt.Printf("Checking for capacity error, got error: %v\n", job.State.Executions[lastExecutionIndex].Status)
return job.State.State == model.JobStateError && strings.Contains(job.State.Executions[lastExecutionIndex].Status, capacityErrorMsg)
}
return false
}

func JobIsRunning(job *model.JobWithInfo) bool {
// the backend counts a Job as running once it is accepted by Bacalhau
if len(job.State.Executions) > 0 {
return job.State.State == model.JobStateInProgress || job.State.Executions[0].State == model.ExecutionStateBidAccepted
lastExecutionIndex := len(job.State.Executions) - 1
return job.State.State == model.JobStateInProgress || job.State.Executions[lastExecutionIndex].State == model.ExecutionStateBidAccepted
} else {
return job.State.State == model.JobStateInProgress
}
Expand All @@ -286,7 +289,8 @@ func JobCancelled(job *model.JobWithInfo) bool {

func JobBidAccepted(job *model.JobWithInfo) bool {
if len(job.State.Executions) > 0 {
return job.State.Executions[0].State == model.ExecutionStateBidAccepted
lastExecutionIndex := len(job.State.Executions) - 1
return job.State.Executions[lastExecutionIndex].State == model.ExecutionStateBidAccepted
}
return false
}
Expand Down