2019-07-19

Kubernetes 部署 Kafka & Zookeeper & Kafka Manager

系统环境：

Kubernetes 版本：1.14.0
kafka 版本：2.3.0
zookeeper 版本：3.4.14
kafka manager 版本：1.3.3
示例部署文件 Github 地址：https://github.com/my-dlq/blog-example/tree/master/kubernetes/kafka-zookeeper-deploy

一、简介

Kafka 简介

Kafka 是由 Apache 软件基金会开发的一个开源流处理平台，由 Scala 和 Java 编写。它是一个分布式、支持分区的、多副本，基于 zookeeper 协调的分布式消息系统。它最大特性是可以实时的处理大量数据以满足各种需求场景，比如基于 Hadoop 的批处理系统、低延时的实时系统、storm/Spark 流式处理引擎，web/nginx 日志，访问日志，消息服务等等。

Zookeeper 简介

ZooKeeper 是一种分布式协调服务，用于管理大型主机。在分布式环境中协调和管理服务是一个复杂的过程。ZooKeeper 通过其简单的架构和 API 解决了这个问题。 ZooKeeper 允许开发人员专注于核心应用程序逻辑，而不必担心应用程序的分布式特性。

Kafka 中主要利用 zookeeper 解决分布式一致性问题。Kafka 使用 Zookeeper 的分布式协调服务将生产者，消费者，消息储存结合在一起。同时借助 Zookeeper，Kafka 能够将生产者、消费者和 Broker 在内的所有组件在无状态的条件下建立起生产者和消费者的订阅关系，实现生产者的负载均衡。

Kafka Manager 简介

Kafka Manager 是目前最受欢迎的 Kafka 集群管理工具，最早由雅虎开源，用户可以在 Web 界面执行一些简单的集群管理操作。

支持以下功能：

管理 Kafka 集群
方便集群状态监控 (包括topics, consumers, offsets, brokers, replica distribution, partition distribution)
方便选择分区副本
配置分区任务，包括选择使用哪些 Brokers
可以对分区任务重分配
提供不同的选项来创建及删除 Topic
Topic List 会指明哪些topic被删除
批量产生分区任务并且和多个topic和brokers关联
批量运行多个主题对应的多个分区
向已经存在的主题中添加分区
对已经存在的 Topic 修改配置
可以在 Broker Level 和 Topic Level 的度量中启用 JMX Polling 功能
可以过滤在 ZooKeeper 上没有 ids/owners/offsets/directories 的 consumer

二、部署过程

这个流程需要部署三个组件，分别为 Zookeeper、Kafka、Kafka Manager：

(1)、Zookeeper： 首先部署 Zookeeper，方便后续部署 Kafka 节点注册到 Zookeeper，用 StatefulSet 方式部署三个节点。
(2)、Kafka： 第二个部署的是 Kafka，设置环境变量来指定 Zookeeper 地址，用 StatefulSet 方式部署。
(3)、Kafka Manager： 最后部署的是 Kafka Manager，用 Deployment 方式部署，然后打开 Web UI 界面来管理、监控 Kafka。

三、Kubernetes 部署 Zookeeper & Kafka & Kafka Manager

1、创建 StorageClass

由于都是使用 StatefulSet 方式部署的有状态服务，所以 Kubernetes 集群需要提前设置一个 StorageClass 方便后续部署时指定存储分配(如果想指定为已经存在的 StorageClass 创建 PV 则跳过此步骤)。

此处用的是 NFS 存储驱动，如果是其它存储需要提前设置好相关配置

创建 StorageClass 部署文件

nfs-storage.yaml

1
apiVersion: storage.k8s.io/v1
2
kind: StorageClass
3
metadata:
4
  name: nfs-storage
5
provisioner: nfs-client    #动态卷分配服务指定的名称
6
parameters:
7
  archiveOnDelete: "true"  #设置为"false"时删除PVC不会保留数据,"true"则保留数据
8
mountOptions:
9
  - hard                   #指定为硬挂载方式
10
  - nfsvers=4              #指定NFS版本

部署 StorageClass

1
$ kubectl apply -f nfs-storage.yaml

2、Kubernetes 部署 Zookeeper

创建 Zookeeper 部署文件

zookeeper.yaml

1
#部署 Service Headless，用于Zookeeper间相互通信
2
apiVersion: v1
3
kind: Service
4
metadata:
5
  name: zookeeper-headless
6
  labels:
7
    app: zookeeper
8
spec:
9
  type: ClusterIP
10
  clusterIP: None
11
  publishNotReadyAddresses: true
12
  ports:
13
  - name: client
