################################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ include "flink-operator.name" . }}
  namespace: {{ .Release.Namespace }}
  labels:
    {{- include "flink-operator.labels" . | nindent 4 }}
    {{- if index (.Values.operatorPod) "labels" }}
      {{- with .Values.operatorPod.labels }}
        {{- toYaml . | nindent 4 }}
      {{- end }}
    {{- end }}
spec:
  replicas: {{ .Values.replicas }}
  strategy:
    {{- toYaml .Values.strategy | nindent 4 }}
  selector:
    matchLabels:
      {{- include "flink-operator.selectorLabels" . | nindent 6 }}
  template:
    metadata:
      labels:
        {{- include "flink-operator.selectorLabels" . | nindent 8 }}
        {{- if index (.Values.operatorPod) "labels" }}
          {{- with .Values.operatorPod.labels }}
            {{- toYaml . | nindent 8 }}
          {{- end }}
        {{- end }}
      annotations:
        kubectl.kubernetes.io/default-container: {{ .Chart.Name }}
        {{- if index (.Values.operatorPod) "annotations" }}
          {{- with .Values.operatorPod.annotations }}
            {{- toYaml . | nindent 8 }}
          {{- end }}
        {{- end }}
    spec:
      {{- with .Values.operatorPod.priorityClassName }}
      priorityClassName: {{ . }}
      {{- end }}
      securityContext:
        {{- toYaml .Values.podSecurityContext | nindent 8 }}
      {{- if .Values.operatorPod.nodeSelector }}
      nodeSelector: {{ toYaml .Values.operatorPod.nodeSelector | nindent 8 }}
      {{- end }}
      {{- if .Values.operatorPod.affinity }}
      affinity: {{ toYaml .Values.operatorPod.affinity | nindent 8 }}
      {{- end }}
      {{- with .Values.operatorPod.tolerations }}
      tolerations:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      {{- with .Values.imagePullSecrets }}
      imagePullSecrets:
        {{- toYaml . | nindent 8 }}
      {{- end }}
      serviceAccountName: {{ include "flink-operator.serviceAccountName" . }}
      {{- if .Values.operatorPod.topologySpreadConstraints }}
      topologySpreadConstraints: {{ toYaml .Values.operatorPod.topologySpreadConstraints | nindent 8 }}
      {{- end }}
      {{- if .Values.operatorPod.initContainers }}
      initContainers:
      {{- toYaml .Values.operatorPod.initContainers | nindent 8 }}
      {{- end }}
      containers:
        - name: {{ .Chart.Name }}
          image: {{ include "flink-operator.imagePath" . }}
          imagePullPolicy: {{ .Values.image.pullPolicy }}
          command: ["/docker-entrypoint.sh", "operator"]
          ports:
          {{- if .Values.metrics.port }}
            - containerPort: {{ .Values.metrics.port }}
              name: metrics
              protocol: TCP
          {{- end }}
          {{- if index .Values "operatorHealth" }}
            - containerPort: {{ .Values.operatorHealth.port }}
              name: health-port
              protocol: TCP
          {{- end }}
          env:
            - name: OPERATOR_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            - name: HOST_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.hostIP
            - name: POD_IP
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
            - name: POD_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
            - name: OPERATOR_NAME
              value: {{ include "flink-operator.name" . }}
            - name: FLINK_CONF_DIR
              value: /opt/flink/conf
            - name: FLINK_PLUGINS_DIR
              value: /opt/flink/plugins
            - name: LOG_CONFIG
              value: {{ .Values.jvmArgs.logConfig }}
            - name: JVM_ARGS
              value: {{ .Values.jvmArgs.operator }}
            {{- if .Values.tls.create }}
            - name: OPERATOR_KEYSTORE_PATH
              value: /opt/flink/tls-cert/keystore.jks
            - name: OPERATOR_TRUSTSTORE_PATH
              value: /opt/flink/tls-cert/truststore.jks
            - name: OPERATOR_KEYSTORE_PASSWORD
              valueFrom:
                secretKeyRef:
                  {{- toYaml .Values.tls.secretKeyRef | nindent 18 }}
                  optional: true
            {{- end }}
            {{- with .Values.operatorPod.env }}
            {{- toYaml . | nindent 12 }}
            {{- end }}
          {{- with .Values.operatorPod.envFrom }}
          envFrom:
            {{- toYaml . | nindent 12 }}
            {{- end }}
          resources:
            {{- toYaml .Values.operatorPod.resources | nindent 12 }}
          securityContext:
            {{- toYaml .Values.operatorSecurityContext | nindent 12 }}
          volumeMounts:
            - name: flink-operator-config-volume
              mountPath: /opt/flink/conf
            {{- if .Values.operatorVolumeMounts.create }}
                {{- toYaml .Values.operatorVolumeMounts.data | nindent 12 }}
            {{- else }}
            - name: flink-artifacts-volume
              mountPath: /opt/flink/artifacts
            {{- end }}
          {{- if .Values.tls.create }}
            - name: flink-operator-cert-secret
              mountPath: /opt/flink/tls-cert
          {{- end }}
          {{- if and (index .Values "operatorHealth") (index .Values.operatorHealth "livenessProbe") }}
          livenessProbe:
            {{- toYaml .Values.operatorHealth.livenessProbe | nindent 12 }}
            httpGet:
              path: /
              port: health-port
          {{- end }}
          {{- if and (index .Values "operatorHealth") (index .Values.operatorHealth "startupProbe") }}
          startupProbe:
            {{- toYaml .Values.operatorHealth.startupProbe | nindent 12 }}
            httpGet:
              path: /
              port: health-port
          {{- end }}
          {{- if .Values.postStart }}
          lifecycle:
            postStart:
              {{- toYaml .Values.postStart | nindent 14 }}
          {{- end }}
        {{- if eq (include "flink-operator.webhook-enabled" .) "true" }}
        - name: flink-webhook
          image: {{ include "flink-operator.imagePath" . }}
          imagePullPolicy: {{ .Values.image.pullPolicy }}
          command: ["/docker-entrypoint.sh", "webhook"]
          env:
            - name: WEBHOOK_KEYSTORE_PASSWORD
              valueFrom:
                secretKeyRef:
                {{- if .Values.webhook.keystore.useDefaultPassword }}
                  name: flink-operator-webhook-secret
                  key: password
                {{- else }}
                  {{- with .Values.webhook.keystore.passwordSecretRef }}
                    {{- toYaml . | nindent 18 }}
                  {{- end }}
                {{- end }}
            - name: WEBHOOK_KEYSTORE_FILE
              value: "/certs/keystore.p12"
            - name: WEBHOOK_KEYSTORE_TYPE
              value: "pkcs12"
            - name: WEBHOOK_SERVER_PORT
              value: "9443"
            - name: LOG_CONFIG
              value: {{ .Values.jvmArgs.logConfig }}
            - name: JVM_ARGS
              value: {{ .Values.jvmArgs.webhook }}
            - name: FLINK_CONF_DIR
              value: /opt/flink/conf
            - name: FLINK_PLUGINS_DIR
              value: /opt/flink/plugins
            - name: OPERATOR_NAMESPACE
              valueFrom:
                fieldRef:
                  fieldPath: metadata.namespace
            {{- with .Values.operatorPod.webhook.container.env }}
            {{- toYaml . | nindent 12 }}
            {{- end }}
          resources:
            {{- toYaml .Values.operatorPod.webhook.resources | nindent 12 }}
          securityContext:
            {{- toYaml .Values.webhookSecurityContext | nindent 12 }}
          volumeMounts:
          - name: keystore
            mountPath: "/certs"
            readOnly: true
          - name: flink-operator-config-volume
            mountPath: /opt/flink/conf
        {{- end }}
        {{- if .Values.operatorPod.sidecarContainers }}
        {{- toYaml .Values.operatorPod.sidecarContainers | nindent 8 }}
        {{- end }}
      {{- if index (.Values.operatorPod) "dnsPolicy" }}
      dnsPolicy: {{ .Values.operatorPod.dnsPolicy | quote }}
      {{- end }}
      {{- if index (.Values.operatorPod) "dnsConfig" }}
      dnsConfig:
        {{- with .Values.operatorPod.dnsConfig }}
          {{- toYaml . | nindent 8 }}
        {{- end }}
      {{- end }}
      volumes:
        - name: flink-operator-config-volume
          configMap:
            name: flink-operator-config
            items:
            {{- if hasKey .Values.defaultConfiguration "flink-conf.yaml" }}
              - key: flink-conf.yaml
                path: flink-conf.yaml
            {{- else }}
              - key: config.yaml
                path: config.yaml
            {{- end }}
              - key: log4j-operator.properties
                path: log4j-operator.properties
              - key: log4j-console.properties
                path: log4j-console.properties
        {{- if .Values.operatorVolumes.create }}
              {{- toYaml .Values.operatorVolumes.data | nindent 8 }}
        {{- else }}
        - name: flink-artifacts-volume
          emptyDir: {}
        {{- end }}
        {{- if eq (include "flink-operator.webhook-enabled" .) "true" }}
        - name: keystore
          secret:
            secretName: webhook-server-cert
            items:
            - key: keystore.p12
              path: keystore.p12
        {{- end }}
        {{- if .Values.tls.create }}
        - name: flink-operator-cert-secret
          secret:
            secretName: {{ .Values.tls.secretName }}
            optional: true
        {{- end }}
