77 lines
2.2 KiB
Go
77 lines
2.2 KiB
Go
package managed_job
|
|
|
|
import (
|
|
"errors"
|
|
"flink-kube-operator/internal/crd/v1alpha1"
|
|
"flink-kube-operator/internal/jar"
|
|
|
|
"gitea.com/logicamp/lc"
|
|
api "github.com/logi-camp/go-flink-client"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// upload jar file and set the jarId for later usages
|
|
func (job *ManagedJob) upload() error {
|
|
jarFile, err := jar.NewJarFile(job.def.Spec.JarURI)
|
|
if err != nil {
|
|
lc.Logger.Debug("[main] error on download jar", zap.Error(err))
|
|
return err
|
|
}
|
|
fileName, err := jarFile.Upload(job.client)
|
|
jarFile.Delete()
|
|
if err != nil {
|
|
lc.Logger.Debug("[main] error on upload jar", zap.Error(err))
|
|
return err
|
|
}
|
|
lc.Logger.Debug("[main] after upload jar", zap.Any("upload-jar-resp", fileName))
|
|
|
|
job.def.Status.JarId = &fileName
|
|
job.crd.Patch(job.def.UID, map[string]interface{}{
|
|
"status": map[string]interface{}{
|
|
"jarId": job.def.Status.JarId,
|
|
},
|
|
})
|
|
return nil
|
|
}
|
|
|
|
// run the job from saved jarId in managedJob
|
|
func (job *ManagedJob) run() error {
|
|
if job.def.Status.JarId == nil {
|
|
err := errors.New("missing jar id")
|
|
lc.Logger.Error("[managed-job] [run]", zap.Error(err))
|
|
return err
|
|
}
|
|
lc.Logger.Info("[managed-job] [run] starting job", zap.String("name", job.def.GetName()))
|
|
runJarResp, err := job.client.RunJar(api.RunOpts{
|
|
JarID: *job.def.Status.JarId,
|
|
AllowNonRestoredState: true,
|
|
EntryClass: job.def.Spec.EntryClass,
|
|
})
|
|
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))
|
|
|
|
// if job.state == nil {
|
|
// job.state = &jobState{}
|
|
// }
|
|
// job.state.JobId = &runJarResp.JobId
|
|
// job.state.Status = v1alpha1.JobStatusCreating
|
|
// job.updateState(*job.state)
|
|
// job.crd.SetJobStatus(job.def.UID, v1alpha1.FlinkJobStatus{
|
|
// JobId: job.state.JobId,
|
|
// })
|
|
job.crd.Patch(job.def.UID, map[string]interface{}{
|
|
"status": map[string]interface{}{
|
|
"jobId": &runJarResp.JobId,
|
|
"jobStatus": v1alpha1.JobStatusCreating,
|
|
"lifeCycleStatus": v1alpha1.LifeCycleStatusInitializing,
|
|
"error": nil,
|
|
},
|
|
})
|
|
//job.updateState(jobState{JobId: &runJarResp.JobId, Status: JobStatusCreating})
|
|
|
|
return err
|
|
}
|