kafka/pulsar on k8s

大数据启示录 2022-06-03

983

pulsar on k8s 部署

环境准备

这里使用的 Kubernetes 集群的版本是 1.22.4，Helm 的版本是 3.7.1。

Pulsar 集群组件和 K8S Node 节点规划

下面做一下 Pulsar 集群各个组件部署节点的规划。使用 Pulsar 官方的 Helm Chart 部署时，可选择部署各个组件。在后边的配置中将禁用监控相关的组件(promethues, grafana 等)，我们这里选择以后尝试使用外部的全局 Prometheus 实现对 Pulsar 集群的监控。

本节选择部署的集群组件如下:

proxy: 无状态, 但 pulsar 的 helm chart 使用 StatefulSet 部署
broker: 无状态, 但 pulsar 的 helm chart 使用 StatefulSet 部署
bookie: 有状态, pulsar 的 helm chart 使用 StatefulSet 部署
zookeeper: 有状态, pulsar 的 helm chart 使用 StatefulSet 部署
recovery: 无状态, 但 pulsar 的 helm chart 使用 StatefulSet 部署
toolset: 无状态, 但 pulsar 的 helm chart 使用 StatefulSet 部署
pulsar-manager: 无状态, pulsar 的 helm chart 使用 Deployment 部署

注意, pulsar-managers 虽然是无状态的，但因为它需要使用 PostgreSQL 数据库，pulsar-managers 的 docker 镜像中内置一个 PostgreSQL, 这个我们在后边的配置中将改为使用集群外部的 PostgreSQL。

下面说一下以上各个组件的部署节点选择。

对于 proxy, broker, recovery, toolset, pulsar-manager 这 5 个无状态组件，可以让 k8s 将其调度到任意节点上。
对于 bookie, zookeeper 这 2 个有状态组件，需要我们根据其存储卷的类型，将其规划到合适的 k8s 节点。

我们在线上环境对于有状态服务的部署，在存储卷的选择上，为了更好的性能，一般都是选择 Local Persistent Volumes 在。因此，如果你在规划一个线上版本的 Pulsar 集群部署的话，对于 bookie 和 zookeeper 肯定需要单独的独立的 k8s 节点，并使用这些节点上创建的 Local PV。例如，一个线上生产可用的 Pulsar 集群可能规划如下:

pulsar zookeeper 集群至少需要 3 个独立的 k8s 节点, 在这些节点上创建 zookeeper 的 local pv
pulsar bookeeper 集群(bookie 节点组成)根据规划的容量需要 N 个独立的 k8s 节点, 在这些节点上创建 bookie 的 local pv。如果后续需要扩容增加 bookie 节点时，只需要有新的创建好 local pv 的 k8s 节点，并对 bookie 的 StatefulSet 扩容即可。
pulsar proxy, broker 等无状态服务，只需要有足够的数量的 k8s 节点，并在需要时按需扩容即可

因本文这里用于实验的 k8s 集群资源有限，所以尽量将上面各组件在 3 个 k8s 节点上混部，将一个用于测试环境的的 Pulsar 集群规划如下:

基于上面测试环境的规划，我们将 node1~node3 三个节点打上 Label 和 Taint:

$ kubectl label node node1 node-role.kubernetes.io/pulsar=pulsar
$ kubectl label node node2 node-role.kubernetes.io/pulsar=pulsar
$ kubectl label node node3 node-role.kubernetes.io/pulsar=pulsar
$ kubectl taint nodes node1 dedicated=pulsar:NoSchedule
$ kubectl taint nodes node2 dedicated=pulsar:NoSchedule
$ kubectl taint nodes node3 dedicated=pulsar:NoSchedule

Label node-role.kubernetes.io/pulsar=pulsar
用于标记节点是专门用于运行 pulsar 集群组件的 k8s 节点。
Taint dedicated=pulsar:NoSchedule
被打到节点上后，默认配置下 k8s 集群中的其他不属于 pulsar 集群组件的 pod 将不会被调度到这 3 个节点上，而后边我们将要部署的 pulsar 组件上将会使用 Toleration 配置允许dedicated=pulsar:NoSchedule
的 Taint。
注意这里只是根据测试环境 Pulsar 集群的规划，做了上面的 Label 和 Taint 的设置，如果是生产环境，这里的 Label 和 Taint 应该做更合理和细粒度的规划，确保实现上面生产可用 Pulsar 集群的 Node 节点规划

Pulsar 集群组件容器镜像准备

前面我们选择要部署 Pulsar 集群的 proxy, broker, bookie, zookeeper, recovery, toolset, pulsar-manager 7 大组件。

其中 proxy, broker, bookie, zookeeper, recovery, toolset 的官方容器镜像都是apachepulsar/pulsar-all^[1]。pulsar-manager 的官方镜像是apachepulsar/pulsar-manager^[2]。

本文使用的 pulsar 官方的 helm chart https://github.com/apache/pulsar-helm-chart/releases。

pulsar-helm-chart 的版本为 2.7.7，该版本中 pulsar 的版本为 2.7.4, pulsar-manager 版本为 v0.1.0:

