如何在Kafka Connect中从OpenShift Secret配置外置数据库连接用户名和密码

我来一步步给你拆解这个操作，不管你是用Strimzi Operator部署的Kafka Connect，还是自己手动维护Deployment，都能轻松搞定：

1. 先创建存储数据库凭证的OpenShift Secret #

首先得把你的数据库用户名和密码存到OpenShift的Secret里，有两种常用方式：

方式一：直接通过命令行创建 #

运行下面的命令，替换成你的实际凭证：

oc create secret generic db-credentials \
  --from-literal=db.username=your-db-user \
  --from-literal=db.password=your-db-pass

方式二：从本地文件读取（适合批量或敏感场景） #

先把用户名和密码分别存到本地文件，比如db-user.txt和db-pass.txt，然后执行：

oc create secret generic db-credentials \
  --from-file=db.username=./db-user.txt \
  --from-file=db.password=./db-pass.txt

2. 让Kafka Connect能访问这个Secret #

这里分两种部署场景来配置：

场景A：用Strimzi Operator部署Kafka Connect #

如果你的Kafka Connect是通过Strimzi Operator管理的，直接修改KafkaConnect自定义资源即可：

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 1
  bootstrapServers: my-kafka-cluster-kafka-bootstrap:9092
  template:
    pod:
      spec:
        containers:
        - name: connect
          # 方式1：把Secret中的值注入为环境变量
          env:
          - name: DB_USERNAME
            valueFrom:
              secretKeyRef:
                name: db-credentials
                key: db.username
          - name: DB_PASSWORD
            valueFrom:
              secretKeyRef:
                name: db-credentials
                key: db.password
          # 方式2：把Secret挂载为文件（适合需要用file变量引用的场景）
          volumeMounts:
          - name: db-secret-volume
            mountPath: /opt/kafka/secrets/db-creds
            readOnly: true
        volumes:
        - name: db-secret-volume
          secret:
            secretName: db-credentials

场景B：手动部署Kafka Connect Deployment #

如果是自己写的Deployment，修改Deployment的spec部分，添加环境变量或卷挂载：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-connect
spec:
  replicas: 1
  template:
    spec:
      containers:
      - name: kafka-connect
        image: confluentinc/cp-kafka-connect:latest
        # 环境变量方式
        env:
        - name: DB_USERNAME
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: db.username
        - name: DB_PASSWORD
          valueFrom:
            secretKeyRef:
              name: db-credentials
              key: db.password
        # 文件挂载方式（可选）
        volumeMounts:
        - name: db-secret
          mountPath: /etc/kafka/secrets/db
          readOnly: true
      volumes:
      - name: db-secret
        secret:
          secretName: db-credentials

另外别忘了给Kafka Connect的ServiceAccount添加读取Secret的权限（如果OpenShift的安全策略限制的话）：

oc policy add-role-to-user view system:serviceaccount:your-namespace:kafka-connect-sa -n your-namespace

替换your-namespace和kafka-connect-sa为你的实际命名空间和ServiceAccount名称。

3. 在Connector配置中引用Secret中的值 #

现在你可以在Connector的配置里，用环境变量或者文件路径来引用凭证，不用写明文了：

方式1：引用环境变量 #

比如JDBC Connector的配置：

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://your-db-host:5432/your-db",
    "connection.user": "${env:DB_USERNAME}",
    "connection.password": "${env:DB_PASSWORD}",
    "topic.prefix": "db-",
    "mode": "incrementing",
    "incrementing.column.name": "id"
  }
}

方式2：引用挂载的文件 #

如果是用文件挂载的方式，配置里这么写：

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://your-db-host:5432/your-db",
    "connection.user": "${file:/opt/kafka/secrets/db-creds/db.username}",
    "connection.password": "${file:/opt/kafka/secrets/db-creds/db.password}",
    "topic.prefix": "db-",
    "mode": "incrementing",
    "incrementing.column.name": "id"
  }
}

最后检查下 #

应用配置完后，重启Kafka Connect的Pod（或者让Operator自动滚动更新），然后创建Connector，查看Connector的状态是否正常。如果有问题，可以查看Pod的日志，确认是否能正确读取到Secret的值。

内容的提问来源于stack exchange，提问作者Althaf Khan