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

feat: add type to get watch crd result

Browse Source
This commit is contained in:
Mohamad Khani 2024-11-30 19:31:36 +03:30
parent 19b874cba6
commit b0ff04126a
2 changed files with 27 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/crd/v1alpha1/flink_job.go
View File

@ -7,9 +7,8 @@ import (
//go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen object paths=$GOFILE //go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen object paths=$GOFILE
type FlinkJobSpec struct { type FlinkJobSpec struct {
Name string `json:"name"` Name string `json:"name"`
Parallelism int `json:"parallelism"` Parallelism int `json:"parallelism"`
Parallelism2 int `json:"parallelism2"`
} }
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

39
internal/crd/watch.go
View File

@ -2,28 +2,46 @@ package crd
import ( import (
"context" "context"
"flink-kube-operator/internal/crd/v1alpha1"
"fmt" "fmt"
"gitea.com/logicamp/lc" "gitea.com/logicamp/lc"
"go.uber.org/zap" "go.uber.org/zap"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch" "k8s.io/apimachinery/pkg/watch"
) )
func (crd Crd) watchFlinkJobs() { func (crd Crd) watchFlinkJobs() {
lc.Logger.Debug("[crd] starting watch") lc.Logger.Debug("[crd] starting watch")
watcher, err := crd.client.Watch(context.Background(), metav1.ListOptions{}) watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
if err != nil { if err != nil {
panic(err) panic(err)
} }
defer watcher.Stop() defer watcher.Stop()
for event := range watcher.ResultChan() { for event := range watcher.ResultChan() {
lc.Logger.Debug("[crd] event received") lc.Logger.Debug("[crd] event received", zap.Any("type", event.Type))
unstructuredJob := event.Object.(*unstructured.Unstructured)
unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
if err != nil {
lc.Logger.Error("cannot create unstructured map", zap.Error(err))
continue
}
job := &v1alpha1.FlinkJob{}
err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
if err != nil {
lc.Logger.Error("cannot convert unstructured to structured", zap.Error(err))
continue
}
switch event.Type { switch event.Type {
case watch.Bookmark:
case watch.Modified:
fmt.Printf("FlinkJob updated: %s\n", job.GetName())
// Handle the new FlinkJob
handleNewFlinkJob(job)
case watch.Added: case watch.Added:
job := event.Object.(*unstructured.Unstructured)
fmt.Printf("New FlinkJob created: %s\n", job.GetName()) fmt.Printf("New FlinkJob created: %s\n", job.GetName())
// Handle the new FlinkJob // Handle the new FlinkJob
handleNewFlinkJob(job) handleNewFlinkJob(job)
@ -31,21 +49,14 @@ func (crd Crd) watchFlinkJobs() {
} }
} }
func handleNewFlinkJob(job *unstructured.Unstructured) { func handleNewFlinkJob(job *v1alpha1.FlinkJob) {
// Extract job details // Extract job details
name := job.GetName() name := job.GetName()
namespace := job.GetNamespace() namespace := job.GetNamespace()
// Get specific fields using unstructured.Unstructured methods
spec, found, err := unstructured.NestedMap(job.Object, "spec")
if err != nil || !found {
fmt.Printf("Error getting spec for job %s: %v\n", name, err)
return
}
// Process job specification // Process job specification
fmt.Printf("Processing FlinkJob %s in namespace %s\n", name, namespace) fmt.Printf("Processing FlinkJob %s in namespace %s kind: %s \n", name, namespace, job.Kind)
lc.Logger.Debug("[crd] [watch]", zap.Any("spec", spec)) lc.Logger.Debug("[crd] [watch]", zap.Any("spec", job))
// Add your custom logic here // Add your custom logic here
} }