Seraj Haqiqi
a36e72877d
- Add jarRef and jarPullSecret fields to FlinkJob CRD (jarUri/basicAuth deprecated) - OCI pull via go-containerregistry with dual auth (K8s pull secret + env vars) - Media type validation on pulled layers - Atomic status patches with runningJarRef/runningJarDigest tracking - NeedsUpgrade/RunningRef/RunningRefPatchData domain helpers - README with usage guide, pushing JARs, and GitHub Actions CI/CD workflow - CONTEXT.md domain glossary
47 lines
1.2 KiB
Go
47 lines
1.2 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() {
|
|
// Init job
|
|
if job.def.Status.LifeCycleStatus == "" && (job.def.Status.JobStatus == "" || job.def.Status.JobStatus == v1alpha1.JobStatusFinished) {
|
|
job.Run(false)
|
|
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.Spec.NeedsUpgrade(job.def.Status) {
|
|
job.upgrade()
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
if job.def.Status.JobStatus == v1alpha1.JobStatusCreating {
|
|
return
|
|
}
|
|
|
|
if job.def.Status.JobStatus == v1alpha1.JobStatusFailed {
|
|
job.def.Status.LifeCycleStatus = v1alpha1.LifeCycleStatusFailed
|
|
job.crd.SetJobStatus(job.def.UID, job.def.Status)
|
|
return
|
|
}
|
|
|
|
pkg.Logger.Warn("[managed-job] [cycle] unhandled job status", zap.String("name", job.def.Name), zap.String("status", string(job.def.Status.JobStatus)), zap.String("namespace", job.def.Namespace))
|
|
}
|