apachepulsar/pulsar-all:2.7.4
apachepulsar/pulsar-manager:v0.1.0

注意因为pulsar-manager:v0.1.0
有这个 ISSUE https://github.com/apache/pulsar-helm-chart/issues/133中描述的问题，所以在后边的部署将镜像pulsar-manager:v0.1.0
更换成了pulsar-manager:v0.2.0
。

为了提高效率，这里将 apachepulsar/pulsar-all:2.7.4 和 apachepulsar/pulsar-manager:v0.2.0 这两个镜像转存到了 k8s 集群所使用的私有镜像仓库中，例如:

harbor.example.com/library/apachepulsar/pulsar-all:2.7.4
harbor.example.com/library/apachepulsar/pulsar-manager:v0.2.0

创建 JWT 认证所需的 K8S Secret

这里部署的 Pulsar 集群需要在安全上开通 JWT 认证。根据前面学习的内容，JWT 支持通过两种不同的秘钥生成和验证 Token:

对称秘钥：

使用单个 Secret key 来生成和验证 Token

非对称秘钥：包含由私钥和公钥组成的一对密钥
使用 Private key 生成 Token
使用 Public key 验证 Token

推荐使用非对称密钥的方式，需要先生成密钥对，再用秘钥生成 token。因为 Pulsar 被部署在 K8S 集群中，在 K8S 集群中存储这些秘钥和 Token 的最好的方式是使用 K8S 的 Secret。

pulsar-helm-chart 专门提供了一个prepare_helm_release.sh
脚本，可以用来生成这些 Secret。

下面我们将 pulsar-helm-chart 的源码 clone 到 K8S 的控制节点上(kubectl 和 helm 可用的节点):

$ git clone -b pulsar-2.7.7 --depth 1 https://github.com/apache/pulsar-helm-chart.git
$ cd pulsar-helm-chart/

执行下面的命令生成秘钥对和 Token 的 Secret 的 Manifest:

$ ./scripts/pulsar/prepare_helm_release.sh \
    -n pulsar \
    -k pulsar \
    -l

上面的命令中:

-n
指定的生成 Secret Manifest 中安装的命名空间，这里我是将其部署到 K8S 中的 pulsar namespace 中，所以指定为 pulsar，当然也可以指定部署到其他的 namespace 中。
-k
指定的是使用 helm 部署时的 helm release 名称，这里指定为 pulsar。
-l
指定只将生成的内容输出达到本地，而不会自动部署到 K8S 中。比较喜欢这种手动的方式，因为一切比较可控。
注意这个脚本还有一个-s,--symmetric
参数，如果给这个参数的话，JWT 认证将使用对称秘钥的方式，这里没有给这个参数，就使用非对称秘钥的方式。

执行上面的脚本会输出以下内容:

generate the token keys for the pulsar cluster
---
The private key and public key are generated to ... successfully.
apiVersion: v1
data:
  PRIVATEKEY: <...>
  PUBLICKEY: <...>
kind: Secret
metadata:
  creationTimestamp: null
  name: pulsar-token-asymmetric-key
  namespace: pulsar
generate the tokens for the super-users: proxy-admin,broker-admin,admin
generate the token for proxy-admin
---
pulsar-token-asymmetric-key
apiVersion: v1
data:
  TOKEN: <...>
  TYPE: YXN5bW1ldHJpYw==
kind: Secret
metadata:
  creationTimestamp: null
  name: pulsar-token-proxy-admin
  namespace: pulsar
generate the token for broker-admin
---
pulsar-token-asymmetric-key
apiVersion: v1
data:
  TOKEN: <...>
  TYPE: YXN5bW1ldHJpYw==
kind: Secret
metadata:
  creationTimestamp: null
  name: pulsar-token-broker-admin
  namespace: pulsar
generate the token for admin
---
pulsar-token-asymmetric-key
apiVersion: v1
data:
  TOKEN:  <...>
  TYPE: YXN5bW1ldHJpYw==
kind: Secret
metadata:
  creationTimestamp: null
  name: pulsar-token-admin
  namespace: pulsar
-------------------------------------


The jwt token secret keys are generated under:
    - 'pulsar-token-asymmetric-key'


The jwt tokens for superusers are generated and stored as below:
    - 'proxy-admin':secret('pulsar-token-proxy-admin')
    - 'broker-admin':secret('pulsar-token-broker-admin')
    - 'admin':secret('pulsar-token-admin')

接下来手动将这 4 个 Secret 使用kubectl apply
创建到 K8S 的 pulsar 命名空间中。创建完成后，可以使用 kubectl 找到它们:

kubectl get secret -n pulsar | grep pulsar-token
pulsar-token-admin                        Opaque                    2      5m
pulsar-token-asymmetric-key               Opaque                    2      5m
pulsar-token-broker-admin                 Opaque                    2      5m
pulsar-token-proxy-admin                  Opaque                    2      5m

创建 Zookeeper 和 Bookie 的 Local PV

根据部署 Pulsar 的 K8S 节点的规划，下面需要为 zookeeper, bookie 所在的节点在 K8S 上创建 Local Persistent Volume。

