k8s部署kafka集群方案收集

博客一

一、概述

在k8s里面部署kafka、zookeeper这种有状态的服务&＃xff0c;不能使用deployment和RC&＃xff0c;k8s提供了一种专门用来部署这种有状态的服务的API–statefulset&＃xff0c;有状态简单来说就是需要持久化数据&＃xff0c;比如日志、数据库数据、服务状态等。
statefulset应用场景&＃xff1a;

• 稳定的持久化存储&＃xff0c;即Pod重新调度后还是能访问到相同的持久化数据&＃xff0c;基于PVC来实现
• 稳定的网络标志&＃xff0c;即Pod重新调度后其PodName和HostName不变&＃xff0c;基于Headless Service&＃xff08;即没有Cluster IP的Service&＃xff09;来实现
• 有序部署&＃xff0c;有序扩展&＃xff0c;即Pod是有顺序的&＃xff0c;在部署或者扩展的时候要依据定义的顺序依次依次进行&＃xff08;即从0到N-1&＃xff0c;在下一个Pod运行之
• 所有之前的Pod必须都是Running和Ready状态&＃xff09;&＃xff0c;基于init containers来实现
• 有序收缩&＃xff0c;有序删除&＃xff08;即从N-1到0&＃xff09;
  statefulset组成&＃xff1a;
• 用于定义网络标志&＃xff08;DNS domain&＃xff09;的Headless Service
• 用于创建PersistentVolumes的volumeClaimTemplates
• 定义具体应用的StatefulSet
  StatefulSet中每个Pod的DNS格式为statefulSetName-{0…N-1}.serviceName.namespace.svc.cluster.local&＃xff0c;其中
• 0…N-1为Pod所在的序号&＃xff0c;从0开始到N-1
• serviceName为Headless Service的名字
• statefulSetName为StatefulSet的名字
• namespace为服务所在的namespace&＃xff0c;Headless Servic和StatefulSet必须在相同的namespace
• .cluster.local为Cluster Domain

二、部署

首先下载相关的部署文件&＃xff1a;git clone https://github.com/cuishuaigit/k8s-kafka.git
部署前需要提前准备nfs提供动态存储&＃xff0c;参考我之前的博客&＃xff1a;https://www.cnblogs.com/cuishuai/p/9152277.html

cd k8s-kafka
kubectl create -f namesapce.yaml
kubectl create -f zk.yaml
kubectl create -f kafka.yaml


详细信息参考&＃xff1a;https://github.com/cuishuaigit/k8s-kafka

博客二

如何快速部署&＃xff0c;借助Helm&＃xff0c;只需少量操作即可部署kafka&＃xff1b;kafka和zookeeper对存储都有需求&＃xff0c;若提前准备了StorageClass&＃xff0c;存储一事则变得十分简单。本次实战涉及到的K8S、Helm、NFS、StorageClass等前置条件&＃xff0c;它们的安装和使用请参考&＃xff1a;
《kubespray2.11安装kubernetes1.15》
《部署和体验Helm(2.16.1版本)》
《Ubuntu16环境安装和使用NFS》
《K8S使用群晖DS218&＃43;的NFS》
《K8S的StorageClass实战(NFS)》

环境信息

本次实战的操作系统和软件的版本信息如下&＃xff1a;
Kubernetes&＃xff1a;1.15
Kubernetes宿主机&＃xff1a;CentOS Linux release 7.7.1908
NFS服务&＃xff1a;IP地址192.168.50.135&＃xff0c;文件夹/volume1/nfs-storageclass-test
Helm&＃xff1a;2.16.1
Kafka&＃xff1a;2.0.1
Zookeeper&＃xff1a;3.5.5
接下来的实战之前&＃xff0c;请您准备好&＃xff1a;K8S、Helm、NFS、StorageClass&＃xff1b;

操作

