This is a playground for KRAFT running with CFK
-
A Kubernetes cluster.
-
Kubectl, with cluster context configured.
-
OpenSSL:
openssl
command line tool. -
CFSSL command:
cfssl
Confluent For Kubernetes
Resoources:
- KraftControllers
- Kafka
- Schema Registry
- Control Center
Networking:
- Internal communication (TLS): Autogenerated Certificates by the Operator.
- External communication (Ingress. TLS Host based routing): Custom Certificates.
External Listener:
- Authentication: mTLS
scripts/1_helm_setup.sh
Confluent For Kubernetes Operator with KRAFT enabled, Namespace: confluent
, and NGINX Ingress Controller.
Replace $DOMAIN
with your domain name in the certificates configuration.
Example of using a fake DNS Domain like nip.io.
Use: kubectl get svc ingress-nginx-controller -o json | jq .status.loadBalancer.ingress
to get the IP address of the ingress controller.
Using a fake DNS Domain: $DOMAIN = <load_balancer_ip>.nip.io
sed -i '' -e 's/$DOMAIN/<load_balancer_ip>.nip.io/g' FILENAME
Generate certificates: scripts/2_certs.sh
All files will be created under the certs/generated
directory.
Create Kubernetes Secrets: scripts/3_secrets.sh
- Kraft Controllers
- Brokers
- Schema Registry
- Control Center
- Demo Topic
Update the $DOMAIN
variable in the crds
files:
-
Control Center ingress:
controlcenter.$DOMAIN
-./scripts/crds/c3-ingress.yaml
-
Kafka ingress:
kafka.$DOMAIN:433
,b0.$DOMAIN
,b1.$DOMAIN
,b2.$DOMAIN
-./scripts/crds/kafkaingress.yaml
-
Kafka external access:
domain: $DOMAIN
-./scripts/crds/kafka.yaml
Deploy: scripts/4_confluent.sh
Control center: Open Browser https://controlcenter.$DOMAIN
- Use the autogenerated certificates.
scripts/config/internal.properties
bootstrap.servers=kafka.confluent.svc.cluster.local:9071
security.protocol=SSL
ssl.truststore.location=/mnt/sslcerts/truststore.jks
ssl.truststore.password=mystorepassword
- Demo producer (
kafka-producer-perf-test
) to topicdemo.topic
, it uses internal listener with SSL.
kubectl apply -f ./crds/producer.yaml
it uses kafka-client-config-secure
secret created previously.
...
volumes:
# This application pod will mount a volume for Kafka client properties from
# the secret `kafka-client-config-secure`
- name: kafka-properties
secret:
secretName: kafka-client-config-secure
# Confluent for Kubernetes, when configured with autogenerated certs, will create a
# JKS keystore and truststore and store that in a Kubernetes secret named `kafka-generated-jks`.
# Here, this client appliation will mount a volume from this secret so that it can use the JKS files.
- name: kafka-ssl-autogenerated
secret:
secretName: kafka-generated-jks
Client certificates: script/5_clients.sh
Use custom certificates. scripts/config/external.properties
security.protocol=SSL
ssl.truststore.type=PEM
ssl.truststore.location=<FULL_PATH>/certs/generated/cacerts.pem
ssl.keystore.type=PKCS12
ssl.keystore.location=<FULL_PATH>/certs/generated/user.p12
ssl.keystore.password=changeme
Test client:
kafka-topics --bootstrap-server kafka.$DOMAIN:443 \
--command-config ./scripts/config/external.properties \
--topic demo.topic \
--describe
librdkafka client:
security.protocol: SSL
ssl.ca.location: "<FULL_PATH>/certs/generated/cacerts.pem"
ssl.certificate.location: "<FULL_PATH>/certs/generated/user.pem"
ssl.key.location: "<FULL_PATH>/certs/generated/user-key.pem"
ssl.key.password: "mystorepassword"
- Create GKE (example):
gcloud beta container --project <PROJECT_ID> clusters create "cluster-gke-1" \
--zone "<ZONE>" --no-enable-basic-auth --cluster-version "1.27.3-gke.100" \
--release-channel "regular" --machine-type "e2-standard-4" --image-type "COS_CONTAINERD" \
--disk-type "pd-balanced" --disk-size "500" --metadata disable-legacy-endpoints=true \
--scopes "https://www.googleapis.com/auth/devstorage.read_only","https://www.googleapis.com/auth/logging.write", \
"https://www.googleapis.com/auth/monitoring","https://www.googleapis.com/auth/servicecontrol", \
"https://www.googleapis.com/auth/service.management.readonly","https://www.googleapis.com/auth/trace.append" \
--max-pods-per-node "110" --num-nodes "3" --logging=SYSTEM,WORKLOAD --monitoring=SYSTEM --enable-ip-alias \
--network "projects/<PROJECT_ID>/global/networks/<NETWORK_ID>" \
--subnetwork "projects/<PROJECT_ID>/regions/<REGION>/subnetworks/<SUBNETID>" \
--no-enable-intra-node-visibility --default-max-pods-per-node "110" \
--no-enable-master-authorized-networks --addons HorizontalPodAutoscaling,HttpLoadBalancing,GcePersistentDiskCsiDriver \
--enable-autoupgrade --enable-autorepair --max-surge-upgrade 1 --max-unavailable-upgrade 0 \
--enable-shielded-nodes --node-locations "<ZONE>"
- Get GKE credentials (example):
gcloud container clusters get-credentials cluster-gke-1 --zone <ZONE> --project <PROJECT_ID>
Adding ACLs to the Kafka cluster.
- Internal authentication using mTLS.
- External authentication using SASL SSL.
ACL Authorizer for Kraft Controller (StandardAuthorizer).
KraftController
CRD:
spec:
authorization:
type: simple
superUsers:
- User:kafka
- User:kraftcontroller
configOverrides:
server:
- authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
ACL Authorizer for Kafka Broker (StandardAuthorizer).
Kafka
CRD:
spec:
authorization:
type: simple
superUsers:
- User:kafka
- User:kraftcontroller
configOverrides:
server:
- authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
Internal authentication using mTLS:
spec:
listeners:
internal:
authentication:
type: mtls
principalMappingRules:
- RULE:.*CN[\s]?=[\s]?([a-zA-Z0-9.]*)?.*/$1/
external:
authentication:
type: plain
jaasConfig:
secretRef: credential
External authentication using SASL SSL:
kafka
user configuration: external.properties
security.protocol=SASL_SSL
ssl.truststore.type=PEM
ssl.truststore.location=<FULL_PATH>/certs/generated/cacerts.pem
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret";
sasl.mechanism=PLAIN
Add permissions to the test
user to produce and consume from the demo.topic
topic.
See users/plain-users.json
.
Configure ACLS (using kafka
user):
kafka-acls --bootstrap-server kafka.$DOMAIN:443 \
--command-config <FULL_PATH>/external.properties \
--add \
--allow-principal User:'test' \
--operation All \
--topic demo.topic
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=demo.topic, patternType=LITERAL)`:
(principal=User:test, host=*, operation=ALL, permissionType=ALLOW)
Configure user: test.properties
security.protocol=SASL_SSL
ssl.truststore.type=PEM
ssl.truststore.location=<FULL_PATH>/certs/generated/cacerts.pem
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="test" password="test123";
sasl.mechanism=PLAIN
test
user can describe the demo.topic
topic:
kafka-topics --bootstrap-server kafka.$DOMAIN:443 --command-config <PATH>/test.properties --topic demo.topic --describe
Topic: demo.topic TopicId: uvihRw2ZRbCUPyPhDIKWZg PartitionCount: 4 ReplicationFactor: 3 Configs: min.insync.replicas=2,cleanup.policy=delete,segment.bytes=1073741824,message.format.version=3.0-IV1
Topic: demo.topic Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 2,1,0
Topic: demo.topic Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 2,1,0
Topic: demo.topic Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 2,1,0
Topic: demo.topic Partition: 3 Leader: 0 Replicas: 0,1,2 Isr: 2,1,0
Testing with unauthorized user:
Error while executing topic command : Topic 'demo.topic' does not exist as expected
Enable DEBUG log level for the authorizer in the Kafka
and in the KraftController
CRD:
spec:
configOverrides:
server:
- authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
log4j:
- log4j.logger.kafka.authorizer.logger=DEBUG, authorizerAppender
- log4j.appender.authorizerAppender=org.apache.log4j.ConsoleAppender
- log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
- log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
- log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
Script: ./monitoring/monitoring.sh
monitoring
namespace- Prometheus deployment
- Grafana deployment
- Grafana ingress
There is no Ingress rules for Prometheus, if you need to access the Prometheus Web:
kubectl port-forward --namespace monitoring svc/prometheus-server 9090:80
Then use http://localhost:9090/ to access the Prometheus Web UI.
Update ./monitoring/grafana/ingress.yaml
with your domain name.
Use [http://monitoring.< $DOMAIN >/](http://monitoring.< $DOMAIN >) to access Grafana.
- Get your 'admin' user password:
kubectl get secret --namespace monitoring grafana -o jsonpath="{.data.admin-password}" | base64 --decode
Dashboards (JSON files) are located in ./monitoring/grafana/dashboards
.
- Kraft Controller (./monitoring/grafana/dashboards/cp-kraft.json)
- Confluent Platform (./monitoring/grafana/dashboards/confluentplatform.json)
- Confluent Platform - Overview (./monitoring/grafana/dashboards/cp-overview.json)
- Confluent Operator (./monitoring/grafana/dashboards/cfk-operator.json)
- Confluent Connect (./monitoring/grafana/dashboards/connect-dashboard.json)