注意每个 zookeeper 节点需要一个 data 的 local volume，每个 bookie 节点需要 journal 和 ledgers 共两个 local volume。

在创建 Local PV 之前，需要确认一下 k8s 中存在 StorageClasslocal-storage
，如果没有可以使用下面的 manifest 创建。

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: local-storage
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer
reclaimPolicy: Retain

注意现在的 K8S 中不在直接提供 local volume 的 provisioner，这里也没有使用 provisioner，因此后续对 local volume 的创建和管理都是需要 K8S 集群管理员的手动进行。也是说目前 Kubernetes 核心中不包含对对本地卷进行动态发放和管理的 provisioner，如果想要体验动态发放和管理的功能，可以试一下由 Rancher 提供的Local Path Provisioner^[3]。

我这里依然使用手动管理的方式，即通过手动在 K8S 节点上创建 Local Volume，手动绑定 Local Volume 与 Pulsar Zookeeper 和 Bookie 的 PVC(PersistentVolumeClaim)之间的关系。

下面，先手动在 node1, node2, node3 上创建 local volume 对应的数据目录:

$ mkdir -p home/puslar/data/zookeeper-data
$ mkdir -p home/puslar/data/bookie-data/ledgers
$ mkdir -p home/puslar/data/bookie-data/journal

zookeeper data 的 local pv 的 manifest 如下:

---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-zookeeper-data-pulsar-zookeeper-0
spec:
  capacity:
    storage: 20Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/zookeeper-data
  claimRef:
    name: pulsar-zookeeper-data-pulsar-zookeeper-0
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node1


---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-zookeeper-data-pulsar-zookeeper-1
spec:
  capacity:
    storage: 20Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/zookeeper-data
  claimRef:
    name: pulsar-zookeeper-data-pulsar-zookeeper-1
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node2




---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-zookeeper-data-pulsar-zookeeper-2
spec:
  capacity:
    storage: 20Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/zookeeper-data
  claimRef:
    name: pulsar-zookeeper-data-pulsar-zookeeper-2
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node3

上面的 manifest 仍中将 3 个 Local PV 通过nodeAffinity
创建并关联到到 node1~node3 上，同时使用claimRef
将这 3 个 Local PV 与即将在 K8S 集群中部署的 zookeeper SatefulSet 中的 PVC 绑定。使用kubectl apply
创建上面的 manifest。

bookie ledgers 和 journal 的 local pv 的 manifest 如下:

---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-ledgers-pulsar-bookie-0
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/ledgers
  claimRef:
    name: pulsar-bookie-ledgers-pulsar-bookie-0
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node1
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-journal-pulsar-bookie-0
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/journal
  claimRef:
    name: pulsar-bookie-journal-pulsar-bookie-0
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node1






---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-ledgers-pulsar-bookie-1
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/ledgers
  claimRef:
    name: pulsar-bookie-ledgers-pulsar-bookie-1
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node2
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-journal-pulsar-bookie-1
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/journal
  claimRef:
    name: pulsar-bookie-journal-pulsar-bookie-1
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node2








---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-ledgers-pulsar-bookie-2
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/ledgers
  claimRef:
    name: pulsar-bookie-ledgers-pulsar-bookie-2
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node3
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pulsar-bookie-journal-pulsar-bookie-2
spec:
  capacity:
    storage: 50Gi
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: home/puslar/data/bookie-data/journal
  claimRef:
    name: pulsar-bookie-journal-pulsar-bookie-2
    namespace: pulsar
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - node3

上面的 manifest 仍中将 6 个 Local PV 通过nodeAffinity
创建并关联到到 node1~node3 上，同时使用claimRef
将这 3 个 Local PV 与即将在 K8S 集群中部署的 zookeeper SatefulSet 中的 PVC 绑定。使用kubectl apply
创建上面的 manifest。

准备 Pulsar Manager 的 PostgreSQL 数据库

这里准备让 Pulsar Manager 使用外部数据库，需要提前在外部的 PostgreSQL 中创建好用户和数据库表结构。

创建数据库和用户:

CREATE USER pulsar_manager WITH PASSWORD '<password>';


CREATE DATABASE pulsar_manager OWNER pulsar_manager;


GRANT ALL PRIVILEGES ON DATABASE pulsar_manager to pulsar_manager;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA pulsar_manager TO pulsar_manager;
ALTER SCHEMA public OWNER to pulsar_manager;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO pulsar_manager;
GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public TO pulsar_manager;

创建表结构(建表脚本可以在 pulsar-manager 的镜像中找到):

CREATE TABLE IF NOT EXISTS environments (
  name varchar(256) NOT NULL,
  broker varchar(1024) NOT NULL,
  CONSTRAINT PK_name PRIMARY KEY (name),
  UNIQUE (broker)
);


CREATE TABLE IF NOT EXISTS topics_stats (
  topic_stats_id BIGSERIAL PRIMARY KEY,
  environment varchar(255) NOT NULL,
  cluster varchar(255) NOT NULL,
  broker varchar(255) NOT NULL,
  tenant varchar(255) NOT NULL,
  namespace varchar(255) NOT NULL,
  bundle varchar(255) NOT NULL,
  persistent varchar(36) NOT NULL,
  topic varchar(255) NOT NULL,
  producer_count BIGINT,
  subscription_count BIGINT,
  msg_rate_in double precision  ,
  msg_throughput_in double precision    ,
  msg_rate_out double precision ,
  msg_throughput_out double precision   ,
  average_msg_size double precision     ,
  storage_size double precision ,
  time_stamp BIGINT
);


