apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ include "flink.fullname" . }}-taskmanager
  labels:
{{ include "flink.labels" . | indent 4 }}
    component: taskmanager
  annotations:
  {{- range $key, $value := .Values.taskmanager.annotations }}
    {{ $key | quote }}: {{ $value | quote }}
  {{- end }}
spec:
  replicas: {{ .Values.taskmanager.replicaCount }}
  podManagementPolicy: {{ .Values.taskmanager.podManagementPolicy }}
  selector:
    matchLabels:
      app.kubernetes.io/name: {{ include "flink.name" . }}
      app.kubernetes.io/instance: {{ .Release.Name }}
      component: taskmanager
  template:
    metadata:
      labels:
        app.kubernetes.io/name: {{ include "flink.name" . }}
        app.kubernetes.io/instance: {{ .Release.Name }}
        component: taskmanager
    spec:
    {{- with .Values.imagePullSecrets }}
      imagePullSecrets:
        {{- toYaml . | nindent 8 }}
    {{- end }}
      serviceAccount: {{ include "taskmanager.serviceAccount" . }}
      containers:
        - name: taskmanager
          image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
          imagePullPolicy: {{ .Values.image.pullPolicy }}
          workingDir: {{ .Values.flink.workDir }}
          command:
            - "/bin/bash"
            - "-c"
            - {{ include "taskmanager.command" . | quote }}
          env:
            - name: JVM_ARGS
              value: {{ tpl .Values.taskmanager.jvmArgs . | quote }}
          {{- if .Values.extraEnvs }}
          {{- toYaml .Values.extraEnvs | nindent 12 }}
          {{- end }}
          {{- if .Values.jobmanager.extraEnvs }}
          {{- toYaml .Values.taskmanager.extraEnvs | nindent 12 }}
          {{- end }}
          ports:
          {{- range $name, $port := .Values.taskmanager.ports }}
            - containerPort: {{ $port }}
              name: {{ $name }}
          {{- end }}
          {{- if .Values.flink.monitoring.enabled }}
            - containerPort: {{ .Values.flink.monitoring.port }}
              name: metrics
          {{- end }}
          livenessProbe:
            tcpSocket:
              port: {{ .Values.taskmanager.ports.rpc }}
            initialDelaySeconds: {{ .Values.taskmanager.livenessProbe.initialDelaySeconds }}
            periodSeconds: {{ .Values.taskmanager.livenessProbe.periodSeconds }}
          volumeMounts:
            - name: flink-config-volume
              mountPath: {{ .Values.flink.workDir }}/conf
          {{- if .Values.taskmanager.persistent.enabled }}
            - name: taskmanager-data
              mountPath: {{ .Values.taskmanager.persistent.mountPath }}
          {{- end }}
          resources:
            {{- toYaml .Values.taskmanager.resources | nindent 12 }}

      volumes:
        - name: flink-config-volume
          configMap:
            name: {{ include "flink.fullname" . }}-config
            items:
              - key: flink-conf-taskmanager.yaml
                path: flink-conf.yaml
              - key: log4j.properties
                path: log4j.properties
              - key: security.properties
                path: security.properties
              {{- if .Values.jobmanager.highAvailability.enabled }}
              - key: masters
                path: masters
              {{- end }}

      {{- with .Values.taskmanager.nodeSelector }}
      nodeSelector:
        {{- toYaml . | nindent 8 }}
      {{- end }}
    {{- with .Values.taskmanager.affinity }}
      affinity:
        {{- toYaml . | nindent 8 }}
    {{- end }}
    {{- with .Values.taskmanager.tolerations }}
      tolerations:
        {{- toYaml . | nindent 8 }}
    {{- end }}
{{- if .Values.taskmanager.persistent.enabled }}
  volumeClaimTemplates:
    - metadata:
        name: taskmanager-data
        annotations:
          volume.alpha.kubernetes.io/storage-class: {{ .Values.taskmanager.persistent.storageClass }}
        labels:
{{ include "flink.labels" . | indent 10 }}
      spec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: {{ .Values.taskmanager.persistent.size }}
{{- end }}