实时计算Flink版：KMS 凭据管理集成（公测）

创建 RAM 角色

1. 登录 RAM 控制台 - 角色页面，单击创建角色。保持默认选项，并单击确定。

2. 填写一个有辨识度的角色名称，例如 FlinkRoleForKMSRead，单击确定。

添加信任策略

角色创建成功后，在角色详情页的信任策略页签下，单击编辑信任策略，在 Principal 中添加 "Service": ["stream.aliyuncs.com"]，信任 Flink 扮演该角色。更多信息，请参见修改 RAM 角色的信任策略。

{
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Effect": "Allow",
      "Principal": {
        "Service": [
          "stream.aliyuncs.com"
        ]
      }
    }
  ],
  "Version": "1"
}

为 RAM 角色授予 KMS 访问权限

扮演该角色后，Flink 将以角色身份调用 KMS，需提前在角色权限策略中授予 KMS 相关权限。

在角色详情页的权限管理页签下，单击新增授权，添加 AliyunKMSSecretUserAccess（获取 KMS 中凭据的权限）权限策略。若凭据使用了用户指定的密钥（CMK）加密，还需有密钥的 kms:Decrypt 权限，可通过添加 AliyunKMSCryptoUserAccess 权限策略来授权。

若期望细粒度控制到密钥及凭据粒度，可自定义权限策略，授予 GetSecretValue、Decrypt 权限，并将资源限制为指定资源。详情请参见 KMS 自定义权限策略参考、创建自定义权限策略。

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "kms:GetSecretValue",
      "Resource": "acs:kms:<region>:<account-id>:secret/<secret-name>"
    },
    {
      "Effect": "Allow",
      "Action": "kms:Decrypt",
      "Resource": "acs:kms:<region>:<account-id>:key/<key-id>"
    }
  ]
}

创建权限策略

登录 RAM 控制台 - 权限策略页面，单击创建权限策略。切换至脚本编辑，编辑权限策略内容。

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ram:ListRoles"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "ram:PassRole",
      "Resource": "acs:ram::<account-id>:role/<role-name>",
      "Condition": {
        "StringEquals": {
          "acs:Service": "stream.aliyuncs.com"
        }
      }
    }
  ]
}

单击确定，将权限策略保存，例如命名为 FlinkPassRolePolicy。更多信息，请参见创建自定义权限策略。

为 RAM 用户授权

1. 登录 RAM 控制台 - 用户页面，找到需要授权的 RAM 用户，单击操作列的新增授权。

2. 在新增授权面板中，搜索并选中步骤三创建的权限策略（例如 FlinkPassRolePolicy），单击确认新增授权。更多信息，请参见管理RAM用户的权限。

Postgres CDC Connector 示例

SQL 作业：以下示例以 Postgres CDC 源表为例，使用 KMS 凭据托管数据库账号密码。KMS 中已创建数据库凭据 my-db-secret，其 SecretData 为 {"AccountName":"alice","AccountPassword":"xxx"}。

CREATE TEMPORARY TABLE postgrescdc_source (
    id INT NOT NULL,
    name STRING,
    description STRING,
    weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = 'secret://kms.my-db-secret.AccountName',
  'password' = 'secret://kms.my-db-secret.AccountPassword',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'slot.name' = '<slot name>',
  'decoding.plugin.name' = 'pgoutput',
  'kms.endpoint' = '<YOUR-KMS-INSTANCE-VPC-ENDPOINT>',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

Postgres CDC 连接器的参数说明、前提条件与类型映射，请参见Postgres CDC 连接器。

参数说明