概述
Apache ZooKeeper 是一个集中式服务,用于维护配置信息、命名、提供分布式同步和提供组服务,ZooKeeper 致力于开发和维护一个开源服务器,以实现高度可靠的分布式协调,其实也可以认为就是一个分布式数据库,只是结构比较特殊,是树状结构。官网文档:https://zookeeper.apache.org/doc/r3.8.0/关于Zookeeper的介绍,
Kafka是最初由Linkedin公司开发,是一个分布式、支持分区的(partition)、多副本的(replica),基于zookeeper协调的分布式消息系统。官方文档:https://kafka.apache.org/documentation
安装helm
文件下载
链接:https://pan.baidu.com/s/1K1IzPjygF4-lepAA-7O2uA 提取码:y30k --来自百度网盘超级会员V4的分享
安装helm
没有网络的情况下使用 helm pull 从有网络的机器上执行并copy内网。
helm pull bitnami/kafka
helm pull bitnami/zookeeper --untar
–untar 参数会直接加压出来
tar -zxvf helm-v3.0.0-linux-amd64.tar.gz mv linux-amd64/helm /usr/local/bin/helm
参考: https://helm.sh/zh/docs/intro/install/
helm install zk
架构图
如果没有外网,文件下载中有zk和kafka的安装包使用. 一定要使用安装包中的镜像版本,否则会有问题.
]#helm install zookeeper -n kafka --set replicaCount=3 --set auth.enabled=false --set allowAnonymousLogin=true /opt/helm/zookeeper NAME: zookeeper LAST DEPLOYED: Fri Aug 12 14:06:02 2022 NAMESPACE: kafka STATUS: deployed REVISION: 1 TEST SUITE: None NOTES: CHART NAME: zookeeper CHART VERSION: 10.0.7 APP VERSION: 3.8.0 ** Please be patient while the chart is being deployed ** ZooKeeper can be accessed via port 2181 on the following DNS name from within your cluster: zookeeper.kafka.svc.cluster.local To connect to your ZooKeeper server run the following commands: export POD_NAME=$(kubectl get pods --namespace kafka -l "app.kubernetes.io/name=zookeeper,app.kubernetes.io/instance=zookeeper,app.kubernetes.io/component=zookeeper" -o jsonpath="{.items[0].metadata.name}") kubectl exec -it $POD_NAME -- zkCli.sh To connect to your ZooKeeper server from outside the cluster execute the following commands: kubectl port-forward --namespace kafka svc/zookeeper 2181:2181 & zkCli.sh 127.0.0.1:2181
确认pod状态
]# kubectl get po -n kafka NAME READY STATUS RESTARTS AGE kafka-manager-57569fcc78-mnhnn 1/1 Running 0 45h zookeeper-0 1/1 Running 0 10m zookeeper-1 1/1 Running 0 10m zookeeper-2 1/1 Running 0 10m
可以使用helm安装提示的命令测试一下
kafka的zk存储结构
zookeeper 是 kafka 不可分割的一部分。接下来就来讲讲zookeeper在kafka中作用。
1)记录和维护broker状态
zookeeper 记录了所有 broker 的存活状态,broker 会向 zookeeper 发送心跳请求来上报自己的状态。
zookeeper 维护了一个正在运行并且属于集群的 broker 列表。
2)控制器(leader )选举
kafka 集群中有多个 broker,其中有一个会被选举为控制器。控制器负责管理整个集群所有分区和副本的状态,例如某个分区的 leader 故障了,控制器会选举新的 leader。从多个 broker 中选出控制器,这个工作就是 zookeeper 负责的。
3)限额权限
kafka 允许一些 client 有不同的生产和消费的限额,这些限额配置信息是保存在 zookeeper 里面的。
4)记录 ISR(已同步的副本)
Kafka会在Zookeeper上针对每个Topic维护一个称为ISR(in-sync replica,已同步的副本)的集合,该集合中是一些分区的副本。只有当这些副本都跟Leader中的副本同步了之后,kafka才会认为消息已提交,并反馈给消息的生产者。如果这个集合有增减,kafka会更新zookeeper上的记录。zookeeper 发现其中有成员不正常,马上移除。
5)node 和 topic 注册
zookeeper 保存了所有 node 和 topic 的注册信息,可以方便的找到每个 broker 持有哪些 topic。
node 和 topic 在 zookeeper 中是以临时节点的形式存在的,只要与 zookeeper 的 session 一关闭,他们的信息就没有了。
6)topic 配置
zookeeper 保存了 topic 相关配置,例如 topic 列表、每个 topic 的 partition 数量、副本的位置等等。
helm install kafka
[root@master2 /opt/helm/zookeeper]#helm install kafka -n kafka --set zookeeper.enabled=false --set replicaCount=3 --set externalZookeeper.servers=zookeeper /opt/helm/kafka/ NAME: kafka LAST DEPLOYED: Fri Aug 12 14:18:47 2022 NAMESPACE: kafka STATUS: deployed REVISION: 1 TEST SUITE: None NOTES: CHART NAME: kafka CHART VERSION: 18.0.8 APP VERSION: 3.2.1 ** Please be patient while the chart is being deployed ** Kafka can be accessed by consumers via port 9092 on the following DNS name from within your cluster: kafka.kafka.svc.cluster.local Each Kafka broker can be accessed by producers via port 9092 on the following DNS name(s) from within your cluster: kafka-0.kafka-headless.kafka.svc.cluster.local:9092 kafka-1.kafka-headless.kafka.svc.cluster.local:9092 kafka-2.kafka-headless.kafka.svc.cluster.local:9092 To create a pod that you can use as a Kafka client run the following commands: kubectl run kafka-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity kubectl exec --tty -i kafka-client --namespace kafka -- bash PRODUCER: kafka-console-producer.sh \ --broker-list kafka-0.kafka-headless.kafka.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \ --topic test CONSUMER: kafka-console-consumer.sh \ --bootstrap-server kafka.kafka.svc.cluster.local:9092 \ --topic test \ --from-beginning
helm install --set的格式和限制
–set zookeeper.enabled=false # 禁用kafka自带zk
–set replicaCount=3 # 三个副本
–set externalZookeeper.servers=zookeeper #外部zk的地址
更多参考: Helm | Values Files
使用kafka image运行一个kafka-client pod
然后依次开启两个kafka-client bash 执行生产者和消费者的脚本进行消息的发送和消费。
kubectl run kafka-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity
kafka‐topics.sh ‐‐create ‐‐zookeeper 192.168.65.60:2181 ‐‐replication‐factor 1 ‐‐partitions 1 ‐‐topic test
生产者
I have no name!@kafka-client:/$ kafka-console-producer.sh \ --broker-list kafka-0.kafka-headless.kafka.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \ --topic test >test >chottest >asduiohfsd >hiushdfsd >tyufgigui
消费者
[root@master1 /nfs/data]#kubectl exec -it kafka-client -n kafka -- /bin/bash I have no name!@kafka-client:/$ I have no name!@kafka-client:/$ kafka-console-consumer.sh \ --bootstrap-server kafka.kafka.svc.cluster.local:9092 \ --topic test \ --from-beginning test chottest asduiohfsd hiushdfsd tyufgigui
错误的测试,不能在kafka节点中创建producer 和consumer
I have no name!@kafka-client:/$ kafka-console-producer.sh \ > --broker-list kafka-0.kafka-headless.default.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \ > --topic test [2022-08-12 15:11:18,722] WARN Couldn't resolve server kafka-0.kafka-headless.default.svc.cluster.local:9092 from bootstrap.servers as DNS resolution failed for kafka-0.kafka-headless.default.svc.cluster.local (org.apache.kafka.clients.ClientUtils)
kafka外部如何访问
上面这个kafka集群仅仅可以K8S内部访问,如何让外部也能访问呢?你第一个想到的肯定是nodePort,同时也回忆一下其他方式,例如LB、外部DNS。
以下内容来自官方内容
Accessing Kafka brokers from outside the cluster
In order to access Kafka Brokers from outside the cluster, an additional listener and advertised listener must be configured. Additionally, a specific service per kafka pod will be created.
There are two ways of configuring external access. Using LoadBalancer services or using NodePort services.
Using LoadBalancer services
You have two alternatives to use LoadBalancer services:
Option A) Use random load balancer IPs using an initContainer that waits for the IPs to be ready and discover them automatically.
externalAccess.enabled=true externalAccess.service.type=LoadBalancer externalAccess.service.ports.external=9094 externalAccess.autoDiscovery.enabled=true serviceAccount.create=true rbac.create=true
Note: This option requires creating RBAC rules on clusters where RBAC policies are enabled.
externalAccess.enabled=true externalAccess.service.type=LoadBalancer externalAccess.service.ports.external=9094 externalAccess.service.loadBalancerIPs[0]='external-ip-1' externalAccess.service.loadBalancerIPs[1]='external-ip-2'}
Note: You need to know in advance the load balancer IPs so each Kafka broker advertised listener is configured with it.
Following the aforementioned steps will also allow to connect the brokers from the outside using the cluster’s default service (when service.type is LoadBalancer or NodePort). Use the property service.externalPort to specify the port used for external connections.
Using NodePort services
You have two alternatives to use NodePort services:
Option A) Use random node ports using an initContainer that discover them automatically.
externalAccess.enabled=true externalAccess.service.type=NodePort externalAccess.autoDiscovery.enabled=true serviceAccount.create=true rbac.create=true
Note: This option requires creating RBAC rules on clusters where RBAC policies are enabled.
Option B) Manually specify the node ports:
externalAccess.enabled=true externalAccess.service.type=NodePort externalAccess.service.nodePorts[0]='node-port-1' externalAccess.service.nodePorts[1]='node-port-2'
Note: You need to know in advance the node ports that will be exposed so each Kafka broker advertised listener is configured with it.
The pod will try to get the external ip of the node using curl -s https://ipinfo.io/ip unless externalAccess.service.domain or externalAccess.service.useHostIPs is provided.
Name resolution with External-DNS
You can use the following values to generate External-DNS annotations which automatically creates DNS records for each ReplicaSet pod:
externalAccess: service: annotations: external-dns.alpha.kubernetes.io/hostname: "{{ .targetPod }}.example.com"
helm 安装zk是保留nodeport
helm install zookeeper-cluster -n kafka --set replicaCount=3,persistence.enabled=false,service.type=NodePort /opt/helm/zookeeper [root@master2 /opt/helm/kafka]#helm install zookeeper-cluster -n kafka --set replicaCount=3,persistence.enabled=false,service.type=NodePort /opt/helm/zookeeper NAME: zookeeper-cluster LAST DEPLOYED: Sat Aug 13 17:21:42 2022 NAMESPACE: kafka STATUS: deployed REVISION: 1 TEST SUITE: None NOTES: CHART NAME: zookeeper CHART VERSION: 10.0.7 APP VERSION: 3.8.0 ** Please be patient while the chart is being deployed ** ZooKeeper can be accessed via port 2181 on the following DNS name from within your cluster: zookeeper-cluster.kafka.svc.cluster.local To connect to your ZooKeeper server run the following commands: export POD_NAME=$(kubectl get pods --namespace kafka -l "app.kubernetes.io/name=zookeeper,app.kubernetes.io/instance=zookeeper-cluster,app.kubernetes.io/component=zookeeper" -o jsonpath="{.items[0].metadata.name}") kubectl exec -it $POD_NAME -- zkCli.sh To connect to your ZooKeeper server from outside the cluster execute the following commands: export NODE_IP=$(kubectl get nodes --namespace kafka -o jsonpath="{.items[0].status.addresses[0].address}") export NODE_PORT=$(kubectl get --namespace kafka -o jsonpath="{.spec.ports[0].nodePort}" services zookeeper-cluster)
helm 安装kafka使用externalAccess
设置服务发布方式为NodePort。关闭Kafka内置的ZooKeeper,配置为使用我们刚刚搭建ZooKeeper。
helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/ [root@master2 /opt/helm/kafka]#helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/ NAME: kafka-cluster LAST DEPLOYED: Sat Aug 13 17:22:01 2022 NAMESPACE: kafka STATUS: deployed REVISION: 1 TEST SUITE: None NOTES: CHART NAME: kafka CHART VERSION: 18.0.8 APP VERSION: 3.2.1 --------------------------------------------------------------------------------------------- WARNING By specifying "serviceType=LoadBalancer" and not configuring the authentication you have most likely exposed the Kafka service externally without any authentication mechanism. For security reasons, we strongly suggest that you switch to "ClusterIP" or "NodePort". As alternative, you can also configure the Kafka authentication. --------------------------------------------------------------------------------------------- ** Please be patient while the chart is being deployed ** Kafka can be accessed by consumers via port 9092 on the following DNS name from within your cluster: kafka-cluster.kafka.svc.cluster.local Each Kafka broker can be accessed by producers via port 9092 on the following DNS name(s) from within your cluster: kafka-cluster-0.kafka-cluster-headless.kafka.svc.cluster.local:9092 kafka-cluster-1.kafka-cluster-headless.kafka.svc.cluster.local:9092 kafka-cluster-2.kafka-cluster-headless.kafka.svc.cluster.local:9092 To create a pod that you can use as a Kafka client run the following commands: kubectl run kafka-cluster-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity kubectl exec --tty -i kafka-cluster-client --namespace kafka -- bash PRODUCER: kafka-console-producer.sh \ --broker-list kafka-cluster-0.kafka-cluster-headless.kafka.svc.cluster.local:9092,kafka-cluster-1.kafka-cluster-headless.kafka.svc.cluster.local:9092,kafka-cluster-2.kafka-cluster-headless.kafka.svc.cluster.local:9092 \ --topic test CONSUMER: kafka-console-consumer.sh \ --bootstrap-server kafka-cluster.kafka.svc.cluster.local:9092 \ --topic test \ --from-beginning To connect to your Kafka server from outside the cluster, follow the instructions below: Kafka brokers domain: You can get the external node IP from the Kafka configuration file with the following commands (Check the EXTERNAL listener) 1. Obtain the pod name: kubectl get pods --namespace kafka -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka-cluster,app.kubernetes.io/component=kafka" 2. Obtain pod configuration: kubectl exec -it KAFKA_POD -- cat /opt/bitnami/kafka/config/server.properties | grep advertised.listeners Kafka brokers port: You will have a different node port for each Kafka broker. You can get the list of configured node ports using the command below: echo "$(kubectl get svc --namespace kafka -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka-cluster,app.kubernetes.io/component=kafka,pod" -o jsonpath='{.items[*].spec.ports[0].nodePort}' | tr ' ' '\n')"
externalAccess的参数
externalAccess.autoDiscovery.enabled | Enable using an init container to auto-detect external IPs/ports by querying the K8s API | FALSE |
externalAccess.autoDiscovery.image.registry | Init container auto-discovery image registry | docker.io |
externalAccess.autoDiscovery.image.repository | Init container auto-discovery image repository | bitnami/kubectl |
externalAccess.autoDiscovery.image.tag | Init container auto-discovery image tag (immutable tags are recommended) | 1.24.3-debian-11-r10 |
externalAccess.autoDiscovery.image.pullPolicy | Init container auto-discovery image pull policy | IfNotPresent |
externalAccess.autoDiscovery.image.pullSecrets | Init container auto-discovery image pull secrets | [] |
externalAccess.autoDiscovery.resources.limits | The resources limits for the auto-discovery init container | {} |
externalAccess.autoDiscovery.resources.requests | The requested resources for the auto-discovery init container | {} |
如何解决重新安装kafka导致应用程序需要重新配置bootstrap?
什么是bootstrap?
bootstrap.servers只是用于客户端启动(bootstrap)的时候有一个可以热启动的一个连接者,一旦启动完毕客户端就应该可以得知当前集群的所有节点的信息,日后集群扩展的时候客户端也能够自动实时的得到新节点的信息,即使bootstrap.servers里面的挂掉了也应该是能正常运行的,除非节点挂掉后客户端也重启了。
所以,bootstrap.servers参数,只在启动客户端kafka连接时有用,且列表里面多个项,只要一个可用就行了。
helm安装的kafka对外暴露了三个nodeport svc.,每次重启svc,这个nodeport就会变。导致应用连接不上。如何解决这个问题呢?
固定nodePort.
安装参数中加入如下三个参数:
externalAccess.service.nodePorts[0]=‘30787’,externalAccess.service.nodePorts[1]=‘31317’,externalAccess.service.nodePorts[2]=‘30113’
helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.nodePorts[0]='30787',externalAccess.service.nodePorts[1]='31317',externalAccess.service.nodePorts[2]='30113',externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/
kafka manager 可视化
访问地址: http://10.50.10.31:30900/
kafka manager yaml文件
注意: spec.env.ZK_HOSTS 一定要是你的zk集群名字。否则会一直连接不上.
apiVersion: v1 kind: Service metadata: name: kafka-manager namespace: kafka labels: app: kafka-manager spec: type: NodePort ports: - name: kafka port: 9000 targetPort: 9000 nodePort: 30900 selector: app: kafka-manager --- apiVersion: apps/v1 kind: Deployment metadata: name: kafka-manager namespace: kafka labels: app: kafka-manager spec: replicas: 1 selector: matchLabels: app: kafka-manager template: metadata: labels: app: kafka-manager spec: containers: - name: kafka-manager image: 10.50.10.185/kafka/ninesun0318/kafka-manager:1.3.3.22 imagePullPolicy: IfNotPresent ports: - name: kafka-manager containerPort: 9000 protocol: TCP env: - name: ZK_HOSTS value: "zookeeper-cluster:2181" # zk集群svc的名字 livenessProbe: httpGet: path: /api/health port: kafka-manager readinessProbe: httpGet: path: /api/health port: kafka-manager
kafka集群压测
kafka-producer-perf-test.sh --topic pressTest --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=10.50.10.33:30787,10.50.10.34:31317,10.50.10.35:30113 acks=1 # --record-size 每条消息大小 # --throughput -1 -1代表不限流 # acks=1 代表发送给broke 至少确认leader写入log成功才算发送成功
测试日志
I have no name!@kafka-client:/$ kafka-producer-perf-test.sh --topic pressTest --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=10.50.10.33:30787,10.50.10.34:31317,10.50.10.35:30113 acks=1 [2022-08-17 07:01:00,460] WARN [Producer clientId=perf-producer-client] Error while fetching metadata with correlation id 1 : {pressTest=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) 132271 records sent, 26454.2 records/sec (25.83 MB/sec), 786.4 ms avg latency, 1099.0 ms max latency. 198150 records sent, 39630.0 records/sec (38.70 MB/sec), 736.4 ms avg latency, 860.0 ms max latency. 180015 records sent, 36003.0 records/sec (35.16 MB/sec), 844.1 ms avg latency, 1048.0 ms max latency. 173220 records sent, 34644.0 records/sec (33.83 MB/sec), 889.0 ms avg latency, 1096.0 ms max latency. 208410 records sent, 41682.0 records/sec (40.71 MB/sec), 719.4 ms avg latency, 832.0 ms max latency. 1000000 records sent, 35870.578951 records/sec (35.03 MB/sec), 792.42 ms avg latency, 1099.00 ms max latency, 779 ms 50th, 1007 ms 95th, 1069 ms 99th, 1097 ms 99.9th.
kafka 清除log设置
安装kafka的时候需要设定一下这几个参数,否则不能达到自动删除topic的效果.
deleteTopicEnable | Switch to enable topic deletion or not | FALSE |
logRetentionBytes | A size-based retention policy for logs | _1073741824 |
logRetentionCheckIntervalMs | The interval at which log segments are checked to see if they can be deleted | 300000 |
logRetentionHours | The minimum age of a log file to be eligible for deletion due to age | 168 |
kafka-cluster-0 1/1 Running 0 4d20h
logRetentionHours 消息保留时间
deleteTopicEnable 是否自动删除,
Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off
如果没有设置 delete.topic.enable=true,则调用kafka 的delete命令无法真正将topic删除,而是显示(marked for deletion)
logRetentionCheckIntervalMs 删除检查周期
参考
https://blog.csdn.net/Hlroliu/article/details/113354423
https://artifacthub.io/packages/helm/bitnami/kafka
踩坑记
这个问题困扰了我两个多小时…
使用helm 安装zk集群,pod一直readliness 和liveness过不去,提示无法找到nc命令, 很是奇怪,这可是官方给的chart,怎么可能在build镜像时没有安装探针检查的命令呢?
后来才发现使用的zk版本不一致导致的。我使用的是3.4 的镜像,但是chart使用的是3.8的。所以以后helm安装软件时一定要注意这一点,镜像版本和chart版本要一致.
10.50.10.185/kafka/docker.io/bitnami/zookeeper:3.8.0-debian-11-r24 # 探针新增了nc命令 10.50.10.185/kafka/docker.io/bitnami/zookeeper:3.4.14-debian-9-r25
另外Stack Overflow上说这个探针失败的原因可能是:
没有设置initialDelaySeconds ,导致容器一致重启。
initialDelaySeconds:10指定容器启动10s之后开始执行
Liveness探测,我们一般会根据应用启动的准备时间来设置。比如某个应用正常启动要花30秒,那么initialDelaySeconds的值就应该大于30,这样容器就不至于一直重启。
kakfa资料多久会删除?
logRetentionHours 168h
kafka log在哪里看?
persistence.mountPath /opt/bitnami/kafka/logs