Compare commits

...

2 Commits

4 changed files with 19 additions and 9 deletions

View File

@ -27,14 +27,14 @@ func (crd Crd) manageFinalizer(jobEventObservable rxgo.Observable) {
// Perform cleanup
pkg.Logger.Debug("[finalizer] stopping managed job", zap.String("name", jobEvent.Job.GetName()))
if err := crd.cleanupResources(jobEvent.Job); err != nil {
pkg.Logger.Error("[crd] [manage-finalizer] cleanup failed", zap.Error(err))
pkg.Logger.Info("[crd] [manage-finalizer] cleanup failed", zap.Error(err))
return
}
// Remove finalizer
controllerutil.RemoveFinalizer(jobEvent.Job, finalizerName)
if err := crd.runtimeClient.Update(context.Background(), jobEvent.Job); err != nil {
pkg.Logger.Error("[crd] [manage-finalizer] failed to remove finalizer", zap.Error(err))
pkg.Logger.Info("[crd] [manage-finalizer] failed to remove finalizer", zap.Error(err))
return
}
pkg.Logger.Debug("[crd] [manage-finalizer] job removed", zap.String("name", jobEvent.Job.GetName()))
@ -49,7 +49,7 @@ func (crd Crd) manageFinalizer(jobEventObservable rxgo.Observable) {
pkg.Logger.Debug("[finalizer] adding job")
// Update the resource to add the finalizer
if err := crd.runtimeClient.Update(context.Background(), jobEvent.Job); err != nil {
pkg.Logger.Error("[finalizer] failed to add", zap.Error(err))
pkg.Logger.Info("[finalizer] failed to add", zap.Error(err))
return
}
}

View File

@ -50,8 +50,11 @@ func NewManager(client *api.Client, crdInstance *crd.Crd) Manager {
go func() {
for event := range crd.FinalizerChannel {
manager := mgr.GetJob(event)
manager.Stop()
delete(mgr.managedJobs, event)
if manager != nil {
err := manager.Stop()
pkg.Logger.Info("[finalizer]", zap.Error(err))
delete(mgr.managedJobs, event)
}
}
}()
@ -129,6 +132,7 @@ func (mgr *Manager) GetJobs() map[types.UID]ManagedJob {
return mgr.managedJobs
}
func (mgr *Manager) GetJob(id types.UID) ManagedJob {
return mgr.managedJobs[id]
func (mgr *Manager) GetJob(id types.UID) *ManagedJob {
job := mgr.managedJobs[id]
return &job
}

View File

@ -33,7 +33,7 @@ func (job *ManagedJob) Pause() error {
if savepointPath != "" {
job.def.Status.LastSavepointPath = &savepointPath
job.def.Status.PauseSavepointTriggerId = nil
job.def.Status.JobStatus = ""
job.def.Status.JobStatus = "FINISHED"
job.def.Status.LastSavepointPath = &savepointPath
lastSavepointDate := time.Now()
job.def.Status.LastSavepointDate = &lastSavepointDate

View File

@ -1,5 +1,11 @@
package managed_job
import "errors"
func (job *ManagedJob) Stop() error {
return job.client.StopJob(*job.def.Status.JobId)
if job.def.Status.JobId != nil {
return job.client.StopJob(*job.def.Status.JobId)
} else {
return errors.New("job Id not found")
}
}