Poll for Task Status-CloudFlow(CloudFlow)-阿里云帮助中心

For long-running tasks that provide no callback mechanism, polling is how you find out when a task is complete. A reliable polling implementation also requires state persistence, so that polling continues even if the polling process fails and restarts. This topic describes the task status polling pattern and how to implement it with CloudFlow.

Scenario

You call Function Compute (FC) to submit a multimedia processing task that runs from one minute to several hours. The task returns no completion callback, so the only way to obtain its status is to poll an API continuously until the task reports a final result.

How the polling workflow works

CloudFlow implements polling as a loop of workflow states instead of a loop inside your application code. The workflow starts the task, waits, polls the task status once, and then decides whether to poll again, carrying the task ID and the polling counter from one state to the next. This is where state persistence comes from: the polling progress belongs to the workflow execution, not to a process that you have to keep alive. Every execution ends in one of two terminal states, so polling stops when the task succeeds or when the retry limit is reached.

Prerequisites

Your Alibaba Cloud account ID and the ID of the region in which you create the functions. The workflow definition references both values in the function ARNs.
An execution role for the workflow.

Procedure

Step 1: Create the FC functions

The two functions in this step simulate the API that the scenario polls.

Create two functions that use the Python 3.10 or later runtime. For detailed instructions, see Create a task function.

Create a function named StartJob. The function simulates starting a long-running task by calling an API, and returns a task ID.

import logging
import uuid
def handler(event, context):
  logger = logging.getLogger()
  id = uuid.uuid4()
  logger.info('Started job with ID %s' % id)
  return {"job_id": str(id)}

Create a function named GetJobStatus. The function simulates calling an API to retrieve the execution result of a specified task. It compares the time elapsed since the function instance was initialized with the delay input value, and returns a status of success or running.

import logging
import time
import json
start_time = int(time.time())
def handler(event, context):
  evt = json.loads(event)
  logger = logging.getLogger()
  job_id = evt["job_id"]
  logger.info('Started job with ID %s' % job_id)
  now = int(time.time())
  status = "running"
  delay = 60
  if "delay" in evt:
    delay = evt["delay"]
  if now - start_time > delay:
    status = "success"
  else:
    status = "running"
  try_count = 0
  if "try_count" in evt:
    try_count = evt["try_count"]
  try_count = try_count + 1
  logger.info('Job %s, status %s, try_count %d' % (job_id, status, try_count))
  return {"job_id": job_id, "job_status":status, "try_count":try_count}

Step 2: Create the workflow

In this step, you create a workflow that starts a task, polls its status, and ends the execution based on the final task status. Only the way you define the workflow differs: you either build it on the CloudFlow Studio canvas or define it from the flow spec at the end of this step. Both describe the same workflow, so choose one of them. The other actions in this step apply in both cases.

Create a workflow in the CloudFlow console. You can use the default parameters.
On the CloudFlow Studio page, drag and configure the following state nodes. Alternatively, skip this action and define the same workflow from the flow spec at the end of this step.
1. StartJob: a Task state that invokes the StartJob function from Step 1 to start a task.
2. Pass: a Pass state that initializes try_count to 0 and extracts job_id from the output of StartJob.
3. Wait: a Wait state that pauses the execution for 10 seconds before each poll.
4. GetJobStatus: a Task state that invokes the GetJobStatus function from Step 1 to get the task execution result.
5. CheckJobComplete: a Choice state that selects the next state based on the result that the GetJobStatus function returns.
  - If success is returned, the execution transitions to JobSucceeded and the workflow execution succeeds.
  - If the number of polling attempts exceeds 3, the task is considered failed and the execution transitions to FailJobFailed.
  - If running is returned, the execution returns to the Wait state and polls again.
6. JobSucceeded and FailJobFailed: the two terminal states of the workflow. JobSucceeded ends the execution successfully, and FailJobFailed ends it with the timeout code.
Click the Workflow Configuration tab and set the execution role.
Click Save.

The following flow spec defines the same workflow as the state nodes in the preceding action. Replace {region} with the ID of the region in which you created the two functions, and replace {accountID} with your Alibaba Cloud account ID. If you define the workflow this way, you still set the execution role and save the workflow.

Type: StateMachine
Name: MyWorkFlow
SpecVersion: v1
StartAt: StartJob
States:
  - Type: Task
    Name: StartJob
    Action: FC:InvokeFunction
    TaskMode: RequestComplete
    Parameters:
      invocationType: Sync
      resourceArn: acs:fc:{region}:{accountID}:functions/StartJob/LATEST
    Next: Pass
  - Type: Pass
    Name: Pass
    Next: Wait
    OutputConstructor:
      $: $Input
    InputConstructor:
      try_count: 0
      job_id.$: $Input.Body.job_id
  - Type: Wait
    Name: Wait
    Seconds: 10
    Next: GetJobStatus
  - Type: Task
    Name: GetJobStatus
    Action: FC:InvokeFunction
    TaskMode: RequestComplete
    Parameters:
      invocationType: Sync
      resourceArn: acs:fc:{region}:{accountID}:functions/GetJobStatus/LATEST
      body:
        job_id.$: $Input.job_id
        try_count.$: $Input.try_count
        delay.$: $Context.Execution.Input.delay
    Next: CheckJobComplete
    OutputConstructor:
      $: jsonMerge($Input, $Output.Body)
  - Type: Choice
    Name: CheckJobComplete
    Branches:
      - Condition: $Input.job_status== "success"
        Next: JobSucceeded
      - Condition: $Input.try_count > 3
        Next: FailJobFailed
      - Condition: $Input.job_status== "running"
        Next: Wait
    Default: FailJobFailed
  - Type: Fail
    Name: FailJobFailed
    Code: timeout
    End: true
  - Type: Succeed
    Name: JobSucceeded
    End: true

Note

In this example, the Wait state pauses for 10 seconds and CheckJobComplete fails the task after more than three polling attempts. The workflow therefore polls for far less time than the multi-hour tasks described in the scenario. To cover longer tasks, increase Seconds in the Wait state and raise the try_count threshold in CheckJobComplete.

Step 3: Start an execution and view the result

Click Execute, and then provide the following JSON object as the execution input:

{
  "delay": 30
}

The delay field simulates the time that the task needs to complete. In this example, the GetJobStatus function returns running until 30 seconds have elapsed and success after that. Adjust the delay value to observe different execution results.

While the execution runs, the graph view on the left of the execution details page shows which state is active. The early polls return running, so CheckJobComplete confirms that the task is not complete yet and sends the execution back to the Wait state, which polls again after another 10 seconds. When the elapsed time exceeds the delay value, GetJobStatus returns success and the execution ends at JobSucceeded.

After the execution completes, every state on the path — StartJob, Pass, Wait, GetJobStatus, CheckJobComplete, and JobSucceeded — is green in the graph view, which indicates that the execution succeeded. The Body section of the Output panel on the right shows job_status: success together with the try_count value that the successful poll reported.

If more than three polls return running, CheckJobComplete routes the execution to FailJobFailed instead. The execution ends with the timeout code, which means that the task did not finish within the polling window that this example allows.