CREATE TABLE IF NOT EXISTS publishers_stats (
  publisher_stats_id BIGSERIAL PRIMARY KEY,
  producer_id BIGINT,
  topic_stats_id BIGINT NOT NULL,
  producer_name varchar(255) NOT NULL,
  msg_rate_in double precision  ,
  msg_throughput_in double precision    ,
  average_msg_size double precision     ,
  address varchar(255),
  connected_since varchar(128),
  client_version varchar(36),
  metadata text,
  time_stamp BIGINT,
  CONSTRAINT fk_publishers_stats_topic_stats_id FOREIGN KEY (topic_stats_id) References topics_stats(topic_stats_id)
);


CREATE TABLE IF NOT EXISTS replications_stats (
  replication_stats_id BIGSERIAL PRIMARY KEY,
  topic_stats_id BIGINT NOT NULL,
  cluster varchar(255) NOT NULL,
  connected BOOLEAN,
  msg_rate_in double precision  ,
  msg_rate_out double precision ,
  msg_rate_expired double precision     ,
  msg_throughput_in double precision    ,
  msg_throughput_out double precision   ,
  msg_rate_redeliver double precision   ,
  replication_backlog BIGINT,
  replication_delay_in_seconds BIGINT,
  inbound_connection varchar(255),
  inbound_connected_since varchar(255),
  outbound_connection varchar(255),
  outbound_connected_since varchar(255),
  time_stamp BIGINT,
  CONSTRAINT FK_replications_stats_topic_stats_id FOREIGN KEY (topic_stats_id) References topics_stats(topic_stats_id)
);


CREATE TABLE IF NOT EXISTS subscriptions_stats (
  subscription_stats_id BIGSERIAL PRIMARY KEY,
  topic_stats_id BIGINT NOT NULL,
  subscription varchar(255) NULL,
  msg_backlog BIGINT,
  msg_rate_expired double precision     ,
  msg_rate_out double precision ,
  msg_throughput_out double precision   ,
  msg_rate_redeliver double precision   ,
  number_of_entries_since_first_not_acked_message BIGINT,
  total_non_contiguous_deleted_messages_range BIGINT,
  subscription_type varchar(16),
  blocked_subscription_on_unacked_msgs BOOLEAN,
  time_stamp BIGINT,
  UNIQUE (topic_stats_id, subscription),
  CONSTRAINT FK_subscriptions_stats_topic_stats_id FOREIGN KEY (topic_stats_id) References topics_stats(topic_stats_id)
);


CREATE TABLE IF NOT EXISTS consumers_stats (
  consumer_stats_id BIGSERIAL PRIMARY KEY,
  consumer varchar(255) NOT NULL,
  topic_stats_id BIGINT NOT NUll,
  replication_stats_id BIGINT,
  subscription_stats_id BIGINT,
  address varchar(255),
  available_permits BIGINT,
  connected_since varchar(255),
  msg_rate_out double precision ,
  msg_throughput_out double precision   ,
  msg_rate_redeliver double precision   ,
  client_version varchar(36),
  time_stamp BIGINT,
  metadata text
);


CREATE TABLE IF NOT EXISTS tokens (
  token_id BIGSERIAL PRIMARY KEY,
  role varchar(256) NOT NULL,
  description varchar(128),
  token varchar(1024) NOT NUll,
  UNIQUE (role)
);


CREATE TABLE IF NOT EXISTS users (
  user_id BIGSERIAL PRIMARY KEY,
  access_token varchar(256),
  name varchar(256) NOT NULL,
  description varchar(128),
  email varchar(256),
  phone_number varchar(48),
  location varchar(256),
  company varchar(256),
  expire BIGINT,
  password varchar(256),
  UNIQUE (name)
);


CREATE TABLE IF NOT EXISTS roles (
  role_id BIGSERIAL PRIMARY KEY,
  role_name varchar(256) NOT NULL,
  role_source varchar(256) NOT NULL,
  description varchar(128),
  resource_id BIGINT NOT NULL,
  resource_type varchar(48) NOT NULL,
  resource_name varchar(48) NOT NULL,
  resource_verbs varchar(256) NOT NULL,
  flag INT NOT NULL
);


CREATE TABLE IF NOT EXISTS tenants (
  tenant_id BIGSERIAL PRIMARY KEY,
  tenant varchar(255) NOT NULL,
  admin_roles varchar(255),
  allowed_clusters varchar(255),
  environment_name varchar(255),
  UNIQUE(tenant)
);


CREATE TABLE IF NOT EXISTS namespaces (
  namespace_id BIGSERIAL PRIMARY KEY,
  tenant varchar(255) NOT NULL,
  namespace varchar(255) NOT NULL,
  UNIQUE(tenant, namespace)
);


