Realtime Compute for Apache Flink:KMS credential management integration (Beta)

1. Crie uma função do RAM

1. Acesse a página RAM console - Roles e clique em Create Role. Mantenha as opções padrão e clique em OK.

2. Insira um nome descritivo para a função, por exemplo, FlinkRoleForKMSRead, e clique em OK.

2. Adicionar uma política de confiança

Depois de criar a função, na página de detalhes da função, clique em aba Trust Policy e clique em Edit Trust Policy. Adicione "Service": ["stream.aliyuncs.com"] a Principal para autorizar o Flink a assumir a função. Para mais detalhes, consulte Modify the trust policy of a RAM role.

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

3. Conceder permissões de acesso ao KMS para a função do RAM

Quando o Flink assume a função, ele chama o KMS usando a função. Conceda antecipadamente as permissões necessárias do KMS à política de permissões da função.

Na página de detalhes da função, clique em aba Permissions e clique em Grant Permission. Adicione a política AliyunKMSSecretUserAccess (que concede permissão para recuperar credenciais no KMS). Se as credenciais forem criptografadas com uma chave mestra do cliente (CMK), conceda também a permissão kms:Decrypt adicionando a política AliyunKMSCryptoUserAccess.

Para controlar as permissões no nível da chave e da credencial, crie uma política personalizada que conceda as permissões GetSecretValue e Decrypt e restrinja o acesso aos recursos especificados. Para mais informações, consulte Referência de política personalizada do KMS e Crie uma política personalizada.

{
  "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>"
    }
  ]
}

Crie uma política de permissões

Acesse a página RAM console - Policies e clique em Create Policy. Alterne para o modo Script e edite o conteúdo da política.

{
  "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"
        }
      }
    }
  ]
}

Clique em OK e salve a política, por exemplo, como FlinkPassRolePolicy. Para obter mais informações, consulte Crie uma política personalizada.

Conceder permissões ao usuário do RAM

1. Acesse a página RAM console - Users. Encontre o usuário do RAM a ser autorizado e clique em Add Permissions na coluna Actions.

2. No painel Add Permissions, pesquise e selecione a política criada na Etapa 3 (por exemplo, FlinkPassRolePolicy) e clique em Confirm e confirme. Para mais detalhes, consulte Conceder permissões a um usuário do RAM.

Exemplo do Postgres CDC Connector

Job SQL: O exemplo a seguir usa uma tabela source do Postgres CDC e hospeda o nome de usuário e a senha do banco de dados com uma credencial do KMS. Uma credencial de banco de dados chamada my-db-secret foi criada no KMS, com o 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.ca' = 'oss://<bucket>/kms-ca.pem',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

Para as descrições de parâmetros, pré-requisitos e mapeamento de tipos do conector Postgres CDC, consulte Postgres CDC connector.

Exemplo do Kafka Connector

A configuração SASL JAAS referencia as credenciais do Kafka. Substitua os valores de username e password em properties.sasl.jaas.config por referências secret://. Os demais parâmetros do Kafka permanecem inalterados. Uma credencial chamada my-kafka-secret foi criada no KMS, com o SecretData {"AccountName":"<SASL username>","AccountPassword":"<SASL password>"}.

Job SQL: O exemplo a seguir usa uma tabela source do Kafka. Configure uma tabela de resultados da mesma maneira.

CREATE TEMPORARY TABLE kafka_source (
  order_id BIGINT,
  user_id BIGINT,
  amount INT,
  payload STRING
) WITH (
  'connector' = 'kafka',
  'topic' = '<topic name>',
  'properties.bootstrap.servers' = '<broker 1>,<broker 2>,<broker 3>',
  'properties.group.id' = '<group id>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  /* Reference KMS credentials in username and password */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";',
  'kms.endpoint' = 'kms.cn-beijing.aliyuncs.com',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

Job YAML de ingestão de dados: O source e o destino especificam os parâmetros do KMS separadamente. O source usa um gateway compartilhado e o destino usa um gateway dedicado, portanto, este exemplo abrange ambos.

source:
  type: kafka
  name: Kafka Source
  topic: <topic name>
  value.format: json
  scan.startup.mode: earliest-offset
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  properties.group.id: <group id>
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # Shared gateway: the endpoint is fixed per region, and kms.ca is not required
  kms.endpoint: kms.cn-beijing.aliyuncs.com
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: <broker 1>,<broker 2>,<broker 3>
  # ApsaraMQ for Kafka does not support idempotent writes. Disable idempotence on VVR 8.0.0 and later
  properties.enable.idempotence: false
  properties.security.protocol: SASL_PLAINTEXT
  properties.sasl.mechanism: PLAIN
  properties.sasl.jaas.config: org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="secret://kms.my-kafka-secret.AccountName" password="secret://kms.my-kafka-secret.AccountPassword";
  # Dedicated gateway: kms.ca is also required
  kms.endpoint: <your-kms-instance-vpc-endpoint>
  kms.ca: oss://flink-fullymanaged-<workspace-id>/artifacts/namespaces/<namespace>/PrivateKmsCA_kst-******.pem
  kms.akless.assume-role.role-name: FlinkRoleForKMSRead

Para os parâmetros completos e configurações de segurança do conector Kafka, consulte Kafka SQL connector e Kafka YAML connector.

Descrição dos parâmetros