feat: if a job failed more than 10 times is marked as canceled

Author: segator

model/model.go

@@ -189,6 +189,16 @@ func (t TaskEvents) GetByEventId(i int) (*TaskEventType, error) {
 func (t TaskEvents) GetLastElement(i int) interface{} {
 	return t[i]
 }
+
+func (t TaskEvents) FilterBy(notification NotificationType, status NotificationStatus) (filteredEvents TaskEvents) {
+	for _, event := range t {
+		if event.NotificationType == notification && event.Status == status {
+			filteredEvents = append(filteredEvents, event)
+		}
+	}
+	return filteredEvents
+
+}
 func (v *Job) AddEvent(notificationType NotificationType, notificationStatus NotificationStatus) (newEvent *TaskEventType) {
 	return v.AddEventComplete(notificationType, notificationStatus, "")
 }

server/repository/repository.go

@@ -21,7 +21,7 @@ type Repository interface {
 	PingServerUpdate(ctx context.Context, pingEventType model.PingEventType) error
 	GetTimeoutJobs(ctx context.Context, timeout time.Duration) ([]*model.TaskEventType, error)
 	GetJobs(ctx context.Context) (*[]model.Job, error)
-	GetJobsByStatus(ctx context.Context, status model.NotificationStatus) (jobs []*model.Job, returnError error)
+	GetJobsByStatus(ctx context.Context, notificationType model.NotificationType, status model.NotificationStatus) (jobs []*model.Job, returnError error)
 	GetJob(ctx context.Context, uuid string) (*model.Job, error)
 	GetJobByPath(ctx context.Context, path string) (*model.Job, error)
 	AddJob(ctx context.Context, video *model.Job) error
@@ -297,16 +297,16 @@ func (s *SQLRepository) RetrieveQueuedJob(ctx context.Context) (video *model.Job
 	return s.queuedJob(ctx, conn)
 }
 
-func (s *SQLRepository) GetJobsByStatus(ctx context.Context, status model.NotificationStatus) (jobs []*model.Job, returnError error) {
+func (s *SQLRepository) GetJobsByStatus(ctx context.Context, notificationType model.NotificationType, status model.NotificationStatus) (jobs []*model.Job, returnError error) {
 	conn, err := s.getConnection()
 	if err != nil {
 		return nil, err
 	}
-	return s.getJobsByStatus(ctx, conn, status)
+	return s.getJobsByStatus(ctx, conn, notificationType, status)
 }
 
-func (s *SQLRepository) getJobsByStatus(ctx context.Context, tx SQLDBOperations, statusFilter model.NotificationStatus) ([]*model.Job, error) {
-	rows, err := tx.QueryContext(ctx, "SELECT v.id, v.source_path,v.source_size, v.target_path,v.target_size,vs.event_time, vs.status, vs.notification_type, vs.message FROM jobs v INNER JOIN job_status vs ON v.id = vs.job_id WHERE vs.status = $1;", statusFilter)
+func (s *SQLRepository) getJobsByStatus(ctx context.Context, tx SQLDBOperations, notificationTypeFilter model.NotificationType, statusFilter model.NotificationStatus) ([]*model.Job, error) {
+	rows, err := tx.QueryContext(ctx, "SELECT v.id, v.source_path,v.source_size, v.target_path,v.target_size,vs.event_time, vs.status, vs.notification_type, vs.message FROM jobs v INNER JOIN job_status vs ON v.id = vs.job_id WHERE vs.notification_type = $1 and vs.status = $2 order by event_time desc;", notificationTypeFilter, statusFilter)
 	if err != nil {
 		return nil, err
 	}

server/scheduler/scheduler.go

@@ -526,7 +526,7 @@ func (r *RuntimeScheduler) jobMaintenance(ctx context.Context) error {
 }
 
 func (r *RuntimeScheduler) queuedJobMaintenance(ctx context.Context) error {
-	queuedJobs, err := r.repo.GetJobsByStatus(ctx, model.QueuedNotificationStatus)
+	queuedJobs, err := r.repo.GetJobsByStatus(ctx, model.JobNotification, model.QueuedNotificationStatus)
 	if err != nil {
 		return err
 	}
@@ -546,22 +546,34 @@ func (r *RuntimeScheduler) queuedJobMaintenance(ctx context.Context) error {
 }
 
 func (r *RuntimeScheduler) failedJobMaintenance(ctx context.Context) error {
-	failedJobs, err := r.repo.GetJobsByStatus(ctx, model.FailedNotificationStatus)
+	failedJobs, err := r.repo.GetJobsByStatus(ctx, model.JobNotification, model.FailedNotificationStatus)
 	if err != nil {
 		return err
 	}
 	for _, failedJob := range failedJobs {
-		if verifyFailureMessage(failedJob.StatusMessage) {
-			jobRequest := &model.JobRequest{
-				SourcePath:  failedJob.SourcePath,
-				TargetPath:  failedJob.TargetPath,
-				ForceFailed: true,
-			}
-			_, err = r.scheduleJobRequest(ctx, jobRequest)
-			if err != nil {
+		if !verifyFailureMessage(failedJob.StatusMessage) {
+			continue
+		}
+
+		failureJobEvents := failedJob.Events.FilterBy(model.JobNotification, model.FailedNotificationStatus)
+		if len(failureJobEvents) > 10 {
+			newEvent := failedJob.AddEventComplete(model.JobNotification, model.CanceledNotificationStatus, "Job canceled by system, because of too many failed attempts")
+			if err = r.repo.AddNewTaskEvent(ctx, newEvent); err != nil {
 				return err
 			}
+			continue
+		}
+
+		jobRequest := &model.JobRequest{
+			SourcePath:  failedJob.SourcePath,
+			TargetPath:  failedJob.TargetPath,
+			ForceFailed: true,
 		}
+		_, err = r.scheduleJobRequest(ctx, jobRequest)
+		if err != nil {
+			return err
+		}
+
 	}
 	return nil
 }
@@ -638,15 +650,10 @@ func verifyFailureMessage(message string) bool {
 	if simpleRegex(`source File duration `, message) {
 		return false
 	}
-	if simpleRegex(`timeout Waiting for PGS Job Done`, message) {
-		return true
-	}
+
 	if simpleRegex(`Disk quota exceeded`, message) || simpleRegex(`No space left on device`, message) {
 		return true
 	}
-	if simpleRegex(`error on process PGS.*no such file or directory`, message) {
-		return true
-	}
 	// if simpleRegex(`At least one output file must be specified`, message) {
 	//	return true
 	//}
@@ -696,6 +703,11 @@ func verifyFailureMessage(message string) bool {
 	if simpleRegex(`connection refused`, message) {
 		return true
 	}
+
+	if simpleRegex(`with 0 items`, message) {
+		return false
+	}
+
 	// if simpleRegex(`srt: Invalid data found when processing input`, message) {
 	//	return true
 	//}
@@ -720,6 +732,7 @@ func verifyFailureMessage(message string) bool {
 	if simpleRegex(`probably corrupt input`, message) {
 		return false
 	}
+
 	return false
 }