Realtime Compute for Apache Flink:KMS 認証情報管理統合 (Beta)

1. RAM ロールの作成

1. RAM コンソール - Roles ページにログインし、[Create Role] をクリックします。既定のオプションのまま [OK] をクリックします。

2. 識別しやすいロール名 (例: FlinkRoleForKMSRead) を入力し、[OK] をクリックします。

2. 信頼ポリシーの追加

ロールが作成された後、ロールの詳細ページで[信頼ポリシー] タブをクリックし、[信頼ポリシーの編集] をクリックします。Principal に "Service": ["stream.aliyuncs.com"] を追加して、Flink がロールを引き受けることを信頼できるようにします。詳細については、「RAM ロールの信頼ポリシーを変更する」をご参照ください。

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

3. RAM ロールへのKMSアクセス権限の付与

Flink がロールを引き受けた後、そのロールとして KMS を呼び出します。必要な KMS 権限を、事前にロールの権限ポリシーに付与してください。

ロールの詳細ページで [Permissions] タブをクリックし、[Grant Permission] をクリックします。AliyunKMSSecretUserAccess ポリシー (KMS 内の認証情報を取得する権限を付与) を追加します。認証情報がカスタマーマスターキー (CMK) で暗号化されている場合は、AliyunKMSCryptoUserAccess ポリシーを追加して kms:Decrypt 権限も付与します。

キーおよび認証情報レベルで権限を制御するには、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 コンソール - Policies ページにログインし、[Create Policy] をクリックします。[Script] モードに切り替え、ポリシー内容を編集します。

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

[OK] をクリックして、たとえば FlinkPassRolePolicy としてポリシーを保存します。詳細については、「カスタムポリシーの作成」をご参照ください。

RAMユーザーへの権限付与

1. RAM コンソール - Users ページにログインします。権限を付与する RAM ユーザーを見つけ、[Actions] 列の [Add Permissions] をクリックします。

2. [権限の追加] パネルで、ステップ 3 で作成したポリシー (例: FlinkPassRolePolicy) を検索して選択し、[確認] をクリックします。詳細については、「RAM ユーザーに権限を付与する」をご参照ください。

Postgres CDCコネクタの例

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.ca' = 'oss://<bucket>/kms-ca.pem',
  'kms.akless.assume-role.role-name' = 'FlinkRoleForKMSRead'
);

Postgres CDC コネクターのパラメーターの説明、前提条件、および型マッピングについては、「Postgres CDC コネクター」をご参照ください。

Kafkaコネクタの例

Kafka の認証情報は SASL JAAS 設定で参照します。properties.sasl.jaas.config 内の username と password の値を secret:// を使った参照に置き換えます。その他の Kafka パラメーターは変更しません。KMS に my-kafka-secret という名前の認証情報が作成済みで、SecretData は {"AccountName":"<SASL username>","AccountPassword":"<SASL password>"} であるものとします。

SQL ジョブ: 次の例では Kafka のソーステーブルを使用します。結果テーブルも同様に設定します。

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',
  /* usernameとpasswordでKMS認証情報を参照します */
  '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'
);

データインジェスト YAML ジョブ: ソースとシンクは、それぞれ個別に KMS パラメーターを指定します。ソースは共有ゲートウェイを使用し、シンクは専用ゲートウェイを使用するため、この例では両方を扱います。

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";
  # 共有ゲートウェイ：エンドポイントはリージョンごとに固定で、kms.caは不要です
  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は冪等書き込みをサポートしていません。VVR 8.0.0以降では冪等性を無効にします
  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";
  # 専用ゲートウェイ：kms.caも必要です
  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

Kafka コネクタの完全なパラメーターとセキュリティ設定については、「Kafka SQL コネクタ」および「Kafka YAML コネクタ」をご参照ください。

パラメーターの説明