72 lines
2.1 KiB
Go
72 lines
2.1 KiB
Go
package managed_job
|
|
|
|
import (
|
|
"flink-kube-operator/internal/crd/v1alpha1"
|
|
"flink-kube-operator/pkg"
|
|
"os"
|
|
"time"
|
|
|
|
api "github.com/logi-camp/go-flink-client"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func (job *ManagedJob) Pause() error {
|
|
var err error
|
|
if job.def.Status.JobId != nil {
|
|
result, stopJobErr := job.client.StopJobWithSavepoint(*job.def.Status.JobId, os.Getenv("SAVEPOINT_PATH"), false)
|
|
if stopJobErr != nil {
|
|
err = stopJobErr
|
|
pkg.Logger.Error("[managed-job] [pause] cannot stop job", zap.Error(err))
|
|
return err
|
|
}
|
|
var savepointPath string
|
|
for {
|
|
trackResult, err := job.client.TrackSavepoint(*job.def.Status.JobId, result.RequestID)
|
|
time.Sleep(time.Millisecond * 500)
|
|
if err == nil && trackResult.Status.Id == api.SavepointStatusInCompleted {
|
|
if trackResult.Operation.Location != "" {
|
|
savepointPath = trackResult.Operation.Location
|
|
}
|
|
break
|
|
}
|
|
}
|
|
if savepointPath != "" {
|
|
job.def.Status.LastSavepointPath = &savepointPath
|
|
job.def.Status.PauseSavepointTriggerId = nil
|
|
job.def.Status.JobStatus = "FINISHED"
|
|
job.def.Status.LastSavepointPath = &savepointPath
|
|
lastSavepointDate := time.Now()
|
|
job.def.Status.LastSavepointDate = &lastSavepointDate
|
|
job.crd.Patch(job.def.UID, map[string]interface{}{
|
|
"status": map[string]interface{}{
|
|
"jobStatus": "FINISHED",
|
|
"lifeCycleStatus": v1alpha1.LifeCycleStatusGracefullyPaused,
|
|
"savepointTriggerId": nil,
|
|
"lastSavepointPath": savepointPath,
|
|
"lastSavepointDate": lastSavepointDate.Format(time.RFC3339),
|
|
},
|
|
})
|
|
pkg.Logger.Info(
|
|
"[managed-job] job paused successfully",
|
|
zap.String("jobName", job.def.GetName()),
|
|
zap.String("savepointPath", savepointPath),
|
|
)
|
|
} else {
|
|
job.crd.Patch(job.def.UID, map[string]interface{}{
|
|
"status": map[string]interface{}{
|
|
"lifeCycleStatus": v1alpha1.LifeCycleStatusGracefulStopFailed,
|
|
"lastSavepointPath": savepointPath,
|
|
"lastSavepointDate": time.Now().Format(time.RFC3339),
|
|
},
|
|
})
|
|
pkg.Logger.Error(
|
|
"[managed-job] error in pausing job",
|
|
zap.Error(err),
|
|
)
|
|
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|