CREATE TABLE IF NOT EXISTS role_binding(
  role_binding_id BIGSERIAL PRIMARY KEY,
  name varchar(256) NOT NULL,
  description varchar(256),
  role_id BIGINT NOT NULL,
  user_id BIGINT NOT NULL
);

上面已经做好了部署的准备工作，下面将使用 Helm 在 K8S 集群中部署 Pulsar 集群。

使用 Helm 在 K8S 中部署 Pulsar

从https://github.com/apache/pulsar-helm-chart/releases下载pulsar helm chart 2.7.7 到 K8S 的控制节点上(kubectl 和 helm 可用)。

https://github.com/apache/pulsar-helm-chart/releases/download/pulsar-2.7.7/pulsar-2.7.7.tgz

定制编写 helm chart 的 values.yaml

定制编写 helm chart 的 values.yaml 文件如下, 定制的内容比较多，具体见下面文件的注释:

auth:
  authentication:
    enabled: true  # 开启jwt认证
    provider: "jwt"
    jwt:
      usingSecretKey: false # jwt认证使用非对称秘钥对
  authorization:
    enabled: true # 开启授权
  superUsers:
    # broker to broker communication
    broker: "broker-admin"
    # proxy to broker communication
    proxy: "proxy-admin"
    # pulsar-admin client to broker/proxy communication
    client: "admin"




components: # 启用的组件
  autorecovery: true
  bookkeeper: true
  broker: true
  functions: true
  proxy: true
  pulsar_manager: true
  toolset: true
  zookeeper: true


monitoring: # 关闭监控组件, 后续尝试使用外部Prometheus对pulsar集群进行监控
  grafana: false
  prometheus: false
  node_exporter: false




volumes:
  local_storage: true # 数据卷使用local storage






proxy: # proxy的配置(这里是测试环境, 将proxy也调度到node1或node2或node3)
  nodeSelector:
    node-role.kubernetes.io/pulsar: pulsar
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"
  configData:
     PULSAR_PREFIX_authenticateMetricsEndpoint: "false"




broker: # broker的配置(这里是测试环境, 将proxy也调度到node1或node2或node3)
  nodeSelector:
    node-role.kubernetes.io/pulsar: pulsar
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"




zookeeper: # broker的配置
  replicaCount: 3
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"
  volumes:
    data: # 配置使用local pv, 需要与前面手动创建的local pv信息一致
      local_storage: true
      size: 20Gi




bookkeeper: # bookkeeper的配置
  replicaCount: 3
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"
  volumes:
    journal: # 配置使用local pv, 需要与前面手动创建的local pv信息一致
      local_storage: true
      size: 50Gi
  ledgers:  # 配置使用local pv, 需要与前面手动创建的local pv信息一致
      local_storage: true
      size: 50Gi


pulsar_manager: # pulsar_manager的配置(这里是测试环境, 将pulsar_manager也调度到node1或node2或node3)
  replicaCount: 1
  admin:
    # 文档中描述这里是pulsar manager web界面登录用户密码，但实际上当使用外部PostgreSQL数据库时，这里需要指定PostgreSQL的数据库和密码，不知道是否是pulsar-helm-chart 2.7.7的问题
    user: pulsar_manager
    password: 05aM3Braz_M4RWpn
  configData:
    DRIVER_CLASS_NAME: org.postgresql.Driver
    URL: jdbc:postgresql://<ip>:5432/pulsar_manager
    # 文档中描述这里PostgreSQL数据库的密码，但实际上这里不能指定USERNAME和PASSWORD, 不知道是否是pulsar-helm-chart 2.7.7的问题
    # USERNAME: pulsar_manager
    # PASSWORD: 05aM3Braz_M4RWpn
    LOG_LEVEL: INFO
    ## 开启JWT认证后, 这里需要指定pulsar-token-admin这个Secret中的JWT Token
    JWT_TOKEN: <jwt token...>




autorecovery: # autorecovery的配置(这里是测试环境, 将autorecovery也调度到node1或node2或node3)
  replicaCount: 1
  nodeSelector:
    node-role.kubernetes.io/pulsar: pulsar
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"


toolset: # toolset的配置(这里是测试环境, 将toolset也调度到node1或node2或node3)
  replicaCount: 1
  nodeSelector:
    node-role.kubernetes.io/pulsar: pulsar
  tolerations:
  - key: "dedicated"
    operator: "Equal"
    value: "pulsar"
    effect: "NoSchedule"




images: # 对个组件使用私有镜像仓库的配置
  imagePullSecrets:
  - regsecret # 私有镜像仓库的image pull secret, 需要提前在k8s命名空间中创建
  autorecovery:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4
  bookie:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4
  broker:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4
  functions:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4
  proxy:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4
  pulsar_manager:
    repository: harbor.example.com/library/apachepulsar/pulsar-manager
    tag: v0.2.0
  zookeeper:
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4


pulsar_metadata:
  component: pulsar-init
  image:
    # the image used for running `pulsar-cluster-initialize` job
    repository: harbor.example.com/library/apachepulsar/pulsar-all
    tag: 2.7.4

