feat: Make Celery workers and queues configurable (#780)

Set to one default worker
PostHog · Jan 16, 2024 · c96fcbb · c96fcbb
1 parent bfa1db1
commit c96fcbb
Show file tree

Hide file tree

Showing 2 changed files with 76 additions and 59 deletions.
diff --git a/charts/posthog/templates/worker-deployment.yaml b/charts/posthog/templates/worker-deployment.yaml
@@ -1,141 +1,155 @@
 {{- if .Values.worker.enabled -}}
+{{- range .Values.worker.consumers }}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: {{ template "posthog.fullname" . }}-worker
-  labels: {{- include "_snippet-metadata-labels-common" . | nindent 4 }}
-  annotations: {{- include "_snippet-metadata-annotations-common" . | nindent 4 }}
+  name: {{ template "posthog.fullname" $ }}-{{ .name }}-worker
+  labels: {{- include "_snippet-metadata-labels-common" $ | nindent 4 }}
+  annotations: {{- include "_snippet-metadata-annotations-common" $ | nindent 4 }}
 spec:
   selector:
     matchLabels:
-        app: {{ template "posthog.fullname" . }}
-        release: "{{ .Release.Name }}"
+        app: {{ template "posthog.fullname" $ }}
+        release: "{{ $.Release.Name }}"
         role: worker
-  {{- if not .Values.worker.hpa.enabled }}
-  replicas: {{ .Values.worker.replicacount }}
+  {{- if not $.Values.worker.hpa.enabled }}
+  replicas: {{ $.Values.worker.replicacount }}
   {{- end }}
 
   strategy:
     type: RollingUpdate
     rollingUpdate:
-      maxSurge: {{ .Values.worker.rollout.maxSurge }}
-      maxUnavailable: {{ .Values.worker.rollout.maxUnavailable }}
+      maxSurge: {{ $.Values.worker.rollout.maxSurge }}
+      maxUnavailable: {{ $.Values.worker.rollout.maxUnavailable }}
 
   template:
     metadata:
       annotations:
-        checksum/secrets.yaml: {{ include (print $.Template.BasePath "/secrets.yaml") . | sha256sum }}
-      {{- if .Values.worker.podAnnotations }}
-{{ toYaml .Values.worker.podAnnotations | indent 8 }}
+        checksum/secrets.yaml: {{ include (print $.Template.BasePath "/secrets.yaml") $ | sha256sum }}
+      {{- if $.Values.worker.podAnnotations }}
+{{ toYaml $.Values.worker.podAnnotations | indent 8 }}
       {{- end }}
       labels:
-        app: {{ template "posthog.fullname" . }}
-        release: "{{ .Release.Name }}"
+        app: {{ template "posthog.fullname" $ }}
+        release: "{{ $.Release.Name }}"
         role: worker
-        {{- if (eq (default .Values.image.tag "none") "latest") }}
+        {{- if (eq (default $.Values.image.tag "none") "latest") }}
         date: "{{ now | unixEpoch }}"
         {{- end }}
-        {{- if .Values.worker.podLabels }}
-{{ toYaml .Values.worker.podLabels | indent 8 }}
+        {{- if $.Values.worker.podLabels }}
+{{ toYaml $.Values.worker.podLabels | indent 8 }}
         {{- end }}
     spec:
-      serviceAccountName: {{ template "posthog.serviceAccountName" . }}
-      {{- include "_snippet-topologySpreadConstraints" . | nindent 6 }}
+      serviceAccountName: {{ template "posthog.serviceAccountName" $ }}
+      {{- include "_snippet-topologySpreadConstraints" $ | nindent 6 }}
 
-      {{- if .Values.worker.affinity }}
+      {{- if $.Values.worker.affinity }}
       affinity:
-{{ toYaml .Values.worker.affinity | indent 8 }}
+{{ toYaml $.Values.worker.affinity | indent 8 }}
       {{- end }}
 
-      {{- if .Values.worker.nodeSelector }}
+      {{- if $.Values.worker.nodeSelector }}
       nodeSelector:
-{{ toYaml .Values.worker.nodeSelector | indent 8 }}
+{{ toYaml $.Values.worker.nodeSelector | indent 8 }}
       {{- end }}
 
-      {{- if .Values.worker.tolerations }}
+      {{- if $.Values.worker.tolerations }}
       tolerations:
-{{ toYaml .Values.worker.tolerations | indent 8 }}
+{{ toYaml $.Values.worker.tolerations | indent 8 }}
       {{- end }}
 
-      {{- if .Values.worker.schedulerName }}
-      schedulerName: "{{ .Values.worker.schedulerName }}"
+      {{- if $.Values.worker.schedulerName }}
+      schedulerName: "{{ $.Values.worker.schedulerName }}"
       {{- end }}
 
-      {{- if .Values.worker.priorityClassName }}
-      priorityClassName: "{{ .Values.worker.priorityClassName }}"
+      {{- if $.Values.worker.priorityClassName }}
+      priorityClassName: "{{ $.Values.worker.priorityClassName }}"
       {{- end }}
 
-      {{- if .Values.image.imagePullSecrets }}
+      {{- if $.Values.image.imagePullSecrets }}
       imagePullSecrets:
-{{ toYaml .Values.image.imagePullSecrets | indent 8 }}
+{{ toYaml $.Values.image.imagePullSecrets | indent 8 }}
       {{- end }}
 
       # I do not know for sure if the old one has been used anywhere, so do both :(
-      {{- if .Values.image.pullSecrets }}
+      {{- if $.Values.image.pullSecrets }}
       imagePullSecrets:
-        {{- range .Values.image.pullSecrets }}
+        {{- range $.Values.image.pullSecrets }}
         - name: {{ . }}
         {{- end }}
       {{- end }}
 
-      {{- if .Values.worker.podSecurityContext.enabled }}
-      securityContext: {{- omit .Values.worker.podSecurityContext "enabled" | toYaml | nindent 8 }}
+      {{- if $.Values.worker.podSecurityContext.enabled }}
+      securityContext: {{- omit $.Values.worker.podSecurityContext "enabled" | toYaml | nindent 8 }}
       {{- end }}
       containers:
-      - name: {{ .Chart.Name }}-workers
-        image: {{ template "posthog.image.fullPath" . }}
-        imagePullPolicy: {{ .Values.image.pullPolicy }}
+      - name: {{ $.Chart.Name }}-workers
+        image: {{ template "posthog.image.fullPath" $ }}
+        imagePullPolicy: {{ $.Values.image.pullPolicy }}
         command:
           - ./bin/docker-worker-celery
           - --with-scheduler
         ports:
-        - containerPort: {{ .Values.service.internalPort }}
+        - containerPort: {{ $.Values.service.internalPort }}
         # Expose the port on which Prometheus /metrics endpoint resides
         - containerPort: 8001
         env:
         # Kafka env variables
-        {{- include "snippet.kafka-env" . | indent 8 }}
+        {{- include "snippet.kafka-env" $ | indent 8 }}
 
         # Object Storage env variables
-        {{- include "snippet.objectstorage-env" . | indent 8 }}
+        {{- include "snippet.objectstorage-env" $ | indent 8 }}
 
         # Redis env variables
-        {{- include "snippet.redis-env" . | indent 8 }}
+        {{- include "snippet.redis-env" $ | indent 8 }}
 
         # statsd env variables
-        {{- include "snippet.statsd-env" . | indent 8 }}
+        {{- include "snippet.statsd-env" $ | indent 8 }}
 
         # prometheus pushgateway if enabled internally
-        {{- if index .Values "prometheus-pushgateway" "enabled" }}
+        {{- if index $.Values "prometheus-pushgateway" "enabled" }}
         - name: PROM_PUSHGATEWAY_ADDRESS
           value: "posthog-prometheus-pushgateway:9091"
         {{- end }}
 
         # prometheus pushgateway if enabled externally
-        {{- if index .Values "external-prometheus-pushgateway" "enabled" }}
+        {{- if index $.Values "external-prometheus-pushgateway" "enabled" }}
         - name: PROM_PUSHGATEWAY_ADDRESS
           value: {{ ".Values.external-prometheus-pushgateway.address" }}
         {{- end }}
 
         - name: PRIMARY_DB
           value: clickhouse
-        {{- include "snippet.posthog-env" . | indent 8 }}
-        {{- include "snippet.posthog-sentry-env" . | indent 8 }}
-        {{- include "snippet.postgresql-env" . | nindent 8 }}
-        {{- include "snippet.clickhouse-env" . | nindent 8 }}
-        {{- include "snippet.email-env" . | nindent 8 }}
-{{- if .Values.env }}
-{{ toYaml .Values.env | indent 8 }}
+        {{- include "snippet.posthog-env" $ | indent 8 }}
+        {{- include "snippet.posthog-sentry-env" $ | indent 8 }}
+        {{- include "snippet.postgresql-env" $ | nindent 8 }}
+        {{- include "snippet.clickhouse-env" $ | nindent 8 }}
+        {{- include "snippet.email-env" $ | nindent 8 }}
+        {{- if .queues }}
+        - name: CELERY_WORKER_QUEUES
+          value: "{{ .queues | join ',' }}"
+        {{- end }}
+        {{- if .concurrency }}
+        - name: CELERY_WORKER_CONCURRENCY
+          value: "{{ .concurrency }}"
+        {{- end }}
+        {{- if .pool }}
+        - name: CELERY_WORKER_POOL
+          value: "{{ .pool }}"
+        {{- end }}
+{{- if $.Values.env }}
+{{ toYaml $.Values.env | indent 8 }}
 {{- end }}
-{{- if .Values.worker.env }}
-{{ toYaml .Values.worker.env | indent 8 }}
+{{- if $.Values.worker.env }}
+{{ toYaml $.Values.worker.env | indent 8 }}
 {{- end }}
         resources:
-{{ toYaml .Values.worker.resources | indent 12 }}
-        {{- if .Values.worker.securityContext.enabled }}
-        securityContext: {{- omit .Values.worker.securityContext "enabled" | toYaml | nindent 12 }}
+{{ toYaml $.Values.worker.resources | indent 12 }}
+        {{- if $.Values.worker.securityContext.enabled }}
+        securityContext: {{- omit $.Values.worker.securityContext "enabled" | toYaml | nindent 12 }}
         {{- end }}
       initContainers:
-      {{- include "_snippet-initContainers-wait-for-service-dependencies" . | indent 8 }}
-      {{- include "_snippet-initContainers-wait-for-migrations" . | indent 8 }}
+      {{- include "_snippet-initContainers-wait-for-service-dependencies" $ | indent 8 }}
+      {{- include "_snippet-initContainers-wait-for-migrations" $ | indent 8 }}
+{{- end }}
 {{- end }}
diff --git a/charts/posthog/values.yaml b/charts/posthog/values.yaml
@@ -301,6 +301,9 @@ worker:
   # -- Count of worker pods to run. This setting is ignored if `worker.hpa.enabled` is set to `true`.
   replicacount: 1
 
+  consumers:
+    - name: default
+
   hpa:
     # -- Whether to create a HorizontalPodAutoscaler for the worker stack.
     enabled: false