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

For long-running tasks that lack a callback mechanism, developers often use polling to check when a task is complete. A reliable polling implementation requires state persistence to ensure that polling continues even if the process fails and restarts. This document shows you how to implement task status polling using CloudFlow.

Scenario

In this scenario, you use a Function Compute function to submit a multimedia processing task. Because the task can take from one minute to several hours to complete, you must continuously poll an API to get its status.

Procedure

Step 1: Create FC functions

In this step, you will create two Python 3.10 (or later) functions to simulate the services for this scenario. For detailed instructions, see Create a task function.

Create a function named StartJob. This 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. This function simulates calling an API to retrieve the task's execution result. It compares the elapsed time since its first invocation 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 a workflow

In this step, you will create a workflow that starts a task, polls its status, and ends based on the final task status.

Create a workflow in the CloudFlow console. You can use the default parameters.

On the CloudFlow Studio page, drag and configure the state nodes as follows.

StartJob step: Invokes the StartJob function from Step 1 to start a task.
Wait step: Sets a wait time of 10 seconds.
GetJobStatus step: Invokes the GetJobStatus function from Step 1 to get the task status.
CheckJobComplete step: Chooses the next step based on the result returned by the GetJobStatus function:
- If "success" is returned, the entire workflow execution succeeds.
- If polling attempts exceed 3, the workflow execution fails.
- If "running" is returned, the workflow returns to the Wait step to continue polling.

The following YAML code defines the logic described above:

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

Step 3: Start execution and view results

After creating the workflow, select the Workflow Configuration tab and set the execution role.
Click Save, then Execute, and provide the following JSON object as input. The delay field simulates the time required for the task to complete. With a 30-second delay, the GetJobStatus function returns "running" before this period ends and "success" afterward. You can adjust the delay value to observe different outcomes.
```
{
  "delay": 30
}
```
- After the execution completes, the graph view on the execution details page shows all workflow steps (StartJob → Pass → Wait → GetJobStatus → CheckJobComplete → JobSucceeded) in green, indicating success. The Output panel shows a final status of job_status: success and try_count: 1 in the Body, indicating that the task succeeded.
- For an execution that takes 30 seconds, the history shows the GetJobStatus function returning "running", causing the CheckJobComplete step to loop back to the Wait step. During this polling cycle, when the GetJobStatus task returns a job_status of running, the workflow enters the CheckJobComplete (Choice state). This state determines the task is incomplete and transitions the workflow to the Wait state. This cycle repeats until the task is complete.