因为当前在 pulsar-helm-chart 2.7.7 中好像不支持为 pulsar-init 设置私有仓库的 imagePullSecret，所以下面为 pulsar namespace 中的 default servcieaccount 添加上 imagePullSecret。

$ kubectl patch serviceaccount default -p '{"imagePullSecrets": [{"name": "regsecret"}]}' -n pulsar

使用 helm install 安装 pulsar

定制完 value.yaml 之后，使用下面的命令向 K8S 集群部署 pulsar。

$ helm install \
    --values values.yaml \
    --set initialize=true \
    --namespace pulsar \
    pulsar pulsar-2.7.7.tgz

安装完成后使用下面的命令查看一下两个初始化 job pulsar-pulsar-init 和 pulsar-bookie-init 的 pod 状态为 Complete:

$ kubectl get pod -n pulsar  | grep init
pulsar-bookie-init--1-h65bp              0/1     Completed   0               5m14s
pulsar-pulsar-init--1-t4thq              0/1     Completed   0               5m5s

使用下面的命令查看一下 pulsar 集群各个组件的 Pod 状态全部都为 Running:

$ kubectl get pod -n pulsar -l cluster=pulsar -o wide
NAME                                     READY   STATUS    RESTARTS      AGE   IP              NODE   NOMINATED NODE   READINESS GATES
pulsar-bookie-0                          1/1     Running   0             14m   10.244.226.91   node1    <none>           <none>
pulsar-bookie-1                          1/1     Running   0             14m   10.244.63.90    node2    <none>           <none>
pulsar-bookie-2                          1/1     Running   0             14m   10.244.46.92    node3    <none>           <none>
pulsar-broker-0                          1/1     Running   0             14m   10.244.226.90   node1    <none>           <none>
pulsar-broker-1                          1/1     Running   0             14m   10.244.63.89    node2    <none>           <none>
pulsar-broker-2                          1/1     Running   0             14m   10.244.46.90    node3    <none>           <none>
pulsar-proxy-0                           1/1     Running   0             14m   10.244.226.93   node1    <none>           <none>
pulsar-proxy-1                           1/1     Running   0             14m   10.244.63.91    node2    <none>           <none>
pulsar-proxy-2                           1/1     Running   0             14m   10.244.46.93    node3    <none>           <none>
pulsar-pulsar-manager-7b98666cff-5626f   1/1     Running   0             14m   10.244.63.88    node2    <none>           <none>
pulsar-recovery-0                        1/1     Running   0             14m   10.244.46.89    node3    <none>           <none>
pulsar-toolset-0                         1/1     Running   0             14m   10.244.46.91    node3    <none>           <none>
pulsar-zookeeper-0                       1/1     Running   0             14m   10.244.226.92   node1    <none>           <none>
pulsar-zookeeper-1                       1/1     Running   0             14m   10.244.63.92    node2    <none>           <none>
pulsar-zookeeper-2                       1/1     Running   0             13m   10.244.46.94    node3    <none>           <none>

如果后边调整了 values.yaml，需要更新部署时，使用下面的命令:

$ helm upgrade pulsar pulsar-2.7.7.tgz \
    --namespace pulsar \
    -f values.yaml

在 toolset pod 中测试创建 tenant, namespace 和 topic

toolset pod 中包含了各种管理和测试 pulsar 的命令行工具，例如 pulsar-admin, pulsar-client 等。

下面进入 toolset pod 中，使用 pulsar-admin 命令行工具测试一下 tenant, namespace 和 topic 的创建，进一步确认 pulsar 集群工作正常。

$ kubectl exec -it -n pulsar pulsar-toolset-0 -- /bin/bash

bin/pulsar-admin tenants create test-tenant

bin/pulsar-admin tenants list
"public"
"pulsar"
"test-tenant"


bin/pulsar-admin namespaces create test-tenant/test-ns

bin/pulsar-admin namespaces list test-tenant
"test-tenant/test-ns"

bin/pulsar-admin topics create-partitioned-topic test-tenant/test-ns/test-topic -p 3

bin/pulsar-admin topics list-partitioned-topics test-tenant/test-ns
"persistent://test-tenant/test-ns/test-topic"

创建 pulsar-manager 的管理员用户并登录查看

下面测试一下 pulsar manager 是否可以使用。

前面使用 helm chart 部署的 pulsar 集群，在 k8s 中创建了下面 7 个 Service。

$ kubectl get svc -l app=pulsar -n pulsar
NAME                    TYPE           CLUSTER-IP       EXTERNAL-IP   PORT(S)                               AGE
pulsar-bookie           ClusterIP      None             <none>        3181/TCP,8000/TCP                     40m
pulsar-broker           ClusterIP      None             <none>        8080/TCP,6650/TCP                     40m
pulsar-proxy            LoadBalancer   10.104.105.137   <pending>     80:31970/TCP,6650:32631/TCP           40m
pulsar-pulsar-manager   LoadBalancer   10.110.207.9     <pending>     9527:32764/TCP                        40m
pulsar-recovery         ClusterIP      None             <none>        8000/TCP                              40m
pulsar-toolset          ClusterIP      None             <none>        <none>                                40m
pulsar-zookeeper        ClusterIP      None             <none>        8000/TCP,2888/TCP,3888/TCP,2181/TCP   40m

