feat: support multiple operator in single namespace

crds.yaml

@@ -2,9 +2,9 @@
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
-  name: flink-jobs.flink.logicamp.tech
+  name: flink-jobs.flink.logicamp.dev
 spec:
-  group: flink.logicamp.tech
+  group: flink.logicamp.dev
   names:
     kind: FlinkJob
     plural: flink-jobs
@@ -30,6 +30,9 @@ spec:
                   type: string
                 name:
                   type: string
+                flinkCluster:
+                  type: string
+                  description: "Identifier of the Flink cluster that manages this job. Defaults to the OPERATOR_ID (usually the Helm release name) if not set."
                 entryClass:
                   type: string
                 parallelism:

example-job.yaml

@@ -1,5 +1,5 @@
 # flink-job-instance.yaml
-apiVersion: flink.logicamp.tech/v1alpha1
+apiVersion: flink.logicamp.dev/v1alpha1
 kind: FlinkJob
 metadata:
   name: my-flink-job

helm/chart/templates/flink/config.yaml

@@ -27,7 +27,7 @@
     {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
     state.checkpoints.dir: file:///opt/flink/checkpoints/
     {{- else if eq .Values.flink.state.checkpoint.storageType "s3" }}
-    state.checkpoints.dir: s3://flink/checkpoints/
+    state.checkpoints.dir: s3://{{ .Release.Name }}-flink/checkpoints/
     {{- end }}
     {{- if semverCompare ">=2.0.0" .Values.flink.version }}
     state.backend.rocksdb.local-dir: /opt/flink/rocksdb
@@ -38,7 +38,7 @@
     {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
     state.savepoints.dir: file:///opt/flink/savepoints/
     {{- else if eq .Values.flink.state.savepoint.storageType "s3" }}
-    state.savepoints.dir: s3://flink/savepoints/
+    state.savepoints.dir: s3://{{ .Release.Name }}-flink/savepoints/
     {{- end }}
     state.backend.incremental: {{ .Values.flink.state.incremental }}
     rest.profiling.enabled: true

helm/chart/templates/flink/job-manager-deploy.yaml

@@ -24,7 +24,8 @@ spec:
       initContainers:
         - name: volume-mount-hack
           image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
-          runAsUser: 0
+          securityContext:
+            runAsUser: 0
           command: ["sh", "-c", "chown -R flink {{ .Values.flink.state.ha.dir }}"]
           volumeMounts:
             - name: flink-ha
@@ -51,17 +52,17 @@ spec:
                   fieldPath: status.podIP
             {{- if or (eq .Values.flink.state.checkpoint.storageType "s3") (eq .Values.flink.state.savepoint.storageType "s3") }}
             - name: S3_ENDPOINT
-              value: "http://minio-service:9000"
+              value: "http://{{ .Release.Name }}-minio:9000"
             - name: AWS_ACCESS_KEY_ID
               valueFrom:
                 secretKeyRef:
-                  name: {{ .Release.Name }}-flink-secrets
-                  key: minio_access_key
+                  name: {{ .Release.Name }}-minio
+                  key: root-user
             - name: AWS_SECRET_ACCESS_KEY
               valueFrom:
                 secretKeyRef:
-                  name: {{ .Release.Name }}-flink-secrets
-                  key: minio_secret_key
+                  name: {{ .Release.Name }}-minio
+                  key: root-password
             {{- end }}
           volumeMounts:
             - name: flink-ha

helm/chart/templates/flink/task-manager-statefulset.yaml

@@ -33,17 +33,17 @@ spec:
               fieldPath: status.podIP
         {{- if or (eq .Values.flink.state.checkpoint.storageType "s3") (eq .Values.flink.state.savepoint.storageType "s3") }}
         - name: S3_ENDPOINT
-          value: "http://minio-service:9000"
+          value: "http://{{ .Release.Name }}-minio:9000"
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
-              name: {{ .Release.Name }}-flink-secrets
-              key: minio_access_key
+              name: {{ .Release.Name }}-minio
+              key: root-user
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
-              name: {{ .Release.Name }}-flink-secrets
-              key: minio_secret_key
+              name: {{ .Release.Name }}-minio
+              key: root-password
         {{- end }}
         volumeMounts:
         - name: rocksdb-storage

helm/chart/templates/operator/hpa.yaml

@@ -8,8 +8,8 @@ metadata:
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
-    kind: Deployment
-    name: {{ include "flink-kube-operator.fullname" . }}
+    kind: StatefulSet
+    name: {{ .Release.Name }}-flink-operator
   minReplicas: {{ .Values.autoscaling.minReplicas }}
   maxReplicas: {{ .Values.autoscaling.maxReplicas }}
   metrics:

helm/chart/templates/operator/role.yaml

@@ -6,7 +6,7 @@ metadata:
     {{- include "flink-kube-operator.labels" . | nindent 4 }}
 rules:
   - apiGroups:
-      - flink.logicamp.tech  # API group of the FlinkJob CRD
+      - flink.logicamp.dev  # API group of the FlinkJob CRD
     resources:
       - flink-jobs           # The plural name of your custom resource
     verbs:

helm/chart/templates/operator/statefulset.yaml

@@ -44,13 +44,15 @@ spec:
               containerPort: {{ .Values.service.port }}
               protocol: TCP
           env:
+            - name: OPERATOR_ID
+              value: {{ .Values.operatorId | default .Release.Name }}
             - name: FLINK_API_URL
               value: {{ .Release.Name }}-flink-job-manager:8081
             - name: NAMESPACE
               value: "{{ .Release.Namespace }}"
             {{- if eq .Values.flink.state.savepoint.storageType "s3" }}
             - name: SAVEPOINT_PATH
-              value: s3://flink/savepoints/
+              value: s3://{{ .Release.Name }}-flink/savepoints/
             - name: S3_ENDPOINT
               value: "http://{{ .Release.Name }}-minio:9000"
             - name: AWS_ACCESS_KEY_ID

helm/chart/values.yaml

@@ -7,11 +7,11 @@ replicaCount: 1
 
 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
-  repository: lcr.logicamp.tech/library/flink-kube-operator
+  repository: logicampdev/flink-kube-operator
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: latest
+  tag: v1.3.0
 
 # This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets: []
@@ -105,6 +105,11 @@ autoscaling:
 config:
   flinkApiUrl: flink:8081
 
+# Identifier for this operator instance. Used to partition FlinkJob CRs via
+# label selector. Defaults to the Helm release name if left empty.
+# Set explicitly when you need stable names across upgrades.
+operatorId: ""
+
 nodeSelector: {}
 
 tolerations: []
@@ -118,7 +123,7 @@ flink:
   # Override to "1.20" for backward compatibility with Flink 1.20.x.
   version: "2.3"
   image:
-    repository: pcr.pishgaman.top/library/flink-kube-operator/flink
+    repository: logicampdev/flink-kube-operator-flink
     tag: 2.3.0-java21-v1.3.0
 
   parallelism:

internal/crd/v1alpha1/flink_job.go

@@ -12,6 +12,7 @@ import (
 type FlinkJobSpec struct {
 	Key                     string          `json:"key"`
 	Name                    string          `json:"name"`
+	FlinkCluster            string          `json:"flinkCluster"`
 	Parallelism             int             `json:"parallelism"`
 	JarURI                  string          `json:"jarUri"`
 	JarURIBasicAuthUsername *string         `json:"jarURIBasicAuthUsername"`

internal/crd/v1alpha1/register.go

@@ -6,7 +6,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
-const GroupName = "flink.logicamp.tech"
+const GroupName = "flink.logicamp.dev"
 const GroupVersion = "v1alpha1"
 const ResourceName = "flink-jobs"
 

internal/managed_job/manager.go

@@ -3,6 +3,7 @@ package managed_job
 import (
 	"flink-kube-operator/internal/crd"
 	"flink-kube-operator/internal/crd/v1alpha1"
+	"os"
 	"time"
 
 	"flink-kube-operator/pkg"
@@ -73,6 +74,8 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 	}
 	pkg.Logger.Debug("[manager] [cycle] overviews", zap.Any("overviews", jobManagerJobOverviews))
 
+	operatorId := os.Getenv("OPERATOR_ID")
+
 	// Loop over job definitions as Kubernetes CRD
 	for _, uid := range crd.GetAllJobKeys() {
 		if lo.Contains(mgr.processingJobsIds, uid) {
@@ -82,6 +85,27 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 
 		// Get job definition from Kubernetes CRD
 		def := crd.GetJob(uid)
+
+		// Auto-claim: if flinkCluster is empty, set it to this operator's ID
+		if def.Spec.FlinkCluster == "" {
+			pkg.Logger.Info("[manager] auto-claiming unowned job", zap.String("name", def.GetName()))
+			crdInstance.Patch(def.GetUID(), map[string]interface{}{
+				"spec": map[string]interface{}{
+					"flinkCluster": operatorId,
+				},
+			})
+			def.Spec.FlinkCluster = operatorId
+		}
+
+		// Skip CRs that belong to another operator
+		if def.Spec.FlinkCluster != operatorId {
+			pkg.Logger.Debug("[manager] skipping job owned by another operator",
+				zap.String("name", def.GetName()),
+				zap.String("flinkCluster", def.Spec.FlinkCluster),
+				zap.String("operatorId", operatorId),
+			)
+			continue
+		}
 		mgr.processingJobsIds = append(mgr.processingJobsIds, uid)
 
 		// Check if job exists in manager managed jobs