第 3 章 使用 Record Encryption 过滤器部署 Apache Kafka 代理的流

流程

1. 下载并提取 Apache Kafka 代理安装工件的 Streams。

   工件包含在安装以及 Apache Kafka 软件下载页面的 Streams 中提供的示例文件。

   文件包含通过 cluster-ip 或 loadbalancer 类型监听程序连接所需的部署配置。

2. 在 Kafka 集群中创建主题：

   oc run -n <my_proxy_namespace> -ti proxy-producer \
     --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 \
     --rm=true \
     --restart=Never \
     -- bin/kafka-topics.sh \
     --bootstrap-server proxy-service:9092 \
     --create -topic my-topic

   在本例中，我们通过交互式 pod 容器创建一个名为 my-topic 的主题。

3. 在 Vault 中为 my-topic 创建密钥：

   vault write -f transit/keys/KEK_my-topic

4. 编辑为代理提供过滤器配置的 ConfigMap。

   Record Encryption 过滤器 配置需要 HashiCorp Vault KMS 的凭证。

   记录加密过滤器配置示例

   filters:
     - type: RecordEncryption
       config:
         kms: VaultKmsService 1
         kmsConfig:
           vaultTransitEngineUrl: http://vault.vault.svc.cluster.local:8200/v1/transit 2
           vaultToken:
             passwordFile: /opt/proxy/encryption/token.txt 3
         selector: TemplateKekSelector 4
         selectorConfig:
           template: "KEK_${topicName}" 5
   	 # ...

   1

   使用的 KMS （密钥管理系统）的类型。在本例中，HashiCorp Vault。

   2

   Vault Transit Engine 服务的 URL。

   3

   包含访问 Vault 服务所需的令牌的文件。如果此位置发生变化，则代理部署配置中需要进行等同的更改。

   4

   要使用的密钥加密密钥(KEK)选择器。${topicName} 是代理理解的字面值。

   5

   基于特定主题名称派生的 KEK 模板。

5. 使用 Record Encryption 过滤器和适当的监听程序为 OpenShift 集群部署 Apache Kafka 代理的流：

   使用 cluster-ip 侦听器部署代理

   cd /examples/proxy/record-encryption/
   oc apply -k cluster-ip

   使用 loadbalancer 侦听器部署代理

   cd /examples/proxy/record-encryption/
   oc apply -k loadbalancer

6. 如果您使用 loadbalancer 侦听器，请更新代理配置以使用所创建的 loadbalancer 服务地址。

   1. 获取代理服务的外部地址：

      LOAD_BALANCER_ADDRESS=$(oc get service -n <my_proxy_namespace> proxy-service --template='{{(index .status.loadBalancer.ingress 0).hostname}}')

   2. 更新代理服务配置中的 brokerAddressPattern 属性以使用代理地址：

      sed -i "s/\(brokerAddressPattern:\).*$/\1 ${LOAD_BALANCER_ADDRESS}/" load-balancer/proxy/proxy-config.yaml

   3. 将更改应用到代理配置并重启代理 pod。

       oc apply -k load-balancer && oc delete pod -n <my_proxy_namespace> --all

7. 检查部署的状态：

   oc get pods -n <my_proxy_namespace>

   输出显示部署名称和就绪

   NAME                      READY  STATUS   RESTARTS
   my-cluster-kafka-0        1/1    Running  0
   my-cluster-kafka-1        1/1    Running  0
   my-cluster-kafka-2        1/1    Running  0
   my-cluster-proxy-<pod_id> 1/1    Running  0

   my-cluster-proxy 是代理的名称。

   标识创建的 pod 的 pod ID。

   使用默认部署，您可以安装单个代理 pod。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

8. 通过代理生成消息，然后直接从 Kafka 集群使用和间接使用，验证加密是否已应用到指定的主题。