从上面命令的输出可以看出，bookie, broker, recovery, toolset, zookeeper 这 5 个 Service 的类型都是 ClusterIP 的，并且 cluser-ip 为 None，都是 Headless 的 Service，因为它们只需要在 k8s 集群内部使用。

pulsar-proxy 和 pulsar-pulsar-manager 为 LoadBalancer 类型，并且都配置了 NodePort，提供了从 K8S 集群外部访问的能力。

从集群外部访问 pulsar-manager 的地址是http://node1:32764
，第一次访问 pulsar manager 之前，需要为其创建一个管理用户:

$ CSRF_TOKEN=$(curl http://node1:32764/pulsar-manager/csrf-token)
$ curl \
   -H 'X-XSRF-TOKEN: $CSRF_TOKEN' \
   -H 'Cookie: XSRF-TOKEN=$CSRF_TOKEN;' \
   -H "Content-Type: application/json" \
   -X PUT http://node1:32764/pulsar-manager/users/superuser \
   -d '{"name": "admin", "password": "pulsar", "description": "test", "email": "username@test.org"}'

上面的命令为 pulsar-manager 创建用户名为 admin, 密码为 pulsar 的管理用户。使用该用户就可以登录 pulsar manager。

备注, 在线上使用时，尽量避免以 NodePort 暴露服务，这里的 pulsar-manager 的 Service 可以修改为 CluserIP 类型，并关闭 NodePort，同时创建 Ingress，以 Ingress+域名的形式暴露出来。看了一下 pulsar-helm-chart 也是支持的，只是目前 pulsar-helm-chart 2.7.7 中创建 Ingress 时，使用的是apiVersion: extensions/v1beta1
API，这个 API 从 k8s 1.19 被标记为废弃，在 k8s 1.22 已被移除。所以要直接是使用 pulsar-helm-chart 创建 Ingress 的话，需要等待 pulsar-helm-chart 的更新。

kafka on k8s

Helm部署kafka

helm repo add incubator http://mirror.azure.cn/kubernetes/charts-incubator


[root@k8s-centos7-master-150 kafka]# helm repo list
NAME            URL
minio           https://helm.min.io/
bitnami         https://charts.bitnami.com/bitnami
incubator       http://mirror.azure.cn/kubernetes/charts-incubator

创建 StorageClass

首先先创建本地存储的 StorageClass

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: local-storage
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer
reclaimPolicy: Retain

执行命令

kubectl apply -f local-storage.yaml 
storageclass.storage.k8s.io/local-storage created

[root@k8s-centos7-master-150 kafka]# kubectl get sc --all-namespaces -o wide
NAME            PROVISIONER                    RECLAIMPOLICY   VOLUMEBINDINGMODE      ALLOWVOLUMEEXPANSION   AGE
local-storage   kubernetes.io/no-provisioner   Delete          WaitForFirstConsumer   false                  83m

创建 Kafka 的 pv

因为要在 k8s-centos7-node3-151 ，k8s-centos7-node4-152 这两个 k8s 节点上部署 2 个 Kafka 的 broker 节点，因此先在三个节点上创建这 2 个kafka broker节点的Local PV

kafka-local-pv.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-pv-0
spec:
  capacity:
    storage: 5Gi 
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: /data/kafka/data-0
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - k8s-centos7-node3-151
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-pv-1
spec:
  capacity:
    storage: 5Gi 
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: /data/kafka/data-1
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - k8s-centos7-node4-152

执行命令:

kubectl apply -f kafka-local-pv.yaml

在 k8s-centos7-node3-151
上创建目录 /data/kafka/data-0

在 k8s-centos7-node4-152
上创建目录 /data/kafka/data-1

[root@k8s-centos7-master-150 kafka]# kubectl get pv,pvc --all-namespaces
NAME                             CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS   CLAIM                            STORAGECLASS    REASON   AGE
persistentvolume/kafka-pv-0      5Gi        RWO            Retain           Bound    default/data-kafka-zookeeper-1   local-storage            53m
persistentvolume/kafka-pv-1      5Gi        RWO            Retain           Bound    default/data-kafka-zookeeper-0   local-storage            53m
persistentvolume/kafka-zk-pv-0   5Gi        RWO            Retain           Bound    default/datadir-kafka-0          local-storage            52m
persistentvolume/kafka-zk-pv-1   5Gi        RWO            Retain           Bound    default/datadir-kafka-1          local-storage            52m


NAMESPACE   NAME                                           STATUS   VOLUME          CAPACITY   ACCESS MODES   STORAGECLASS    AGE
default     persistentvolumeclaim/data-kafka-zookeeper-0   Bound    kafka-pv-1      5Gi        RWO            local-storage   39m
default     persistentvolumeclaim/data-kafka-zookeeper-1   Bound    kafka-pv-0      5Gi        RWO            local-storage   26m
default     persistentvolumeclaim/datadir-kafka-0          Bound    kafka-zk-pv-0   5Gi        RWO            local-storage   39m
default     persistentvolumeclaim/datadir-kafka-1          Bound    kafka-zk-pv-1   5Gi        RWO            local-storage   20m

创建zookeeper 的PV

因为要在 k8s-centos7-node3-151 ，k8s-centos7-node4-152 这两个 k8s 节点上部署 2 个 Zookeeper 节点，因此先在三个节点上创建这 2 个 Zookeeper 节点的 Local PV

zookeeper-local-pv.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-zk-pv-0
spec:
  capacity:
    storage: 5Gi 
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: /data/kafka/zkdata-0
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - k8s-centos7-node3-151
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-zk-pv-1
spec:
  capacity:
    storage: 5Gi 
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Retain
  storageClassName: local-storage
  local:
    path: /data/kafka/zkdata-1
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - k8s-centos7-node4-152

执行命令:

kubectl apply -f zookeeper-local-pv.yaml

同理，也要在相应的节点上创建目录，不再赘述

[root@k8s-centos7-master-150 kafka]# kubectl get pv,pvc --all-namespaces
NAME                             CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS   CLAIM                            STORAGECLASS    REASON   AGE
persistentvolume/kafka-pv-0      5Gi        RWO            Retain           Bound    default/data-kafka-zookeeper-1   local-storage            57m
persistentvolume/kafka-pv-1      5Gi        RWO            Retain           Bound    default/data-kafka-zookeeper-0   local-storage            57m
persistentvolume/kafka-zk-pv-0   5Gi        RWO            Retain           Bound    default/datadir-kafka-0          local-storage            56m
persistentvolume/kafka-zk-pv-1   5Gi        RWO            Retain           Bound    default/datadir-kafka-1          local-storage            56m


NAMESPACE   NAME                                           STATUS   VOLUME          CAPACITY   ACCESS MODES   STORAGECLASS    AGE
default     persistentvolumeclaim/data-kafka-zookeeper-0   Bound    kafka-pv-1      5Gi        RWO            local-storage   42m
default     persistentvolumeclaim/data-kafka-zookeeper-1   Bound    kafka-pv-0      5Gi        RWO            local-storage   30m
default     persistentvolumeclaim/datadir-kafka-0          Bound    kafka-zk-pv-0   5Gi        RWO            local-storage   42m
default     persistentvolumeclaim/datadir-kafka-1          Bound    kafka-zk-pv-1   5Gi        RWO            local-storage   24m

部署 Kafka-values.yaml

kafka-values.yaml

image:
  repository: zenko/kafka-manager
  tag: 1.3.3.22
zkHosts: kafka-zookeeper:2181
basicAuth:
  enabled: true
  username: admin
  password: admin
ingress:
  enabled: true
  hosts: 
   - km.hongda.com
  tls:
    - secretName: hongda-com-tls-secret
      hosts:
      - km.hongda.com

执行命令：

helm install kafka -f kafka-values.yaml incubator/kafka

结果

# pod，svc
[root@k8s-centos7-master-150 kafka]# kubectl get po,svc -o wide
NAME                    READY   STATUS    RESTARTS   AGE   IP               NODE                    NOMINATED NODE   READINESS GATES
pod/kafka-0             1/1     Running   2          33m   10.244.182.130   k8s-centos7-node3-151   <none>           <none>
pod/kafka-1             1/1     Running   0          26m   10.244.216.4     k8s-centos7-node4-152   <none>           <none>
pod/kafka-zookeeper-0   1/1     Running   0          33m   10.244.216.3     k8s-centos7-node4-152   <none>           <none>
pod/kafka-zookeeper-1   1/1     Running   0          32m   10.244.182.131   k8s-centos7-node3-151   <none>           <none>


NAME                               TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)                      AGE   SELECTOR
service/kafka                      ClusterIP   10.106.236.254   <none>        9092/TCP                     33m   app.kubernetes.io/component=kafka-broker,app.kubernetes.io/instance=kafka,app.kubernetes.io/name=kafka
service/kafka-headless             ClusterIP   None             <none>        9092/TCP                     33m   app.kubernetes.io/component=kafka-broker,app.kubernetes.io/instance=kafka,app.kubernetes.io/name=kafka
service/kafka-zookeeper            ClusterIP   10.96.218.152    <none>        2181/TCP                     33m   app=zookeeper,release=kafka
service/kafka-zookeeper-headless   ClusterIP   None             <none>        2181/TCP,3888/TCP,2888/TCP   33m   app=zookeeper,release=kafka
service/kubernetes                 ClusterIP   10.96.0.1        <none>        443/TCP                      18h   <none>


# statefulset
[root@k8s-centos7-master-150 kafka]# kubectl get statefulset
NAME              READY   AGE
kafka             2/2     35m
kafka-zookeeper   2/2     35m

Kafka可视化–安装 Kafka Manager

Helm的官方repo中已经提供了Kafka Manager的Chart。

kafka-manager-values.yaml

image:
  repository: zenko/kafka-manager
  tag: 1.3.3.22
zkHosts: kafka-zookeeper:2181
basicAuth:
  enabled: true
  username: admin
  password: admin
ingress:
  enabled: true
  hosts: 
   - km.hongda.com
  tls:
    - secretName: hongda-com-tls-secret
      hosts:
      - km.hongda.com