46 lines
1.3 KiB
Go
46 lines
1.3 KiB
Go
package managed_job
|
|
|
|
import (
|
|
"flink-kube-operator/internal/crd/v1alpha1"
|
|
"time"
|
|
|
|
"flink-kube-operator/pkg"
|
|
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func (job *ManagedJob) Cycle() {
|
|
pkg.Logger.Debug("[managed-job] [new] check cycle", zap.String("jobKey", string(job.def.UID)))
|
|
|
|
// Init job
|
|
if job.def.Status.LifeCycleStatus == "" && job.def.Status.JobStatus == "" {
|
|
job.run()
|
|
return
|
|
}
|
|
|
|
if job.def.Status.JobStatus == v1alpha1.JobStatusRunning {
|
|
if (job.def.Spec.SavepointInterval.Duration != 0) && ((job.def.Status.LastSavepointDate == nil) || time.Now().Add(-job.def.Spec.SavepointInterval.Duration).After(*job.def.Status.LastSavepointDate)) {
|
|
if job.def.Status.SavepointTriggerId == nil {
|
|
job.createSavepoint()
|
|
} else {
|
|
job.trackSavepoint()
|
|
}
|
|
}
|
|
if job.def.Status.RunningJarURI != nil && job.def.Spec.JarURI != *job.def.Status.RunningJarURI {
|
|
job.upgrade()
|
|
}
|
|
|
|
return
|
|
}
|
|
// if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath != nil {
|
|
// //job.restore()
|
|
// return
|
|
// }
|
|
// if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath == nil {
|
|
// //job.restore()
|
|
// return
|
|
// }
|
|
|
|
pkg.Logger.Warn("[managed-job] [cycle]", zap.String("unhanded job status", string(job.def.Status.JobStatus)))
|
|
}
|