14
    port: 2181
15
    targetPort: client
16
  - name: follower
17
    port: 2888
18
    targetPort: follower
19
  - name: election
20
    port: 3888
21
    targetPort: election
22
  selector:
23
    app: zookeeper
24
---
25
#部署 Service，用于外部访问 Zookeeper
26
apiVersion: v1
27
kind: Service
28
metadata:
29
  name: zookeeper
30
  labels:
31
    app: zookeeper
32
spec:
33
  type: ClusterIP
34
  ports:
35
  - name: client
36
    port: 2181
37
    targetPort: client
38
  - name: follower
39
    port: 2888
40
    targetPort: follower
41
  - name: election
42
    port: 3888
43
    targetPort: election
44
  selector:
45
    app: zookeeper
46
---
47
apiVersion: apps/v1
48
kind: StatefulSet
49
metadata:
50
  name: zookeeper
51
  labels:
52
    app: zookeeper
53
spec:
54
  serviceName: zookeeper-headless
55
  replicas: 3
56
  podManagementPolicy: Parallel
57
  updateStrategy:
58
    type: RollingUpdate
59
  selector:
60
    matchLabels:
61
      app: zookeeper
62
  template:
63
    metadata:
64
      name: zookeeper
65
      labels:
66
        app: zookeeper
67
    spec:
68
      securityContext:
69
        fsGroup: 1001
70
      containers:
71
      - name: zookeeper
72
        image: docker.io/bitnami/zookeeper:3.4.14-debian-9-r25
73
        imagePullPolicy: IfNotPresent
74
        securityContext:
75
          runAsUser: 1001
76
        command:
77
         - bash
78
         - -ec
79
         - |
80
            # Execute entrypoint as usual after obtaining ZOO_SERVER_ID based on POD hostname
81
            HOSTNAME=`hostname -s`
82
            if [[ $HOSTNAME =~ (.*)-([0-9]+)$ ]]; then
83
              ORD=${BASH_REMATCH[2]}
84
              export ZOO_SERVER_ID=$((ORD+1))
85
            else
86
              echo "Failed to get index from hostname $HOST"
87
              exit 1
88
            fi
89
            . /opt/bitnami/base/functions
90
            . /opt/bitnami/base/helpers
91
            print_welcome_page
92
            . /init.sh
93
            nami_initialize zookeeper
94
            exec tini -- /run.sh
95
        resources:
96
          limits:
97
            cpu: 500m
98
            memory: 512Mi
99
          requests:
100
            cpu: 250m
101
            memory: 256Mi
102
        env:
103
        - name: ZOO_PORT_NUMBER
104
          value: "2181"
105
        - name: ZOO_TICK_TIME
106
          value: "2000"
107
        - name: ZOO_INIT_LIMIT
108
          value: "10"
109
        - name: ZOO_SYNC_LIMIT
110
          value: "5"
111
        - name: ZOO_MAX_CLIENT_CNXNS
112
          value: "60"
113
        - name: ZOO_SERVERS
114
          value: "
115
                  zookeeper-0.zookeeper-headless:2888:3888,
116
                  zookeeper-1.zookeeper-headless:2888:3888,
117
                  zookeeper-2.zookeeper-headless:2888:3888
118
                 "
119
        - name: ZOO_ENABLE_AUTH
120
          value: "no"
121
        - name: ZOO_HEAP_SIZE
122
          value: "1024"
123
        - name: ZOO_LOG_LEVEL
124
          value: "ERROR"
125
        - name: ALLOW_ANONYMOUS_LOGIN
126
          value: "yes"
127
        ports:
128
        - name: client
129
          containerPort: 2181
130
        - name: follower
131
          containerPort: 2888
132
        - name: election
133
          containerPort: 3888
134
        livenessProbe:
135
          tcpSocket:
136
            port: client
137
          initialDelaySeconds: 30
138
          periodSeconds: 10
139
          timeoutSeconds: 5
140
          successThreshold: 1
141
          failureThreshold: 6
142
        readinessProbe:
143
          tcpSocket:
144
            port: client
145
          initialDelaySeconds: 5
146
          periodSeconds: 10
147
          timeoutSeconds: 5
148
          successThreshold: 1
149
          failureThreshold: 6
150
        volumeMounts:
151
        - name: data
152
          mountPath: /bitnami/zookeeper
153
  volumeClaimTemplates:
154
    - metadata:
155
        name: data
156
        annotations:
157
      spec:
158
        storageClassName: nfs-storage    #指定为上面创建的 storageclass
159
        accessModes:
160
          - ReadWriteOnce
