如何在Flink Kubernetes Operator中安全传递S3密钥至flink-conf.yaml

在AWS EKS集群中使用Flink Kubernetes Pod Operator部署Flink作业，需将AWS S3配置为检查点（checkpointing）和保存点（savepoint）的存储位置。要将S3访问密钥/秘密密钥传入flink-conf.yaml，但不愿将明文凭据写入配置并提交至Git。

已创建如下Secret资源：

apiVersion: v1
kind: Secret
metadata:
  namespace: flink-stage
  name: flink-secrets
type: Opaque
data:
  # base64编码后的内容
  s3.access_key: <key>
  s3.secret_key: <secret>

尝试通过Secret引用配置时触发校验错误，使用的FlinkDeployment配置如下：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: nrt-sessionizer
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    s3.access-key:
      valueFrom:
        secretKeyRef:
          key: s3.access_key
          name: flink-secrets
    s3.secret-key:
      valueFrom:
        secretKeyRef:
          key: s3.secret_key
          name: flink-secrets

错误信息：

error: error validating "infra/flink-k8/overlays/prod": error validating data:
 [ValidationError(FlinkDeployment.spec.flinkConfiguration.s3.access-key): invalid type for org.apache.flink.v1beta1.FlinkDeployment.spec.flinkConfiguration: got "map", expected "string",
 ValidationError(FlinkDeployment.spec.flinkConfiguration.s3.secret-key): invalid type for org.apache.flink.v1beta1.FlinkDeployment.spec.flinkConfiguration: got "map", expected "string"]; 
if you choose to ignore these errors, turn validation off with --validate=false

注意：使用InitContainer修改flink-conf.yaml的方案不可行，因为ConfigMaps是只读的。

方案1：通过环境变量注入密钥，Flink自动读取 #

Flink支持从环境变量加载配置项，只需将Secret中的密钥挂载为环境变量，Flink会自动识别并转换为对应配置。

修改FlinkDeployment的podTemplate添加环境变量：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: nrt-sessionizer
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: FLINK_PROPERTIES_s3_access_key
              valueFrom:
                secretKeyRef:
                  name: flink-secrets
                  key: s3.access_key
            - name: FLINK_PROPERTIES_s3_secret_key
              valueFrom:
                secretKeyRef:
                  name: flink-secrets
                  key: s3.secret_key

原理：Flink启动时会自动读取前缀为FLINK_PROPERTIES_的环境变量，将下划线替换为点后转为配置项（如FLINK_PROPERTIES_s3_access_key对应s3.access-key）。

方案2：使用IRSA服务账号IAM角色（推荐） #

在AWS EKS中，推荐使用IAM Roles for Service Accounts (IRSA) 授予Flink Pod S3访问权限，无需手动管理密钥。

1. 创建关联IAM角色的服务账号：
   • 为Kubernetes服务账号创建IAM角色，信任策略允许EKS OIDC提供商执行AssumeRoleWithWebIdentity
   • 为IAM角色附加S3权限策略，包含目标桶的s3:GetObject、s3:PutObject等必要权限
2. 修改FlinkDeployment使用该服务账号：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: nrt-sessionizer
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    state.checkpoints.dir: s3://your-bucket/checkpoints
    state.savepoints.dir: s3://your-bucket/savepoints
    s3.path.style.access: "true" # 根据S3配置调整
  serviceAccount: flink-sa # 关联IAM角色的服务账号名称

该方案无需管理密钥，符合AWS安全最佳实践。

方案3：将Secret挂载为文件，Flink从文件读取配置 #

Flink支持解析带file:前缀的配置值，自动读取对应文件内容作为配置项。

修改FlinkDeployment：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: nrt-sessionizer
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    s3.access-key: "file:/opt/flink/secrets/s3.access_key"
    s3.secret-key: "file:/opt/flink/secrets/s3.secret_key"
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
            - name: flink-secrets-volume
              mountPath: /opt/flink/secrets
              readOnly: true
      volumes:
        - name: flink-secrets-volume
          secret:
            secretName: flink-secrets

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