Introduction
ksqlDB is a database purpose-built to help developers create stream processing applications on top of Apache Kafka.
Setup
Setup secret for kafka keystore/truststore
Create a strimzi user (ksqldb)
| Code Block |
|---|
| language | yml |
|---|
| title | Strimzi User |
|---|
|
apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
name: ksqldb
namespace: kafka
labels:
strimzi.io/cluster: my-cluster
spec:
authentication:
type: tls |
Generates keystire/truststore
| Code Block |
|---|
| language | text |
|---|
| title | generate_stores.sh |
|---|
|
#!/bin/sh
if [ -z "$1" ]
then
echo "No argument supplied"
exit 1
fi
kafkauser=$1
WORKDIR=$(dirname "$(realpath "$0")")
rm $WORKDIR/ca.crt $WORKDIR/user.crt $WORKDIR/user.key $WORKDIR/user-keystore.jks $WORKDIR/user.p12 $WORKDIR/user.password $WORKDIR/user-truststore.jks 2>/dev/null
kubectl delete secret ${kafkauser}-jks -n kafka 2>/dev/null
kubectl get secret my-cluster-cluster-ca-cert -n kafka -o jsonpath='{.data.ca\.crt}' | base64 --decode > $WORKDIR/ca.crt
kubectl get secret ${kafkauser} -n kafka -o jsonpath='{.data.user\.key}' | base64 --decode > $WORKDIR/user.key
kubectl get secret ${kafkauser} -n kafka -o jsonpath='{.data.user\.crt}' | base64 --decode > $WORKDIR/user.crt
kubectl get secret ${kafkauser} -n kafka -o jsonpath='{.data.user\.p12}' | base64 --decode > $WORKDIR/user.p12
kubectl get secret ${kafkauser} -n kafka -o jsonpath='{.data.user\.password}' | base64 --decode > $WORKDIR/user.password
export PASSWORD=`cat ${WORKDIR}/user.password`
keytool -import -trustcacerts -file $WORKDIR/ca.crt -keystore $WORKDIR/user-truststore.jks -storepass $PASSWORD -noprompt
keytool -importkeystore -srckeystore $WORKDIR/user.p12 -srcstorepass ${PASSWORD} -srcstoretype pkcs12 -destkeystore $WORKDIR/user-keystore.jks -deststorepass ${PASSWORD} -deststoretype jks
kubectl create secret generic ${kafkauser}-jks -n kafka --from-literal=keystore_password=$PASSWORD --from-file=user-keystore.jks=${WORKDIR}/user-keystore.jks --from-literal=truststore_password=$PASSWORD --from-file=user-truststore.jks=${WORKDIR}/user-truststore.jks --from-literal=key_password=$PASSWORD |
Run ./generate_stores.sh ksqldb
This will create a secret called "ksqldb-jks" which contains the keystore and truststore needed to connect to Kafka over SSL.
ksqldb-server
| Code Block |
|---|
| language | yml |
|---|
| title | ksqldb-server |
|---|
|
apiVersion: apps/v1
kind: Deployment
metadata:
name: ksqldb-server
namespace: kafka
spec:
selector:
matchLabels:
app: ksqldb-server
template:
metadata:
labels:
app: ksqldb-server
version: v1
spec:
containers:
- name: ksqldb-server
image: confluentinc/ksqldb-server:0.28.2
imagePullPolicy: IfNotPresent
env:
- name: KEYSTORE_PASSWORD
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: keystore_password
- name: TRUSTSTORE_PASSWORD
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: truststore_password
- name: KEY_PASSWORD
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: key_password
- name: KSQL_BOOTSTRAP_SERVERS
value: my-cluster-kafka-bootstrap.kafka:9093
- name: KSQL_LISTENERS
value: http://0.0.0.0:8088
- name: KSQL_KSQL_SERVICE_ID
value: ksql_service_2_
- name: KSQL_SECURITY_PROTOCOL
value: SSL
- name: KSQL_OPTS
value: "-Dssl.keystore.location=/var/private/ssl/user-keystore.jks -Dssl.keystore.password=$(KEYSTORE_PASSWORD) -Dssl.key.password=$(KEY_PASSWORD) -Dssl.truststore.location=/var/private/ssl/user-truststore.jks -Dssl.truststore.password=$(TRUSTSTORE_PASSWORD) -Dlisteners=http://0.0.0.0:8088/"
- name: KSQL_KSQL_EXTENSION_DIR
value: /opt/ksqldb-udfs
volumeMounts:
- name: jks
mountPath: /var/private/ssl
readOnly: true
volumes:
- name: jks
secret:
secretName: ksqldb-jks
---
apiVersion: v1
kind: Service
metadata:
name: ksqldb-server
namespace: kafka
labels:
app: ksqldb-server
service: ksqldb-server
spec:
type: LoadBalancer
selector:
app: ksqldb-server
ports:
- port: 8088
name: http-80
|
The above YAML file deploys the ksqldb server to your cluster.
It has been configured to connect to your kafka cluster over SSL using the "ksqldb-jks" secret created in the previous step.
ksqldb-cli
| Code Block |
|---|
| language | yml |
|---|
| title | ksqldb-cli |
|---|
|
apiVersion: apps/v1
kind: Deployment
metadata:
name: ksqldb-cli
namespace: kafka
spec:
selector:
matchLabels:
app: ksqldb-cli
template:
metadata:
labels:
app: ksqldb-cli
version: v1
spec:
containers:
- name: ksqldb-cli
image: confluentinc/ksqldb-cli:0.28.2
imagePullPolicy: IfNotPresent
env:
- name: KEYSTORE_PASSWORD
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: keystore_password
- name: TRUSTSTORE_PASSWORD
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: truststore_password
valueFrom:
secretKeyRef:
name: ksqldb-jks
key: key_password
- name: KSQL_BOOTSTRAP_SERVERS
value: my-cluster-kafka-bootstrap.kafka:9093
- name: KSQL_SECURITY_PROTOCOL
value: SSL
- name: KSQL_OPTS
value: "-Dssl.keystore.location=/var/private/ssl/user-keystore.jks -Dssl.keystore.password=$(KEYSTORE_PASSWORD) -Dssl.key.password=$(KEY_PASSWORD) -Dssl.truststore.location=/var/private/ssl/user-truststore.jks -Dssl.truststore.password=$(TRUSTSTORE_PASSWORD) -Dlisteners=http://0.0.0.0:8088/"
volumeMounts:
- name: jks
mountPath: /var/private/ssl
readOnly: true
volumes:
- name: jks
secret:
secretName: ksqldb-jks |
...