161
        resources:
162
          requests:
163
            storage: 5Gi

部署 Zookeeper

-n：指定应用启动的 Namespace,替换自己集群的 Namespace

1
$ kubectl apply -f zookeeper.yaml -n mydlqcloud

3、Kubernetes 部署 Kafka

创建 Kafka 部署文件

kafka.yaml

1
#部署 Service Headless，用于Kafka间相互通信
2
apiVersion: v1
3
kind: Service
4
metadata:
5
  name: kafka-headless
6
  labels:
7
    app: kafka
8
spec:
9
  type: ClusterIP
10
  clusterIP: None
11
  ports:
12
  - name: kafka
13
    port: 9092
14
    targetPort: kafka
15
  selector:
16
    app: kafka
17
---
18
#部署 Service，用于外部访问 Kafka
19
apiVersion: v1
20
kind: Service
21
metadata:
22
  name: kafka
23
  labels:
24
    app: kafka
25
spec:
26
  type: ClusterIP
27
  ports:
28
  - name: kafka
29
    port: 9092
30
    targetPort: kafka
31
  selector:
32
    app: kafka
33
---
34
apiVersion: apps/v1
35
kind: StatefulSet
36
metadata:
37
  name: "kafka"
38
  labels:
39
    app: kafka
40
spec:
41
  selector:
42
    matchLabels:
43
      app: kafka
44
  serviceName: kafka-headless
45
  podManagementPolicy: "Parallel"
46
  replicas: 3
47
  updateStrategy:
48
    type: "RollingUpdate"
49
  template:
50
    metadata:
51
      name: "kafka"
52
      labels:
53
        app: kafka
54
    spec:
55
      securityContext:
56
        fsGroup: 1001
57
        runAsUser: 1001
58
      containers:
59
      - name: kafka
60
        image: "docker.io/bitnami/kafka:2.3.0-debian-9-r4"
61
        imagePullPolicy: "IfNotPresent"
62
        resources:
63
          limits:
64
            cpu: 500m
65
            memory: 512Mi
66
          requests:
67
            cpu: 250m
68
            memory: 256Mi
69
        env:
70
        - name: MY_POD_IP
71
          valueFrom:
72
            fieldRef:
73
              fieldPath: status.podIP
74
        - name: MY_POD_NAME
75
          valueFrom:
76
            fieldRef:
77
              fieldPath: metadata.name
78
        - name: KAFKA_CFG_ZOOKEEPER_CONNECT
79
          value: "zookeeper"                 #Zookeeper Service 名称
80
        - name: KAFKA_PORT_NUMBER
81
          value: "9092"
82
        - name: KAFKA_CFG_LISTENERS
83
          value: "PLAINTEXT://:$(KAFKA_PORT_NUMBER)"
84
        - name: KAFKA_CFG_ADVERTISED_LISTENERS
85
          value: 'PLAINTEXT://$(MY_POD_NAME).kafka-headless:$(KAFKA_PORT_NUMBER)'
86
        - name: ALLOW_PLAINTEXT_LISTENER
87
          value: "yes"
88
        - name: KAFKA_HEAP_OPTS
89
          value: "-Xmx512m -Xms512m"
90
        - name: KAFKA_CFG_LOGS_DIRS
91
          value: /opt/bitnami/kafka/data
92
        - name: JMX_PORT
93
          value: "9988"
94
        ports:
95
        - name: kafka
96
          containerPort: 9092
97
        livenessProbe:
98
          tcpSocket:
99
            port: kafka
100
          initialDelaySeconds: 10
101
          periodSeconds: 10
102
          timeoutSeconds: 5
103
          successThreshold: 1
104
          failureThreshold: 2
105
        readinessProbe:
106
          tcpSocket:
107
            port: kafka
108
          initialDelaySeconds: 5
109
          periodSeconds: 10
110
          timeoutSeconds: 5
111
          successThreshold: 1
112
          failureThreshold: 6
113
        volumeMounts:
114
        - name: data
115
          mountPath: /bitnami/kafka
116
  volumeClaimTemplates:
117
    - metadata:
118
        name: data
119
      spec:
120
        storageClassName: nfs-storage    #指定为上面创建的 storageclass
121
        accessModes:
122
          - "ReadWriteOnce"
123
        resources:
124
          requests:
125
            storage: 5Gi

部署 Kafka

-n：指定应用启动的 Namespace,替换自己集群的 Namespace

1
$ kubectl apply -f kafka.yaml -n mydlqcloud

4、Kubernetes 部署 Kafka Manager

创建 Kafka Manager 部署文件

kafka-manager.yaml

