From 7fb18cd45fc07f351ac3219aef7af3193469c340 Mon Sep 17 00:00:00 2001 From: Mohammadreza Khani Date: Thu, 23 Jul 2026 13:51:33 +0330 Subject: [PATCH] feat: support multiple operator in single namespace --- crds.yaml | 7 ++++-- example-job.yaml | 2 +- helm/chart/templates/flink/config.yaml | 4 ++-- .../templates/flink/job-manager-deploy.yaml | 13 +++++----- .../flink/task-manager-statefulset.yaml | 10 ++++---- helm/chart/templates/operator/hpa.yaml | 4 ++-- helm/chart/templates/operator/role.yaml | 2 +- .../chart/templates/operator/statefulset.yaml | 4 +++- helm/chart/values.yaml | 11 ++++++--- internal/crd/v1alpha1/flink_job.go | 1 + internal/crd/v1alpha1/register.go | 2 +- internal/managed_job/manager.go | 24 +++++++++++++++++++ 12 files changed, 60 insertions(+), 24 deletions(-) diff --git a/crds.yaml b/crds.yaml index 5de3b21..7cb09eb 100644 --- a/crds.yaml +++ b/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: diff --git a/example-job.yaml b/example-job.yaml index a1b3068..8cc8e5e 100644 --- a/example-job.yaml +++ b/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 diff --git a/helm/chart/templates/flink/config.yaml b/helm/chart/templates/flink/config.yaml index cdec79e..2ebed22 100644 --- a/helm/chart/templates/flink/config.yaml +++ b/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 diff --git a/helm/chart/templates/flink/job-manager-deploy.yaml b/helm/chart/templates/flink/job-manager-deploy.yaml index 5fcb51d..19aa367 100644 --- a/helm/chart/templates/flink/job-manager-deploy.yaml +++ b/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 diff --git a/helm/chart/templates/flink/task-manager-statefulset.yaml b/helm/chart/templates/flink/task-manager-statefulset.yaml index cad2862..203e611 100644 --- a/helm/chart/templates/flink/task-manager-statefulset.yaml +++ b/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 diff --git a/helm/chart/templates/operator/hpa.yaml b/helm/chart/templates/operator/hpa.yaml index 3a7bba4..bbdbc28 100644 --- a/helm/chart/templates/operator/hpa.yaml +++ b/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: diff --git a/helm/chart/templates/operator/role.yaml b/helm/chart/templates/operator/role.yaml index 2f4ec2b..f291664 100644 --- a/helm/chart/templates/operator/role.yaml +++ b/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: diff --git a/helm/chart/templates/operator/statefulset.yaml b/helm/chart/templates/operator/statefulset.yaml index 9ea5a66..af2eedc 100644 --- a/helm/chart/templates/operator/statefulset.yaml +++ b/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 diff --git a/helm/chart/values.yaml b/helm/chart/values.yaml index 5fadded..f8acb32 100644 --- a/helm/chart/values.yaml +++ b/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: diff --git a/internal/crd/v1alpha1/flink_job.go b/internal/crd/v1alpha1/flink_job.go index e0dfe8a..53e8af6 100644 --- a/internal/crd/v1alpha1/flink_job.go +++ b/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"` diff --git a/internal/crd/v1alpha1/register.go b/internal/crd/v1alpha1/register.go index c90c3ed..fc42ddc 100644 --- a/internal/crd/v1alpha1/register.go +++ b/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" diff --git a/internal/managed_job/manager.go b/internal/managed_job/manager.go index 6235778..3748c5c 100644 --- a/internal/managed_job/manager.go +++ b/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