Python と boto3 で Amazon Simple Workflow Service (SWF) を試す

試す為のソース

試す

cp config.json.sample config.json

で任意の各 swf の項目の名前を記述する。
AWS_PROFILE は SWF へのアクセス権限のあるものを記述する。

{
  "swf": {
    "DOMAIN": "test-domain",
    "WORKFLOW": "test-workflow",
    "TASKNAME": "test-taskname",
    "VERSION": "0.1",
    "TASKLIST": "test-tasklist"
  },
  "aws": {
    "profile": "AWS_PROFILE"
  }
}

Domain / Workflow Type / Activity Type を作る

作成

python testcreate.py

Domain 確認

aws swf describe-domain --name test-domain --profile AWS_PROFILE
{
    "domainInfo": {
        "status": "REGISTERED", 
        "name": "test-domain", 
        "description": "Test SWF domain"
    }, 
    "configuration": {
        "workflowExecutionRetentionPeriodInDays": "10"
    }
}

Workflow Type 確認

List

aws swf list-workflow-types --domain "test-domain" --registration-status REGISTERED --profile=AWS_PROFILE
{
    "typeInfos": [
        {
            "status": "REGISTERED", 
            "creationDate": 1472616047.186, 
            "workflowType": {
                "version": "0.1", 
                "name": "test-workflow"
            }, 
            "description": "Test workflow"
        }
    ]
}

Describe

aws swf describe-workflow-type --domain "test-domain" --workflow-type "name=test-workflow,version=0.1" --profile=AWS_PROFILE
{
    "configuration": {
        "defaultTaskStartToCloseTimeout": "NONE", 
        "defaultExecutionStartToCloseTimeout": "250", 
        "defaultTaskList": {
            "name": "test-tasklist"
        }, 
        "defaultChildPolicy": "TERMINATE"
    }, 
    "typeInfo": {
        "status": "REGISTERED", 
        "creationDate": 1472616047.186, 
        "workflowType": {
            "version": "0.1", 
            "name": "test-workflow"
        }, 
        "description": "Test workflow"
    }
}

Activity Type 確認

List

aws swf list-activity-types --domain test-domain --registration-status REGISTERED --profile AWS_PROFILE  
{
    "typeInfos": [
        {
            "status": "REGISTERED", 
            "creationDate": 1472616047.336, 
            "activityType": {
                "version": "0.1", 
                "name": "test-taskname"
            }, 
            "description": "Test worker"
        }
    ]
}

Describe

aws swf describe-activity-type --domain "test-domain" --activity-type "name=test-taskname,version=0.1" --profile=AWS_PROFILE

{
    "configuration": {
        "defaultTaskStartToCloseTimeout": "NONE", 
        "defaultTaskList": {
            "name": "test-tasklist"
        }
    }, 
    "typeInfo": {
        "status": "REGISTERED", 
        "creationDate": 1472616047.336, 
        "activityType": {
            "version": "0.1", 
            "name": "test-taskname"
        }, 
        "description": "Test worker"
    }
}

Decider を起動する

python testdecider.py
Listening for Decision Tasks
Poll timed out, no new task.  Repoll

Worker を起動する

python testworker.py   
Listening for Worker Tasks
Poll timed out, no new task.  Repoll

WorkFlow を作る

Domain と workflowId と workflowType の name と version、taskList の name を指定して、Workflow を作成する。

作成

python testrequest.py 
Workflow requested:  {u'runId': u'abcdefghijklmnopqrstuvwxyz1234567890abcdefghij', 'ResponseMetadata': {'HTTPStatusCode': 200, 'RequestId': 'xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx', 'HTTPHeaders': {'x-amzn-requestid': 'xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxx', 'content-length': '58', 'content-type': 'application/x-amz-json-1.0'}}}

確認

aws swf list-open-workflow-executions --domain "test-domain" --start-time-filter "oldestDate=1475593200,latestDate=1475679600" --profile=AWS_PROFILE
{
    "executionInfos": [
        {
            "startTimestamp": 1475654057.8, 
            "execution": {
                "workflowId": "test-1001", 
                "runId": "u'abcdefghijklmnopqrstuvwxyz1234567890abcdefghij"
            }, 
            "executionStatus": "OPEN", 
            "cancelRequested": false, 
            "workflowType": {
                "version": "0.1", 
                "name": "test-workflow"
            }
        }
    ]
}

Decider の反応を確認する