1
apiVersion: v1
2
kind: Service
3
metadata:
4
  name: kafka-manager
5
  labels:
6
    app: kafka-manager
7
spec:
8
  type: NodePort
9
  ports:
10
  - name: kafka
11
    port: 9000
12
    targetPort: 9000
13
    nodePort: 30900
14
  selector:
15
    app: kafka-manager
16
---
17
apiVersion: apps/v1
18
kind: Deployment
19
metadata:
20
  name: kafka-manager
21
  labels:
22
    app: kafka-manager
23
spec:
24
  replicas: 1
25
  selector:
26
    matchLabels:
27
      app: kafka-manager
28
  template:
29
    metadata:
30
      labels:
31
        app: kafka-manager
32
    spec:
33
      containers:
34
      - name: kafka-manager
35
        image: zenko/kafka-manager:1.3.3.22
36
        imagePullPolicy: IfNotPresent
37
        ports:
38
        - name: kafka-manager
39
          containerPort: 9000
40
          protocol: TCP
41
        env:
42
        - name: ZK_HOSTS
43
          value: "zookeeper:2181"
44
        livenessProbe:
45
          httpGet:
46
            path: /api/health
47
            port: kafka-manager
48
        readinessProbe:
49
          httpGet:
50
            path: /api/health
51
            port: kafka-manager
52
        resources:
53
          limits:
54
            cpu: 500m
55
            memory: 512Mi
56
          requests:
57
            cpu: 250m
58
            memory: 256Mi

部署 Kafka Manager

-n：指定应用启动的 Namespace,替换自己集群的 Namespace

1
$ kubectl apply -f kafka-manager.yaml -n mydlqcloud

四、进入 Kafka Manager 管理 Kafka 集群

这里的 Kubernetes 集群地址为：192.168.2.11，并且在上面设置 Kafka-Manager 网络策略为 NodePort 方式，且设置端口为 30900，这里输入地址：http://192.168.2.11:30900 访问 Kafka Manager。

进入后先配置 Kafka Manager，增加一个 Zookeeper 地址。

配置三个必填参数:

Cluster Name：自定义一个名称，任意输入即可。
Zookeeper Hosts：输入 Zookeeper 地址，这里设置为 Zookeeper 服务名+端口。
Kafka Version：选择 kafka 版本。

配置完成后就可以看到新增了一条记录，点进去就可以查看相关集群信息。

五、附录：镜像参数配置

Zookeeper 镜像可配置变量参数

镜像 github 地址：https://github.com/bitnami/bitnami-docker-zookeeper

参数名称	描述
ZOO_PORT_NUMBER	Zookeeper客户端端口。默认值：2181
ZOO_SERVER_ID	集合中服务器的ID。默认值：1
ZOO_TICK_TIME	ZooKeeper用于心跳的基本时间单位（以毫秒为单位）。默认值：2000
ZOO_INIT_LIMIT	ZooKeeper用于限制仲裁中ZooKeeper服务器连接到领导者的时间长度。默认值：10
ZOO_SYNC_LIMIT	服务器与领导者的过时距离。默认值：5
ZOO_MAX_CLIENT_CNXNS	限制单个客户端可能对ZooKeeper集合的单个成员进行的并发连接数。默认60
ZOO_SERVERS	逗号，空格或冒号分隔的服务器列表。示例：zoo1：2888：3888，zoo2：2888：3888。没有默认值。
ZOO_CLIENT_USER	将使用Zookeeper客户端进行身份验证的用户。默认值：无默认值。
ZOO_CLIENT_PASSWORD	将使用Zookeeper客户端进行身份验证的密码。没有默认值。
ZOO_SERVER_USERS	逗号，分号或空格分隔的要创建的用户列表。示例：user1，user2，admin。没有默认值
ZOO_SERVER_PASSWORDS	逗号，半精或空格分隔的密码列表，在创建时分配给用户。示例：pass4user1，pass4user2，pass4admin。没有默认值
ZOO_ENABLE_AUTH	启用Zookeeper身份验证。它使用SASL / Digest-MD5。默认值：否
ZOO_RECONFIG_ENABLED	启用ZooKeeper动态重配置。默认值：否
ZOO_HEAP_SIZE	Java堆选项（Xmx和XM）的大小（MB）。如果通过Xmx配置Xm，则忽略此env var JVMFLAGS。默认值：1024
ZOO_LOG_LEVEL	Zookeeper日志级别。可用级别为：ALL，DEBUG，INFO，WARN，ERROR，FATAL，OFF，TRACE。默认值：INFO
ALLOW_ANONYMOUS_LOGIN	如果设置为true，则允许接受来自未经身份验证的用户的连接。默认值：否
JVMFLAGS	ZooKeeper进程的默认JVMFLAGS。没有默认值

