apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ .Release.Name }}-flink-job-manager
  labels:
    app: {{ .Release.Name }}-flink-operator
    component: {{ .Release.Name }}-flink-job-manager
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels:
      app: {{ .Release.Name }}-flink-operator
      component: {{ .Release.Name }}-flink-job-manager
  template:
    metadata:
      labels:
        app: {{ .Release.Name }}-flink-operator
        component: {{ .Release.Name }}-flink-job-manager
    spec:
      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
      initContainers:
        - name: volume-mount-hack
          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
          runAsUser: 0
          command: ["sh", "-c", "chown -R flink {{ .Values.flink.state.ha.dir }}"]
          volumeMounts:
            - name: flink-ha
              mountPath: {{ .Values.flink.state.ha.dir }}
      containers:
        - name: jobmanager
          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
          imagePullPolicy: Always
          args: ["jobmanager"]
          ports:
            - containerPort: 6123  # JobManager RPC port
              name: rpc
            - containerPort: 6124  # JobManager blob server port
              name: blob
            - containerPort: 6125  # JobManager queryable state port
              name: query
            - containerPort: 8081  # JobManager Web UI port
              name: ui
          env:
            {{- include "flink.env" . | nindent 12 }}
            - name: POD_IP
              valueFrom:
                fieldRef:
                  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"
            - name: AWS_ACCESS_KEY_ID
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-flink-secrets
                  key: minio_access_key
            - name: AWS_SECRET_ACCESS_KEY
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-flink-secrets
                  key: minio_secret_key
            {{- end }}
          volumeMounts:
            - name: flink-ha
              mountPath: {{ .Values.flink.state.ha.dir }}
            {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
            - name: flink-checkpoint
              mountPath: /opt/flink/checkpoints
            {{- end }}
            {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
            - name: flink-savepoint
              mountPath: /opt/flink/savepoints
            {{- end }}
      volumes:
        - name: flink-ha
          persistentVolumeClaim:
            claimName: {{ .Release.Name }}-flink-ha-pvc
        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
        - name: flink-checkpoint
          persistentVolumeClaim:
            claimName: {{ .Release.Name }}-flink-checkpoint-pvc
        {{- end }}
        {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
        - name: flink-savepoint
          persistentVolumeClaim:
            claimName: {{ .Release.Name }}-flink-savepoint-pvc
        {{- end }}
      {{- with .Values.nodeSelector }}
      nodeSelector:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      {{- with .Values.affinity }}
      affinity:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      {{- with .Values.tolerations }}
      tolerations:
        {{- toYaml . | nindent 8 }}
      {{- end }}