Message delivery-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Message delivery streams startup logs, resource usage, running events, and job status changes from Flink jobs in a target namespace to an external message queue or storage system in real time. By persisting this data, you can query historical records for troubleshooting, performance optimization, and auditing. This topic describes how to configure message delivery and view the delivered messages.

Background

This feature supports cross-region delivery. The following table describes the types of information delivered and their delivery triggers.

Category	Description	Delivery trigger
Job startup logs	Logs generated during the entire job startup process, from the initialization of the Flink environment to the startup of the JobManager and the generation of the Flink execution graph.	Delivered once when the job starts successfully or reaches a final state (such as failed or finished).
Resource information	Changes in resource usage within a namespace, including the consumption and allocation of CPU and memory. Changes in resource usage for a queue, including the consumption and allocation of CPU and memory. Important This information is for resource capacity tracking only and does not provide alerts.	Delivered every 30 seconds while the namespace is operating normally.
Runtime events	The startup status at each stage of the job startup process.	Triggered when a job runtime event occurs.
Job resource consumption	Resource usage of running streaming jobs. This does not include batch jobs or jobs running on a session cluster.	Delivered every 10 minutes while the job is running.
Job state change	Job state change events, including the previous and new states.	Triggered when the job state changes.

Notes

You can deliver messages only to Simple Log Service (SLS). Before you begin, you must create an SLS project and a Logstore. For instructions, see Use LoongCollector to collect and analyze text logs on ECS instances.
Message delivery is free of charge. However, you are charged for using SLS features such as Logstore indexing. For more information, see Billing overview.
Querying and analyzing logs in SLS requires an index. Enabling an index is optional and incurs fees for index traffic and storage. For more information about the fees, see Billing overview.
You can configure server-side encryption for the Logstore. Messages delivered to the Logstore are also encrypted. To learn how, see Data encryption.
Message delivery currently supports only job startup logs to SLS. You must configure job runtime logs separately to output them to OSS, SLS, or Kafka. For more information, see Configure job log output.
Changes to the message delivery configuration can take up to 30 seconds to take effect.

Procedure

Step 1: Configure message delivery

Go to the message delivery configuration page
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left-side navigation pane, click O&M > Configurations.

Configure Log Service delivery parameters

On the Message Delivery tab, turn on the Deliver to SLS switch.

Configure the Log Service parameters.

Parameter	Description
Authorization mode	STS Token: This mode delivers messages only to a Log Service project in the same region as your Flink workspace. Therefore, you only need to configure the Log Service project and Logstore. AccessKey: This mode allows message delivery to a Log Service project in a different region. Therefore, you must configure the endpoint, AccessKey ID, and AccessKey Secret.
Log Service project	The name of the Log Service project.
Log Service Logstore	The name of the Log Service Logstore.
endpoint	The Log Service endpoint. If Authorization mode is set to STS Token, the system automatically uses the endpoint for the region where your Flink workspace is located. If you set Authorization mode to AccessKey, you must manually configure this parameter.
Delivery scope	The message content to deliver. For more information, see Field reference later in this topic.
AccessKey ID	Your Alibaba Cloud account's AccessKey ID and AccessKey Secret. Important To prevent your AccessKey credentials from being exposed, we recommend that you manage your AccessKey ID and AccessKey Secret as variables. You can click the drop-down arrow to select an existing variable, or click the icon next to the parameter to create a new variable. For more information about how to manage variables and view your AccessKey ID and AccessKey Secret, see Variable management and How do I view AccessKey credentials?
AccessKey Secret

Click Save.

Step 2: View delivered messages

Procedure

Click Open the SLS console to the right of Log Service project.
View the information in the raw logs.

The raw logs are JSON-format messages that contain resource quota information (RESOURCE_QUOTA). These logs include fields such as namespaceTotalCpu (20.0), namespaceTotalMemory (0.08 TiB), namespaceUsedCpu (8.0), namespaceUsedMemory (23.00 GiB), and resourceQueueName (default-queue). These fields record the CPU and memory quotas and usage for the namespace and resource queue.

Field descriptions

Startup logs (JOB_START_LOG)

Parameter	Description
messageType	Message type. Fixed to JOB_START_LOG (job startup log).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
length	The total length of the log.
offset	For sharded logs, this field specifies the start offset of the log entry.
content	Details of the job startup log.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Resource details (JOB_RESOURCE_QUOTA)

Parameter	Description
messageType	Message type. Fixed to RESOURCE_QUOTA (resource usage).
namespaceTotalCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
namespaceTotalCpu	The total CPU resources for the namespace, measured in CUs.
namespaceTotalMemory	The total memory resources for the namespace.
namespaceUsedCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
namespaceUsedCpu	The CPU resources used by the namespace, measured in CUs.
namespaceUsedMemory	The memory resources used by the namespace.
resourceQueueName	The name of the resource queue.
resourceQueueTotalCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
resourceQueueTotalCpu	The total CPU resources for the resource queue, measured in CUs.
resourceQueueTotalMemory	The total memory resources for the resource queue.
resourceQueueUsedCpuMemory	This field is deprecated. Use the corresponding `...Cpu` and `...Memory` fields instead.
resourceQueueUsedCpu	The CPU resources used by the resource queue, measured in CUs.
resourceQueueUsedMemory	The memory resources used by the resource queue.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Runtime events (JOB_EVENT)

Parameter	Description
messageType	Message type. Fixed to JOB_EVENT (runtime event).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
eventId	The ID of the event.
eventName	The name of the event.
content	Details of the event.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Job resource usage (JOB_RESOURCE_USAGE)

Parameter	Description
messageType	Message type. Fixed to JOB_RESOURCE_USAGE (job resource usage).
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
tag	The job tag. This field is empty if no tag is configured.
jobUsedCpu	The number of CUs used by the job.
jobUsedMemory	The amount of memory used by the job.
workspace	The ID of the workspace.
namespace	The name of the namespace.
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated.

Job status changes (JOB_STATUS_CHANGED)

Parameter	Description
messageType	Message type. Fixed to JOB_STATUS_CHANGED (job status change).
messageId	The unique ID of the message.
timestamp	The timestamp when the message was generated, in milliseconds.
workspace	The ID of the workspace.
namespace	The name of the namespace.
deploymentId	The ID of the job deployment.
deploymentName	The name of the job deployment.
jobId	The ID of the job instance.
previousStatus	The job status before the change.
currentStatus	The job status after the change.
tag	The job tag. This field is empty if no tag is configured.
additionalInfo	Additional information.

References

To configure logging for a single job, see Configure job log output.
To view logs in the Realtime Compute for Apache Flink development console, see View startup and running logs, View running events, View running exception logs, and View historical job instance logs.