feat(managed-job): add restore from failed state

internal/managed_job/cycle.go

@@ -45,7 +45,7 @@ func (job *ManagedJob) cycle() {
 		return
 	}
 	// Check for set running or error state
-	if job.state.Status == JobStatusCreating {
+	if job.state.Status == JobStatusCreating || job.state.Status == JobStatusFailing {
 		err := job.checkStatus()
 		if errors.Is(err, ErrNoJobId) {
 			job.state = nil
@@ -67,5 +67,10 @@ func (job *ManagedJob) cycle() {
 		}
 		return
 	}
+	if job.state.Status == JobStatusFailed {
+		job.restore()
+		return
+	}
+	lc.Logger.Warn("[managed-job] [cycle]", zap.String("unhanded job status", string(job.state.Status)))
 
 }

internal/managed_job/restore.go

@@ -0,0 +1,34 @@
+package managed_job
+
+import (
+	"gitea.com/logicamp/lc"
+	api "github.com/logi-camp/go-flink-client"
+	"go.uber.org/zap"
+)
+
+// restore the job from savepoint and jarId in managedJob
+func (job *ManagedJob) restore() error {
+	if job.state.LastSavepointPath == nil {
+		lc.Logger.Error("[managed-job] [restore]", zap.Error(ErrNoSavepointPath))
+		return ErrNoSavepointPath
+	}
+	lc.Logger.Debug("[managed-job] [restore] restoring", zap.String("savepointPath", *job.state.LastSavepointPath))
+	runJarResp, err := job.client.RunJar(api.RunOpts{
+		JarID:                 job.jarId,
+		AllowNonRestoredState: true,
+		EntryClass:            job.def.Spec.EntryClass,
+		SavepointPath:         *job.state.LastSavepointPath,
+	})
+	if err != nil {
+		lc.Logger.Error("[managed-job] [run]", zap.Error(err))
+		return err
+	}
+	lc.Logger.Debug("[main] after run jar", zap.Any("run-jar-resp", runJarResp))
+
+	job.state.JobId = &runJarResp.JobId
+	job.state.Status = JobStatusCreating
+	job.state.Error = nil
+	job.updateState(*job.state)
+
+	return err
+}

internal/managed_job/savepoint.go

@@ -33,7 +33,7 @@ func (job ManagedJob) trackSavepoint() error {
 		return ErrNoSavepointTriggerId
 	}
 	resp, err := job.client.TrackSavepoint(*job.state.JobId, *job.state.SavepointTriggerId)
-	lc.Logger.Debug("[managed-job] [savepoint] track savepoint", zap.Any("resp", resp), zap.Error(err))
+	lc.Logger.Debug("[managed-job] [savepoint] track savepoint", zap.Any("status.Id", resp.Status.Id), zap.Any("failureCause.stacktrace", resp.Operation.FailureCause.StackTrace), zap.Error(err))
 	if err != nil {
 		if strings.IndexAny(err.Error(), "http status not 2xx: 404") == 0 {
 			job.removeSavepointTriggerId()

internal/managed_job/state.go

@@ -45,7 +45,7 @@ func (job *ManagedJob) setError(errMsg string) {
 }
 
 func (job *ManagedJob) setSavepointLocation(savepointId string) {
-	job.state.LastSavepointLocation = &savepointId
+	job.state.LastSavepointPath = &savepointId
 	job.state.SavepointTriggerId = nil
 	n := time.Now()
 	job.state.LastSavepointDate = &n

internal/managed_job/type.go

@@ -10,22 +10,31 @@ type JobStatus string
 var (
 	ErrNoJobId              = errors.New("[managed-job] no job id")
 	ErrNoSavepointTriggerId = errors.New("[managed-job] no savepoint trigger id")
+	ErrNoSavepointPath      = errors.New("[managed-job] no savepoint path")
 )
 
 const (
-	JobStatusRunning     JobStatus = "RUNNING"
-	JobStatusCreating    JobStatus = "CREATING"
-	JobStatusNotFound    JobStatus = "NotFound"
-	JobStatusError       JobStatus = "ERROR"
-	JobStatusReconciling JobStatus = "RECONCILING"
+	JobStatusInitializing JobStatus = "INITIALIZING"
+	JobStatusRunning      JobStatus = "RUNNING"
+	JobStatusCreating     JobStatus = "CREATING"
+	JobStatusNotFound     JobStatus = "NotFound"
+	JobStatusError        JobStatus = "ERROR"
+	JobStatusReconciling  JobStatus = "RECONCILING"
+	JobStatusFailed       JobStatus = "FAILED"
+	JobStatusFailing      JobStatus = "FAILING"
+	JobStatusRestarting   JobStatus = "RESTARTING"
+	JobStatusFinished     JobStatus = "FINISHED"
+	JobStatusCanceled     JobStatus = "CANCELED"
+	JobStatusCancelling   JobStatus = "CANCELLING"
+	JobStatusSuspended    JobStatus = "SUSPENDED"
 )
 
 type jobState struct {
-	Status                JobStatus  `json:"status"`
-	Error                 *string    `json:"error"`
-	Info                  *string    `json:"info"`
-	JobId                 *string    `json:"job_id"`
-	LastSavepointLocation *string    `json:"last_savepoint_location"`
-	SavepointTriggerId    *string    `json:"savepoint_trigger_id"`
-	LastSavepointDate     *time.Time `json:"last_savepoint_time"`
+	Status             JobStatus  `json:"status"`
+	Error              *string    `json:"error"`
+	Info               *string    `json:"info"`
+	JobId              *string    `json:"job_id"`
+	LastSavepointPath  *string    `json:"last_savepoint_location"`
+	SavepointTriggerId *string    `json:"savepoint_trigger_id"`
+	LastSavepointDate  *time.Time `json:"last_savepoint_time"`
 }