logicamp/flink-kube-operator
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve crash operator in some cases

Browse Source
This commit is contained in:
Mohamad Khani 2025-01-17 21:11:17 +03:30
parent 322877ad8e
commit 896d45e15a
3 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/crd/finalizer.go
View File

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

12
internal/managed_job/manager.go
View File

@ -50,8 +50,11 @@ func NewManager(client *api.Client, crdInstance *crd.Crd) Manager {
go func() { go func() {
for event := range crd.FinalizerChannel { for event := range crd.FinalizerChannel {
manager := mgr.GetJob(event) manager := mgr.GetJob(event)
manager.Stop() if manager != nil {
delete(mgr.managedJobs, event) 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 return mgr.managedJobs
} }
func (mgr *Manager) GetJob(id types.UID) ManagedJob { func (mgr *Manager) GetJob(id types.UID) *ManagedJob {
return mgr.managedJobs[id] job := mgr.managedJobs[id]
return &job
} }

8
internal/managed_job/remove.go
View File

@ -1,5 +1,11 @@
package managed_job package managed_job
import "errors"
func (job *ManagedJob) Stop() error { 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")
}
} }