charts/pulsar/templates/broker-statefulset.yaml

#
# 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.
#

{{- if .Values.components.broker }}
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: "{{ template "pulsar.fullname" . }}-{{ .Values.broker.component }}"
  namespace: {{ template "pulsar.namespace" . }}
  labels:
    {{- include "pulsar.standardLabels" . | nindent 4 }}
    component: {{ .Values.broker.component }}
spec:
  serviceName: "{{ template "pulsar.fullname" . }}-{{ .Values.broker.component }}"
  {{- if not .Values.broker.autoscaling.enabled }}
  replicas: {{ .Values.broker.replicaCount }}
  {{- end }}
  selector:
    matchLabels:
      {{- include "pulsar.matchLabels" . | nindent 6 }}
      component: {{ .Values.broker.component }}
  updateStrategy:
    type: RollingUpdate
  podManagementPolicy: Parallel
  template:
    metadata:
      labels:
        {{- include "pulsar.template.labels" . | nindent 8 }}
        component: {{ .Values.broker.component }}
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "{{ .Values.broker.ports.http }}"
        {{- if .Values.broker.restartPodsOnConfigMapChange }}
        checksum/config: {{ include (print $.Template.BasePath "/broker-configmap.yaml") . | sha256sum }}
        {{- end }}
{{- with .Values.broker.annotations }}
{{ toYaml . | indent 8 }}
{{- end }}
    spec:
      serviceAccountName: "{{ template "pulsar.fullname" . }}-{{ .Values.broker.component }}-acct"
    {{- if .Values.broker.nodeSelector }}
      nodeSelector:
{{ toYaml .Values.broker.nodeSelector | indent 8 }}
    {{- end }}
    {{- if .Values.broker.tolerations }}
      tolerations:
{{ toYaml .Values.broker.tolerations | indent 8 }}
    {{- end }}
      affinity:
        {{- if and .Values.affinity.anti_affinity .Values.broker.affinity.anti_affinity}}
        podAntiAffinity:
          {{ if eq .Values.broker.affinity.type "requiredDuringSchedulingIgnoredDuringExecution"}}
          {{ .Values.broker.affinity.type }}:
          - labelSelector:
              matchExpressions:
              - key: "app"
                operator: In
                values:
                - "{{ template "pulsar.name" . }}"
              - key: "release"
                operator: In
                values:
                - {{ .Release.Name }}
              - key: "component"
                operator: In
                values:
                - {{ .Values.broker.component }}
            topologyKey: {{ .Values.broker.affinity.anti_affinity_topology_key }}
        {{ else }}
          {{ .Values.broker.affinity.type }}:
            - weight: 100
              podAffinityTerm:
                labelSelector:
                  matchExpressions:
                    - key: "app"
                      operator: In
                      values:
                      - "{{ template "pulsar.name" . }}"
                    - key: "release"
                      operator: In
                      values:
                      - {{ .Release.Name }}
                    - key: "component"
                      operator: In
                      values:
                      - {{ .Values.broker.component }}
                topologyKey: {{ .Values.broker.affinity.anti_affinity_topology_key }}
        {{ end }}
        {{- end }}
      terminationGracePeriodSeconds: {{ .Values.broker.gracePeriod }}
      initContainers:
      # This init container will wait for zookeeper to be ready before
      # deploying the bookies
      - name: wait-zookeeper-ready
        image: "{{ template "pulsar.imageFullName" (dict "image" .Values.images.broker "root" .) }}"
        imagePullPolicy: {{ .Values.images.broker.pullPolicy }}
        command: ["sh", "-c"]
        args:
          - >-
            {{- include "pulsar.broker.zookeeper.tls.settings" . | nindent 12 }}
            {{- if .Values.pulsar_metadata.configurationStore }}
            until bin/bookkeeper org.apache.zookeeper.ZooKeeperMain -server {{ template "pulsar.configurationStore.connect" . }} get {{ .Values.configurationStoreMetadataPrefix }}/admin/clusters/{{ template "pulsar.cluster.name" . }}; do
            {{- end }}
            {{- if not .Values.pulsar_metadata.configurationStore }}
            until bin/bookkeeper org.apache.zookeeper.ZooKeeperMain -server {{ template "pulsar.zookeeper.connect" . }} get {{ .Values.metadataPrefix }}/admin/clusters/{{ template "pulsar.cluster.name" . }}; do
            {{- end }}
              echo "pulsar cluster {{ template "pulsar.cluster.name" . }} isn't initialized yet ... check in 3 seconds ..." && sleep 3;
            done;
      {{- if and .Values.rbac.enabled .Values.rbac.psp }}
        securityContext:
          readOnlyRootFilesystem: false
      {{- end }}
        volumeMounts:
        {{- include "pulsar.broker.certs.volumeMounts" . | nindent 8 }}
      # This init container will wait for bookkeeper to be ready before
      # deploying the broker
      - name: wait-bookkeeper-ready
        image: "{{ template "pulsar.imageFullName" (dict "image" .Values.images.broker "root" .) }}"
        imagePullPolicy: {{ .Values.images.broker.pullPolicy }}
        command: ["sh", "-c"]
        args:
          - >
            {{- include "pulsar.broker.zookeeper.tls.settings" . | nindent 12 }}
            bin/apply-config-from-env.py conf/bookkeeper.conf;
            until bin/bookkeeper shell whatisinstanceid; do
              echo "bookkeeper cluster is not initialized yet. backoff for 3 seconds ...";
              sleep 3;
            done;
            echo "bookkeeper cluster is already initialized";
            bookieServiceNumber="$(nslookup -timeout=10 {{ template "pulsar.fullname" . }}-{{ .Values.bookkeeper.component }} | grep Name | wc -l)";
            until [ ${bookieServiceNumber} -ge {{ .Values.broker.configData.managedLedgerDefaultEnsembleSize }} ]; do
              echo "bookkeeper cluster {{ template "pulsar.cluster.name" . }} isn't ready yet ... check in 10 seconds ...";
              sleep 10;
              bookieServiceNumber="$(nslookup -timeout=10 {{ template "pulsar.fullname" . }}-{{ .Values.bookkeeper.component }} | grep Name | wc -l)";
            done;
            echo "bookkeeper cluster is ready";
        envFrom:
          - configMapRef:
              name: "{{ template "pulsar.fullname" . }}-{{ .Values.bookkeeper.component }}"
      {{- if and .Values.rbac.enabled .Values.rbac.psp }}
        securityContext:
          readOnlyRootFilesystem: false
      {{- end }}
        volumeMounts:
        {{- include "pulsar.broker.certs.volumeMounts" . | nindent 10 }}
      containers:
      - name: "{{ template "pulsar.fullname" . }}-{{ .Values.broker.component }}"
        image: "{{ template "pulsar.imageFullName" (dict "image" .Values.images.broker "root" .) }}"
        imagePullPolicy: {{ .Values.images.broker.pullPolicy }}
        {{- if .Values.broker.probe.liveness.enabled }}
        livenessProbe:
          httpGet:
            path: /status.html
            port: {{ .Values.broker.ports.http }}
          initialDelaySeconds: {{ .Values.broker.probe.liveness.initialDelaySeconds }}
          periodSeconds: {{ .Values.broker.probe.liveness.periodSeconds }}
          timeoutSeconds: {{ .Values.broker.probe.liveness.timeoutSeconds }}
          failureThreshold: {{ .Values.broker.probe.liveness.failureThreshold }}
        {{- end }}
        {{- if .Values.broker.probe.readiness.enabled }}
        readinessProbe:
          httpGet:
            path: /status.html
            port: {{ .Values.broker.ports.http }}
          initialDelaySeconds: {{ .Values.broker.probe.readiness.initialDelaySeconds }}
          periodSeconds: {{ .Values.broker.probe.readiness.periodSeconds }}
          timeoutSeconds: {{ .Values.broker.probe.readiness.timeoutSeconds }}
          failureThreshold: {{ .Values.broker.probe.readiness.failureThreshold }}
        {{- end }}
        {{- if .Values.broker.probe.startup.enabled }}
        startupProbe:
          httpGet:
            path: /status.html
            port: {{ .Values.broker.ports.http }}
          initialDelaySeconds: {{ .Values.broker.probe.startup.initialDelaySeconds }}
          periodSeconds: {{ .Values.broker.probe.startup.periodSeconds }}
          timeoutSeconds: {{ .Values.broker.probe.startup.timeoutSeconds }}
          failureThreshold: {{ .Values.broker.probe.startup.failureThreshold }}
        {{- end }}
      {{- if .Values.broker.resources }}
        resources:
{{ toYaml .Values.broker.resources | indent 10 }}
      {{- end }}
        command: ["sh", "-c"]
        args:
        - >
        {{- if .Values.broker.additionalCommand }}
          {{ .Values.broker.additionalCommand }}
        {{- end }}
          bin/apply-config-from-env.py conf/broker.conf;
          bin/gen-yml-from-env.py conf/functions_worker.yml;
          echo "OK" > status;
          {{- include "pulsar.broker.zookeeper.tls.settings" . | nindent 10 }}
          bin/pulsar zookeeper-shell -server {{ template "pulsar.zookeeper.connect" . }} get {{ template "pulsar.broker.znode" . }};
          while [ $? -eq 0 ]; do
            echo "broker {{ template "pulsar.broker.hostname" . }} znode still exists ... check in 10 seconds ...";
            sleep 10;
            bin/pulsar zookeeper-shell -server {{ template "pulsar.zookeeper.connect" . }} get {{ template "pulsar.broker.znode" . }};
          done;
          cat conf/pulsar_env.sh;
          OPTS="${OPTS} -Dlog4j2.formatMsgNoLookups=true" exec bin/pulsar broker;
        ports:
        # prometheus needs to access /metrics endpoint
        - name: http
          containerPort: {{ .Values.broker.ports.http }}
        {{- if or (not .Values.tls.enabled) (not .Values.tls.broker.enabled) }}
        - name: "{{ .Values.tcpPrefix }}pulsar"
          containerPort: {{ .Values.broker.ports.pulsar }}
        {{- end }}
        {{- if and .Values.tls.enabled .Values.tls.broker.enabled }}
        - name: https
          containerPort: {{ .Values.broker.ports.https }}
        - name: "{{ .Values.tlsPrefix }}pulsarssl"
          containerPort: {{ .Values.broker.ports.pulsarssl }}
        {{- end }}
{{- if .Values.broker.extreEnvs }}
        env:
{{ toYaml .Values.broker.extreEnvs | indent 8 }}
{{- end }}
        envFrom:
        - configMapRef:
            name: "{{ template "pulsar.fullname" . }}-{{ .Values.broker.component }}"
        volumeMounts:
          {{- if .Values.auth.authentication.enabled }}
          {{- if eq .Values.auth.authentication.provider "jwt" }}
          - mountPath: "/pulsar/keys"
            name: token-keys
            readOnly: true
          - mountPath: "/pulsar/tokens"
            name: broker-token
            readOnly: true
          {{- end }}
          {{- end }}
          {{- if .Values.broker.extraVolumeMounts }}
{{ toYaml .Values.broker.extraVolumeMounts | indent 10 }}
          {{- end }}
          {{- include "pulsar.broker.certs.volumeMounts" . | nindent 10 }}
      {{- if and .Values.rbac.enabled .Values.rbac.psp }}
        securityContext:
          readOnlyRootFilesystem: false
      {{- end }}
      volumes:
      {{- if .Values.broker.extraVolumes }}
{{ toYaml .Values.broker.extraVolumes | indent 6 }}
      {{- end }}
      {{- if .Values.auth.authentication.enabled }}
      {{- if eq .Values.auth.authentication.provider "jwt" }}
      - name: token-keys
        secret:
          {{- if not .Values.auth.authentication.jwt.usingSecretKey }}
          secretName: "{{ .Release.Name }}-token-asymmetric-key"
          {{- end}}
          {{- if .Values.auth.authentication.jwt.usingSecretKey }}
          secretName: "{{ .Release.Name }}-token-symmetric-key"
          {{- end}}
          items:
            {{- if .Values.auth.authentication.jwt.usingSecretKey }}
            - key: SECRETKEY
              path: token/secret.key
            {{- else }}
            - key: PUBLICKEY
              path: token/public.key
            {{- end}}
      - name: broker-token
        secret:
          secretName: "{{ .Release.Name }}-token-{{ .Values.auth.superUsers.broker }}"
          items:
            - key: TOKEN
              path: broker/token
      {{- end}}
      {{- end}}
      {{- include "pulsar.broker.certs.volumes" . | nindent 6 }}
      {{- include "pulsar.imagePullSecrets" . | nindent 6}}
{{- end }}