Poll for task status - CloudFlow - Alibaba Cloud Documentation Center

This topic describes how to poll for task status in Serverless workflow.

Overview

If no callback is specified after a long-running task is completed, developers usually poll the task status to check whether the task is completed. Reliable polling requires task status to be persistent. Therefore, even if the current polling process fails, the polling continues after the process resumes. In this example, assume that a user calls Function Compute to submit a multimedia processing task that takes one minute to several hours. The task execution status can be queried by calling the corresponding API. This topic describes how to implement a common and reliable flow for polling task status in Serverless workflow.

Implementation in Serverless workflow

The following tutorial shows how to orchestrate two functions deployed in Function Compute as a flow for polling task status in the following three steps:

Create a function in Function Compute
Create a flow in Serverless workflow
Execute the flow and view the result

Step 1: Create a function in Function Compute

Create a service named fnf-demo in Function Compute, and create two functions (StartJob and GetJobStatus) in Python 2.7 in this service. For more information, see Quickly create a function.

The StartJob function is used to simulate calling an API to start a long-running task and return the 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)}

The GetJobStatus function is used to simulate calling an API to query the execution result of the specified task. It compares the value of the current time minus the time when the function is first executed with the value specified in delay and then returns the task status "success" or "running" accordingly.

import logging
import uuid
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"

  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 flow in Serverless workflow

The following steps show the main logic of this flow:

StartJob: Serverless Workflow calls the StartJob function to start a task.
Wait10s: Serverless Workflow waits for 10s.
GetJobStatus: Serverless Workflow calls the GetJobStatus function to query the status of the current task.

CheckJobComplete: Serverless Workflow checks the result returned by the GetJobStatus function.

The result "success" indicates that the flow is completed.
If the polling requests are sent three or more times, Serverless Workflow considers that the task fails, and then the flow fails.
If the result "running" is returned, the system goes back to the Wait10s step.

version: v1
type: flow
steps:
  - type: task
    name: StartJob
    resourceArn: acs:fc:cn-hangzhou:{accountID}:services/fnf-demo/functions/StartJob
  - type: pass
    name: Init
    outputMappings:
      - target: try_count
        source: 0
  - type: wait
    name: Wait10s
    duration: 10
  - type: task
    name: GetJobStatus
    resourceArn: acs:fc:cn-hangzhou:{accountID}:services/fnf-demo/functions/GetJobStatus
    inputMappings:
      - target: job_id
        source: $local.job_id
      - target: delay
        source: $input.delay
      - target: try_count
        source: $local.try_count
  - type: choice
    name: CheckJobComplete
    inputMappings:
      - target: status
        source: $local.job_status
      - target: try_count
        source: $local.try_count
    choices:
        - condition: $.status == "success"
          goto: JobSucceeded
        - condition: $.try_count > 3
          goto: JobFailed
        - condition: $.status == "running"
          goto: Wait10s
  - type: succeed
    name: JobSucceeded
  - type: fail
    name: JobFailed

Step 3: Execute the flow and view the result

In the Serverless Workflow console, find the target flow, click Start Execution, and then enter the following JSON object as the input of this execution. The value of delay indicates the time that the task takes to run. In this example, it is set to 20, which means that the GetJobStatus function returns "success" 20s later after the task is started, before which "running" is returned. You can change the value of delay to view different execution results.

{
  "delay": 20
}

The following figure is the visual display of the polling flow from start to end.
As shown in the following figure, the task takes 20s to run. When the GetJobStatus function is called for the first time, "running" is returned. Therefore, when CheckJobComplete is called, the system proceeds to the Wait10s step to wait 10s before the next query is initiated. The "success" result is returned for the second query, and the flow ends.