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

Genel Bakış

KMS kimlik bilgisi yönetimini entegre ederek, Flink işlerinde KMS üzerinde barındırılan kimlik bilgilerine (veritabanı kullanıcı adları ve şifreleri gibi) başvurabilirsiniz. Bu sayede iş kodunda düz metin kimlik bilgileri açıkta kalmaz. Çalışma zamanında Flink, geçici bir STS belirteci almak için bir RAM rolünü üstlenir, gerçek kimlik bilgisi değerini almak üzere KMS'i çağırır ve ardından hedef veri kaynağına bağlanır.

Temel avantajlar:

• Hassas bilgilerin korunması: İşlerde düz metin user-pwd açıkta kalmaz. Kimlik bilgileri KMS'de şifrelenmiş olarak barındırılır.

• Bakım gerektirmeyen kimlik bilgileri: KMS kimlik bilgisi döndürme özelliği sayesinde, güncellemelerden sonra Flink en son kimlik bilgisi değerini otomatik olarak alır. İş kodunun değiştirilmesine gerek kalmaz.

Notlar

KMS'deki kimlik bilgilerinin yanlışlıkla silinmesi veya RAM'deki bir RAM rolünün izinlerinin ve güven ilkesinin yanlışlıkla silinmesi ya da değiştirilmesi, Flink işlerinin KMS kimlik bilgilerini alamamasına ve dolayısıyla başarısız olmasına neden olabilir. Realtime Compute for Apache Flink Hizmet Seviyesi Sözleşmesi uyarınca, müşterilerin kaynakları hatalı kullanmasından kaynaklanan kullanılamazlık durumu SLA kapsamında değildir.

Sınırlamalar

• Yalnızca Postgres CDC Connector (VVR 11.8 ve sonrası) ile Kafka Connector (VVR 11.9 ve sonrası), KMS kimlik bilgilerini kullanarak cloud kaynaklara erişimi destekler.

• KMS'e erişmek için başvurulan RAM rollerinin sayısı konusunda bir sınır bulunmaktadır. Ayrıntılar için bkz. Cloud kaynaklara erişmek için RAM rolü kullanma.

İzin hazırlığı

• Adım 1 (KMS kimlik bilgisi oluşturma): Operatörün KMS kimlik bilgisi yönetimi izinlerine sahip olması gerekir.

• Adım 2 ve Adım 3: Operatöre AliyunRAMFullAccess ilkesi verilmeli veya RAM rolü yönetimi ile izin yönetimi izinleri ayrı ayrı tanınmalıdır. Daha fazla bilgi için bkz. Özel ilke oluşturma.

• Adım 4 ve Adım 5 (Flink'te rol referans gösterme ve iş geliştirme): Operatörün Flink ad alanı üzerinde düzenleyici veya daha yüksek düzeyde izinlere sahip olması ya da RAM rollerini referans gösterme, dosyaları yönetme ve iş geliştirme için ayrıntılı izinlere sahip olması gerekir. Ayrıntılar için bkz. Geliştirme konsoluna erişim izinleri verme.

Adım 1: KMS kimlik bilgisi oluşturma

KMS, birden fazla türde kimlik bilgisi sunar. Aşağıdaki adımlarda örnek olarak bir RDS kimlik bilgisi kullanılmıştır:

1. Key Management Service konsolu üzerinde oturum açın. Üst menü çubuğundan bölgeyi seçin ve sol gezinme bölmesinde Resources > Secrets seçeneğini tıklayın.

2. Self-Managed Secrets sekmesinde, Secret Type alanında Database Secret seçeneğini tıklayın.

3. Sağ taraftaki listenin üst kısmında, Create Database Secret altında Create Single Secret seçeneğini tıklayın, yapılandırmayı tamamlayın ve OK seçeneğini tıklayın.

Daha fazla bilgi için bkz. Secrets Manager'a giriş ve ApsaraDB RDS gizli dizilerini yönetme ve kullanma.

Adım 2: RAM rolü oluşturma ve yapılandırma

Bu rol, Flink hizmetinin KMS'e erişmek için kullandığı kimlik kimlik bilgisidir. Flink, geçici kimlik bilgilerini almak için bu rolü üstlenir ve kimlik bilgisi değerlerini almak üzere rol adına KMS'i çağırır. KMS'e kimlik bilgisi içermeyen erişim hakkında ayrıntılı bilgi için Cloud kaynaklara erişmek için RAM rolü kullanma bölümünün ekine bakın.

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

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

Adım 3: RAM kullanıcıya izin verme

Bir izin ilkesi oluşturun ve bunu RAM kullanıcıya atayın. Bu işlem, kullanıcının Adım 2'de oluşturulan RAM rolünü Flink hizmetine aktarmasına (pass) olanak tanır.

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

Adım 4: Flink'te RAM rolünü referans gösterme

Adım 2'de oluşturulan RAM rolünü Flink ad alanına referans gösterin, böylece işler KMS'e erişmek için bu rolü kullanabilir.

1. Realtime Compute konsolunda oturum açın ve hedef çalışma alanına gidin.

2. Sol gezinme bölmesinde Security > Security seçeneğini tıklayın.

3. RAM Roles sekmesini tıklayın.

4. Reference RAM Role seçeneğini tıklayın.

5. Açılan iletişim kutusunda, Adım 2'de oluşturulan RAM rolünü seçin (örneğin, FlinkRoleForKMSRead). Rol adına göre bulanık arama desteklenir.

   Not

   Rol listesi, mevcut ad alanına sahip olan Alibaba Cloud hesabı altındaki normal hizmet rollerini içerir. Operatörün mevcut ad alanıyla aynı Alibaba Cloud hesabına ait olması ve rolleri görüntülemek için ram:ListRoles iznine sahip olması gerekir.

6. Test seçeneğini tıklayın. Sistem aşağıdaki iki öğeyi otomatik olarak denetler:

   • PassRole izin denetimi: Mevcut kullanıcının seçilen RAM rolü üzerinde ram:PassRole izni vardır.

   • Güven ilkesi denetimi: Seçilen RAM rolü, mevcut Flink ad alanının rolü üstlenmesine güvenir.

7. Denetimler başarılı olduktan sonra OK seçeneğini tıklayın.

Daha fazla bilgi için bkz. Cloud kaynaklara erişmek için RAM rolü kullanma.

Adım 5: İşlerde KMS kimlik bilgilerini kullanma

Önceki adımlar tamamlandıktan sonra, iş kodunda secret:// başvuru formatını kullanarak kimlik bilgilerini belirtin ve KMS bağlantı parametrelerini yapılandırın. Flink, çalışma zamanında veri kaynağına erişmek için kimlik bilgisi değerlerini otomatik olarak alır.

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'
);

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'
);

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