---
{{- if .Values.defaultConfiguration.create }}
apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-operator-config
  namespace: {{ .Release.Namespace }}
  labels:
    {{- include "flink-operator.labels" . | nindent 4 }}
data:
  config.yaml: |+
{{- if .Values.defaultConfiguration.append }}
  {{- $.Files.Get "conf/flink-conf.yaml"  | nindent 4 -}}
{{- end }}
{{- if hasKey (.Values.defaultConfiguration) "config.yaml" }}
  {{- index (.Values.defaultConfiguration) "config.yaml" | nindent 4 -}}
{{- end }}
{{- if .Values.watchNamespaces }}
    kubernetes.operator.watched.namespaces: {{ join "," .Values.watchNamespaces  }}
{{- end }}
{{- if index .Values "operatorHealth" }}
    kubernetes.operator.health.probe.enabled: true
    kubernetes.operator.health.probe.port: {{ .Values.operatorHealth.port }}
{{- end }}
  flink-conf.yaml: |+
{{- if .Values.defaultConfiguration.append }}
  {{- $.Files.Get "conf/flink-conf.yaml"  | nindent 4 -}}
{{- end }}
{{- if hasKey (.Values.defaultConfiguration) "flink-conf.yaml" }}
  {{- index (.Values.defaultConfiguration) "flink-conf.yaml" | nindent 4 -}}
{{- end }}
{{- if .Values.watchNamespaces }}
    kubernetes.operator.watched.namespaces: {{ join "," .Values.watchNamespaces  }}
{{- end }}
{{- if index .Values "operatorHealth" }}
    kubernetes.operator.health.probe.enabled: true
    kubernetes.operator.health.probe.port: {{ .Values.operatorHealth.port }}
{{- end }}
  log4j-operator.properties: |+
{{- if .Values.defaultConfiguration.append }}
  {{- $.Files.Get "conf/log4j-operator.properties"  | nindent 4 -}}
{{- end }}
{{- if index (.Values.defaultConfiguration) "log4j-operator.properties" }}
  {{- index (.Values.defaultConfiguration) "log4j-operator.properties" | nindent 4 -}}
{{- end }}
  log4j-console.properties: |+
{{- if .Values.defaultConfiguration.append }}
  {{- $.Files.Get "conf/log4j-console.properties"  | nindent 4 -}}
{{- end }}
{{- if index (.Values.defaultConfiguration) "log4j-console.properties" }}
  {{- index (.Values.defaultConfiguration) "log4j-console.properties" | nindent 4 -}}
{{- end }}
{{- end }}