添加helm仓库(该仓库中有kafka)&＃xff1a;helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator
下载kafka的chart&＃xff1a;helm fetch incubator/kafka
下载成功后当前目录有个压缩包&＃xff1a;kafka-0.20.8.tgz&＃xff0c;解压&＃xff1a;tar -zxvf kafka-0.20.8.tgz
进入解压后的kafka目录&＃xff0c;编辑values.yaml文件&＃xff0c;下面是具体的修改点&＃xff1a;
首先要设置在K8S之外的也能使用kafka服务&＃xff0c;修改external.enabled的值&＃xff0c;改为true&＃xff1a;

找到configurationOverrides&＃xff0c;下图两个黄框中的内容原本是注释的&＃xff0c;请删除注释符号&＃xff0c;另外&＃xff0c;如果您之前设置过跨网络访问kafka&＃xff0c;就能理解下面写入K8S宿主机IP的原因了&＃xff1a;

接下来设置数据卷&＃xff0c;找到persistence&＃xff0c;按需要调整大小&＃xff0c;再设置已准备好的storageclass的名称&＃xff1a;

再设置zookeeper的数据卷&＃xff1a;

设置完成&＃xff0c;开始部署&＃xff0c;先创建namespace&＃xff0c;执行&＃xff1a;kubectl create namespace kafka-test
在kafka目录下执行&＃xff1a;helm install --name-template kafka -f values.yaml . --namespace kafka-test
如果前面的配置没有问题&＃xff0c;控制台提示如下所示&＃xff1a;

kafka启动依赖zookeeper&＃xff0c;整个启动会耗时数分钟&＃xff0c;期间可见zookeeper和kafka的pod逐渐启动&＃xff1a;

查看服务&＃xff1a;kubectl get services -n kafka-test&＃xff0c;如下图红框所示&＃xff0c;通过宿主机IP:31090、宿主机IP:31091、宿主机IP:31092即可从外部访问kafka&＃xff1a;

查看kafka版本&＃xff1a;kubectl exec kafka-0 -n kafka-test – sh -c ‘ls /usr/share/java/kafka/kafka_*.jar’ &＃xff0c;如下图红框所示&＃xff0c;scala版本2.11&＃xff0c;kafka版本2.0.1&＃xff1a;

kafka启动成功后&＃xff0c;咱们来验证服务是否正常&＃xff1b;

对外暴露zookeeper

为了远程操作kafka&＃xff0c;有时需要连接到zookeeper&＃xff0c;所以需要将zookeeper也暴露出来&＃xff1b;
创建文件zookeeper-nodeport-svc.yaml&＃xff0c;内容如下&＃xff1a;

apiVersion: v1
kind: Service
metadata:
name: zookeeper-nodeport
namespace: kafka-test
spec:
type: NodePort
ports:
- port: 2181
nodePort: 32181
selector:
app: zookeeper
release: kafka


执行命令&＃xff1a;kubectl apply -f zookeeper-nodeport-svc.yaml
查看服务&＃xff0c;发现已经可以通过宿主机IP:32181访问zookeeper了&＃xff0c;如下图&＃xff1a;

验证kafka服务
找一台电脑安装kafka包&＃xff0c;就能通过里面自带的命令远程连接和操作K8S的kafka了&＃xff1a;
访问kafka官网&＃xff1a;http://kafka.apache.org/downloads &＃xff0c;刚才确定了scala版本2.11&＃xff0c;kafka版本2.0.1&＃xff0c;因此下载下图红框中的版本&＃xff1a;

下载后解压&＃xff0c;进入目录kafka_2.11-2.0.1/bin
查看当前topic&＃xff1a;./kafka-topics.sh --list --zookeeper 192.168.50.135:32181
如下图&＃xff0c;空空如也&＃xff1a;

创建topic&＃xff1a;./kafka-topics.sh --create --zookeeper 192.168.50.135:32181 --replication-factor 1 --partitions 1 --topic test001
如下图&＃xff0c;创建成功后再查看topic终于有内容了&＃xff1a;

查看名为test001的topic&＃xff1a;./kafka-topics.sh --describe --zookeeper 192.168.50.135:32181 --topic test001

进入创建消息的交互模式&＃xff1a;./kafka-console-producer.sh --broker-list 192.168.50.135:31090 --topic test001
进入交互模式后&＃xff0c;输入任何字符串再输入回车&＃xff0c;就会将当前内容作为一条消息发送出去&＃xff1a;

