From beeca3fa360b0d5a3e97c3d767e1203d444efb94 Mon Sep 17 00:00:00 2001 From: Jacob Cox Date: Tue, 14 Apr 2026 08:47:13 -0700 Subject: [PATCH 1/2] bumped orchestrator version to v5.2.7 --- manticore/versions/1.3.0/Chart.yaml | 2 +- manticore/versions/1.3.0/values.yaml | 10 +++++----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/manticore/versions/1.3.0/Chart.yaml b/manticore/versions/1.3.0/Chart.yaml index f965831d..03a2df54 100644 --- a/manticore/versions/1.3.0/Chart.yaml +++ b/manticore/versions/1.3.0/Chart.yaml @@ -8,6 +8,6 @@ appVersion: "15.1.0" annotations: created: "2026-01-05" - lastModified: "2026-04-09" + lastModified: "2026-04-14" category: "search" createsGvc: false \ No newline at end of file diff --git a/manticore/versions/1.3.0/values.yaml b/manticore/versions/1.3.0/values.yaml index 317c2762..24aced5b 100644 --- a/manticore/versions/1.3.0/values.yaml +++ b/manticore/versions/1.3.0/values.yaml @@ -88,7 +88,7 @@ manticore: # Orchestrator Configuration # ============================================================================= orchestrator: - version: v5.2.3 + version: v5.2.7 image: ghcr.io/controlplane-com/manticore-orchestrator/manticore-cpln-api logLevel: debug # debug, info, warn, error resources: @@ -106,7 +106,7 @@ orchestrator: # Orchestrator API api: - version: v5.2.3 + version: v5.2.7 image: ghcr.io/controlplane-com/manticore-orchestrator/manticore-cpln-api logLevel: debug importPollInterval: 30s @@ -122,7 +122,7 @@ orchestrator: # Agent sidecar agent: - version: v5.2.3 + version: v5.2.7 image: ghcr.io/controlplane-com/manticore-orchestrator/manticore-cpln-agent # REQUIRED: Generate with `openssl rand -base64 32` token: "6Gl5uO9KkKAh1u+ymoBW98WCtjTFpljuhpLdKb+tNAA=" @@ -140,7 +140,7 @@ orchestrator: # Web UI - See README.md "Authentication" section for security notes ui: - version: v5.2.3 + version: v5.2.7 image: ghcr.io/controlplane-com/manticore-orchestrator/manticore-cpln-ui resources: cpu: 0.25 @@ -155,7 +155,7 @@ orchestrator: # Backup Configuration (Optional) - Cron job runs logical backup on delta table to S3 bucket. backup: enabled: false - version: v5.2.3 + version: v5.2.7 image: ghcr.io/controlplane-com/manticore-orchestrator/manticore-cpln-backup cloudAccountName: my-backup-cloud-account s3Bucket: my-backup-bucket # S3 bucket for backups From c2c5c30158d055fc3c0e129bef8f7f0e7b5f41dd Mon Sep 17 00:00:00 2001 From: Igor Koshchyts Date: Mon, 13 Apr 2026 17:54:58 -0700 Subject: [PATCH 2/2] add kafka 3.5.0 chart and update RELEASES.md --- kafka/RELEASES.md | 6 + kafka/versions/3.5.0/.helmignore | 1 + kafka/versions/3.5.0/Chart.yaml | 12 + kafka/versions/3.5.0/README.md | 169 ++++ kafka/versions/3.5.0/templates/_helpers.tpl | 894 ++++++++++++++++++ kafka/versions/3.5.0/templates/domain.yaml | 64 ++ kafka/versions/3.5.0/templates/identity.yaml | 4 + kafka/versions/3.5.0/templates/kafbat-ui.yaml | 114 +++ .../3.5.0/templates/kafka-connectors.yaml | 224 +++++ .../3.5.0/templates/kafka-rest-proxy.yaml | 169 ++++ kafka/versions/3.5.0/templates/policy.yaml | 16 + .../secret-controller-configuration.yaml | 77 ++ .../versions/3.5.0/templates/secret-init.yaml | 184 ++++ .../3.5.0/templates/secret-secrets.yaml | 12 + .../versions/3.5.0/templates/volumesets.yaml | 31 + .../templates/workload-kafka-client.yaml | 46 + .../templates/workload-kafka-cluster.yaml | 258 +++++ .../3.5.0/templates/workload-kafka-ui.yaml | 66 ++ kafka/versions/3.5.0/values.yaml | 483 ++++++++++ 19 files changed, 2830 insertions(+) create mode 100644 kafka/versions/3.5.0/.helmignore create mode 100644 kafka/versions/3.5.0/Chart.yaml create mode 100644 kafka/versions/3.5.0/README.md create mode 100644 kafka/versions/3.5.0/templates/_helpers.tpl create mode 100644 kafka/versions/3.5.0/templates/domain.yaml create mode 100644 kafka/versions/3.5.0/templates/identity.yaml create mode 100644 kafka/versions/3.5.0/templates/kafbat-ui.yaml create mode 100644 kafka/versions/3.5.0/templates/kafka-connectors.yaml create mode 100644 kafka/versions/3.5.0/templates/kafka-rest-proxy.yaml create mode 100644 kafka/versions/3.5.0/templates/policy.yaml create mode 100644 kafka/versions/3.5.0/templates/secret-controller-configuration.yaml create mode 100644 kafka/versions/3.5.0/templates/secret-init.yaml create mode 100644 kafka/versions/3.5.0/templates/secret-secrets.yaml create mode 100644 kafka/versions/3.5.0/templates/volumesets.yaml create mode 100644 kafka/versions/3.5.0/templates/workload-kafka-client.yaml create mode 100644 kafka/versions/3.5.0/templates/workload-kafka-cluster.yaml create mode 100644 kafka/versions/3.5.0/templates/workload-kafka-ui.yaml create mode 100644 kafka/versions/3.5.0/values.yaml diff --git a/kafka/RELEASES.md b/kafka/RELEASES.md index 302b7598..9375966c 100644 --- a/kafka/RELEASES.md +++ b/kafka/RELEASES.md @@ -1,3 +1,9 @@ +# Release Notes - Version 3.5.0 + +## What's New + +- **Kafka Cluster Parallel Scaling Policy**: Changed the default `scalingPolicy` for the Kafka cluster stateful workload from `OrderedReady` to `Parallel` + # Release Notes - Version 3.4.0 ## What's New diff --git a/kafka/versions/3.5.0/.helmignore b/kafka/versions/3.5.0/.helmignore new file mode 100644 index 00000000..7d101009 --- /dev/null +++ b/kafka/versions/3.5.0/.helmignore @@ -0,0 +1 @@ +values.yaml \ No newline at end of file diff --git a/kafka/versions/3.5.0/Chart.yaml b/kafka/versions/3.5.0/Chart.yaml new file mode 100644 index 00000000..1e7cea62 --- /dev/null +++ b/kafka/versions/3.5.0/Chart.yaml @@ -0,0 +1,12 @@ +apiVersion: v2 +name: kafka +description: Kafka cluster app for Control Plane +type: application +version: 3.5.0 +appVersion: "3.9" + +annotations: + created: "2026-04-13" + lastModified: "2026-04-13" + category: "event-streaming" + createsGvc: false \ No newline at end of file diff --git a/kafka/versions/3.5.0/README.md b/kafka/versions/3.5.0/README.md new file mode 100644 index 00000000..47dc2aa5 --- /dev/null +++ b/kafka/versions/3.5.0/README.md @@ -0,0 +1,169 @@ +## Kafka App + +### How to connect to the cluster + +You can connect to Kafka from the same GVC in which it's deployed using the following methods: + +- To connect using the cluster's general address, use `{kafka-cluster-workload-name}:9092`. + +- To connect to a specific replica, use one of the following addresses based on the replica you wish to connect to: + - `{kafka-cluster-workload-name}-0.{kafka-cluster-workload-name}:9092` + - `{kafka-cluster-workload-name}-1.{kafka-cluster-workload-name}:9092` + - `{kafka-cluster-workload-name}-2.{kafka-cluster-workload-name}:9092` + +- If you're configuring your Kafka for external access, you'll need to provide a domain name for the public address of the listener you want to use. Prerequisites: + - Make sure the dedicated load balancer is enabled on the GVC. See [Configure Domain documentation](https://docs.controlplane.com/guides/configure-domain#dedicated-load-balancing). + - Make sure to register your [Apex domain](https://docs.controlplane.com/reference/domain#apex-domain-considerations) name with Control Plane and set up a DNS record for the Kafka public address CNAME with the canonical GVC endpoint in your DNS provider. + +### Test Kafka Cluster with Kafka Client + +1. To activate the Kafka client, make sure `kafka_client` is uncommented in your values file. If necessary, reinstall the chart with the command: + ```bash + cpln helm install kafka-dev -f values-example.yaml + ``` + +2. To connect to the `kafka-client` workload, navigate through the UI to the appropriate GVC and select the `kafka-client` workload. In the workload details, find and use the **Connect** feature to establish a connection, which can be done either via the UI or by utilizing the CLI command provided there. + +3. Once connected, you can write and consume messages through the `kafka-client` workload. If it's `PLAINTEXT`, producer and consumer configurations should be omitted below: + +```BASH +# Change to bin directory +cd /opt/kafka/bin + +# Create client.properties +echo "security.protocol=SASL_PLAINTEXT +sasl.mechanism=PLAIN +sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"your-admin-password\";" > ./client.properties + +# Produce messages to the 'controlplane' topic +kafka-console-producer.sh --bootstrap-server {kafka-cluster-workload-name}:9092 --topic controlplane --producer.config ./client.properties + +# Consume messages from the 'controlplane' topic +kafka-console-consumer.sh --bootstrap-server {kafka-cluster-workload-name}:9092 --topic controlplane --from-beginning --consumer.config ./client.properties +``` + +### Public Listener Domain Configuration + +When configuring Kafka for external access via a public listener, you can choose between two domain routing modes: + +#### **Direct Replica Routing Mode (Recommended)** + +The recommended approach with automatic replica endpoint generation: + +```yaml +kafka: + listeners: + public: + protocol: SASL_PLAINTEXT + name: PUBLIC + directReplicaRouting: + enabled: true + containerPort: 9095 # ports 9091, 9093 and 9094 are reserved + publicAddress: kafka.example.com + sasl: + users: "public-user" + passwords: "your-password" +``` + +**Behavior:** +- Single domain configuration with the specified container port +- DNS01 certificate challenge for automatic SSL +- Platform automatically generates replica-specific subdomains in format: `{replica-name}-{location}.{publicAddress}` +- Replica-aware routing reduces cross-zone traffic costs in multi-zone deployments +- Connection endpoints (auto-generated examples): + - `kafka-cluster-0-aws-us-east-1.kafka.example.com:9095` + - `kafka-cluster-1-aws-us-east-1.kafka.example.com:9095` + - `kafka-cluster-2-aws-us-east-1.kafka.example.com:9095` + +**Prerequisites for Direct Routing:** +- DNS provider must support CNAME records +- Create DNS records for each replica and the ACME challenge record: + 1. `CNAME kafka-cluster-0-aws-us-east-1.kafka.example.com → kafka-cluster--0.aws-us-east-1.controlplane.us` + 2. `CNAME kafka-cluster-1-aws-us-east-1.kafka.example.com → kafka-cluster--1.aws-us-east-1.controlplane.us` + 3. `CNAME kafka-cluster-2-aws-us-east-1.kafka.example.com → kafka-cluster--2.aws-us-east-1.controlplane.us` + 4. `CNAME _acme-challenge.kafka → _acme-challenge.cpln.app` (for certificate validation) + +#### **Multi-Port Routing** + +Each replica gets its own port. Not recommended for multi-zone clusters: + +```yaml +kafka: + listeners: + public: + protocol: SASL_PLAINTEXT + name: PUBLIC + publicAddress: kafka.example.com + sasl: + users: "public-user" + passwords: "your-password" +``` + +**Behavior:** +- Creates ports 3000, 3001, 3002 (one per replica) +- Each port routes to a specific replica +- Custom TLS cipher suites configuration +- Connection format: `kafka.example.com:3000`, `kafka.example.com:3001`, etc. +- **Note**: Not recommended for multi-zone deployments as cross-zone traffic charges may occur + +**Which Mode to Use:** +- Use **Direct Replica Routing** for new deployments that require automatic SSL with zone-aware routing and per-replica hostnames +- Avoid using **Multi-Port Routing** unless you have specific use cases or existing clients configured with port numbers (3000-300X) + +**Configuration Rules:** +- Cannot use both `publicAddress` and `directReplicaRouting.enabled: true` in the same listener +- When `directReplicaRouting.enabled: true`, both `containerPort` and `publicAddress` must be specified within the `directReplicaRouting` section +- Only one listener can have a public address configured across all listeners +- Direct Replica Routing automatically creates DNS entries in format: `{replica-name}-{location}.{publicAddress}:{containerPort}` + +### Enable Custom Encryption using AWS Key Management Service (KMS) + +Custom encryption for volumes can be configured by setting the values under `kafka.volumes.customEncryption`. + +A key must be created in AWS before proceeding with the template. + +In the values file, set `enabled` to `true` and add the proper `region` and `keyId`. + +**Important** - To finish configuring in AWS once the template is installed: + +1. Navigate in the console to the created volume +2. Click on `spec` +3. Follow the `AWS Custom Encryption Instructions` +4. Repeat for each encrypted volume created + +### Kafbat configuration example + +Full configuration Docs: https://ui.docs.kafbat.io/configuration/configuration-file + +```YAML +kafka: + clusters: + - name: "apache-kafka" + bootstrapServers: "kafka-dev-cluster.kafka-dev.cpln.local:9092" + kafkaConnect: + - name: kafka-dev-connect-connect-cluster + address: http://kafka-dev-connect-connect-cluster.kafka-dev.cpln.local:8083 + properties: + security.protocol: "SASL_PLAINTEXT" + sasl.mechanism: "PLAIN" + sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"your-admin-password\";" + +management: + health: + ldap: + enabled: false + +auth: + type: "LOGIN_FORM" +spring: + security: + user: + name: "admin" + password: "adminPassword" + +server: + port: 8080 +``` + +### Release Notes +See [RELEASES.md](https://github.com/controlplane-com/templates/blob/main/kafka/RELEASES.md) diff --git a/kafka/versions/3.5.0/templates/_helpers.tpl b/kafka/versions/3.5.0/templates/_helpers.tpl new file mode 100644 index 00000000..61c9433e --- /dev/null +++ b/kafka/versions/3.5.0/templates/_helpers.tpl @@ -0,0 +1,894 @@ +{{/* +Name +*/}} +{{- define "kafka.name" -}} +{{- printf "%s" .Release.Name -}} +{{- end }} + +{{/* +Cluster Workload Name +*/}} +{{- define "kafka.clusterName" -}} +{{- printf "%s-%s" (include "kafka.name" .) .Values.kafka.name -}} +{{- end }} + +{{/* +Convert .Values.kafka.memory to appropriate JVM heap size settings. +*/}} +{{- define "kafka.heap.opts" -}} +{{- $memory := default "512Mi" .Values.kafka.memory }} +{{- $memoryInMi := 0 }} +{{- if hasSuffix "Gi" $memory }} + {{- $value := trimSuffix "Gi" $memory | float64 }} + {{- $memoryInMi = mul $value 1024 | int }} +{{- else if hasSuffix "Mi" $memory }} + {{- $memoryInMi = trimSuffix "Mi" $memory | int }} +{{- else }} + {{- $memoryInMi = 512 }} # Default to 512Mi if no suffix +{{- end }} +{{- $heapSize := div (mul $memoryInMi 60) 100 | int }} +-Xmx{{ $heapSize }}m -Xms{{ $heapSize }}m +{{- end }} + +{{- define "kafka.validateListenerConfig" -}} + {{- if not .name -}} + {{- fail "Error: 'name' must be provided for the listener" -}} + {{- end -}} + {{- if not .protocol -}} + {{- fail "Error: 'protocol' must be provided for the listener" -}} + {{- end -}} + {{- $hasValidConfig := or .publicAddress .containerPort (and .directReplicaRouting .directReplicaRouting.enabled .directReplicaRouting.containerPort .directReplicaRouting.publicAddress) -}} + {{- if not $hasValidConfig -}} + {{- fail "Error: At least one of 'publicAddress', 'containerPort', or valid 'directReplicaRouting' (with enabled: true, containerPort, and publicAddress) must be provided for the listener" -}} + {{- end -}} + {{- if and .publicAddress .containerPort -}} + {{- fail "Error: When publicAddress is set for the listener, containerPort should not be specified as it will be automatically set to port range 3000-3004" -}} + {{- end -}} + {{- if .containerPort -}} + {{- $port := .containerPort | printf "%s" }} + {{- if or (eq $port "9091") (eq $port "9093") (eq $port "9094") -}} + {{- fail "Error: containerPort cannot be 9091, 9093, or 9094 for listener" -}} + {{- end -}} + {{- end -}} + {{- if and .directReplicaRouting .directReplicaRouting.enabled -}} + {{- if .directReplicaRouting.containerPort -}} + {{- $port := .directReplicaRouting.containerPort | printf "%s" }} + {{- if or (eq $port "9091") (eq $port "9093") (eq $port "9094") -}} + {{- fail "Error: directReplicaRouting.containerPort cannot be 9091, 9093, or 9094 for listener" -}} + {{- end -}} + {{- end -}} + {{- end -}} +{{- end -}} + +{{- define "kafka.validateAdminExists" -}} +{{- $adminFound := false -}} +{{- $saslPlaintextExists := false -}} +{{- range .Values.kafka.listeners -}} + {{- if eq .protocol "SASL_PLAINTEXT" -}} + {{- $saslPlaintextExists = true -}} + {{- if and .sasl .sasl.admin -}} + {{- $adminFound = true -}} + {{- end -}} + {{- end -}} +{{- end -}} +{{- if and $saslPlaintextExists (not $adminFound) -}} + {{- fail "Error: At least one SASL_PLAINTEXT listener must have an admin user configured in sasl.admin" -}} +{{- end -}} +{{- end -}} + +{{- define "kafka.validateAuthConfig" -}} +{{- if eq .protocol "SASL_PLAINTEXT" -}} + {{- if not .sasl -}} + {{- fail (printf "Error: SASL_PLAINTEXT protocol requires sasl configuration to be enabled for listener '%s'" .name) -}} + {{- else if not .sasl.users -}} + {{- fail (printf "Error: SASL_PLAINTEXT protocol requires at least one user to be defined in sasl.users for listener '%s'" .name) -}} + {{- else -}} + {{- $userCount := len (splitList "," .sasl.users) -}} + {{- if not .sasl.passwords -}} + {{- fail (printf "Error: sasl.passwords must be provided when sasl.users is defined for listener '%s'" .name) -}} + {{- else -}} + {{- $passwordCount := len (splitList "," .sasl.passwords) -}} + {{- if ne $userCount $passwordCount -}} + {{- fail (printf "Error: Number of users (%d) does not match number of passwords (%d) for listener '%s'" $userCount $passwordCount .name) -}} + {{- end -}} + {{- end -}} + {{- end -}} +{{- end -}} +{{- end -}} + +{{- define "kafka.validateReplicas" -}} +{{- $replicas := .Values.kafka.replicas | int }} +{{- if or (gt $replicas 5) (eq $replicas 2) -}} + {{- fail "Invalid value for kafka.replicas. It must be less than or equal to 5 and not equal to 2." -}} +{{- end -}} +{{- end -}} + +{{- define "kafka.validateOnePublicAddress" -}} +{{- $publicAddressCount := 0 -}} +{{- range .Values.kafka.listeners }} + {{- if .publicAddress }} + {{- $publicAddressCount = add $publicAddressCount 1 -}} + {{- end }} + {{- if and .directReplicaRouting .directReplicaRouting.enabled .directReplicaRouting.publicAddress }} + {{- $publicAddressCount = add $publicAddressCount 1 -}} + {{- end }} +{{- end }} +{{- if gt $publicAddressCount 1 -}} + {{- fail "There must be at most one listener with a publicAddress set (either listener.publicAddress or listener.directReplicaRouting.publicAddress)." -}} +{{- end }} +{{- end -}} + +{{- define "kafka.validatedirectReplicaRoutingConfig" -}} +{{- range $key, $listener := .Values.kafka.listeners }} + {{- if and $listener.publicAddress $listener.directReplicaRouting }} + {{- if $listener.directReplicaRouting.enabled }} + {{- fail (printf "Error in listener '%s': Cannot have both 'publicAddress' at listener level and 'directReplicaRouting.enabled: true'. Use either legacy mode (publicAddress only) or new mode (directReplicaRouting with enabled: true)." $key) -}} + {{- end }} + {{- end }} + {{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }} + {{- if not $listener.directReplicaRouting.publicAddress }} + {{- fail (printf "Error in listener '%s': When directReplicaRouting.enabled is true, directReplicaRouting.publicAddress must be specified." $key) -}} + {{- end }} + {{- if not $listener.directReplicaRouting.containerPort }} + {{- fail (printf "Error in listener '%s': When directReplicaRouting.enabled is true, directReplicaRouting.containerPort must be specified." $key) -}} + {{- end }} + {{- end }} +{{- end }} +{{- end -}} + +{{- define "kafka.validateKafkaImage" -}} +{{- $image := .Values.kafka.image -}} +{{- if contains "bitnami" $image -}} + {{- fail (printf "Error: This chart does not support Bitnami images, please use Apache Kafka images instead. Current value: %s" $image) -}} +{{- end -}} +{{- end -}} + +{{- define "kafka.validateImage" -}} +{{- $image := .image -}} +{{- if contains "bitnami" $image -}} + {{- fail (printf "Error: This chart does not support Bitnami images, please use Apache Kafka images instead. Current value: %s" $image) -}} +{{- end -}} +{{- end -}} + +{{- define "kafka.clientBootstrapAddress" -}} +{{- $clusterName := include "kafka.clusterName" . -}} +{{- $bootstrapAddress := "" -}} +{{- $listenerName := "" -}} + +{{- if .listenerName -}} + {{- $listenerName = .listenerName -}} +{{- else if .Values.kafka_connectors -}} + {{- range .Values.kafka_connectors -}} + {{- if .listener -}} + {{- $listenerName = .listener -}} + {{- end -}} + {{- end -}} +{{- end -}} + +{{- if $listenerName -}} + {{- if hasKey .Values.kafka.listeners $listenerName -}} + {{- $listener := index .Values.kafka.listeners $listenerName -}} + {{- if $listener.publicAddress -}} + {{- $bootstrapAddress = printf "%s:3000" $listener.publicAddress -}} + {{- else -}} + {{- $containerPort := $listener.containerPort | int -}} + {{- $bootstrapAddress = printf "%s:%d" $clusterName $containerPort -}} + {{- end -}} + {{- else -}} + {{- $bootstrapAddress = include "kafka.bootstrapAddress" . -}} + {{- end -}} +{{- else -}} + {{- $bootstrapAddress = include "kafka.bootstrapAddress" . -}} +{{- end -}} + +{{- $bootstrapAddress -}} +{{- end -}} + +{{- define "kafka.propertiesMapToList" -}} +{{- range $key, $value := . -}} +{{ $key }}={{ $value }} +{{- end -}} +{{- end -}} + + + +{{- define "kafka.connectors.download.script" -}} +#!/bin/sh +set -e{{- if .verbose }}x{{- end }} + +download_file() { + local url=$1 + local output_file=$2 + + if echo "$url" | grep -q "@.*jfrog"; then + echo "Handling JFrog redirect for: $url" + local redirect_url=$(wget -S --spider "$url" 2>&1 | grep 'Location:' | awk '{print $2}') + if [ -n "$redirect_url" ]; then + echo "Downloading from redirect URL..." + wget -q "$redirect_url" -O "$output_file" + else + echo "Failed to get redirect URL, trying direct download..." + wget -q "$url" -O "$output_file" + fi + else + wget -q "$url" -O "$output_file" + fi +} + +# Function to download and extract artifacts +download_and_extract() { + local artifact_type=$1 + local artifact_url=$2 + local plugin_path=$3 + local plugin_name=$4 + local temp_dir=$(mktemp -d) + + echo "Downloading artifact from $artifact_url" + + if [ "$artifact_type" == "jar" ]; then + # For jar files, create a directory for the plugin if it doesn't exist + local plugin_dir="$plugin_path/$plugin_name" + mkdir -p "$plugin_dir" + + # Download jar file to the plugin-specific directory + download_file "$artifact_url" "$plugin_dir/${plugin_name}.jar" + echo "Downloaded JAR file to $plugin_dir/${plugin_name}.jar" + else + # For archives, download to temp dir and extract + local archive_file="$temp_dir/archive.${artifact_type}" + download_file "$artifact_url" "$archive_file" + + echo "Extracting $artifact_type archive to $plugin_path" + case "$artifact_type" in + "tgz"|"tar.gz") + tar -xzf "$archive_file" -C "$plugin_path" + ;; + "tar") + tar -xf "$archive_file" -C "$plugin_path" + ;; + "zip") + unzip -o "$archive_file" -d "$plugin_path" + ;; + *) + echo "Unsupported archive type: $artifact_type" + ;; + esac + + rm -rf "$temp_dir" + fi +} +# Process each Kafka connector +echo "Setting up Kafka connector plugins" + +# Download and extract artifacts for each enabled plugin +{{- range .plugins }} +{{- if eq .enabled true }} +echo "Processing plugin: {{ .name }}" +{{- $pluginName := .name }} +{{- range .artifacts }} +download_and_extract "{{ .type }}" "{{ .url }}" "{{ $.plugins_folder }}" "{{ $pluginName }}" +{{- end }} +{{- else }} +echo "Skipping disabled plugin: {{ .name }}" +{{- end }} +{{- end }} + +echo "All Kafka connector plugins have been downloaded and extracted." +echo "Sleeping..." +sleep infinity +{{- end }} + +{{- define "kafka.connectors.run.script" -}} +#!/bin/bash +set -e{{- if .verbose }}x{{- end }} + +# Function to create or update a connector +create_or_update_connector() { + local connector_name=$1 + local config=$2 + local cluster_connectors=$3 + + echo "Checking if connector $connector_name exists in cluster..." + + # Check if connector exists in the cluster-wide list (reliable in distributed mode) + local exists=false + if echo "$cluster_connectors" | grep -q "\"$connector_name\""; then + exists=true + echo "Connector $connector_name found in cluster list" + else + echo "Connector $connector_name does not exist in cluster" + fi + + if [ "$exists" = true ]; then + echo "Connector $connector_name exists. Updating configuration..." + + # Extract just the config part from the full connector JSON + # Remove the "name" line, remove everything up to and including "config": {, remove last two lines (both closing braces) + local config_content=$(echo "$config" | sed '/^[[:space:]]*"name":/d' | sed '1,/^[[:space:]]*"config":[[:space:]]*{/d' | sed '$d' | sed '$d') + local update_config="{${config_content}}" + + echo "Updating connector $connector_name with config:" + echo "$update_config" + + # Update the connector using PUT with nc (BusyBox wget doesn't support PUT) + local content_length=$(echo -n "$update_config" | wc -c | xargs) + local response + response=$(echo -e "PUT /connectors/$connector_name/config HTTP/1.1\r\nHost: localhost:8083\r\nContent-Type: application/json\r\nContent-Length: $content_length\r\nConnection: close\r\n\r\n$update_config" | nc localhost 8083 2>&1) + echo "HTTP response for $connector_name update: $(echo "$response" | head -1)" + if ! echo "$response" | grep -q "HTTP/1\.. 2"; then + echo "ERROR: Failed to update connector $connector_name. Response: $(echo "$response" | head -5)" + else + echo "Connector $connector_name updated successfully" + fi + else + echo "Connector $connector_name does not exist. Creating..." + + echo "Creating connector $connector_name with config:" + echo "$config" + + # Create the connector using POST (this may also update if connector exists) + wget -q -O /dev/null "http://localhost:8083/connectors" \ + --header="Content-Type: application/json" \ + --post-data="$config" + + echo "Connector $connector_name created successfully" + + # Add a delay after creation to allow connector to initialize + sleep 2 + fi +} + +# Function to check and setup truststore for SSL connections +truststore_init() { + local hostname=$1 + local port=$2 + local alias=$3 + local jdbc_props=$4 + + echo "Setting up truststore for $hostname:$port with alias $alias" + + # Parse JDBC connection properties + local truststore_path + local truststore_password + + # First check if ssl.truststore.location is provided in the config + if [[ -n "${SSL_TRUSTSTORE_LOCATION}" ]]; then + truststore_path="${SSL_TRUSTSTORE_LOCATION}" + echo "Using ssl.truststore.location from config: $truststore_path" + # Then check if it's in JDBC properties + elif [[ "$jdbc_props" =~ ssl\.truststore\.location=([^;]+) ]]; then + truststore_path="${BASH_REMATCH[1]}" + echo "Using ssl.truststore.location from JDBC properties: $truststore_path" + elif [[ "$jdbc_props" =~ trustStorePath=([^;]+) ]]; then + truststore_path="${BASH_REMATCH[1]}" + echo "Using trustStorePath from JDBC properties: $truststore_path" + else + # Use default path if not specified + truststore_path="/tmp/kafka.client.truststore.jks" + echo "No truststore path specified, using default: $truststore_path" + fi + + # Check if ssl.truststore.password is provided + if [[ -n "${SSL_TRUSTSTORE_PASSWORD}" ]]; then + truststore_password="${SSL_TRUSTSTORE_PASSWORD}" + echo "Using ssl.truststore.password from config" + # Then check if it's in JDBC properties + elif [[ "$jdbc_props" =~ ssl\.truststore\.password=([^;]+) ]]; then + truststore_password="${BASH_REMATCH[1]}" + echo "Using ssl.truststore.password from JDBC properties" + elif [[ "$jdbc_props" =~ trustStorePassword=([^;]+) ]]; then + truststore_password="${BASH_REMATCH[1]}" + echo "Using trustStorePassword from JDBC properties" + else + # Generate random password if not specified + truststore_password=$(openssl rand -base64 12) + export SSL_TRUSTSTORE_PASSWORD="${truststore_password}" + echo "Generated random ssl.truststore.password: ${truststore_password}" + fi + + # Create certs directory if it doesn't exist + mkdir -p $(dirname "$truststore_path") + + # Download CA certificate + echo "Downloading CA certificate for $hostname:$port" + echo | openssl s_client -connect $hostname:$port -showcerts 2>/dev/null | \ + openssl x509 -outform PEM > $(dirname "$truststore_path")/$alias.pem || \ + echo "WARNING: Failed to download certificate from $hostname:$port, truststore may be incomplete" + + # Create truststore if it doesn't exist or override existing one + echo "Creating new truststore from $JAVA_HOME/lib/security/cacerts" + if [[ -f "$JAVA_HOME/lib/security/cacerts" ]]; then + cp "$JAVA_HOME/lib/security/cacerts" "$truststore_path" || \ + echo "WARNING: Failed to copy cacerts to $truststore_path, truststore setup may be incomplete" + else + echo "WARNING: $JAVA_HOME/lib/security/cacerts not found, skipping truststore creation for $hostname" + return 0 + fi + + # Change the default password to our password + echo "Setting truststore password" + keytool -storepasswd -keystore "$truststore_path" \ + -storepass "changeit" -new "${truststore_password}" || \ + echo "WARNING: Failed to change truststore password for $hostname, continuing with default password" + + # Import certificate into truststore (only if cert file is non-empty) + local cert_file="$(dirname "$truststore_path")/$alias.pem" + if [[ -s "$cert_file" ]]; then + echo "Importing certificate into truststore" + keytool -import -noprompt -alias $alias -file "$cert_file" \ + -keystore "$truststore_path" -storepass "${truststore_password}" || \ + echo "WARNING: Failed to import certificate for $alias, truststore may be incomplete" + else + echo "WARNING: Skipping certificate import for $alias, cert file is empty or missing" + fi + + echo "Truststore setup completed for $hostname" +} + +# Function to setup multi-domain truststore from values configuration +setup_multi_domain_truststore() { + local plugin_name="$1" + local ssl_truststore_config="$2" + + echo "Setting up multi-domain truststore for plugin: $plugin_name" + + # Parse the ssl_truststore configuration (passed as JSON-like string) + local generate=$(echo "$ssl_truststore_config" | grep -o '"generate"[[:space:]]*:[[:space:]]*true' | wc -l) + + if [[ $generate -eq 0 ]]; then + echo "Multi-domain truststore generation disabled for $plugin_name" + return 0 + fi + + echo "Multi-domain truststore generation enabled for $plugin_name" + + # Extract truststore path (REQUIRED) + local truststore_path=$(echo "$ssl_truststore_config" | grep -o '"truststore_path"[[:space:]]*:[[:space:]]*"[^"]*"' | sed 's/.*"truststore_path"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/') + if [[ -z "$truststore_path" ]]; then + echo "ERROR: ssl_truststore.truststore_path is required when ssl_truststore.generate is true for plugin $plugin_name" + exit 1 + fi + + # Extract password environment variable name (REQUIRED) + local password_env=$(echo "$ssl_truststore_config" | grep -o '"truststore_password_env"[[:space:]]*:[[:space:]]*"[^"]*"' | sed 's/.*"truststore_password_env"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/') + if [[ -z "$password_env" ]]; then + echo "ERROR: ssl_truststore.truststore_password_env is required when ssl_truststore.generate is true for plugin $plugin_name" + exit 1 + fi + + # Check if password already exists + if [[ -n "${!password_env}" ]]; then + echo "Using existing password from environment variable: $password_env" + local truststore_password="${!password_env}" + else + # Generate random password + local truststore_password=$(openssl rand -base64 12) + export "$password_env"="$truststore_password" + echo "Generated random password for $password_env: $truststore_password" + fi + + # Create truststore directory if it doesn't exist + mkdir -p $(dirname "$truststore_path") + + # Create truststore if it doesn't exist or if we're starting fresh + if [[ ! -f "$truststore_path" ]]; then + echo "Creating new multi-domain truststore at: $truststore_path" + + # Validate JAVA_HOME exists + if [[ -z "$JAVA_HOME" ]]; then + echo "ERROR: JAVA_HOME environment variable is not set, required for truststore creation for plugin $plugin_name" + exit 1 + fi + + # Validate cacerts file exists + if [[ ! -f "$JAVA_HOME/lib/security/cacerts" ]]; then + echo "ERROR: Java cacerts file not found at $JAVA_HOME/lib/security/cacerts for plugin $plugin_name" + exit 1 + fi + + # Copy cacerts as base truststore + if ! cp "$JAVA_HOME/lib/security/cacerts" "$truststore_path"; then + echo "ERROR: Failed to create truststore file at $truststore_path for plugin $plugin_name" + exit 1 + fi + + # Change the default password to our password + echo "Setting truststore password" + if ! keytool -storepasswd -keystore "$truststore_path" \ + -storepass "changeit" -new "$truststore_password" >/dev/null 2>&1; then + echo "ERROR: Failed to set truststore password for plugin $plugin_name" + exit 1 + fi + fi + + # Extract and process hostnames (REQUIRED) + local hostnames=$(echo "$ssl_truststore_config" | grep -o '"hostnames"[[:space:]]*:[[:space:]]*\[[^]]*\]' | sed 's/.*"hostnames"[[:space:]]*:[[:space:]]*\[\([^]]*\)\].*/\1/' | tr ',' '\n') + + if [[ -z "$hostnames" ]]; then + echo "ERROR: ssl_truststore.hostnames is required and must be a non-empty array when ssl_truststore.generate is true for plugin $plugin_name" + exit 1 + fi + + # Validate that hostnames array is not empty + local hostname_count=$(echo "$hostnames" | grep -v '^$' | wc -l) + if [[ $hostname_count -eq 0 ]]; then + echo "ERROR: ssl_truststore.hostnames must contain at least one hostname when ssl_truststore.generate is true for plugin $plugin_name" + exit 1 + fi + + # Download and import certificates for each hostname + while IFS= read -r hostname_entry; do + if [[ -n "$hostname_entry" ]]; then + # Clean up the hostname (remove quotes and whitespace) + local clean_hostname=$(echo "$hostname_entry" | sed 's/[[:space:]]*"\([^"]*\)".*/\1/' | xargs) + + if [[ -n "$clean_hostname" ]]; then + local hostname=$(echo "$clean_hostname" | cut -d':' -f1) + local port=$(echo "$clean_hostname" | cut -d':' -f2) + + # Validate hostname:port format + if [[ -z "$hostname" || -z "$port" || "$hostname" == "$port" ]]; then + echo "ERROR: Invalid hostname format '$clean_hostname' in ssl_truststore.hostnames for plugin $plugin_name. Expected format: 'hostname:port'" + exit 1 + fi + + # Validate port is numeric + if ! [[ "$port" =~ ^[0-9]+$ ]]; then + echo "ERROR: Invalid port '$port' in hostname '$clean_hostname' for plugin $plugin_name. Port must be numeric." + exit 1 + fi + + local alias="$plugin_name-$(echo $hostname | tr '.' '-')" + + echo "Downloading certificate for $hostname:$port with alias $alias" + + # Download CA certificate + local cert_file="$(dirname "$truststore_path")/$alias.pem" + if ! echo | openssl s_client -connect $hostname:$port -showcerts 2>/dev/null | \ + openssl x509 -outform PEM > "$cert_file"; then + echo "ERROR: Failed to download certificate from $hostname:$port for plugin $plugin_name" + exit 1 + fi + + # Validate certificate file is not empty + if [[ ! -s "$cert_file" ]]; then + echo "ERROR: Downloaded certificate from $hostname:$port is empty for plugin $plugin_name" + exit 1 + fi + + # Import certificate into truststore (skip if already exists) + if keytool -list -keystore "$truststore_path" -storepass "$truststore_password" -alias "$alias" >/dev/null 2>&1; then + echo "Certificate with alias $alias already exists in truststore, skipping" + else + echo "Importing certificate with alias $alias into truststore" + if ! keytool -import -noprompt -alias "$alias" -file "$cert_file" \ + -keystore "$truststore_path" -storepass "$truststore_password" >/dev/null 2>&1; then + echo "ERROR: Failed to import certificate with alias $alias into truststore for plugin $plugin_name" + exit 1 + fi + fi + fi + fi + done <<< "$hostnames" + + + + echo "Multi-domain truststore setup completed for $plugin_name at: $truststore_path" +} + +# Function to setup connectors in the background +setup_connectors() { + echo "Starting connector setup process..." + + # Wait for Kafka Connect to start + echo "Waiting for Kafka Connect to start..." + until wget -q http://localhost:8083/ -O /dev/null; do + echo "Waiting for Kafka Connect REST API..." + sleep 5 + done + + echo "Kafka Connect REST API is up. Waiting for connector plugins to load..." + # Wait for connector plugins to be available (indicates full initialization) + local retry_count=0 + local max_retries=12 + until wget -q -O - http://localhost:8083/connector-plugins 2>/dev/null | grep -q "class" || [ $retry_count -ge $max_retries ]; do + echo "Waiting for connector plugins to load... (attempt $((retry_count+1))/$max_retries)" + sleep 5 + retry_count=$((retry_count+1)) + done + + echo "Kafka Connect plugins loaded. Now waiting for existing connectors to be restored from connect-config topic..." + # Poll for connectors to be restored, but with a timeout + local connector_wait=0 + local max_connector_wait=20 + local prev_count=-1 + while [ $connector_wait -lt $max_connector_wait ]; do + INSTALLED_CONNECTORS=$(wget -q -O - http://localhost:8083/connectors 2>/dev/null || echo "[]") + local current_count=$(echo "$INSTALLED_CONNECTORS" | tr -d '[]"' | tr ',' '\n' | grep -v '^$' | wc -l | xargs) + + if [ "$current_count" != "$prev_count" ]; then + echo "Connectors being restored... Found $current_count connector(s) so far: $INSTALLED_CONNECTORS" + prev_count=$current_count + connector_wait=0 # Reset wait counter when we see changes + else + if [ $connector_wait -eq 0 ] && [ "$current_count" -gt 0 ]; then + echo "Connector count stable at $current_count. Waiting 5 more seconds to ensure restoration is complete..." + fi + connector_wait=$((connector_wait+1)) + fi + + sleep 1 + done + + # Get final list of currently installed connectors + echo "Fetching final list of installed connectors..." + INSTALLED_CONNECTORS=$(wget -q -O - http://localhost:8083/connectors 2>/dev/null || echo "[]") + echo "Installed connectors: $INSTALLED_CONNECTORS" + + # Build list of desired connectors from values file + DESIRED_CONNECTORS=({{- range .plugins }} "{{ .name }}"{{- end }}) + echo "Desired connectors from values: ${DESIRED_CONNECTORS[@]}" + echo "Number of desired connectors: ${#DESIRED_CONNECTORS[@]}" + + # Remove connectors that are not in the desired list + if [[ "$INSTALLED_CONNECTORS" != "[]" && "$INSTALLED_CONNECTORS" != "" ]]; then + echo "$INSTALLED_CONNECTORS" | tr -d '[]"' | tr ',' '\n' | while IFS= read -r connector; do + connector=$(echo "$connector" | xargs) # trim whitespace + if [[ -n "$connector" ]]; then + found=false + for desired in "${DESIRED_CONNECTORS[@]}"; do + if [[ "$connector" == "$desired" ]]; then + found=true + break + fi + done + if [[ "$found" == "false" ]]; then + echo "Connector '$connector' is not enabled. Removing..." + (echo -e "DELETE /connectors/$connector HTTP/1.1\r\nHost: localhost:8083\r\nConnection: close\r\n\r\n" | nc localhost 8083 > /dev/null 2>&1) || true + fi + fi + done + fi + + # Create/update connectors + {{- range .plugins }} + echo "Processing connector: {{ .name }}" + {{- if hasKey . "enabled" }} + {{- if not .enabled }} + echo "Connector {{ .name }} is disabled. Removing if it exists..." + (echo -e "DELETE /connectors/{{ .name }} HTTP/1.1\r\nHost: localhost:8083\r\nConnection: close\r\n\r\n" | nc localhost 8083 > /dev/null 2>&1) || true + + # Wait a bit between connectors to allow Kafka Connect API to stabilize + sleep 3 + {{- else }} + echo "Creating/updating connector: {{ .name }}" + +{{- if hasKey . "ssl_truststore" }} +# Setup multi-domain truststore if configured +SSL_TRUSTSTORE_CONFIG='{"generate":{{ if hasKey .ssl_truststore "generate" }}{{ .ssl_truststore.generate }}{{ else }}false{{ end }}{{- if hasKey .ssl_truststore "truststore_path" }},"truststore_path":"{{ .ssl_truststore.truststore_path }}"{{- end }}{{- if hasKey .ssl_truststore "truststore_password_env" }},"truststore_password_env":"{{ .ssl_truststore.truststore_password_env }}"{{- end }}{{- if hasKey .ssl_truststore "hostnames" }},"hostnames":[{{- range $i, $hostname := .ssl_truststore.hostnames }}{{- if $i }},{{- end }}"{{ $hostname }}"{{- end }}]{{- end }}}' +setup_multi_domain_truststore "{{ .name }}" "$SSL_TRUSTSTORE_CONFIG" +{{- end }} + +{{- if and (hasKey .config "ssl") (eq .config.ssl "true") }} +# Check if SSL is enabled +# Export ssl.truststore.location if it exists +{{- if hasKey .config "ssl.truststore.location" }} +export SSL_TRUSTSTORE_LOCATION={{ index .config "ssl.truststore.location" | quote }} +{{- end }} +# Export ssl.truststore.password if it exists +{{- if hasKey .config "ssl.truststore.password" }} +export SSL_TRUSTSTORE_PASSWORD={{ index .config "ssl.truststore.password" | quote }} +{{- end }} +# Setup truststore +truststore_init "{{ .config.hostname }}" "{{ .config.port }}" "{{ .name }}" "{{ default "" .config.jdbcConnectionProperties }}" +{{- end }} + +CONFIG=$(cat << 'EOF' +{ + "name": "{{ .name }}", + "config": { + {{- $first := true }} + {{- range $key, $value := .config }} + {{- if $first }}{{ $first = false }}{{ else }},{{ end }} + "{{ $key }}": "{{ $value }}" + {{- end }} + {{- if and (hasKey .config "ssl") (eq .config.ssl "true") (not (hasKey .config "ssl.truststore.password")) }} + ,"ssl.truststore.password": "${SSL_TRUSTSTORE_PASSWORD}" + {{- end }} + } +} +EOF +) + +# If we have a generated password, replace it in the config +if [[ -n "${SSL_TRUSTSTORE_PASSWORD}" && ! "{{ if hasKey .config "ssl.truststore.password" }}true{{ else }}false{{ end }}" == "true" ]]; then + CONFIG=$(echo "$CONFIG" | sed "s|\${SSL_TRUSTSTORE_PASSWORD}|${SSL_TRUSTSTORE_PASSWORD}|g") +fi + + {{- if hasKey . "ssl_truststore" }} + {{- if hasKey .ssl_truststore "truststore_password_env" }} + # Replace plugin-specific truststore password if it exists + PLUGIN_PASSWORD_VAR="{{ .ssl_truststore.truststore_password_env }}" + echo "DEBUG: Looking for password in environment variable: $PLUGIN_PASSWORD_VAR" + echo "DEBUG: Password value: ${!PLUGIN_PASSWORD_VAR}" + if [[ -n "${!PLUGIN_PASSWORD_VAR}" ]]; then + echo "DEBUG: Replacing \${${PLUGIN_PASSWORD_VAR}} with password in config" + CONFIG=$(echo "$CONFIG" | sed "s|\${${PLUGIN_PASSWORD_VAR}}|${!PLUGIN_PASSWORD_VAR}|g") + echo "DEBUG: Config after replacement:" + echo "$CONFIG" + else + echo "DEBUG: No password found in $PLUGIN_PASSWORD_VAR" + fi + {{- end }} + {{- end }} + +# Try to create connector with retry logic +max_retries=5 +retry_count=0 +while [ $retry_count -lt $max_retries ]; do + if create_or_update_connector "{{ .name }}" "$CONFIG" "$INSTALLED_CONNECTORS"; then + echo "Successfully created/updated connector {{ .name }} on attempt $((retry_count+1))" + break + else + retry_count=$((retry_count+1)) + if [ $retry_count -lt $max_retries ]; then + echo "Failed to create/update connector {{ .name }}, retrying in 10 seconds (attempt $retry_count/$max_retries)..." + sleep 10 + else + echo "Failed to create/update connector {{ .name }} after $max_retries attempts" + fi + fi +done + +# Wait a bit between connectors to allow Kafka Connect API to stabilize +sleep 3 +{{- end }} + {{- else }} + echo "Creating/updating connector: {{ .name }}" + {{- if and (hasKey .config "ssl") (eq .config.ssl "true") }} + # Check if SSL is enabled + # Export ssl.truststore.location if it exists + {{- if hasKey .config "ssl.truststore.location" }} + export SSL_TRUSTSTORE_LOCATION={{ index .config "ssl.truststore.location" | quote }} + {{- end }} + # Export ssl.truststore.password if it exists + {{- if hasKey .config "ssl.truststore.password" }} + export SSL_TRUSTSTORE_PASSWORD={{ index .config "ssl.truststore.password" | quote }} + {{- end }} + # Setup truststore + truststore_init "{{ .config.hostname }}" "{{ .config.port }}" "{{ .name }}" "{{ default "" .config.jdbcConnectionProperties }}" + {{- end }} + + CONFIG=$(cat << 'EOF' +{ + "name": "{{ .name }}", + "config": { + {{- $first := true }} + {{- range $key, $value := .config }} + {{- if $first }}{{ $first = false }}{{ else }},{{ end }} + "{{ $key }}": "{{ $value }}" + {{- end }} + {{- if and (hasKey .config "ssl") (eq .config.ssl "true") (not (hasKey .config "ssl.truststore.password")) }} + {{- if not $first }},{{ end }} + "ssl.truststore.password": "${SSL_TRUSTSTORE_PASSWORD}" + {{- end }} + } +} +EOF +) + + # If we have a generated password, replace it in the config + if [[ -n "${SSL_TRUSTSTORE_PASSWORD}" && ! "{{ if hasKey .config "ssl.truststore.password" }}true{{ else }}false{{ end }}" == "true" ]]; then + CONFIG=$(echo "$CONFIG" | sed "s|\${SSL_TRUSTSTORE_PASSWORD}|${SSL_TRUSTSTORE_PASSWORD}|g") + fi + + {{- if hasKey . "ssl_truststore" }} +{{- if hasKey .ssl_truststore "truststore_password_env" }} +# Replace plugin-specific truststore password if it exists +PLUGIN_PASSWORD_VAR="{{ .ssl_truststore.truststore_password_env }}" +echo "DEBUG: Looking for password in environment variable: $PLUGIN_PASSWORD_VAR" +echo "DEBUG: Password value: ${!PLUGIN_PASSWORD_VAR}" +if [[ -n "${!PLUGIN_PASSWORD_VAR}" ]]; then + echo "DEBUG: Replacing \${${PLUGIN_PASSWORD_VAR}} with password in config" + CONFIG=$(echo "$CONFIG" | sed "s|\${${PLUGIN_PASSWORD_VAR}}|${!PLUGIN_PASSWORD_VAR}|g") + echo "DEBUG: Config after replacement:" + echo "$CONFIG" +else + echo "DEBUG: No password found in $PLUGIN_PASSWORD_VAR" +fi +{{- end }} +{{- end }} + + # Try to create connector with retry logic +max_retries=5 +retry_count=0 +while [ $retry_count -lt $max_retries ]; do + if create_or_update_connector "{{ .name }}" "$CONFIG" "$INSTALLED_CONNECTORS"; then + echo "Successfully created/updated connector {{ .name }} on attempt $((retry_count+1))" + break + else + retry_count=$((retry_count+1)) + if [ $retry_count -lt $max_retries ]; then + echo "Failed to create/update connector {{ .name }}, retrying in 10 seconds (attempt $retry_count/$max_retries)..." + sleep 10 + else + echo "Failed to create/update connector {{ .name }} after $max_retries attempts" + fi + fi +done + +# Wait a bit between connectors to allow Kafka Connect API to stabilize +sleep 3 + {{- end }} + {{- end }} + + echo "All Kafka connectors have been configured and started." +} + +# Signal handler for graceful shutdown +cleanup() { + echo "Received shutdown signal, stopping Kafka Connect..." + if [[ -n $KAFKA_PID ]]; then + kill -TERM $KAFKA_PID + wait $KAFKA_PID + fi + exit 0 +} + +# Set up signal handlers +trap cleanup SIGTERM SIGINT + +echo "Starting Kafka Connect distributed worker..." + +# Updating rest.advertised.host.name dynamically +POD_ID=$(echo "$POD_NAME" | rev | cut -d'-' -f 1 | rev) +WORKLOAD_NAME=$(echo $CPLN_WORKLOAD | sed 's|.*/workload/\([^/]*\)$|\1|') +cp /opt/kafka/config/connect-distributed.properties /opt/kafka/config/connect-distributed-updated.properties +echo "" >> /opt/kafka/config/connect-distributed-updated.properties +echo "rest.advertised.host.name=${WORKLOAD_NAME}-${POD_ID}.${WORKLOAD_NAME}" >> /opt/kafka/config/connect-distributed-updated.properties + +# Start the connector setup process in the background +setup_connectors & +SETUP_PID=$! + +# Start Kafka Connect in the foreground +echo "Starting Kafka Connect in foreground mode..." +exec /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed-updated.properties & +KAFKA_PID=$! + +# Wait for either process to finish +wait $KAFKA_PID +{{- end }} + + +{{/* Labeling */}} + +{{/* +Create chart name and version as used by the chart label. +*/}} +{{- define "kafka.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "kafka.tags" -}} +helm.sh/chart: {{ include "kafka.chart" . }} +{{ include "kafka.selectorLabels" . }} +{{- if .Chart.AppVersion }} +app.cpln.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.cpln.io/managed-by: {{ .Release.Service }} +cpln/marketplace: "true" +cpln/marketplace-template: kafka +cpln/marketplace-template-version: {{ .Chart.Version }} +{{- end }} + +{{/* +Selector labels +*/}} +{{- define "kafka.selectorLabels" -}} +app.cpln.io/name: {{ .Release.Name }} +app.cpln.io/instance: {{ .Release.Name }} +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/domain.yaml b/kafka/versions/3.5.0/templates/domain.yaml new file mode 100644 index 00000000..acf09fa9 --- /dev/null +++ b/kafka/versions/3.5.0/templates/domain.yaml @@ -0,0 +1,64 @@ +{{- include "kafka.validateOnePublicAddress" . }} +{{- include "kafka.validatedirectReplicaRoutingConfig" . }} +{{- range $key, $listener := .Values.kafka.listeners }} +{{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }} +--- +kind: domain +name: {{ $listener.directReplicaRouting.publicAddress }} +description: {{ $listener.directReplicaRouting.publicAddress }} +spec: + acceptAllHosts: false + acceptAllSubdomains: false + certChallengeType: dns01 + dnsMode: cname + ports: + - number: {{ $listener.directReplicaRouting.containerPort }} + protocol: tcp + routes: + - port: {{ $listener.directReplicaRouting.containerPort }} + prefix: / + workloadLink: //gvc/{{ $.Values.global.cpln.gvc }}/workload/{{ include "kafka.clusterName" $ }} + tls: + cipherSuites: + - ECDHE-ECDSA-AES256-GCM-SHA384 + - ECDHE-ECDSA-CHACHA20-POLY1305 + - ECDHE-ECDSA-AES128-GCM-SHA256 + - ECDHE-RSA-AES256-GCM-SHA384 + - ECDHE-RSA-CHACHA20-POLY1305 + - ECDHE-RSA-AES128-GCM-SHA256 + - AES256-GCM-SHA384 + - AES128-GCM-SHA256 + minProtocolVersion: TLSV1_2 + workloadLink: //gvc/{{ $.Values.global.cpln.gvc }}/workload/{{ include "kafka.clusterName" $ }} +{{- else if $listener.publicAddress }} +--- +kind: domain +name: {{ $listener.publicAddress }} +description: {{ $listener.publicAddress }} +spec: + acceptAllHosts: false + dnsMode: cname + ports: + {{- $replicaCount := $.Values.kafka.replicas | int }} + {{- range $i := until $replicaCount }} + - number: {{ add 3000 $i }} + protocol: tcp + routes: + - port: {{ add 3000 $i }} + prefix: / + replica: {{ $i }} + workloadLink: //gvc/{{ $.Values.global.cpln.gvc }}/workload/{{ include "kafka.clusterName" $ }} + tls: + cipherSuites: + - ECDHE-ECDSA-AES256-GCM-SHA384 + - ECDHE-ECDSA-CHACHA20-POLY1305 + - ECDHE-ECDSA-AES128-GCM-SHA256 + - ECDHE-RSA-AES256-GCM-SHA384 + - ECDHE-RSA-CHACHA20-POLY1305 + - ECDHE-RSA-AES128-GCM-SHA256 + - AES256-GCM-SHA384 + - AES128-GCM-SHA256 + minProtocolVersion: TLSV1_2 + {{- end }} +{{- end }} +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/identity.yaml b/kafka/versions/3.5.0/templates/identity.yaml new file mode 100644 index 00000000..571ed3b6 --- /dev/null +++ b/kafka/versions/3.5.0/templates/identity.yaml @@ -0,0 +1,4 @@ +kind: identity +name: {{ include "kafka.name" . }} +description: {{ include "kafka.clusterName" . }} identity +gvc: {{ .Values.global.cpln.gvc }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/kafbat-ui.yaml b/kafka/versions/3.5.0/templates/kafbat-ui.yaml new file mode 100644 index 00000000..992c5ba1 --- /dev/null +++ b/kafka/versions/3.5.0/templates/kafbat-ui.yaml @@ -0,0 +1,114 @@ +{{- if .Values.kafbat_ui.enabled }} +{{- if .Values.kafbat_ui.domain }} +kind: domain +name: {{ .Values.kafbat_ui.domain }} +description: {{ .Values.kafbat_ui.domain }} +spec: + acceptAllHosts: false + dnsMode: cname + ports: + - number: 443 + protocol: http2 + routes: + - port: 8080 + prefix: / + workloadLink: //gvc/{{ $.Values.global.cpln.gvc }}/workload/{{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} + tls: + cipherSuites: + - ECDHE-ECDSA-AES256-GCM-SHA384 + - ECDHE-ECDSA-CHACHA20-POLY1305 + - ECDHE-ECDSA-AES128-GCM-SHA256 + - ECDHE-RSA-AES256-GCM-SHA384 + - ECDHE-RSA-CHACHA20-POLY1305 + - ECDHE-RSA-AES128-GCM-SHA256 + - AES256-GCM-SHA384 + - AES128-GCM-SHA256 + minProtocolVersion: TLSV1_2 +--- +{{- end }} +kind: policy +name: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +description: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +tags: {{- include "kafka.tags" . | nindent 2 }} +bindings: + - permissions: + - reveal + principalLinks: + - //gvc/{{ $.Values.global.cpln.gvc }}/identity/{{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +targetKind: secret +targetLinks: + - //secret/{{ .Values.kafbat_ui.configuration_secret }} +--- +kind: identity +name: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +description: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +gvc: {{ $.Values.global.cpln.gvc }} +--- +kind: workload +name: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +description: {{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} +tags: + {{- if .Values.kafbat_ui.deletionProtection }} + cpln/protected: true + {{- end }} + {{- include "kafka.tags" . | nindent 2 }} +spec: + type: standard + containers: + - name: kafbat-ui + cpu: {{ .Values.kafbat_ui.cpu }} + {{- if .Values.kafbat_ui.minCpu }} + minCpu: '{{ .Values.kafbat_ui.minCpu }}' + {{- end }} + env: + - name: SPRING_CONFIG_ADDITIONAL-LOCATION + value: /etc/config.yaml + image: {{ .Values.kafbat_ui.image }} + inheritEnv: false + memory: {{ .Values.kafbat_ui.memory }} + {{- if .Values.kafbat_ui.minMemory }} + minMemory: {{ .Values.kafbat_ui.minMemory }} + {{- end }} + ports: + - number: 8080 + volumes: + - path: /etc/config.yaml + recoveryPolicy: retain + uri: cpln://secret/{{ .Values.kafbat_ui.configuration_secret }} + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: {{ .Values.kafbat_ui.replicas }} + metric: disabled + minScale: {{ .Values.kafbat_ui.replicas }} + scaleToZeroDelay: 300 + target: 100 + {{- if or .Values.kafbat_ui.minCpu .Values.kafbat_ui.minMemory }} + capacityAI: true + {{- else }} + capacityAI: false + {{- end }} + debug: false + suspend: false + timeoutSeconds: {{ .Values.kafbat_ui.timeoutSeconds }} +{{- if .Values.kafbat_ui.firewall }} + firewallConfig: + {{- if or (hasKey .Values.kafbat_ui.firewall "external_inboundAllowCIDR") (hasKey .Values.kafbat_ui.firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .Values.kafbat_ui.firewall.external_inboundAllowCIDR }}{{ .Values.kafbat_ui.firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .Values.kafbat_ui.firewall.external_outboundAllowCIDR }}{{ .Values.kafbat_ui.firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .Values.kafbat_ui.firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "[]" .Values.kafbat_ui.firewall.internal_inboundAllowType }} + {{- end }} +{{- end }} + identityLink: //gvc/{{ $.Values.global.cpln.gvc }}/identity/{{ include "kafka.name" $ }}-{{ .Values.kafbat_ui.name }} + loadBalancer: + direct: + enabled: false + ports: [] + securityOptions: + filesystemGroupId: 101 + supportDynamicTags: false +{{- end }} diff --git a/kafka/versions/3.5.0/templates/kafka-connectors.yaml b/kafka/versions/3.5.0/templates/kafka-connectors.yaml new file mode 100644 index 00000000..4d0b8300 --- /dev/null +++ b/kafka/versions/3.5.0/templates/kafka-connectors.yaml @@ -0,0 +1,224 @@ +{{- if .Values.kafka_connectors }} +{{- range .Values.kafka_connectors }} +{{- include "kafka.validateImage" . -}} +kind: policy +name: {{ include "kafka.name" $ }}-connect-{{ .name }} +description: {{ include "kafka.name" $ }}-connect-{{ .name }} +tags: {{- include "kafka.tags" $ | nindent 2 }} +bindings: + - permissions: + - reveal + principalLinks: + - //gvc/{{ $.Values.global.cpln.gvc }}/identity/{{ include "kafka.name" $ }}-connect-{{ .name }} +targetKind: secret +targetLinks: + - //secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-props + - //secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-init + - //secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-download + {{- if and .extraVolumes (ne (len .extraVolumes) 0) }} + {{- range .extraVolumes }} + {{- if contains "/secret/" .uri }} + {{- $secretName := regexReplaceAll ".*//secret/([^.]+).*" .uri "${1}" }} + - //secret/{{ $secretName }} + {{- end }} + {{- end }} + {{- end }} + {{- /* Check for secrets in ssl_truststore configuration values */ -}} + {{- if hasKey . "ssl_truststore" }} + {{- range $key, $value := .ssl_truststore }} + {{- if and (kindIs "string" $value) (contains "//secret/" $value) }} + {{- $secretName := regexReplaceAll ".*//secret/([^.]+).*" $value "${1}" }} + - //secret/{{ $secretName }} + {{- end }} + {{- end }} + {{- end }} + {{- /* Check for secrets in env variables */ -}} + {{- if hasKey . "env" }} + {{- range .env }} + {{- if and (hasKey . "value") (contains "//secret/" .value) }} + {{- $secretName := regexReplaceAll ".*//secret/([^.]+).*" .value "${1}" }} + - //secret/{{ $secretName }} + {{- end }} + {{- end }} + {{- end }} + {{- /* Check for secrets in connector_properties */ -}} + {{- if hasKey . "connector_properties" }} + {{- range $key, $value := .connector_properties }} + {{- if and (kindIs "string" $value) (contains "//secret/" $value) }} + {{- $secretName := regexReplaceAll ".*//secret/([^.]+).*" $value "${1}" }} + - //secret/{{ $secretName }} + {{- end }} + {{- end }} + {{- end }} +--- +kind: secret +name: {{ include "kafka.name" $ }}-connect-{{ .name }}-download +type: opaque +data: + encoding: plain + payload: | + {{- include "kafka.connectors.download.script" (dict "plugins" .plugins "plugins_folder" .plugins_folder "verbose" .verbose) | nindent 4 }} +--- +kind: secret +name: {{ include "kafka.name" $ }}-connect-{{ .name }}-init +type: opaque +data: + encoding: plain + payload: | + {{- include "kafka.connectors.run.script" (dict "plugins" .plugins "plugins_folder" .plugins_folder "verbose" .verbose) | nindent 4 }} +--- +kind: secret +name: {{ include "kafka.name" $ }}-connect-{{ .name }}-props +description: {{ include "kafka.name" $ }}-connect-{{ .name }}-props +tags: {{- include "kafka.tags" $ | nindent 2 }} +type: opaque +data: + encoding: plain + payload: |- + {{- if not (hasKey .connector_properties "bootstrap.servers") }} + bootstrap.servers={{ include "kafka.clientBootstrapAddress" $ }} + {{- end }} + {{- range $key, $value := .connector_properties }} + {{ $key }}={{ $value }} + {{- end }} +--- +kind: identity +name: {{ include "kafka.name" $ }}-connect-{{ .name }} +description: {{ include "kafka.name" $ }}-connect-{{ .name }} +gvc: {{ $.Values.global.cpln.gvc }} +--- +kind: volumeset +name: {{ include "kafka.name" $ }}-connect-{{ .name }} +description: {{ include "kafka.name" $ }}-connect-{{ .name }} +tags: {{- include "kafka.tags" $ | nindent 2 }} +spec: + fileSystemType: {{ dig "volumes" "fileSystemType" "ext4" . }} + initialCapacity: {{ dig "volumes" "initialCapacity" 10 . }} + performanceClass: {{ dig "volumes" "performanceClass" "general-purpose-ssd" . }} + {{- if and .volumes .volumes.customEncryption .volumes.customEncryption.enabled }} + customEncryption: + regions: + {{ .volumes.customEncryption.region }}: + keyId: '{{ .volumes.customEncryption.keyId }}' + {{- end }} + {{- if and .volumes .volumes.snapshots }} + snapshots: + createFinalSnapshot: {{ .volumes.snapshots.createFinalSnapshot | default true }} + retentionDuration: {{ .volumes.snapshots.retentionDuration | default "7d" }} + {{- if .volumes.snapshots.schedule }} + schedule: {{ .volumes.snapshots.schedule }} + {{- end }} + {{- else }} + snapshots: + createFinalSnapshot: true + retentionDuration: 7d + {{- end }} +--- +kind: workload +name: {{ include "kafka.name" $ }}-connect-{{ .name }} +description: {{ include "kafka.name" $ }}-connect-{{ .name }} +gvc: {{ $.Values.global.cpln.gvc }} +tags: + {{- if .deletionProtection }} + cpln/protected: true + {{- end }} + {{- include "kafka.tags" $ | nindent 2 }} +spec: + type: stateful + containers: + - name: kafka-connect + {{- if and .env (ne (len .env) 0) }} + env: + {{- toYaml .env | nindent 8 }} + {{- end }} + args: + - '-c' + - sleep 60 && cp /opt/kafka/init.sh /opt/kafka/init-run.sh && chmod +x /opt/kafka/init-run.sh && /opt/kafka/init-run.sh + command: /bin/bash + cpu: {{ .cpu }} + {{- if .minCpu }} + minCpu: {{ .minCpu }} + {{- end }} + image: {{ .image }} + inheritEnv: false + memory: {{ .memory }} + {{- if .minMemory }} + minMemory: {{ .minMemory }} + {{- end }} + ports: + - number: 8083 + protocol: http + volumes: + - path: /opt/kafka/plugins + recoveryPolicy: retain + uri: cpln://volumeset/{{ include "kafka.name" $ }}-connect-{{ .name }} + - path: /opt/kafka/config/connect-distributed.properties + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-props + - path: /opt/kafka/init.sh + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-init + {{- if .extraVolumes }} + {{- toYaml .extraVolumes | nindent 8 }} + {{- end }} + - name: plugins-downloader + args: + - '-c' + - cp /opt/kafka/download.sh /opt/kafka/download-run.sh && chmod +x /opt/kafka/download-run.sh && /opt/kafka/download-run.sh + command: /bin/sh + cpu: 80m + image: busybox:musl + inheritEnv: false + memory: 120Mi + ports: [] + volumes: + - path: /opt/kafka/plugins + recoveryPolicy: retain + uri: cpln://volumeset/{{ include "kafka.name" $ }}-connect-{{ .name }} + - path: /opt/kafka/download.sh + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" $ }}-connect-{{ .name }}-download + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: {{ .replicas }} + metric: cpu + minScale: {{ .replicas }} + scaleToZeroDelay: 300 + target: 100 + capacityAI: false + debug: false + {{- if .multiZone }} + multiZone: + enabled: true + {{- else }} + multiZone: + enabled: false + {{- end }} + suspend: false + timeoutSeconds: {{ .timeoutSeconds }} +{{- if .firewall }} + firewallConfig: + {{- if or (hasKey .firewall "external_inboundAllowCIDR") (hasKey .firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .firewall.external_inboundAllowCIDR }}{{ .firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .firewall.external_outboundAllowCIDR }}{{ .firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "none" .firewall.internal_inboundAllowType }} + {{- if hasKey .firewall "inboundAllowWorkload" }} + inboundAllowWorkload: {{ .firewall.inboundAllowWorkload | toYaml | nindent 8 }} + {{- end }} + {{- end }} +{{- end }} + identityLink: //gvc/{{ $.Values.global.cpln.gvc }}/identity/{{ include "kafka.name" $ }}-connect-{{ .name }} + loadBalancer: + direct: + enabled: false + ports: [] + securityOptions: + filesystemGroupId: 1001 + supportDynamicTags: false +{{- end }} +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/kafka-rest-proxy.yaml b/kafka/versions/3.5.0/templates/kafka-rest-proxy.yaml new file mode 100644 index 00000000..177f8c2b --- /dev/null +++ b/kafka/versions/3.5.0/templates/kafka-rest-proxy.yaml @@ -0,0 +1,169 @@ +{{- if .Values.kafka_rest_proxy.enabled }} +{{- if .Values.kafka_rest_proxy.password_properties }} +kind: secret +name: {{ include "kafka.name" . }}-rest-password-properties +description: {{ include "kafka.name" . }}-rest-password-properties +tags: {{- include "kafka.tags" . | nindent 2 }} +type: opaque +data: + encoding: plain + payload: |- + {{- range $key, $value := .Values.kafka_rest_proxy.password_properties }} + {{ $key }}: {{ $value }} + {{- end }} +{{- end }} +--- +kind: secret +name: {{ include "kafka.name" . }}-rest-properties +description: {{ include "kafka.name" . }}-rest-properties +tags: {{- include "kafka.tags" . | nindent 2 }} +type: opaque +data: + encoding: plain + payload: |- + {{- range $key, $value := .Values.kafka_rest_proxy.properties }} + {{ $key }}={{ $value }} + {{- end }} +--- +kind: secret +name: {{ include "kafka.name" . }}-rest-jaas-conf +description: {{ include "kafka.name" . }}-rest-jaas-conf +tags: {{- include "kafka.tags" . | nindent 2 }} +type: opaque +data: + encoding: plain + payload: >- + {{- .Values.kafka_rest_proxy.jaas_conf | nindent 4 }} +--- +kind: identity +name: {{ include "kafka.name" . }}-rest-proxy-identity +description: Identity for Kafka Rest Proxy {{ include "kafka.name" . }} +gvc: {{ .Values.global.cpln.gvc }} +--- +kind: policy +name: {{ include "kafka.name" . }}-rest-proxy-policy +origin: default +bindings: + - permissions: + - reveal + principalLinks: + - //gvc/{{ .Values.global.cpln.gvc }}/identity/{{ include "kafka.name" . }}-rest-proxy-identity +targetKind: secret +targetLinks: +{{- if .Values.kafka_rest_proxy.password_properties }} + - //secret/{{ include "kafka.name" . }}-rest-password-properties +{{- end }} + - //secret/{{ include "kafka.name" . }}-rest-properties + - //secret/{{ include "kafka.name" . }}-rest-jaas-conf +--- +kind: workload +name: {{ include "kafka.name" . }}-{{ .Values.kafka_rest_proxy.name }} +description: Kafka Rest Proxy +gvc: {{ .Values.global.cpln.gvc }} +tags: + {{- if .Values.kafka_rest_proxy.deletionProtection }} + cpln/protected: true + {{- end }} + cpln/marketplace: "true" + cpln/marketplace-template: kafka + cpln/marketplace-template-version: {{ .Chart.Version }} +spec: + type: standard + containers: + - name: rest-proxy + args: + - '-c' + - >- + KAFKAREST_OPTS="-Djava.security.auth.login.config=/etc/kafka-rest/kafka-rest.jaas.conf" + kafka-rest-start /etc/kafka-rest/kafka-rest.properties + command: /bin/bash + cpu: {{ .Values.kafka_rest_proxy.cpu }} + image: {{ .Values.kafka_rest_proxy.image }} + inheritEnv: false + memory: {{ .Values.kafka_rest_proxy.memory }} + {{- if and .Values.kafka_rest_proxy.capacityAI .Values.kafka_rest_proxy.capacityAI.enabled }} + {{- if .Values.kafka_rest_proxy.capacityAI.minCpu }} + minCpu: {{ .Values.kafka_rest_proxy.capacityAI.minCpu }} + {{- end }} + {{- if .Values.kafka_rest_proxy.capacityAI.minMemory }} + minMemory: {{ .Values.kafka_rest_proxy.capacityAI.minMemory }} + {{- end }} + {{- end }} + ports: + - number: 8082 + protocol: http + volumes: + {{- if .Values.kafka_rest_proxy.password_properties }} + - path: /etc/kafka-rest/password.properties + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" . }}-rest-password-properties + {{- end }} + - path: /etc/kafka-rest/kafka-rest.jaas.conf + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" . }}-rest-jaas-conf + - path: /etc/kafka-rest/kafka-rest.properties + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" . }}-rest-properties + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: {{ .Values.kafka_rest_proxy.replicas }} + metric: disabled + minScale: {{ .Values.kafka_rest_proxy.replicas }} + scaleToZeroDelay: 300 + target: 100 + capacityAI: {{ .Values.kafka_rest_proxy.capacityAI.enabled }} + debug: false + suspend: false + timeoutSeconds: {{ .Values.kafka_rest_proxy.timeoutSeconds }} +{{- if .Values.kafka_rest_proxy.firewall }} + firewallConfig: + {{- if or (hasKey .Values.kafka_rest_proxy.firewall "external_inboundAllowCIDR") (hasKey .Values.kafka_rest_proxy.firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .Values.kafka_rest_proxy.firewall.external_inboundAllowCIDR }}{{ .Values.kafka_rest_proxy.firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .Values.kafka_rest_proxy.firewall.external_outboundAllowCIDR }}{{ .Values.kafka_rest_proxy.firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .Values.kafka_rest_proxy.firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "[]" .Values.kafka_rest_proxy.firewall.internal_inboundAllowType }} + {{- if .Values.kafka_rest_proxy.firewall.inboundAllowWorkload }} + inboundAllowWorkload: {{ .Values.kafka_rest_proxy.firewall.inboundAllowWorkload | toYaml | nindent 8 }} + {{- end }} + {{- end }} +{{- end }} + identityLink: //identity/{{ include "kafka.name" . }}-rest-proxy-identity + loadBalancer: + direct: + enabled: false + ports: [] + securityOptions: + filesystemGroupId: 1000 + supportDynamicTags: false +{{ if .Values.kafka_rest_proxy.domain }} +--- +kind: domain +name: {{ .Values.kafka_rest_proxy.domain }} +description: {{ .Values.kafka_rest_proxy.domain }} +spec: + acceptAllHosts: false + dnsMode: cname + ports: + - number: 443 + protocol: http2 + routes: + - port: 8082 + prefix: / + workloadLink: //gvc/{{ .Values.global.cpln.gvc }}/workload/{{ include "kafka.name" . }}-{{ .Values.kafka_rest_proxy.name }} + tls: + cipherSuites: + - ECDHE-ECDSA-AES256-GCM-SHA384 + - ECDHE-ECDSA-CHACHA20-POLY1305 + - ECDHE-ECDSA-AES128-GCM-SHA256 + - ECDHE-RSA-AES256-GCM-SHA384 + - ECDHE-RSA-CHACHA20-POLY1305 + - ECDHE-RSA-AES128-GCM-SHA256 + - AES256-GCM-SHA384 + - AES128-GCM-SHA256 + minProtocolVersion: TLSV1_2 +{{- end }} +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/policy.yaml b/kafka/versions/3.5.0/templates/policy.yaml new file mode 100644 index 00000000..fb9a338b --- /dev/null +++ b/kafka/versions/3.5.0/templates/policy.yaml @@ -0,0 +1,16 @@ +kind: policy +name: {{ include "kafka.name" . }} +origin: default +bindings: + - permissions: + - reveal + principalLinks: + - //gvc/{{ .Values.global.cpln.gvc }}/identity/{{ include "kafka.name" . }} +targetKind: secret +targetLinks: + - //secret/{{ include "kafka.name" . }}-controller-configuration + - //secret/{{ include "kafka.name" . }}-init + - //secret/{{ include "kafka.name" . }}-secrets +{{- if .Values.jmx_exporter }} + - //secret/{{ include "kafka.name" . }}-jmx-exporter-conf +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/secret-controller-configuration.yaml b/kafka/versions/3.5.0/templates/secret-controller-configuration.yaml new file mode 100644 index 00000000..97da6557 --- /dev/null +++ b/kafka/versions/3.5.0/templates/secret-controller-configuration.yaml @@ -0,0 +1,77 @@ +kind: secret +name: {{ include "kafka.name" . }}-controller-configuration +type: opaque +data: + encoding: plain + payload: | + {{- include "kafka.validateReplicas" . }} + + # Listeners configuration + listeners-placeholder + advertised.listeners=INTERNAL://advertised-address-placeholder:9094,CONTROLLER://advertised-controller-address-placeholder:9093{{- range .Values.kafka.listeners }}{{- include "kafka.validateListenerConfig" . }},{{ .name | upper }}://advertised-{{ .name | lower }}-address-placeholder{{- end }} + listener.security.protocol.map=INTERNAL:SASL_PLAINTEXT,CONTROLLER:SASL_PLAINTEXT{{- range .Values.kafka.listeners }},{{ .name | upper }}:{{ .protocol }}{{- end }} + + # KRaft process roles + process.roles=process-roles-placeholder + + #node.id= + controller.listener.names=CONTROLLER + {{$replicaCount := int .Values.kafka.replicas -}} + {{- if eq $replicaCount 2 -}} + {{- fail "Invalid number of Kraft replicas: must not be 2" -}} + {{- end -}} + controller.quorum.voters= {{- $result := "" }} + {{- range $i := until $replicaCount }} + {{- if and (ge $i 0) (lt $i 5) }} + {{- if $i }} + {{- $result = print $result "," }} + {{- end }} + {{- $result = print $result (printf "%d@%s-%s-%d.%s-%s:9093" $i $.Release.Name $.Values.kafka.name $i $.Release.Name $.Values.kafka.name ) }} + {{- end }} + {{- end }} + {{- $result }} + + # Kraft Controller listener SASL settings + sasl.mechanism.controller.protocol=PLAIN + listener.name.controller.sasl.enabled.mechanisms=PLAIN + listener.name.controller.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="controller_user" password="controller-password-placeholder" user_controller_user="controller-password-placeholder"; + log.dirs={{ .Values.kafka.logDirs }} + sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512 + + # Interbroker configuration + inter.broker.listener.name=INTERNAL + sasl.mechanism.inter.broker.protocol=PLAIN + + # Listeners SASL JAAS configuration +{{- include "kafka.validateAdminExists" . }} +{{- range .Values.kafka.listeners }} + {{- include "kafka.validateAuthConfig" . }} + {{- if .sasl }} + {{- $adminConfig := "" }} + {{- if .sasl.admin }} + {{- $adminConfig = printf "user_%s=\"%s\"" .sasl.admin.username .sasl.admin.password }} + {{- end }} + listener.name.{{ .name | lower }}.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required {{- if $adminConfig }} {{ $adminConfig }}{{- end }}{{- $users := .sasl.users | split "," }}{{- $passwords := .sasl.passwords | split "," }}{{- range $index, $user := $users }}{{- $password := index $passwords $index }} user_{{ $user }}="{{ $password }}"{{- end }}; + listener.name.{{ .name | lower }}.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required {{- if $adminConfig }} {{ $adminConfig }}{{- end }}{{- range $index, $user := $users }}{{- $password := index $passwords $index }} user_{{ $user }}="{{ $password }}"{{- end }}; + listener.name.{{ .name | lower }}.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required {{- if $adminConfig }} {{ $adminConfig }}{{- end }}{{- range $index, $user := $users }}{{- $password := index $passwords $index }} user_{{ $user }}="{{ $password }}"{{- end }}; + {{- end }} +{{- end }} + listener.name.internal.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="inter_broker_user" password="interbroker-password-placeholder" user_inter_broker_user="interbroker-password-placeholder"{{- range .Values.kafka.listeners }}{{- if and .sasl .sasl.admin }} user_{{ .sasl.admin.username }}="{{ .sasl.admin.password }}"{{- break }}{{- end }}{{- end }}; + listener.name.internal.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="inter_broker_user" password="interbroker-password-placeholder"; + listener.name.internal.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="inter_broker_user" password="interbroker-password-placeholder"; + # End of SASL JAAS configuration + + + {{- if .Values.kafka.acl }} + + # Enable ACL + authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer + super.users=User:controller_user;User:inter_broker_user{{- if .Values.kafka.acl.superUsers }};{{ .Values.kafka.acl.superUsers }}{{- end }} + allow.everyone.if.no.acl.found={{ .Values.kafka.acl.allowEveryoneIfNoAclFound | default "false" }} + # End of ACL configuration + {{- end }} + + # Extra configurations + {{- range $key, $value := .Values.kafka.extra_configurations }} + {{ $key }}={{ $value }} + {{- end }} diff --git a/kafka/versions/3.5.0/templates/secret-init.yaml b/kafka/versions/3.5.0/templates/secret-init.yaml new file mode 100644 index 00000000..d7f9ae4a --- /dev/null +++ b/kafka/versions/3.5.0/templates/secret-init.yaml @@ -0,0 +1,184 @@ +{{- include "kafka.validatedirectReplicaRoutingConfig" . }} +kind: secret +name: {{ include "kafka.name" . }}-init +type: opaque +data: + encoding: plain + payload: | + #!/bin/bash + + set -o errexit + set -o nounset + set -o pipefail + + WORKLOAD_NAME=$(echo $CPLN_WORKLOAD | sed 's|.*/workload/\([^/]*\)$|\1|') + + error(){ + local message="${1:?missing message}" + echo "ERROR: ${message}" + exit 1 + } + + retry_while() { + local -r cmd="${1:?cmd is missing}" + local -r retries="${2:-12}" + local -r sleep_time="${3:-5}" + local return_value=1 + + read -r -a command <<< "$cmd" + for ((i = 1 ; i <= retries ; i+=1 )); do + "${command[@]}" && return_value=0 && break + sleep "$sleep_time" + done + return $return_value + } + + replace_in_file() { + local filename="${1:?filename is required}" + local match_regex="${2:?match regex is required}" + local substitute_regex="${3:?substitute regex is required}" + local posix_regex=${4:-true} + + local result + + # We should avoid using 'sed in-place' substitutions + # 1) They are not compatible with files mounted from ConfigMap(s) + # 2) We found incompatibility issues with Debian10 and "in-place" substitutions + local -r del=$'\001' # Use a non-printable character as a 'sed' delimiter to avoid issues + if [[ $posix_regex = true ]]; then + result="$(sed -E "s${del}${match_regex}${del}${substitute_regex}${del}g" "$filename")" + else + result="$(sed "s${del}${match_regex}${del}${substitute_regex}${del}g" "$filename")" + fi + echo "$result" > "$filename" + } + + kafka_conf_set() { + local file="${1:?missing file}" + local key="${2:?missing key}" + local value="${3:?missing value}" + + # Check if the value was set before + if grep -q "^[#\\s]*$key\s*=.*" "$file"; then + # Update the existing key + replace_in_file "$file" "^[#\\s]*${key}\s*=.*" "${key}=${value}" false + else + # Add a new key + printf '\n%s=%s' "$key" "$value" >>"$file" + fi + } + + replace_placeholder() { + local placeholder="${1:?missing placeholder value}" + local password="${2:?missing password value}" + sed -i "s|$placeholder|$password|g" "$KAFKA_CONFIG_FILE" + } + + configure_external_access() { + # Configure external hostname + if [[ -f "/shared/external-host.txt" ]]; then + host=$(cat "/shared/external-host.txt") + elif [[ -n "${EXTERNAL_ACCESS_HOST:-}" ]]; then + host="$EXTERNAL_ACCESS_HOST" + elif [[ -n "${EXTERNAL_ACCESS_HOSTS_LIST:-}" ]]; then + read -r -a hosts <<<"$(tr ',' ' ' <<<"${EXTERNAL_ACCESS_HOSTS_LIST}")" + host="${hosts[$POD_ID]}" + elif [[ "$EXTERNAL_ACCESS_HOST_USE_PUBLIC_IP" =~ ^(yes|true)$ ]]; then + host=$(curl -s https://ipinfo.io/ip) + else + error "External access hostname not provided" + fi + + # Configure external port + if [[ -f "/shared/external-port.txt" ]]; then + port=$(cat "/shared/external-port.txt") + elif [[ -n "${EXTERNAL_ACCESS_PORT:-}" ]]; then + if [[ "${EXTERNAL_ACCESS_PORT_AUTOINCREMENT:-}" =~ ^(yes|true)$ ]]; then + port="$((EXTERNAL_ACCESS_PORT + POD_ID))" + else + port="$EXTERNAL_ACCESS_PORT" + fi + elif [[ -n "${EXTERNAL_ACCESS_PORTS_LIST:-}" ]]; then + read -r -a ports <<<"$(tr ',' ' ' <<<"${EXTERNAL_ACCESS_PORTS_LIST}")" + port="${ports[$POD_ID]}" + else + error "External access port not provided" + fi + # Configure Kafka advertised listeners + sed -i -E "s|^(advertised\.listeners=\S+)$|\1,EXTERNAL://${host}:${port}|" "$KAFKA_CONFIG_FILE" + } + + configure_kafka_sasl() { + + # Replace placeholders with passwords + replace_placeholder "interbroker-password-placeholder" "$KAFKA_INTER_BROKER_PASSWORD" + replace_placeholder "controller-password-placeholder" "$KAFKA_CONTROLLER_PASSWORD" + } + + export KAFKA_CONFIG_FILE=${KAFKA_CONFIG_FILE:-/mnt/shared/config/server.properties} + cp /configmaps/server.properties $KAFKA_CONFIG_FILE + + # Get pod ID and role, last and second last fields in the pod name respectively + POD_ID=$(echo "$POD_NAME" | rev | cut -d'-' -f 1 | rev) + export KAFKA_CFG_NODE_ID="$POD_ID" + LOCATION_NAME=$(echo "$CPLN_LOCATION" | sed 's|.*/location/\([^/]*\)$|\1|') + + # Configure POD Role + if [ "$POD_ID" -le 4 ]; then + replace_placeholder "process-roles-placeholder" "controller,broker" + replace_placeholder "advertised-controller-address-placeholder" "${POD_NAME}.${WORKLOAD_NAME}.${CPLN_GVC_ALIAS}.svc.cluster.local" + replace_placeholder "listeners-placeholder" "listeners=INTERNAL://:9094,CONTROLLER://:9093{{- range $i, $key := keys .Values.kafka.listeners | sortAlpha }} + {{- $listener := index $.Values.kafka.listeners $key -}} + {{- include "kafka.validateListenerConfig" $listener -}},{{ $listener.name | upper }}://:{{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }}{{ $listener.directReplicaRouting.containerPort }}{{- else if $listener.publicAddress }}300${POD_ID}{{- else }}{{ $listener.containerPort }}{{- end }}{{- end }}" + else + replace_placeholder "process-roles-placeholder" "broker" + replace_placeholder ",CONTROLLER://advertised-controller-address-placeholder:9093" "" + replace_placeholder "listeners-placeholder" "listeners=INTERNAL://:9094{{- range $i, $key := keys .Values.kafka.listeners | sortAlpha }} + {{- $listener := index $.Values.kafka.listeners $key -}} + {{- include "kafka.validateListenerConfig" $listener -}},{{ $listener.name | upper }}://:{{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }}{{ $listener.directReplicaRouting.containerPort }}{{- else if $listener.publicAddress }}300${POD_ID}{{- else }}{{ $listener.containerPort }}{{- end }}{{- end }}" + fi + + # Configure node.id and/or broker.id + ID=$((POD_ID + KAFKA_MIN_ID)) + kafka_conf_set "$KAFKA_CONFIG_FILE" "node.id" "$ID" + + replace_placeholder "advertised-address-placeholder" "${POD_NAME}.${WORKLOAD_NAME}.${CPLN_GVC_ALIAS}.svc.cluster.local" + + {{- range $key, $listener := .Values.kafka.listeners }} + {{- include "kafka.validateListenerConfig" . }} + {{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }} + replace_placeholder "advertised-{{ $listener.name | lower }}-address-placeholder" "${POD_NAME}-${LOCATION_NAME}.{{ $listener.directReplicaRouting.publicAddress }}:{{ $listener.directReplicaRouting.containerPort }}" + {{- else if $listener.publicAddress }} + replace_placeholder "advertised-{{ $listener.name | lower }}-address-placeholder" "{{ $listener.publicAddress }}:300${POD_ID}" + {{- else }} + replace_placeholder "advertised-{{ $listener.name | lower }}-address-placeholder" "${POD_NAME}.${WORKLOAD_NAME}.${CPLN_GVC_ALIAS}.svc.cluster.local:{{ .containerPort }}" + {{- end }} + {{- end }} + + if [[ "${EXTERNAL_ACCESS_ENABLED:-false}" =~ ^(yes|true)$ ]]; then + configure_external_access + fi + + configure_kafka_sasl + + # Initialize log directories for Apache Kafka + {{- $root := . -}} + {{- $logDirs := split "," $root.Values.kafka.logDirs }} + {{- $counter := 0 }} + {{- range $path := $logDirs }} + # Create log directory if it doesn't exist + mkdir -p {{ $path }} + + # Remove lost+found if it exists (common with mounted volumes) + rm -rf {{ $path }}/lost+found 2>/dev/null || true + + # Ensure proper ownership for Apache Kafka (runs as appuser) + chown -R $(id -u):$(id -g) {{ $path }} 2>/dev/null || true + {{- $counter = add $counter 1 }} + {{- end }} + + # Ensure data directory exists (Apache Kafka default) + mkdir -p /var/lib/kafka/data + chown -R $(id -u):$(id -g) /var/lib/kafka/data 2>/dev/null || true + + /etc/kafka/docker/run diff --git a/kafka/versions/3.5.0/templates/secret-secrets.yaml b/kafka/versions/3.5.0/templates/secret-secrets.yaml new file mode 100644 index 00000000..86060a2d --- /dev/null +++ b/kafka/versions/3.5.0/templates/secret-secrets.yaml @@ -0,0 +1,12 @@ +kind: secret +name: {{ include "kafka.name" . }}-secrets +type: dictionary +data: + kraft-cluster-id: {{ .Values.kafka.secrets.kraft_cluster_id }} + {{- range $key, $listener := .Values.kafka.listeners }} + {{- if and $listener.sasl $listener.sasl.admin }} + {{ $listener.name | lower }}-admin-password: {{ $listener.sasl.admin.password }} + {{- end }} + {{- end }} + inter-broker-password: {{ .Values.kafka.secrets.inter_broker_password }} + controller-password: {{ .Values.kafka.secrets.controller_password }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/volumesets.yaml b/kafka/versions/3.5.0/templates/volumesets.yaml new file mode 100644 index 00000000..dba17d86 --- /dev/null +++ b/kafka/versions/3.5.0/templates/volumesets.yaml @@ -0,0 +1,31 @@ +{{- $root := . -}} +{{- $logDirs := split "," $root.Values.kafka.logDirs }} +{{- $counter := 0 }} +{{- range $index, $path := $logDirs }} +kind: volumeset +name: {{ include "kafka.name" $root }}-logs-{{ $counter }} +description: {{ include "kafka.name" $root }} logs {{ $counter }} +gvc: {{ $root.Values.global.cpln.gvc }} +spec: + initialCapacity: {{ $root.Values.kafka.volumes.logs.initialCapacity }} + performanceClass: {{ $root.Values.kafka.volumes.logs.performanceClass }} + fileSystemType: {{ $root.Values.kafka.volumes.logs.fileSystemType }} + {{- if and $root.Values.kafka.volumes.logs.customEncryption $root.Values.kafka.volumes.logs.customEncryption.enabled }} + customEncryption: + regions: + {{ $root.Values.kafka.volumes.logs.customEncryption.region }}: + keyId: '{{ $root.Values.kafka.volumes.logs.customEncryption.keyId }}' + {{- end }} + autoscaling: + maxCapacity: {{ $root.Values.kafka.volumes.logs.autoscaling.maxCapacity }} + minFreePercentage: {{ $root.Values.kafka.volumes.logs.autoscaling.minFreePercentage }} + scalingFactor: {{ $root.Values.kafka.volumes.logs.autoscaling.scalingFactor }} +{{- if $root.Values.kafka.volumes.logs.snapshots }} + snapshots: + createFinalSnapshot: {{ $root.Values.kafka.volumes.logs.snapshots.createFinalSnapshot }} + retentionDuration: {{ $root.Values.kafka.volumes.logs.snapshots.retentionDuration }} + schedule: {{ $root.Values.kafka.volumes.logs.snapshots.schedule }} +{{- end }} +--- +{{- $counter = add $counter 1 }} +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/workload-kafka-client.yaml b/kafka/versions/3.5.0/templates/workload-kafka-client.yaml new file mode 100644 index 00000000..08a0a78a --- /dev/null +++ b/kafka/versions/3.5.0/templates/workload-kafka-client.yaml @@ -0,0 +1,46 @@ +{{- if .Values.kafka_client }} +kind: workload +name: {{ include "kafka.name" . }}-{{ .Values.kafka_client.name }} +gvc: {{ .Values.global.cpln.gvc }} +spec: + type: standard + containers: + - name: kafka + args: + - '-c' + - sleep infinity + command: /bin/bash + cpu: {{ .Values.kafka_client.cpu }} + image: {{ .Values.kafka_client.image }} + inheritEnv: false + memory: {{ .Values.kafka_client.memory }} + ports: + - number: 9092 + protocol: tcp + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: 3 + metric: cpu + minScale: 1 + scaleToZeroDelay: 300 + target: 100 + capacityAI: false + debug: false + suspend: false + timeoutSeconds: 5 +{{- if .Values.kafka_client.firewall }} + firewallConfig: + {{- if or (hasKey .Values.kafka_client.firewall "external_inboundAllowCIDR") (hasKey .Values.kafka_client.firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .Values.kafka_client.firewall.external_inboundAllowCIDR }}{{ .Values.kafka_client.firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .Values.kafka_client.firewall.external_outboundAllowCIDR }}{{ .Values.kafka_client.firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .Values.kafka_client.firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "[]" .Values.kafka_client.firewall.internal_inboundAllowType }} + {{- end }} +{{- end }} + localOptions: [] + supportDynamicTags: false +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/workload-kafka-cluster.yaml b/kafka/versions/3.5.0/templates/workload-kafka-cluster.yaml new file mode 100644 index 00000000..44733ff6 --- /dev/null +++ b/kafka/versions/3.5.0/templates/workload-kafka-cluster.yaml @@ -0,0 +1,258 @@ +{{- include "kafka.validateKafkaImage" . -}} +{{- include "kafka.validatedirectReplicaRoutingConfig" . -}} +{{- if .Values.jmx_exporter }} +kind: secret +name: {{ include "kafka.name" . }}-jmx-exporter-conf +type: opaque +data: + encoding: plain + payload: |- + {{ .Values.jmx_exporter.config | toYaml | nindent 4 }} +--- +{{- end }} +kind: workload +name: {{ include "kafka.clusterName" . }} +gvc: {{ .Values.global.cpln.gvc }} +tags: + {{- if .Values.kafka.deletionProtection }} + cpln/protected: true + {{- end }} + {{- include "kafka.tags" . | nindent 2 }} +spec: + type: stateful + containers: + - name: kafka + args: + - '-c' + - >- + cp /scripts/kafka-init.sh /tmp/ && chmod +x /tmp/kafka-init.sh && + /tmp/kafka-init.sh + command: /bin/bash + cpu: '{{ .Values.kafka.cpu }}' + {{- if .Values.kafka.minCpu }} + minCpu: '{{ .Values.kafka.minCpu }}' + {{- end }} + env: + {{- if .Values.kafka.env }} +{{ toYaml .Values.kafka.env | indent 8 }} + {{- end }} + {{- if .Values.jmx_exporter }} + - name: JMX_PORT + value: {{ .Values.jmx_exporter.kafkaJmxPort | quote }} + {{- end }} + - name: KAFKA_CONTROLLER_PASSWORD + value: 'cpln://secret/{{ include "kafka.name" . }}-secrets.controller-password' + - name: KAFKA_CONTROLLER_USER + value: controller_user + - name: KAFKA_HEAP_OPTS + value: "{{ .Values.kafka.overrideHeapOpts | default (include "kafka.heap.opts" .) | trim }}" + - name: KAFKA_INTER_BROKER_PASSWORD + value: 'cpln://secret/{{ include "kafka.name" . }}-secrets.inter-broker-password' + - name: KAFKA_INTER_BROKER_USER + value: inter_broker_user + - name: KAFKA_KRAFT_BOOTSTRAP_SCRAM_USERS + value: 'true' + - name: CLUSTER_ID + value: 'cpln://secret/{{ include "kafka.name" . }}-secrets.kraft-cluster-id' + - name: KAFKA_MIN_ID + value: '0' + image: {{ .Values.kafka.image }} + inheritEnv: false + livenessProbe: + failureThreshold: 5 + initialDelaySeconds: 60 + periodSeconds: 15 + successThreshold: 1 + tcpSocket: + port: 9093 + timeoutSeconds: 15 + memory: {{ .Values.kafka.memory }} + {{- if .Values.kafka.minMemory }} + minMemory: {{ .Values.kafka.minMemory }} + {{- end }} + ports: +{{ range $key, $listener := .Values.kafka.listeners }} +{{- include "kafka.validateListenerConfig" $listener }} +{{- if and $listener.directReplicaRouting $listener.directReplicaRouting.enabled }} + - number: {{ $listener.directReplicaRouting.containerPort }} + protocol: tcp +{{- else if $listener.publicAddress }} + {{- $startPort := 3000 }} + {{- $replicas := $.Values.kafka.replicas | int }} + {{- range $replicaIndex := until $replicas }} + - number: {{ add $startPort $replicaIndex }} + protocol: tcp + {{- end }} +{{- else }} + - number: {{ $listener.containerPort }} + protocol: tcp +{{- end }} +{{- end }} + - number: 9093 + protocol: tcp + - number: 9094 + protocol: tcp +{{- if .Values.jmx_exporter }} + - number: {{ .Values.jmx_exporter.kafkaJmxPort }} + protocol: tcp +{{- end }} + readinessProbe: + failureThreshold: 20 + initialDelaySeconds: 20 + periodSeconds: 10 + successThreshold: 6 + tcpSocket: + port: 9093 + timeoutSeconds: 5 + volumes: + {{- $root := . -}} + {{- $logDirs := split "," $root.Values.kafka.logDirs }} + {{- $counter := 0 }} + {{- range $path := $logDirs }} + - path: {{ $path | trim }} + recoveryPolicy: retain + uri: 'cpln://volumeset/{{ include "kafka.name" $root }}-logs-{{ $counter }}' + {{- $counter = add $counter 1 }} + {{- end }} + - path: /configmaps/server.properties + recoveryPolicy: retain + uri: 'cpln://secret/{{ include "kafka.name" $root }}-controller-configuration' + - path: /scripts/kafka-init.sh + recoveryPolicy: retain + uri: 'cpln://secret/{{ include "kafka.name" $root }}-init' +{{- if .Values.kafka_exporter }} + - name: kafka-exporter + args: + - '-c' + - >- +{{- $listenerName := .Values.kafka_exporter.listener }} +{{- if not (hasKey .Values.kafka.listeners $listenerName) }} + {{- fail (printf "Error: Listener '%s' specified in kafka_exporter.listener does not exist" $listenerName) }} +{{- end }} +{{- $listener := index .Values.kafka.listeners $listenerName }} +{{- $port := 3000 }} +{{- if $listener.containerPort }} + {{- $port = $listener.containerPort }} +{{- else }} + {{- $port = "$(echo $((3000 + $POD_ID)))" }} +{{- end }} +{{- if eq $listener.protocol "SASL_PLAINTEXT" }} + {{- if not (and $listener.sasl $listener.sasl.admin) }} + {{- fail (printf "Error: SASL_PLAINTEXT listener '%s' must have sasl.admin configured for kafka_exporter" $listenerName) }} + {{- end }} + sleep 60 && POD_ID=$(echo "$POD_NAME" | rev | cut -d'-' -f 1 | rev) && kafka_exporter --kafka.server=localhost:{{ if not $listener.containerPort }}$(echo $((3000 + $POD_ID))){{ else }}{{ $port }}{{ end }} + --sasl.enabled --sasl.username={{ $listener.sasl.admin.username }} --sasl.mechanism=plain + --sasl.password=${KAFKA_CLIENT_PASSWORDS} --web.listen-address=:9308 +{{- else if eq $listener.protocol "PLAINTEXT" }} + sleep 60 && POD_ID=$(echo "$POD_NAME" | rev | cut -d'-' -f 1 | rev) && kafka_exporter --kafka.server=localhost:{{ if not $listener.containerPort }}$(echo $((3000 + $POD_ID))){{ else }}{{ $port }}{{ end }} + --no-sasl.handshake --web.listen-address=:9308 +{{- else }} + sleep 60 && POD_ID=$(echo "$POD_NAME" | rev | cut -d'-' -f 1 | rev) && kafka_exporter --kafka.server=localhost:{{ if not $listener.containerPort }}$(echo $((3000 + $POD_ID))){{ else }}{{ $port }}{{ end }} + --no-sasl.handshake --web.listen-address=:9308 +{{- end }} + command: /bin/sh + cpu: {{ .Values.kafka_exporter.cpu }} + metrics: + path: /metrics + port: 9308 + dropMetrics: {{- if .Values.kafka_exporter.dropMetrics }}{{ .Values.kafka_exporter.dropMetrics | toYaml | nindent 8 }}{{- else }} []{{- end }} + env: +{{- if .Values.kafka_exporter.env }} +{{ toYaml .Values.kafka_exporter.env | indent 8 }} +{{- end }} +{{- $listenerName := .Values.kafka_exporter.listener }} +{{- if not (hasKey .Values.kafka.listeners $listenerName) }} + {{- fail (printf "Error: Listener '%s' specified in kafka_exporter.listener does not exist" $listenerName) }} +{{- end }} +{{- $listener := index .Values.kafka.listeners $listenerName }} +{{- if eq $listener.protocol "SASL_PLAINTEXT" }} + - name: KAFKA_CLIENT_PASSWORDS + value: 'cpln://secret/{{ include "kafka.name" $ }}-secrets.{{ $listener.name | lower }}-admin-password' +{{- end }} + image: {{ .Values.kafka_exporter.image }} + inheritEnv: false + memory: {{ .Values.kafka_exporter.memory }} + ports: + - number: 9308 + protocol: tcp +{{- end }} +{{- if .Values.jmx_exporter }} + - name: jmx-exporter + command: java + args: + - -XX:MaxRAMPercentage=100 + - -XshowSettings:vm + - -jar + - jmx_prometheus_standalone.jar + - {{ .Values.jmx_exporter.exporterPort | quote }} + - /etc/jmx-kafka/jmx-kafka-prometheus.yml + cpu: {{ .Values.jmx_exporter.cpu }} + {{- if .Values.jmx_exporter.minCpu }} + minCpu: '{{ .Values.jmx_exporter.minCpu }}' + {{- end }} + metrics: + path: /metrics + port: {{ .Values.jmx_exporter.exporterPort }} + dropMetrics: {{- if .Values.jmx_exporter.dropMetrics }}{{ .Values.jmx_exporter.dropMetrics | toYaml | nindent 8 }}{{- else }} []{{- end }} + image: {{ .Values.jmx_exporter.image }} + inheritEnv: false + memory: {{ .Values.jmx_exporter.memory }} + {{- if .Values.jmx_exporter.minMemory }} + minMemory: {{ .Values.jmx_exporter.minMemory }} + {{- end }} + ports: + - number: {{ .Values.jmx_exporter.exporterPort }} + protocol: tcp + volumes: + - path: /etc/jmx-kafka/jmx-kafka-prometheus.yml + recoveryPolicy: retain + uri: cpln://secret/{{ include "kafka.name" . }}-jmx-exporter-conf +{{- end }} + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: {{ .Values.kafka.replicas }} + metric: disabled + minScale: {{ .Values.kafka.replicas }} + scaleToZeroDelay: 300 + target: 95 + capacityAI: false + debug: false + {{- if .Values.kafka.multiZone }} + multiZone: + enabled: true + {{- else }} + multiZone: + enabled: false + {{- end }} + suspend: {{ .Values.kafka.suspend }} + timeoutSeconds: 30 +{{- if .Values.kafka.firewall }} + firewallConfig: + {{- if or (hasKey .Values.kafka.firewall "external_inboundAllowCIDR") (hasKey .Values.kafka.firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .Values.kafka.firewall.external_inboundAllowCIDR }}{{ .Values.kafka.firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .Values.kafka.firewall.external_outboundAllowCIDR }}{{ .Values.kafka.firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .Values.kafka.firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "[]" .Values.kafka.firewall.internal_inboundAllowType }} + {{- if .Values.kafka.firewall.inboundAllowWorkload }} + inboundAllowWorkload: {{ .Values.kafka.firewall.inboundAllowWorkload | toYaml | nindent 8 }} + {{- end }} + {{- end }} +{{- end }} + loadBalancer: + direct: + enabled: false + ports: [] + replicaDirect: true + identityLink: //identity/{{ include "kafka.name" . }} + rolloutOptions: + maxSurgeReplicas: 25% + maxUnavailableReplicas: '1' + minReadySeconds: {{ .Values.kafka.minReadySeconds }} + scalingPolicy: Parallel + securityOptions: + filesystemGroupId: 1001 + supportDynamicTags: false \ No newline at end of file diff --git a/kafka/versions/3.5.0/templates/workload-kafka-ui.yaml b/kafka/versions/3.5.0/templates/workload-kafka-ui.yaml new file mode 100644 index 00000000..dc2726df --- /dev/null +++ b/kafka/versions/3.5.0/templates/workload-kafka-ui.yaml @@ -0,0 +1,66 @@ +{{- if and .Values.kafka_ui .Values.kafka_ui.enabled }} +kind: workload +name: {{ include "kafka.name" . }}-{{ .Values.kafka_ui.name }} +description: kafka-ui +gvc: {{ .Values.global.cpln.gvc }} +spec: + type: standard + containers: + - name: kafka-ui + cpu: {{ .Values.kafka_ui.cpu }} + env: + - name: KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS + value: "{{- $replicas := int .Values.kafka.replicas -}}{{- $bootstrapServers := list -}}{{- range $i := until $replicas -}}{{- if $i -}},{{- end -}}{{- printf "%s-%s-%d.%s-%s:9092" $.Release.Name $.Values.kafka.name $i $.Release.Name $.Values.kafka.name -}}{{- end }}" + - name: KAFKA_CLUSTERS_0_NAME + value: {{ include "kafka.name" . }} + - name: KAFKA_CLUSTERS_0_PROPERTIES_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM + value: '' + - name: LOGGING_LEVEL_ROOT + value: INFO +{{- $listenerName := .Values.kafka_ui.listener }} +{{- $listener := index .Values.kafka.listeners $listenerName }} +{{- if eq $listener.protocol "SASL_PLAINTEXT" }} + {{- if not (and $listener.sasl $listener.sasl.admin) }} + {{- fail (printf "Error: SASL_PLAINTEXT listener '%s' must have sasl.admin configured for kafka_exporter" $listenerName) }} + {{- end }} + - name: KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL + value: {{ $listener.protocol }} + - name: KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM + value: PLAIN + - name: KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG + value: >- + org.apache.kafka.common.security.plain.PlainLoginModule required username="{{ $listener.sasl.admin.username }}" password="{{ $listener.sasl.admin.password }}"; +{{- end }} + image: 'provectuslabs/kafka-ui:latest' + inheritEnv: false + memory: {{ .Values.kafka_ui.memory }} + ports: + - number: 8080 + protocol: http + defaultOptions: + autoscaling: + maxConcurrency: 0 + maxScale: 1 + metric: cpu + minScale: 1 + scaleToZeroDelay: 300 + target: 100 + capacityAI: false + debug: false + suspend: false + timeoutSeconds: 5 +{{- if .Values.kafka_ui.firewall }} + firewallConfig: + {{- if or (hasKey .Values.kafka_ui.firewall "external_inboundAllowCIDR") (hasKey .Values.kafka_ui.firewall "external_outboundAllowCIDR") }} + external: + inboundAllowCIDR: {{- if .Values.kafka_ui.firewall.external_inboundAllowCIDR }}{{ .Values.kafka_ui.firewall.external_inboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + outboundAllowCIDR: {{- if .Values.kafka_ui.firewall.external_outboundAllowCIDR }}{{ .Values.kafka_ui.firewall.external_outboundAllowCIDR | splitList "," | toYaml | nindent 8 }}{{- else }} []{{- end }} + {{- end }} + {{- if hasKey .Values.kafka_ui.firewall "internal_inboundAllowType" }} + internal: + inboundAllowType: {{ default "[]" .Values.kafka_ui.firewall.internal_inboundAllowType }} + {{- end }} +{{- end }} + localOptions: [] + supportDynamicTags: false +{{- end }} \ No newline at end of file diff --git a/kafka/versions/3.5.0/values.yaml b/kafka/versions/3.5.0/values.yaml new file mode 100644 index 00000000..43e448ce --- /dev/null +++ b/kafka/versions/3.5.0/values.yaml @@ -0,0 +1,483 @@ +kafka: + name: cluster + image: apache/kafka:3.9.1 + suspend: false + deletionProtection: false + replicas: 3 # must not be 2 + minReadySeconds: 0 + debug: false + multiZone: false # If true: It's recommended to enable multi-zone on the Dedicated Load Balancer setting on GVC to reduce the cross-zone traffic + logDirs: /opt/kafka/logs-0,/opt/kafka/logs-1 + env: [] # If you need to set environment variables, add them here + volumes: + logs: + initialCapacity: 10 # In GB + performanceClass: general-purpose-ssd # general-purpose-ssd / high-throughput-ssd (Min 1000GB) + fileSystemType: ext4 # ext4 / xfs + snapshots: + createFinalSnapshot: true + retentionDuration: 7d + schedule: 0 0 * * * # UTC + autoscaling: + maxCapacity: 1000 # In GB + minFreePercentage: 20 + scalingFactor: 1.2 + # customEncryption: + # enabled: false + # region: aws-us-east-2 # Replace with the appropriate region + # keyId: arn:aws:kms:us-east-2:1234567890:key/d411f35a-1d31-4515-9934-4f193e042d80 # Replace with your AWS KMS key ARN + cpu: 1000m # For millicores us 'm' like 500m + memory: 2000Mi # Gi / Mi + minCpu: 250m # For millicores us 'm' like 500m + minMemory: 2000Mi # Gi / Mi + # overrideHeapOpts: "-Xmx1024m -Xms1024m" # Override the default heap Options settings + # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall + firewall: + internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc(Recommended) + # external_inboundAllowCIDR: 0.0.0.0/0 # Provide a comma-separated list + # # You can specify additional workloads with either same-gvc or workload-list: + # inboundAllowWorkload: + # - //gvc/main-kafka/workload/main-kafka-kafbat-ui + # - //gvc/client-gvc/workload/client + # external_outboundAllowCIDR: "111.222.333.444/16,111.222.444.333/32" # Provide a comma-separated list + listeners: + # @param listeners.client.name Name for the Kafka client listener + # @param listeners.client.containerPort Port for the Kafka client listener. Except ports 9091,9093,9094 + # @param listeners.client.protocol Security protocol for the Kafka client listener. Allowed values are 'PLAINTEXT', 'SASL_PLAINTEXT' + # @param listeners.client.publicAddress DNS address for public access to brokers. Must be the same as kafka.replicas + client: + protocol: SASL_PLAINTEXT + name: CLIENT + containerPort: 9092 # If publicAddress is enabled, Client automatically set to port range 3000-3004 + sasl: + ## @param listeners.client.sasl.users Comma-separated list of usernames for client communications when SASL is enabled + ## @param listeners.client.passwords Comma-separated list of passwords for client communications when SASL is enabled, must match the number of client.sasl.users + ## @param listeners.client.admin Admin username and password for client communications when SASL is enabled + admin: + username: admin + password: "your-admin-password" + users: "user" + passwords: "your-user-password" + # public: + # protocol: SASL_PLAINTEXT # TLS enforced, Kafka clients should use SASL_SSL to access 'publicAddress' if provided + # name: PUBLIC + # # containerPort: 9095 # Uncomment only when no directReplicaRouting or publicAddress is provided + # # Use directReplicaRouting for automatic public replica endpoints with DNS01 cert challenge + # directReplicaRouting: + # enabled: true + # containerPort: 9095 # ports 9093 and 9094 are reserved for controller and inter-broker communication + # publicAddress: kafka.example.com # Make sure Dedicate Load Balancer is enabled on the GVC + # sasl: + # ## @param listeners.client.sasl.users Comma-separated list of usernames for client communications when SASL is enabled + # ## @param listeners.client.brokersAddresses Comma-separated list of passwords for client communications when SASL is enabled, must match the number of client.sasl.users + # ## @param listeners.client.admin Admin username and password for client communications when SASL is enabled + # # admin: + # # username: admin + # # password: tgtgtg + # users: "public-user" + # passwords: "your-public-user-password" + acl: + superUsers: "User:admin" # User:admin;User:connectors (for multiple users) + allowEveryoneIfNoAclFound: false + secrets: + kraft_cluster_id: your-kraft-cluster-id # Example:bkdDtS1Rsf536si7BGM0JY + inter_broker_password: your-inter-broker-password # Example: HfcgCHp32e + controller_password: your-controller-password # Example: ayd8iJwqXe + extra_configurations: + default.replication.factor: 3 # default.replication.factor Can't be greater than the number of cluster replicas + auto.create.topics.enable: true # auto.create.topics.enable + log.retention.hours: 168 # The number of hours to keep a log file before deleting it (in hours) + +kafka_exporter: + name: exporter + image: danielqsj/kafka-exporter:v1.9.0 + debug: false + cpu: 50m + memory: 128Mi + listener: client + env: [] # If you need to set environment variables, add them here + dropMetrics: [] # e.g., ["kafka_consumergroup.*", "^kafka_topic_partition_current_offset"] + +jmx_exporter: + name: jmx-exporter + image: ghcr.io/controlplane-com/bitnami/jmx-exporter + kafkaJmxPort: 5557 # Ensure this port matches the port in the jmxUrl below + exporterPort: 5556 + debug: false + cpu: 250m + memory: 256Mi + minCpu: 80m + minMemory: 125Mi + listener: client + dropMetrics: [] # e.g., ["kafka_consumergroup.*", "^kafka_topic_partition_current_offset"] + config: + jmxUrl: service:jmx:rmi:///jndi/rmi://127.0.0.1:5557/jmxrmi + lowercaseOutputName: true + lowercaseOutputLabelNames: true + ssl: false + whitelistObjectNames: + - kafka.controller:* + - kafka.server:* + - java.lang:* + - kafka.network:* + - kafka.log:* + - kafka.producer:* + - kafka.consumer:* + rules: + - labels: + request: "$3" + name: kafka_request_count + pattern: kafka.network<>(Count) + - labels: + request: "$3" + stat: "$4" + name: kafka_request_metrics_totaltimems + pattern: kafka.network<>(.+) + - labels: + request: "$3" + component: "$2" + stat: "$4" + name: kafka_request_latency_ms + pattern: kafka.network<>(.+) + - labels: + client_type: "$3" + metric: "$2" + stat: "$4" + name: kafka_client_metrics + pattern: kafka.network<>(.+) + - labels: + client_id: "$1" + metric: "$2" + name: kafka_consumer_metrics + pattern: kafka.consumer<>(.+) + - labels: + client_id: "$1" + metric: "$2" + name: kafka_producer_metrics + pattern: kafka.producer<>(.+) + - name: kafka_server_$1_$2_$3 + pattern: kafka.server<>(Count|Value) + - name: java_lang_$1_$2 + pattern: java.lang<>(.+) + +kafbat_ui: + enabled: true + deletionProtection: false + name: kafbat-ui + image: ghcr.io/kafbat/kafka-ui + cpu: 300m + memory: 1000Mi + minCpu: 100m + minMemory: 400Mi + replicas: 1 + timeoutSeconds: 30 + configuration_secret: kafka-kafbat-ui-config # Pre-create a secret with the configuration; Example in README + # Domain name for the UI. + # Make sure the required DNS records are created in your DNS server + # https://docs.controlplane.com/guides/configure-domain#subdomain-e-g-sample-domain-com-cname-mode-path-based-routing + # domain: kafbat-ui.example.com # Domain name for the UI. + # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall + firewall: + # internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc + external_inboundAllowCIDR: "0.0.0.0/0" # Provide a comma-separated list + external_outboundAllowCIDR: "0.0.0.0/0" # Provide a comma-separated list + +# kafka_connectors: +# - name: cluster +# image: apache/kafka:3.9.1 +# multiZone: true +# cpu: 400m +# memory: 1500Mi +# minCpu: 100m +# minMemory: 375Mi +# plugins_folder: /opt/kafka/plugins +# timeoutSeconds: 15 +# replicas: 1 +# verbose: false +# extraVolumes: [] +# # Volume configuration for Kafka Connect (Optional - defaults shown below) +# volumes: +# initialCapacity: 10 # In GB (Default: 10) +# performanceClass: general-purpose-ssd # general-purpose-ssd / high-throughput-ssd (Min 1000GB) (Default: general-purpose-ssd) +# fileSystemType: ext4 # ext4 / xfs (Default: ext4) +# snapshots: +# createFinalSnapshot: true # Default: true +# retentionDuration: 7d # Default: 7d +# # schedule: 0 0 * * * # UTC (Optional) +# # customEncryption: +# # enabled: true # Encrypting is only possible for new volumes. Existing volumes cannot be re-encrypted after creation. +# # region: aws-us-east-2 # Replace with the appropriate region +# # keyId: arn:aws:kms:us-east-2:1234567890:key/fewf2f43-1d31-2332-9934-efhg4334gfe # Replace with your AWS KMS key ARN +# env: +# - name: KAFKA_HEAP_OPTS +# value: '-Xms900m -Xmx900m' # set to 50%-75% of the memory +# # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall +# firewall: +# external_inboundAllowCIDR: 0.0.0.0/0 # Provide a comma-separated list +# internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc +# # You can specify additional workloads with either same-gvc or workload-list: +# inboundAllowWorkload: +# - //gvc/main-kafka/workload/main-kafka-kafbat-ui +# - //gvc/client-gvc/workload/client +# external_outboundAllowCIDR: "0.0.0.0/0" # Provide a comma-separated list +# listener: client # Provide the listener name to connect to +# connector_properties: +# # bootstrap.servers: "kafka-dev-cluster:9092" # Optional. If not set, the bootstrap address will be the cluster name or publicAddress +# group.id: "connect-cluster" +# security.protocol: "SASL_PLAINTEXT" +# sasl.mechanism: "PLAIN" +# sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"your-admin-password\";" +# consumer.security.protocol: "SASL_PLAINTEXT" +# consumer.sasl.mechanism: "PLAIN" +# consumer.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"your-admin-password\";" +# producer.security.protocol: "SASL_PLAINTEXT" +# producer.sasl.mechanism: "PLAIN" +# producer.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"your-admin-password\";" +# key.converter.schemas.enable: "false" +# value.converter.schemas.enable: "false" +# offset.storage.topic: "connect-offsets" +# offset.storage.replication.factor: "3" +# config.storage.topic: "connect-configs" +# config.storage.replication.factor: "3" +# status.storage.topic: "connect-status" +# status.storage.replication.factor: "3" +# offset.flush.interval.ms: "10000" +# plugin.path: "/opt/kafka/plugins" +# key.converter: "org.apache.kafka.connect.storage.StringConverter" +# value.converter: "org.apache.kafka.connect.converters.ByteArrayConverter" +# plugins: +# - name: kafka-mirror-test-1 +# enabled: true +# config: +# connector.class: org.apache.kafka.connect.mirror.MirrorSourceConnector +# tasks.max: '1' +# offset-syncs.topic.location: source +# source.cluster.alias: remote +# target.cluster.alias: local +# source.bootstrap.servers: kafka-dev-cluster:9092 +# target.bootstrap.servers: kafka-dev-cluster:9092 +# source.consumer.bootstrap.servers: kafka-dev-cluster:9092 +# target.consumer.bootstrap.servers: kafka-dev-cluster:9092 +# source.producer.bootstrap.servers: kafka-dev-cluster:9092 +# target.producer.bootstrap.servers: kafka-dev-cluster:9092 +# source.admin.bootstrap.servers: kafka-dev-cluster:9092 +# target.admin.bootstrap.servers: kafka-dev-cluster:9092 +# replication.policy.class: org.apache.kafka.connect.mirror.DefaultReplicationPolicy +# topics: mirror-test-1-a,mirror-test-1-b +# groups: .* +# sync.topic.configs.enabled: 'true' +# sync.topic.acls.enabled: 'false' +# refresh.topics.interval.seconds: '60' +# refresh.groups.interval.seconds: '60' +# replication.factor: '3' +# offset.syncs.topic.replication.factor: '3' +# checkpoints.topic.replication.factor: '3' +# heartbeats.topic.replication.factor: '3' +# source.consumer.security.protocol: SASL_PLAINTEXT +# source.consumer.sasl.mechanism: PLAIN +# source.consumer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# source.producer.security.protocol: SASL_PLAINTEXT +# source.producer.sasl.mechanism: PLAIN +# source.producer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# target.producer.security.protocol: SASL_PLAINTEXT +# target.producer.sasl.mechanism: PLAIN +# target.producer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# target.consumer.security.protocol: SASL_PLAINTEXT +# target.consumer.sasl.mechanism: PLAIN +# target.consumer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# admin.security.protocol: SASL_PLAINTEXT +# admin.sasl.mechanism: PLAIN +# admin.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# consumer.security.protocol: SASL_PLAINTEXT +# consumer.sasl.mechanism: PLAIN +# consumer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# producer.security.protocol: SASL_PLAINTEXT +# producer.sasl.mechanism: PLAIN +# producer.sasl.jaas.config: >- +# org.apache.kafka.common.security.plain.PlainLoginModule required +# username='admin' password='your-admin-password'; +# - name: "camel-s3-sink" +# enabled: true +# artifacts: +# - type: tgz +# url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-aws-s3-sink-kafka-connector/4.8.5/camel-aws-s3-sink-kafka-connector-4.8.5-package.tar.gz +# config: +# "connector.class": "org.apache.camel.kafkaconnector.awss3sink.CamelAwss3sinkSinkConnector" +# "tasks.max": "1" +# "topics": "your-topic" +# "camel.kamelet.aws-s3-sink.useSessionCredentials": "false" +# "camel.kamelet.aws-s3-sink.bucketNameOrArn": "your-bucket-name" +# "camel.kamelet.aws-s3-sink.keyName": "your-topic-sink-${exchangeId}.txt" +# "camel.kamelet.aws-s3-sink.region": "your-region" +# "camel.kamelet.aws-s3-sink.autoCreateBucket": "true" +# "camel.kamelet.aws-s3-sink.accessKey": "your-access-key" +# "camel.kamelet.aws-s3-sink.secretKey": "your-secret-key" +# - name: "clickhouse-sink" +# enabled: true +# ssl_truststore: +# generate: true +# truststore_path: /tmp/kafka.autogenerated.truststore.jks +# truststore_password_env: "SSL_CLICKHOUSE_SINK_TRUSTSTORE_PASSWORD" +# hostnames: +# - domain1.clickhouse-sink.com +# - domain2.clickhouse-sink.com +# artifacts: +# - type: zip +# url: https://github.com/ClickHouse/clickhouse-kafka-connect/releases/download/v1.2.8/clickhouse-kafka-connect-v1.2.8.zip +# config: +# "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector" +# "tasks.max": "1" +# "topics": "your-topic" +# "security.protocol": "SASL_PLAINTEXT" # Connect to Kafka cluster using PLAINTEXT protocol - Internal connection mTLS encrypted +# "hostname": "your-hostname" +# "username": "your-username" +# "database": "your-database" +# "password": "your-password" +# "port": "8443" +# "value.converter.schemas.enable": "false" +# "ssl": "true" # Connect to ClickHouse using SSL protocol +# "value.converter": "org.apache.kafka.connect.json.JsonConverter" +# "key.converter": "org.apache.kafka.connect.storage.StringConverter" +# "errors.retry.timeout": "30" +# "schemas.enable": "false" +# "jdbcConnectionProperties": "?sslmode=STRICT" +# "ssl.truststore.location": "/tmp/kafka.autogenerated.truststore.jks" +# "ssl.truststore.password": "${SSL_CLICKHOUSE_SINK_TRUSTSTORE_PASSWORD}" +# "errors.tolerance": "all" +# "errors.log.enable": "true" +# "errors.log.include.messages": "true" +# - name: "snowflake-sink" +# enabled: true +# artifacts: +# - type: jar +# url: https://repo1.maven.org/maven2/com/snowflake/snowflake-kafka-connector/3.1.1/snowflake-kafka-connector-3.1.1.jar +# - type: jar +# url: https://repo1.maven.org/maven2/org/bouncycastle/bc-fips/2.1.0/bc-fips-2.1.0.jar +# - type: jar +# url: https://repo1.maven.org/maven2/org/bouncycastle/bcpkix-fips/2.1.9/bcpkix-fips-2.1.9.jar +# config: +# "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector" +# "tasks.max": "1" +# "topics": "your-topic" +# "key.converter": "org.apache.kafka.connect.storage.StringConverter" +# "value.converter": "com.snowflake.kafka.connector.records.SnowflakeJsonConverter" +# "value.converter.schemas.enable": "false" +# "security.protocol": "SASL_PLAINTEXT" # Connect to Kafka cluster using SASL_PLAINTEXT protocol - Internal connection mTLS encrypted +# "snowflake.url.name": "your-snowflake-url" +# "snowflake.user.name": "your-snowflake-username" +# "snowflake.private.key": "your-snowflake-private-key" +# "snowflake.private.key.passphrase": "your-snowflake-private-key-passphrase" +# "snowflake.warehouse.name": "your-snowflake-warehouse-name" +# "snowflake.database.name": "your-snowflake-database-name" +# "snowflake.schema.name": "your-snowflake-schema-name" +# "snowflake.topic2table.map": "your-topic:your-table" +# "snowflake.role.name": "your-snowflake-role-name" +# "snowflake.enable.schematization": "false" +# "snowflake.disable.ssl.certificate.verification": "true" +# "snowflake.log.enable": "true" +# "snowflake.log.level": "DEBUG" +# "buffer.count.records": "10000" +# "buffer.flush.time": "120" +# "buffer.size.bytes": "10000000" +# "errors.tolerance": "all" +# "errors.log.enable": "true" +# "errors.log.include.messages": "true" + +kafka_rest_proxy: + enabled: true + deletionProtection: false + name: rest-proxy + image: confluentinc/cp-kafka-rest:latest + cpu: 500m + memory: 1000Mi + capacityAI: + enabled: true + minCpu: 125m # This only applied when capacityAI is enabled + minMemory: 200Mi # This only applied when capacityAI is enabled + replicas: 1 + timeoutSeconds: 15 + # domain: kafka-rest.example.com # Domain name for the Kafka Rest Proxy. + + # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall + firewall: + # internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc(Recommended) + external_inboundAllowCIDR: 0.0.0.0/0 # Provide a comma-separated list + # # You can specify additional workloads with either same-gvc or workload-list: + # inboundAllowWorkload: + # - //gvc/main-kafka/workload/main-kafka-kafbat-ui + # - //gvc/client-gvc/workload/client + external_outboundAllowCIDR: "0.0.0.0/0" # Provide a comma-separated list + properties: + # host.name: kafka-rest.example.com + bootstrap.servers: SASL_PLAINTEXT://kafka-dev-cluster:9092 + resource.extension: ALL + api.v3.enable: true + api.v2.enable: true + client.sasl.mechanism: PLAIN + api.compatibility.mode: BOTH + log4j.opts: -Dlog4j.configuration=file:/tmp/log4j.properties + listeners: http://0.0.0.0:8082 + authentication.realm: KafkaRest + authentication.method: BASIC + authentication.roles: user + client.security.protocol: SASL_PLAINTEXT + + # JAAS configuration for Kafka client and Kafka Rest Proxy + # https://docs.confluent.io/platform/current/kafka-rest/production-deployment/confluent-server/security.html#authentication-between-the-admin-rest-and-ak-brokers + jaas_conf: + KafkaClient { + org.apache.kafka.common.security.plain.PlainLoginModule required + username="admin" + password="your-admin-password"; + }; + KafkaRest { + org.eclipse.jetty.jaas.spi.PropertyFileLoginModule required + debug="true" + file="/etc/kafka-rest/password.properties"; + }; + + # # Password properties for Kafka Rest Proxy + # # Required when authentication.method is set to BASIC + # # https://docs.confluent.io/platform/current/kafka-rest/production-deployment/confluent-server/security.html#password-properties + password_properties: + user: your-user-password,user + user1: password213,user + user2: password214,user + +kafka_client: + name: client + image: apache/kafka:3.9.1 + cpu: 500m + memory: 1000Mi + # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall + firewall: + # internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc + # external_inboundAllowCIDR: 0.0.0.0/0 # Provide a comma-separated list + external_outboundAllowCIDR: "0.0.0.0/0" # Provide a comma-separated list + +# DEPRECATED NOTICE https://github.com/provectus/kafka-ui +# PLEASE USE KAFBAT UI INSTEAD +kafka_ui: + enabled: false + name: ui + image: provectuslabs/kafka-ui:latest + cpu: 200m + memory: 600Mi + listener: client + # To disable all traffic, comment out the corresponding rule. Docs: https://docs.controlplane.com/concepts/security#firewall + firewall: {} + # internal_inboundAllowType: "same-gvc" # Options: same-org / same-gvc + # external_inboundAllowCIDR: 0.0.0.0/0 # Provide a comma-separated list + # external_outboundAllowCIDR: "111.222.333.444/16,111.222.444.333/32" # Provide a comma-separated list