python testdecider.py
Listening for Decision Tasks
Poll timed out, no new task.  Repoll
Dispatching task to worker {u'workflowId': u'test-1001', u'runId': u'u'abcdefghijklmnopqrstuvwxyz1234567890abcdefghij'} {u'version': u'0.1', u'name': u'test-workflow'}
Task Dispatched: 
xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
Task Completed!
Poll timed out, no new task.  Repoll

Worker の反応を確認する

python testworker.py   
Listening for Worker Tasks
Poll timed out, no new task.  Repoll
New task arrived
Task Done

Workflow Execution の History を見る

aws swf get-workflow-execution-history --domain "test-domain" --execution "workflowId=test-1001,runId=abcdefghijklmnopqrstuvwxyz1234567890abcdefghij" --profile=AWS_PROFILE
{
    "events": [
        {
            "eventId": 1, 
            "eventType": "WorkflowExecutionStarted", 
            "workflowExecutionStartedEventAttributes": {
                "taskList": {
                    "name": "test-tasklist"
                }, 
                "parentInitiatedEventId": 0, 
                "taskStartToCloseTimeout": "NONE", 
                "childPolicy": "TERMINATE", 
                "executionStartToCloseTimeout": "250", 
                "input": "", 
                "workflowType": {
                    "version": "0.1", 
                    "name": "test-workflow"
                }
            }, 
            "eventTimestamp": 1475654722.825
        }, 
        {
            "eventId": 2, 
            "eventType": "DecisionTaskScheduled", 
            "decisionTaskScheduledEventAttributes": {
                "startToCloseTimeout": "NONE", 
                "taskList": {
                    "name": "test-tasklist"
                }
            }, 
            "eventTimestamp": 1475654722.825
        }, 
        {
            "eventId": 3, 
            "eventType": "DecisionTaskStarted", 
            "eventTimestamp": 1475654722.903, 
            "decisionTaskStartedEventAttributes": {
                "scheduledEventId": 2, 
                "identity": "decider-1"
            }
        }, 
        {
            "eventId": 4, 
            "eventType": "DecisionTaskCompleted", 
            "decisionTaskCompletedEventAttributes": {
                "startedEventId": 3, 
                "scheduledEventId": 2
            }, 
            "eventTimestamp": 1475654723.059
        }, 
        {
            "eventId": 5, 
            "eventType": "ActivityTaskScheduled", 
            "activityTaskScheduledEventAttributes": {
                "taskList": {
                    "name": "test-tasklist"
                }, 
                "scheduleToCloseTimeout": "NONE", 
                "activityType": {
                    "version": "0.1", 
                    "name": "test-taskname"
                }, 
                "decisionTaskCompletedEventId": 4, 
                "heartbeatTimeout": "NONE", 
                "activityId": "activityid-yyyyyyy-yyyyyyy-yyyyyyy-yyyyyyy-yyyyyyy", 
                "scheduleToStartTimeout": "NONE", 
                "startToCloseTimeout": "NONE", 
                "input": ""
            }, 
            "eventTimestamp": 1475654723.059
        }, 
        {
            "eventId": 6, 
            "eventType": "ActivityTaskStarted", 
            "eventTimestamp": 1475654723.109, 
            "activityTaskStartedEventAttributes": {
                "scheduledEventId": 5, 
                "identity": "worker-1"
            }
        }, 
        {
            "eventId": 7, 
            "eventType": "ActivityTaskCompleted", 
            "activityTaskCompletedEventAttributes": {
                "startedEventId": 6, 
                "scheduledEventId": 5, 
                "result": "success"
            }, 
            "eventTimestamp": 1475654723.147
        }, 
        {
            "eventId": 8, 
            "eventType": "DecisionTaskScheduled", 
            "decisionTaskScheduledEventAttributes": {
                "startToCloseTimeout": "NONE", 
                "taskList": {
                    "name": "test-tasklist"
                }
            }, 
            "eventTimestamp": 1475654723.147
        }, 
        {
            "eventId": 9, 
            "eventType": "DecisionTaskStarted", 
            "eventTimestamp": 1475654723.19, 
            "decisionTaskStartedEventAttributes": {
                "scheduledEventId": 8, 
                "identity": "decider-1"
            }
        }, 
        {
            "eventId": 10, 
            "eventType": "DecisionTaskCompleted", 
            "decisionTaskCompletedEventAttributes": {
                "startedEventId": 9, 
                "scheduledEventId": 8
            }, 
            "eventTimestamp": 1475654723.241
        }, 
        {
            "eventId": 11, 
            "eventType": "WorkflowExecutionCompleted", 
            "workflowExecutionCompletedEventAttributes": {
                "result": "success", 
                "decisionTaskCompletedEventId": 10
            }, 
            "eventTimestamp": 1475654723.241
        }
    ]
}