再打开一个窗口&＃xff0c;执行命令消费消息&＃xff1a;./kafka-console-consumer.sh --bootstrap-server 192.168.50.135:31090 --topic test001 --from-beginning

再打开一个窗口&＃xff0c;执行命令查看消费者group&＃xff1a;./kafka-consumer-groups.sh --bootstrap-server 192.168.50.135:31090 --list
如下图可见groupid等于console-consumer-21022

执行命令查看groupid等于console-consumer-21022的消费情况&＃xff1a;./kafka-consumer-groups.sh --group console-consumer-21022 --describe --bootstrap-server 192.168.50.135:31090
如下图所示&＃xff1a;

远程连接kafka体验基本功能完毕&＃xff0c;查看、收发消息都正常&＃xff0c;证明本次部署成功&＃xff1b;

kafkacat连接

kafkacat是个客户端工具&＃xff0c;我这里是在MacBook Pro上用brew安装的&＃xff1b;
我这里K8S服务器IP是192.168.50.135&＃xff0c;因此执行此命令查看kafka信息&＃xff1a;kafkacat -b 192.168.50.135:31090 -L&＃xff0c;如下图&＃xff0c;可以看到broker信息&＃xff0c;以及topic信息(一个是test001&＃xff0c;还有一个是consumer的offset)&＃xff0c;把端口换成31091和31092会连接到另外两个broker&＃xff0c;也能得到相同信息&＃xff1a;

清理资源

本次实战创建了很多资源&＃xff1a;rbac、role、serviceaccount、pod、deployment、service&＃xff0c;下面的脚本可以将这些资源清理掉(只剩NFS的文件没有被清理掉)&＃xff1a;

helm del --purge kafka
kubectl delete service zookeeper-nodeport -n kafka-test
kubectl delete storageclass managed-nfs-storage
kubectl delete deployment nfs-client-provisioner -n kafka-test
kubectl delete clusterrolebinding run-nfs-client-provisioner
kubectl delete serviceaccount nfs-client-provisioner -n kafka-test
kubectl delete role leader-locking-nfs-client-provisioner -n kafka-test
kubectl delete rolebinding leader-locking-nfs-client-provisioner -n kafka-test
kubectl delete clusterrole nfs-client-provisioner-runner
kubectl delete namespace kafka-test


详细信息参考&＃xff1a;https://github.com/zq2599/blog_demos

在Kubernetes上运行Kafka合适吗&＃xff1f;

Kubernetes设计的初衷是运行无状态工作负载。这些通常采用微服务架构的工作负载&＃xff0c;是轻量级&＃xff0c;可水平扩展&＃xff0c;遵循十二要素应用程序&＃xff0c;可以处理环形断路和随机Monkey测试。
另一方面&＃xff0c;Kafka本质上是一个分布式数据库。这意味着你必须处理状态&＃xff0c;它比微服务更重量级。Kubernetes支持有状态的工作负载&＃xff0c;但你必须谨慎对待它&＃xff0c;正如Kelsey Hightower在最近的两条推文中指出的那样&＃xff1a;

现在你应该在Kubernetes上运行Kafka吗&＃xff1f;我的反问是&＃xff1a;没有它&＃xff0c;Kafka会跑得更好吗&＃xff1f;这就是为什么我要指出Kafka和Kubernetes之间的相互补充性以及你可能遇到的陷阱。

运行时

让我们先看一下基本的东西——运行时本身。

进程

Kafka brokers对CPU很友好。TLS可能会引入一些开销。如果Kafka客户端使用加密&＃xff0c;则需要更多CPU&＃xff0c;但这不会影响brokers。

内存

Kafka brokers是内存消耗大户。JVM堆通常可以限制为4-5 GB&＃xff0c;但由于Kafka大量使用页面缓存&＃xff0c;因此还需要足够的系统内存。在Kubernetes中&＃xff0c;可以相应地设置容器资源限制和请求。

存储

