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

fix: restore watcher if channel is closed

Browse Source
This commit is contained in:
Mohamad Khani 2025-03-12 23:27:40 +03:30
parent 6f91ad607f
commit 4d6b06efe7
1 changed files with 44 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
internal/crd/watch.go
View File

@ -17,52 +17,54 @@ import (
func (crd Crd) watchFlinkJobs(ch chan FlinkJobCrdEvent) {
go func() {
pkg.Logger.Debug("[crd] starting watch")
watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
if err != nil {
panic(err)
}
defer watcher.Stop()
namespace := os.Getenv("NAMESPACE")
pkg.Logger.Debug("[crd] [watch]", zap.String("namespace", namespace))
for event := range watcher.ResultChan() {
unstructuredJob := event.Object.(*unstructured.Unstructured)
unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
for {
pkg.Logger.Debug("[crd] starting watch")
watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
if err != nil {
pkg.Logger.Error("[crd] [watch]cannot create unstructured map", zap.Error(err))
continue
panic(err)
}
job := &v1alpha1.FlinkJob{}
err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
if err != nil {
pkg.Logger.Error("[crd] [watch]cannot convert unstructured to structured", zap.Error(err))
continue
}
if job.Namespace != namespace {
continue
}
go func() {
ch <- FlinkJobCrdEvent{
EventType: event.Type,
Job: job,
namespace := os.Getenv("NAMESPACE")
pkg.Logger.Debug("[crd] [watch]", zap.String("namespace", namespace))
for event := range watcher.ResultChan() {
unstructuredJob := event.Object.(*unstructured.Unstructured)
unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
if err != nil {
pkg.Logger.Error("[crd] [watch]cannot create unstructured map", zap.Error(err))
continue
}
}()
pkg.Logger.Debug("[crd] [watch] change in", zap.String("name", job.Name), zap.String("operation", string(event.Type)))
switch event.Type {
case watch.Bookmark:
case watch.Modified:
//pkg.Logger.Info("[crd] [watch] flink job modified", zap.String("jobName", job.GetName()))
crd.repsert(job)
case watch.Added:
//pkg.Logger.Info("[crd] [watch] new flink job created")
crd.repsert(job)
case watch.Deleted:
crd.remove(job.UID)
}
}
job := &v1alpha1.FlinkJob{}
err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
if err != nil {
pkg.Logger.Error("[crd] [watch]cannot convert unstructured to structured", zap.Error(err))
continue
}
if job.Namespace != namespace {
continue
}
go func() {
ch <- FlinkJobCrdEvent{
EventType: event.Type,
Job: job,
}
}()
pkg.Logger.Debug("[crd] [watch] change in", zap.String("name", job.Name), zap.String("operation", string(event.Type)))
switch event.Type {
case watch.Bookmark:
case watch.Modified:
//pkg.Logger.Info("[crd] [watch] flink job modified", zap.String("jobName", job.GetName()))
crd.repsert(job)
case watch.Added:
//pkg.Logger.Info("[crd] [watch] new flink job created")
crd.repsert(job)
case watch.Deleted:
crd.remove(job.UID)
}
}
defer watcher.Stop()
pkg.Logger.Warn("[crd] [watch] Watcher stopped, restarting...")
}
}()
}