feat(experimentation): write ingestion events straight to ClickHouse without Kafka

charts/flagsmith/templates/experimentation/_helpers.tpl

@@ -51,8 +51,37 @@ Usage: (dict "root" . "username" <string> "passwordRef" <secretKeyRef body>)
 {{- end }}
 {{- end -}}
 
+{{/*
+True when events go through Kafka and warehouse delivery, rather than straight to ClickHouse.
+*/}}
+{{- define "flagsmith.experimentation.usesKafka" -}}
+{{- if and .Values.experimentation.enabled (eq .Values.experimentation.sink "kafka") -}}
+true
+{{- end -}}
+{{- end -}}
+
+{{- define "flagsmith.experimentation.clickhousePasswordSecretRef" -}}
+{{- with .Values.experimentation.clickhouse.passwordFromExistingSecret -}}
+{{- if .enabled -}}
+name: {{ required "experimentation.clickhouse.passwordFromExistingSecret.name is required" .name }}
+key: {{ required "experimentation.clickhouse.passwordFromExistingSecret.key is required" .key }}
+{{- else -}}
+name: {{ include "flagsmith.experimentation.secretName" $ }}
+key: CLICKHOUSE_PASSWORD
+{{- end -}}
+{{- end -}}
+{{- end -}}
+
 {{- define "flagsmith.experimentation.validate" -}}
 {{- $e := .Values.experimentation -}}
+{{- if not (has $e.sink (list "kafka" "clickhouse")) -}}
+{{- fail "experimentation.sink must be kafka or clickhouse" -}}
+{{- end -}}
+{{- if eq $e.sink "clickhouse" -}}
+{{- if not $e.clickhouse.url -}}
+{{- fail "experimentation.clickhouse.url is required when experimentation.sink is clickhouse" -}}
+{{- end -}}
+{{- else -}}
 {{- if not $e.kafka.bootstrapServers -}}
 {{- fail "experimentation.kafka.bootstrapServers is required" -}}
 {{- end -}}
@@ -68,6 +97,26 @@ Usage: (dict "root" . "username" <string> "passwordRef" <secretKeyRef body>)
 {{- end -}}
 {{- end -}}
 {{- end -}}