容器中的存储是短暂的——重启后数据将丢失。可以对Kafka数据使用emptyDir卷&＃xff0c;这将产生相同的效果&＃xff1a;brokers的数据将在停机后丢失。您的消息在其他broker上作为副本还是可以使用的。因此&＃xff0c;重新启动后&＃xff0c;失败的broker必须得复制所有的数据&＃xff0c;这可能是一个耗时过程。
这就是你应该使用持久存储的原因。使用XFS或ext4的非本地持久性块存储更合适。我警告你&＃xff1a;不要使用NFS。NFS v3和v4都不会起作用。简而言之&＃xff0c;Kafka broker会因为NFS“愚蠢重命名”问题而无法删除数据目录&＃xff0c;自行终止。如果你仍然不相信我&＃xff0c;那么请仔细阅读这篇博文。存储必须是非本地的&＃xff0c;以便Kubernetes在重新启动或重新定位时可以更灵活地选择另一个节点。

网络

与大多数分布式系统一样&＃xff0c;Kafka性能在很大程度上取决于低网络延迟和高带宽。不要试图将所有代理放在同一节点上&＃xff0c;因为这会降低可用性。如果Kubernetes节点出现故障&＃xff0c;那么整个Kafka集群都会出现故障。不要跨数据中心扩展Kafka集群。这同样适用于Kubernetes集群。不同的可用区域是一个很好的权衡。

配置清单

Kubernetes网站包含一个非常好的教程&＃xff0c;介绍如何使用清单设置ZooKeeper。由于ZooKeeper是Kafka的一部分&＃xff0c;因此可以通过这个了解哪些Kubernetes概念被应用在这里。一旦理解&＃xff0c;您也可以对Kafka集群使用相同的概念。
Pod&＃xff1a;Pod是Kubernetes中最小的可部署单元。它包含您的工作负载&＃xff0c;它代表群集中的一个进程。一个Pod包含一个或多个容器。整体中的每个ZooKeeper服务器和Kafka集群中的每个Kafka broker都将在一个单独的Pod中运行。
StatefulSet&＃xff1a;StatefulSet是一个Kubernetes对象&＃xff0c;用于处理需要协调的多个有状态工作负载。StatefulSets保证Pod的有序性和唯一性的。
Headless Services&＃xff1a;服务通过逻辑名称将Pod与客户端分离。Kubernetes负责负载平衡。但是&＃xff0c;对于ZooKeeper和Kafka等有状态工作负载&＃xff0c;客户端必须与特定实例进行通信。这就是 Headless Services发挥作用的地方&＃xff1a;作为客户端&＃xff0c;仍然可以获得逻辑名称&＃xff0c;但不必直接访问Pod。
持久卷&＃xff1a;如上所述&＃xff0c;需要配置非本地持久块存储。
Yolean提供了一套全面的清单&＃xff0c;可以帮助您开始使用Kubernetes上的Kafka。

Helm Charts

Helm是Kubernetes的包管理器&＃xff0c;类似yum&＃xff0c;apt&＃xff0c;Homebrew或Chocolatey等OS包管理器。它允许您安装Helm Charts中描述的预定义软件包。精心设计的Helm Charts能简化所有参数正确配置的复杂任务&＃xff0c;以便在Kubernetes上运行Kafka。有几张图表适用于Kafka的的可供选择&＃xff1a;一个是处于演进状态的官方图表&＃xff0c;一个来自Confluent&＃xff0c;另一个来自Bitnami&＃xff0c;仅举几例。

Operators

由于Helm的一些限制&＃xff0c;另一种工具变得非常流行&＃xff1a;Kubernetes Operators。Operators不仅可以为Kubernetes打包软件&＃xff0c;还可以为Kubernetes部署和管理一个软件。
评价很高的Operators名单中提到Kafka有两个&＃xff0c;其中一个是Strimzi&＃xff0c;Strimzi使得在几分钟内启动Kafka集群变得非常容易&＃xff0c;几乎不需要任何配置&＃xff0c;它增加了一些漂亮的功能&＃xff0c;如群集间点对点TLS加密。Confluent还宣布即将推出新的Operator。

性能

