feat: separate ha and checkpoint pvc

Dockerfile

@@ -20,8 +20,8 @@ COPY . .
 
 # Build
 ENV GOCACHE=/root/.cache/go-build
-RUN --mount=type=cache,target="/go" --mount=type=cache,target="/root/.cache/go-build" CGO_ENABLED=1 GOOS=linux go build -ldflags '-s -w' -o /flink-kube-operator ./cmd/operator
-RUN upx -q -5 /flink-kube-operator
+RUN --mount=type=cache,target="/go" --mount=type=cache,target="/root/.cache/go-build" CGO_ENABLED=1 GOOS=linux go build -ldflags '-s -w' -o /flink-kube-operator ./cmd/operator \
+  && upx -q -9 /flink-kube-operator
 
 FROM public.ecr.aws/docker/library/busybox:1.37.0 AS final
 

helm/chart/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: flink-kube-operator
 description: Helm chart for flink kube operator
 type: application
-version: 0.1.1
+version: 0.1.2
 appVersion: "0.1.0"

helm/chart/templates/flink/data.pvc.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Values.flink.state.data.pvcName }}
+spec:
+  accessModes:
+    - ReadWriteOnce
+  resources:
+    requests:
+      storage: {{ .Values.flink.state.data.size }}  # Use size defined in values.yaml

helm/chart/templates/flink/deploy.yaml

@@ -39,29 +39,33 @@ spec:
                 taskmanager.numberOfTaskSlots: {{ .Values.flink.taskManager.numberOfTaskSlots }}
                 parallelism.default: {{ .Values.flink.parallelism.default }}
                 state.backend: {{ .Values.flink.state.backend }}
-                state.savepoints.dir: {{ .Values.flink.state.savepoints.dir }}
                 rest.port: 8081
                 rootLogger.level = DEBUG
                 rootLogger.appenderRef.console.ref = ConsoleAppender
-                web.upload.dir: /opt/flink/data/web-upload
-                state.checkpoints.dir: file:///tmp/flink-checkpoints
                 high-availability.type: kubernetes
-                high-availability.storageDir: file:///opt/flink/ha
-                kubernetes.cluster-id: cluster-one
                 kubernetes.namespace: {{ .Release.Namespace }}
+                kubernetes.cluster-id: cluster-one
+                web.upload.dir: file://{{ .Values.flink.state.data.dir }}/web-upload
+                state.checkpoints.dir: file://{{ .Values.flink.state.data.dir }}/checkpoints
+                state.backend.rocksdb.localdir: file://{{ .Values.flink.state.data.dir }}/rocksdb
+                high-availability.storageDir: file://{{ .Values.flink.state.data.dir }}
+                state.savepoints.dir: file://{{ .Values.flink.state.savepoints.dir }}
           volumeMounts:
             - name: flink-data
-              mountPath: /opt/flink/data
-              subPath: data
+              mountPath: {{ .Values.flink.state.data.dir }}
+            - name: flink-ha
+              mountPath: {{ .Values.flink.state.ha.dir }}
             - name: flink-savepoints
-              mountPath: /opt/flink/savepoints
-            - name: flink-savepoints
-              mountPath: /opt/flink/ha
-              subPath: ha
+              mountPath: {{ .Values.flink.state.savepoints.dir }}
+
 
       volumes:
         - name: flink-data
-          emptyDir: {}  # Temporary storage for internal data
+          persistentVolumeClaim:
+            claimName: {{ .Values.flink.state.data.pvcName }}  # PVC for savepoints persistence
         - name: flink-savepoints
           persistentVolumeClaim:
             claimName: {{ .Values.flink.state.savepoints.pvcName }}  # PVC for savepoints persistence
+        - name: flink-ha
+          persistentVolumeClaim:
+            claimName: {{ .Values.flink.state.ha.pvcName }}  # PVC for savepoints persistence

helm/chart/templates/flink/savepoint.pvc.yaml

@@ -7,4 +7,4 @@ spec:
     - ReadWriteOnce
   resources:
     requests:
-      storage: {{ .Values.flink.persistence.size }}  # Use size defined in values.yaml
+      storage: {{ .Values.flink.state.savepoints.size }}  # Use size defined in values.yaml

helm/chart/values.yaml

@@ -125,14 +125,17 @@ flink:
   state:
     backend: rocksdb  # Use RocksDB for state backend
     savepoints:
-      dir: "file:///opt/flink/savepoints"  # Directory to store savepoints
+      dir: "/opt/flink/savepoints"  # Directory to store savepoints
       pvcName: flink-savepoints-pvc  # PVC for savepoints persistence
+      size: 10Gi  # PVC size for savepoints storage
+    data:
+      dir: "/opt/flink/data"  # Directory to store checkpoints/web-upload/rocksdb
+      pvcName: flink-data-pvc  # PVC for checkpoints/web-upload/rocksdb
+      size: 10Gi  # PVC size for checkpoints/web-upload/rocksdb
+    ha:
+      dir: "/opt/flink/ha"  # Directory to store ha data
+      pvcName: flink-ha-pvc # PVC for ha
+      size: 10Gi # PVC size for ha
 
   taskManager:
     numberOfTaskSlots: 100  # Number of task slots for TaskManager
-
-  persistence:
-    enabled: true
-    size: 10Gi  # PVC size for savepoints storage
-    accessModes:
-      - ReadWriteOnce

helm/index.yaml

@@ -3,7 +3,17 @@ entries:
   flink-kube-operator:
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-19T00:39:44.4857163+03:30"
+    created: "2024-12-20T17:12:30.023189281+03:30"
+    description: Helm chart for flink kube operator
+    digest: 89345b1a9a79aa18b646705aeb8cfdc547629600cb8a00708a3f64d188f296f2
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.2.tgz
+    version: 0.1.2
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2024-12-20T17:12:30.022644192+03:30"
     description: Helm chart for flink kube operator
     digest: 1d2af9af6b9889cc2962d627946464766f1b65b05629073b7fffb9a98cd957e2
     name: flink-kube-operator
@@ -13,7 +23,7 @@ entries:
     version: 0.1.1
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-19T00:39:44.485286485+03:30"
+    created: "2024-12-20T17:12:30.022187811+03:30"
     description: Helm chart for flink kube operator
     digest: 0890d955904e6a3b2155c086a933b27e45266d896fb69eaad0e811dea40414da
     name: flink-kube-operator
@@ -21,4 +31,4 @@ entries:
     urls:
     - flink-kube-operator-0.1.0.tgz
     version: 0.1.0
-generated: "2024-12-19T00:39:44.48463577+03:30"
+generated: "2024-12-20T17:12:30.021533384+03:30"

internal/managed_job/run.go

@@ -78,7 +78,6 @@ func (job *ManagedJob) run(restoreMode bool) error {
 			shouldUpload = false
 			continue
 		}
-		return nil
 	}
 
 	// job.def.Status.JobId = &runJarResp.JobId