Kafka 镜像可配置变量参数

镜像 github 地址：https://github.com/bitnami/bitnami-docker-kafka

参数名称	描述
KAFKA_CFG_ZOOKEEPER_CONNECT	Zookeeper集群地址,例如”zookeeper:2181”
KAFKA_PORT_NUMBER	Kafka端口,例如”9092”
ALLOW_PLAINTEXT_LISTENER	是否启用Plaintext侦听器,默认”false”
KAFKA_CFG_LISTENERS	Kafka 监听列表，broker对外提供服务时绑定的IP和端口
KAFKA_CFG_ADVERTISED_LISTENERS	给客户端用的发布至zookeeper的监听，broker 会上送此地址到zookeeper，zookeeper会将此地址提供给消费者，消费者根据此地址获取消息。
KAFKA_HEAP_OPTS	Java JVM堆内存大小配置，例如”-Xmx512m -Xms512m”
KAFKA_CFG_LOGS_DIRS	Kafka 日志存储目录
JMX_PORT	JMX端口配置，设置此参数才能开启JMX,例如设置为”9988”
KAFKA_CFG_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM	是否启用基于主机名的认证认证，例如想开启可以设置为”https”
KAFKA_CFG_BROKER_ID	Broker ID值，每个节点值都唯一。默认为”-1”,自动生成BrokerId
KAFKA_CFG_DELETE_TOPIC_ENABLE	是否允许删除Topic，默认”false”
KAFKA_CFG_LOG_FLUSH_INTERVAL_MESSAGES	此项配置指定时间间隔，强制进行fsync日志，默认”10000”
KAFKA_CFG_LOG_FLUSH_INTERVAL_MS	此项配置用来置顶强制进行fsync日志到磁盘的时间间隔”1000”
KAFKA_CFG_LOG_RETENTION_BYTES	每个Topic下每个Partition保存数据的总量，超过限制都会删除一个段文件,默认”1073741824”
KAFKA_CFG_LOG_RETENTION_CHECK_INTERVALS_MS	检查日志分段文件的间隔时间，以确定是否文件属性是否到达删除要求,默认”300000”
KAFKA_CFG_LOG_RETENTION_HOURS	每个日志文件删除之前保存的时间，默认数据保存时间对所有topic都一样,默认”168”
KAFKA_CFG_LOG_MESSAGE_FORMAT_VERSION	指定broker将用于将消息添加到日志文件的消息格式版本
KAFKA_CFG_MAX_MESSAGE_BYTES	Kafka允许的最大记录批大小。默认”1000000”
KAFKA_CFG_SEGMENT_BYTES	日志段文件的最大大小。当达到这个大小时，将创建一个新的日志段。默认”1073741824”
KAFKA_CFG_DEFAULT_REPLICATION_FACTOR	对replica的数目进行配置，默认值为”1”，表示不对topic进行备份。如果配置为2，表示除了leader节点，对于topic里的每一个partition，都会有一个额外的备份。
KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR	Topic的offset的备份份数。建议设置更高的数字保证更高的可用性
KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR	事务Topic的复制因子(设置得更高以确保可用性)
KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR	事务Topic的副本数
KAFKA_CFG_NUM_IO_THREADS	用来处理请求的I/O线程的数目
KAFKA_CFG_NUM_NETWORK_THREADS	用来处理网络请求的网络线程数目
KAFKA_CFG_NUM_PARTITIONS	每个主题的日志分区的默认数量
KAFKA_CFG_NUM_RECOVERY_THREADS_PER_DATA_DIR	每个数据目录中的线程数，用于在启动时日志恢复，并在关闭时刷新。默认”1”
KAFKA_CFG_SOCKET_RECEIVE_BUFFER_BYTES	SO_RCVBUFF缓存大小，server进行socket连接时所用
KAFKA_CFG_SOCKET_REQUEST_MAX_BYTES	允许的最大请求尺寸，这将避免server溢出，它应该小于Java heap size，默认”104857600”
KAFKA_CFG_SOCKET_SEND_BUFFER_BYTES	Kafka追加消息的最大尺寸
KAFKA_CFG_ZOOKEEPER_CONNECT_TIMEOUT_MS	连接到zookeeper的超时时间(ms)

---END---

如果本文对你有帮助，可以关注我的公众号 "小豆丁技术栈" 了解最新动态，顺便也请帮忙 Github 点颗星哦，感谢~