+{{- end -}}
+
+{{/*
+ClickHouse env for the ingestion service when it writes straight to ClickHouse.
+*/}}
+{{- define "flagsmith.experimentation.clickhouseEnv" -}}
+{{- $clickhouse := .Values.experimentation.clickhouse -}}
+- name: CLICKHOUSE_URL
+  value: {{ $clickhouse.url | quote }}
+- name: CLICKHOUSE_USER
+  value: {{ $clickhouse.user | quote }}
+- name: CLICKHOUSE_DATABASE
+  value: {{ $clickhouse.database | quote }}
+{{- if or $clickhouse.password $clickhouse.passwordFromExistingSecret.enabled }}
+- name: CLICKHOUSE_PASSWORD
+  valueFrom:
+    secretKeyRef:
+      {{- include "flagsmith.experimentation.clickhousePasswordSecretRef" . | nindent 6 }}
+{{- end }}
+{{- end -}}
 
 {{/*
 The secret the API encrypts warehouse connection credentials with, and warehouse delivery decrypts them with.

charts/flagsmith/templates/experimentation/ingestion-deployment.yaml

@@ -59,10 +59,16 @@ spec:
             - name: http
               containerPort: 8080
           env:
+            - name: SINK
+              value: {{ .Values.experimentation.sink | quote }}
+            {{- if include "flagsmith.experimentation.usesKafka" . }}
             {{- include "flagsmith.experimentation.kafkaEnv" (dict "root" . "username" .Values.experimentation.kafka.username "passwordRef" (include "flagsmith.experimentation.kafkaPasswordSecretRef" .)) | nindent 12 }}
             # One topic for every environment: warehouse delivery routes each to its warehouse.
             - name: KAFKA_EXTERNAL_TOPIC
               value: {{ .Values.experimentation.kafka.topic | quote }}
+            {{- else }}
+            {{- include "flagsmith.experimentation.clickhouseEnv" . | nindent 12 }}
+            {{- end }}
             - name: DATABASE_URL
               valueFrom:
                 secretKeyRef:

charts/flagsmith/templates/experimentation/init-configmap.yaml

@@ -1,4 +1,4 @@
-{{- if .Values.experimentation.enabled }}
+{{- if include "flagsmith.experimentation.usesKafka" . }}
 apiVersion: v1
 kind: ConfigMap
 metadata:

charts/flagsmith/templates/experimentation/init-job.yaml

@@ -1,4 +1,4 @@
-{{- if .Values.experimentation.enabled }}
+{{- if include "flagsmith.experimentation.usesKafka" . }}
 {{- $job := .Values.jobs.experimentationInit }}
 {{- $kafka := .Values.experimentation.kafka }}
 {{- $adminPasswordRef := include "flagsmith.experimentation.kafkaPasswordSecretRef" . }}

charts/flagsmith/templates/experimentation/secret.yaml

@@ -2,7 +2,11 @@
 {{- include "flagsmith.experimentation.validate" . }}
 {{- $e := .Values.experimentation }}
 {{- $data := dict }}
-{{- if and (eq $e.kafka.auth "scram") (not $e.kafka.passwordFromExistingSecret.enabled) }}
+{{- if eq $e.sink "clickhouse" }}
+{{- if and $e.clickhouse.password (not $e.clickhouse.passwordFromExistingSecret.enabled) }}
+{{- $_ := set $data "CLICKHOUSE_PASSWORD" $e.clickhouse.password }}
+{{- end }}
+{{- else if and (eq $e.kafka.auth "scram") (not $e.kafka.passwordFromExistingSecret.enabled) }}
 {{- $_ := set $data "KAFKA_PASSWORD" $e.kafka.password }}
 {{- end }}
 {{- with $data }}

charts/flagsmith/templates/experimentation/warehouse-delivery-deployment.yaml

@@ -1,4 +1,4 @@
-{{- if .Values.experimentation.enabled }}
+{{- if include "flagsmith.experimentation.usesKafka" . }}
 {{- $delivery := .Values.experimentation.warehouseDelivery }}
 apiVersion: apps/v1
 kind: Deployment

charts/flagsmith/templates/experimentation/warehouse-delivery-pdb.yaml

@@ -1,4 +1,4 @@
-{{- if and .Values.experimentation.enabled .Values.pdb.warehouseDelivery.enabled }}
+{{- if and (include "flagsmith.experimentation.usesKafka" .) .Values.pdb.warehouseDelivery.enabled }}
 apiVersion: policy/v1
 kind: PodDisruptionBudget
 metadata:

charts/flagsmith/unittests/experimentation/ingestion-deployment_test.yaml

@@ -16,6 +16,9 @@ tests:
   - it: sends every event to the one topic, and reads keys from the API database
     template: experimentation/ingestion-deployment.yaml
     asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content: {name: SINK, value: kafka}
       - contains:
           path: spec.template.spec.containers[0].env
           content: {name: KAFKA_EXTERNAL_TOPIC, value: flagsmith_warehouse_events}
@@ -42,3 +45,38 @@ tests:
           content:
             name: DATABASE_URL
             valueFrom: {secretKeyRef: {name: ingestion-db, key: url}}
+
+  - it: writes events straight to ClickHouse with the clickhouse sink
+    template: experimentation/ingestion-deployment.yaml
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse: {url: https://clickhouse.example.com:8443, password: clickhouse-secret}
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content: {name: SINK, value: clickhouse}
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content: {name: CLICKHOUSE_URL, value: https://clickhouse.example.com:8443}
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: CLICKHOUSE_PASSWORD
+            valueFrom: {secretKeyRef: {name: flagsmith-experimentation, key: CLICKHOUSE_PASSWORD}}
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content: {name: KAFKA_BOOTSTRAP_SERVERS, value: kafka-1:9096,kafka-2:9096}
+
+  - it: reads the ClickHouse password from an existing secret
+    template: experimentation/ingestion-deployment.yaml
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse:
+        url: https://clickhouse.example.com:8443
+        passwordFromExistingSecret: {enabled: true, name: clickhouse, key: password}
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: CLICKHOUSE_PASSWORD
+            valueFrom: {secretKeyRef: {name: clickhouse, key: password}}

charts/flagsmith/unittests/experimentation/init-configmap_test.yaml

@@ -17,3 +17,11 @@ tests:
       - notMatchRegex:
           path: data["init.sh"]
           pattern: clickhouse
+
+  - it: renders nothing with the clickhouse sink
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse.url: https://clickhouse.example.com:8443
+    asserts:
+      - hasDocuments:
+          count: 0

charts/flagsmith/unittests/experimentation/init-job_test.yaml

@@ -43,3 +43,11 @@ tests:
       - contains:
           path: spec.template.spec.containers[0].env
           content: {name: KAFKA_TOPICS, value: events events-retry}
+
+  - it: renders nothing with the clickhouse sink
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse.url: https://clickhouse.example.com:8443
+    asserts:
+      - hasDocuments:
+          count: 0

charts/flagsmith/unittests/experimentation/secret_test.yaml

@@ -12,6 +12,17 @@ tests:
           path: data.KAFKA_PASSWORD
           value: a2Fma2Etc2VjcmV0
 
+  - it: stores the ClickHouse password with the clickhouse sink
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse: {url: https://clickhouse.example.com:8443, password: clickhouse-secret}
+    asserts:
+      - equal:
+          path: data.CLICKHOUSE_PASSWORD
+          value: Y2xpY2tob3VzZS1zZWNyZXQ=
+      - notExists:
+          path: data.KAFKA_PASSWORD
+
   - it: renders nothing when every value comes from an existing secret
     set:
       experimentation.kafka.passwordFromExistingSecret: {enabled: true, name: kafka, key: password}

charts/flagsmith/unittests/experimentation/validation_test.yaml

@@ -12,3 +12,17 @@ tests:
     asserts:
       - failedTemplate:
           errorMessage: experimentation.kafka.password or experimentation.kafka.passwordFromExistingSecret is required when experimentation.kafka.auth is scram
+
+  - it: requires a known sink
+    set:
+      experimentation.sink: s3
+    asserts:
+      - failedTemplate:
+          errorMessage: experimentation.sink must be kafka or clickhouse
+
+  - it: requires a ClickHouse URL with the clickhouse sink
+    set:
+      experimentation.sink: clickhouse
+    asserts:
+      - failedTemplate:
+          errorMessage: experimentation.clickhouse.url is required when experimentation.sink is clickhouse

charts/flagsmith/unittests/experimentation/warehouse-delivery-deployment_test.yaml

@@ -81,3 +81,12 @@ tests:
     asserts:
       - failedTemplate:
           errorMessage: experimentation.warehouseCredentialsSecretFromExistingSecret.name is required
+
+  - it: renders nothing with the clickhouse sink
+    template: experimentation/warehouse-delivery-deployment.yaml
+    set:
+      experimentation.sink: clickhouse
+      experimentation.clickhouse.url: https://clickhouse.example.com:8443
+    asserts:
+      - hasDocuments:
+          count: 0

charts/flagsmith/unittests/experimentation/warehouse-delivery-pdb_test.yaml

@@ -0,0 +1,16 @@
+suite: warehouse delivery PDB
+templates:
+  - experimentation/warehouse-delivery-pdb.yaml
+release:
+  name: flagsmith
+values:
+  - ../values/experimentation.yaml
+tests:
+  - it: renders nothing with the clickhouse sink
+    set:
+      pdb.warehouseDelivery.enabled: true
+      experimentation.sink: clickhouse
+      experimentation.clickhouse.url: https://clickhouse.example.com:8443
+    asserts:
+      - hasDocuments:
+          count: 0

charts/flagsmith/values.schema.json

@@ -471,6 +471,49 @@
         "experimentation": {
             "type": "object",
             "properties": {
+                "clickhouse": {
+                    "type": "object",
+                    "properties": {
+                        "database": {
+                            "type": "string"
+                        },
+                        "password": {
+                            "type": [
+                                "string",
+                                "null"
+                            ]
+                        },
+                        "passwordFromExistingSecret": {
+                            "type": "object",
+                            "properties": {
+                                "enabled": {
+                                    "type": "boolean"
+                                },
+                                "key": {
+                                    "type": [
+                                        "string",
+                                        "null"
+                                    ]
+                                },
+                                "name": {
+                                    "type": [
+                                        "string",
+                                        "null"
+                                    ]
+                                }
+                            }
+                        },
+                        "url": {
+                            "type": [
+                                "string",
+                                "null"
+                            ]
+                        },
+                        "user": {
+                            "type": "string"
+                        }
+                    }
+                },
                 "enabled": {
                     "type": "boolean"
                 },
@@ -704,6 +747,9 @@
                         }
                     }
                 },
+                "sink": {
+                    "type": "string"
+                },
                 "warehouseCredentialsSecretFromExistingSecret": {
                     "type": "object",
                     "properties": {

charts/flagsmith/values.yaml

@@ -663,6 +663,23 @@ experimentation:
     name: ""
     key: ""
 
+  # Where the ingestion service writes events. clickhouse writes straight to one ClickHouse, with no Kafka or
+  # warehouse delivery.
+  sink: kafka # kafka | clickhouse
+
+  # Used when sink is clickhouse. Every environment's events go to this ClickHouse, so connect it in the dashboard for
+  # each environment. Run the dashboard's ClickHouse setup SQL before deploying. Events are buffered in memory only.
+  clickhouse:
+    # https://clickhouse.example.com:8443
+    url: null # @schema type:[string, null]
+    user: default
+    database: flagsmith_exp
+    password: null # @schema type:[string, null]
+    passwordFromExistingSecret:
+      enabled: false
+      name: null # @schema type:[string, null]
+      key: null # @schema type:[string, null]
+
   kafka:
     # host1:9096,host2:9096
     bootstrapServers: null # @schema type:[string, null]
@@ -688,7 +705,7 @@ experimentation:
       key: null # @schema type:[string, null]
     image:
       repository: flagsmith/events-ingestion-api
-      tag: "0.5"
+      tag: "0.6"
       imagePullPolicy: ""
       imagePullSecrets: []
     resources: