Realtime Compute for Apache Flink:Secure access best practices

Step 1: Create a RAM user

Create a RAM user using your Alibaba Cloud account or as a RAM administrator.

In the Access Mode section, select OpenAPI Access. An AccessKey ID and AccessKey secret are automatically generated.

Step 2: Create a custom policy for Simple Log Service

1. In the RAM console, go to Permissions > Policies in the left navigation pane.

2. On the Policies page, click Create Policy.

3. On the Create Policy page, click the JSON tab and replace the existing content with one of the following policy templates. Then click Next to edit policy information.

   Replace <Project name> and <Logstore name> with your actual values. For more on writing custom policies, see Examples of using custom policies to grant permissions to a RAM user.

   Grant the read-only permissions on a specific Logstore

   Read-only access to a specific Logstore:

   {
       "Version": "1",
       "Statement": [
           {
               "Action": "log:ListProject",
               "Resource": "acs:log:*:*:project/*",
               "Effect": "Allow"
           },
           {
               "Action": "log:List*",
               "Resource": "acs:log:*:*:project/<Project name>/logstore/*",
               "Effect": "Allow"
           },
           {
               "Action": [
                   "log:Get*",
                   "log:List*"
               ],
               "Resource": "acs:log:*:*:project/<Project name>/logstore/<Logstore name>",
               "Effect": "Allow"
           }
       ]
   }

   Grant the write permissions on a specific Logstore

   Write access to a specific Logstore:

   {
     "Version":"1",
     "Statement":[
       {
         "Effect":"Allow",
         "Action":[
           "log:PostLogStoreLogs"
         ],
         "Resource":[
           "acs:log:*:*:project/<Project name>/logstore/<Logstore name>"
         ]
       }
     ]
   }

   The logstore keyword in the resource ARN covers both Logstores and Metricstores. The write policy above applies to Metricstores as well.

4. Enter a name and description for the policy, then click OK.

Step 3: Attach the policy to the RAM user

Use the custom policy you just created to grant permissions to the RAM user. See Grant permissions to a RAM user for step-by-step instructions.

Step 4: Store the AccessKey pair as Flink variables

Create two variables in Flink — one for the AccessKey ID and one for the AccessKey secret — using the credentials from Step 1. After the variables are created, your SQL references the variable names rather than the credentials directly.

See Manage variables and keys for instructions. In this example, the variables are named slslak (AccessKey ID) and slsaks (AccessKey secret).

Step 5: Use the variables in a Flink deployment

Reference variables in the ${secret_values.<variable_name>} format inside your SQL connector configuration. The following example reads from a Simple Log Service Logstore and writes to a blackhole sink for testing:

CREATE TEMPORARY TABLE sls_input(
   `__source__` STRING METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
   `__topic__` STRING METADATA VIRTUAL,
   deploymentName STRING,
   `level`STRING,
   `location` STRING,
   message STRING,
   thread STRING,
   `time`STRING
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-beijing-intranet.log.aliyuncs.com',
  'accessId' = '${secret_values.slsak}',
  'accessKey' = '${secret_values.slsaks}',
  'starttime' = '2024-08-30 15:39:00',
  'project' ='test',
  'logstore' ='flinktest'
);

CREATE TEMPORARY TABLE blackhole_sink(
   `__source__` STRING,
   `__topic__` STRING,
   deploymentName STRING,
   `level` STRING,
   `location` STRING,
   message STRING,
   thread STRING,
   `time` STRING,
   receive_time BIGINT
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT `__source__`,
   `__topic__`,
   deploymentName,
   `level`,
   `location`,
   message,
   thread,
   `time`,
   cast(__tag__['__receive_time__'] as bigint) as receive_time
FROM sls_input;

The accessId and accessKey fields resolve to the actual credentials at runtime — they are never stored in plaintext in your SQL draft.