运行性能测试以对Kafka安装进行基准测试非常重要。在您遇到麻烦之前&＃xff0c;它会为您提供有关可能的瓶颈的地方。幸运的是&＃xff0c;Kafka已经提供了两个性能测试工具&＃xff1a;kafka-producer-perf-test.sh和kafka-consumer-perf-test.sh。记得经常使用它们。作为参考&＃xff0c;可以使用Jay Kreps的博客结果&＃xff0c;或者 Stéphane Maarek在 Amazon MSK的评论。

运维监控

可见性非常重要&＃xff0c;否则您将不知道发生了什么。如今&＃xff0c;有一种不错的工具可以用云原生方式监控指标。Prometheus和Grafana是两种流行的工具。Prometheus可以直接从JMX导出器收集所有Java进程&＃xff08;Kafka&＃xff0c;ZooKeeper&＃xff0c;Kafka Connect&＃xff09;的指标。添加cAdvisor指标可为提供有关Kubernetes资源使用情况的其他信息。
Strimzi为Kafka提供了一个优雅的Grafana仪表板示例。它以非常直观的方式可视化关键指标&＃xff0c;如未复制的和离线分区。它通过资源使用和性能以及稳定性指标来补充这些指标。因此&＃xff0c;可以免费获得基本的Kafka集群监控&＃xff01;
可以通过客户端监控&＃xff08;消费者和生产者指标&＃xff09;&＃xff0c;使用Burrow滞后监控&＃xff0c;使用Kafka Monitor进行端到端监控&＃xff0c;来完成这个任务

日志记录

日志记录是另一个关键部分。确保Kafka安装中的所有容器都记录到标准输出&＃xff08;stdout&＃xff09;和标准错误输出&＃xff08;stderr&＃xff09;&＃xff0c;并确保Kubernetes集群将所有日志聚合到中央日志记录设施中如Elasticsearch中。

健康检查

Kubernetes使用活跃度和就绪探测器来确定Pod是否健康。如果活跃度探测失败&＃xff0c;Kubernetes将终止容器并在相应设置重启策略时自动重启。如果准备就绪探测失败&＃xff0c;那么Kubernetes将通过服务从服务请求中删除该Pod。这意味着在这种情况下不再需要人工干预&＃xff0c;这是一大优点。

滚动更新

StatefulSets支持自动更新&＃xff1a;滚动更新策略将一次更新一个Kafka Pod。通过这种方式&＃xff0c;可以实现零停机时间&＃xff0c;这是Kubernetes带来的另一大优势。

扩展

扩展Kafka集群并非易事。但是&＃xff0c;Kubernetes可以很容易地将Pod缩放到一定数量的副本&＃xff0c;这意味着可以声明式地定义所需数量的Kafka brokers。困难的部分是在放大或缩小之前重新分配部分。同样&＃xff0c;Kubernetes可以帮助您完成这项任务。

管理

通过在Pod中打开shell&＃xff0c;可以使用现有的shell脚本完成Kafka群集的管理任务&＃xff0c;例如创建主题和重新分配分区。这不是一个很好的解决方案。Strimzi支持与另一个Operator管理主题。这还有改进的余地。

备份和还原

现在Kafka的可用性还取决于Kubernetes的可用性。如果Kubernetes群集出现故障&＃xff0c;那么在最坏的情况下Kafka群集也会故障。墨菲定律告诉我们&＃xff0c;这也会发生在你身上&＃xff0c;你会丢失数据。要降低此风险&＃xff0c;请确保您具有备份想法。MirrorMaker是一种可选方案&＃xff0c;另一种可能是利用S3进行连接备份&＃xff0c;如Zalando的博客文章所述。

结论

对于中小型Kafka集群&＃xff0c;我肯定会选择Kubernetes&＃xff0c;因为它提供了更大的灵活性并简化了操作。如果您在延迟和/或吞吐量方面具有非常高的非功能性要求&＃xff0c;则不同的部署选项可能更有益。

https://github.com/kubernetes-retired/contrib/tree/master/statefulsets/kafka
https://cloud.tencent.com/developer/article/1005492
https://www.kubernetes.org.cn/5032.html