Introduction to Simple Workflow Service (SWF)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
require 'aws-sdk-swf'
require_relative 'workflow_base.rb'

class RepairsWorkflow < WorkflowBase
  WORKFLOW_NAME = 'repairs-workflow'

  def initialize
    super
    register_workflow
    poll
  end

  def register_workflow
    begin
      @swf.register_workflow_type({
        domain: DOMAIN_NAME,
        name: WORKFLOW_NAME,
        version: VERSION,
        # Maximum time the decider can take to make a decision about this task
        default_task_start_to_close_timeout: '15'
      })
      puts "Workflow #{WORKFLOW_NAME} registered"
    rescue Aws::SWF::Errors::TypeAlreadyExistsFault
      puts "Workflow #{WORKFLOW_NAME} already exists"
    end
  end

  def poll
    while true
      options = {
        domain: DOMAIN_NAME,
        task_list: {
          name: TASK_LIST_NAME
        }
      }
      task = @swf.poll_for_decision_task(options)

      if task.task_token == nil
        puts "Polling expired for decision task. Trying again"
        next
      end

      process_events(task.workflow_execution.workflow_id, task.task_token, task.events)
    end
  end

  def process_events(workflow_id, token, events)
    # events contains all the event history for this workflow execution.
    # The lowest indexes [0] contains the oldest events. The first event is
    # always WorkflowExecutionStarted.
    # `events` can span multiple pages, in which case we would need to make
    # extra requests to get the other pages. We are not taking care of that
    # scenario in this example
    pending = 0
    in_progress = 0
    completed = 0
    results = {}
    events.each do | event |
      case event.event_type
        # This means at some point an activity was scheduled. It will either fail
        # or be grabbed by a worker
        when 'ActivityTaskScheduled'
          pending += 1
        # Activity couldn't be scheduled. This means something is wrong in the
        # workflow
        when 'ScheduleActivityTaskFailed'
          puts "There was a failure scheduling an activity"
        # A worker grabbed the task
        when 'ActivityTaskStarted'
          pending -= 1
          in_progress += 1
        # A task finished unsuccessfully
        when 'ActivityTaskFailed', 'ActivityTaskTimedOut'
          in_progress -= 1
        # A task completed successfully
        when 'ActivityTaskCompleted'
          if event.activity_task_completed_event_attributes
            parts = event.activity_task_completed_event_attributes.result.split(':')
            results[parts[0]] = parts[1]
          end
          completed += 1
          in_progress -= 1
      end
    end

    decide_task(workflow_id, token, completed, pending, in_progress, results)
  end

  # - If there are no `pending`, `in_progress` or `completed` activities, we
  #   need to schedule 'probe_machines'
  # - If 1 task has completed and the result includes a machine, we need to
  #   schedule 'drain_machine'. If there is no machine in the result, we can
  #   finish the workflow
  # - If 2 tasks have completed we need to schedule both 'fix_machine' and
  #   'reimage_machine'
  # - If 3 tasks have completed, it means that one of our parallel tasks has
  #   completed, but not the other. If it's still in progress, or pending we
  #   don't need to do anything. If that's not the case, we need to reschedule
  #   the missing task
  # - If 4 tasks have completed we need to schedule 'enable_machine'
  # - If 5 tasks have completed we need to can finish the workflow
  def decide_task(workflow_id, token, completed, pending, in_progress, results)
    puts "Completed: #{completed} Pending: #{pending} In progress: #{in_progress}"
    if completed == 0 && pending == 0 && in_progress == 0
      schedule_tasks(token, ['probe_machines'])
      return
    end

    if completed == 1 && pending == 0 && in_progress == 0
      if !results['probe_machines']
        # If there are no results, we can terminate the workflow
        end_execution(workflow_id)
      else
        schedule_tasks(token, ['drain_machine'], results['probe_machines'])
      end
      return
    end

    if completed == 2 && pending == 0 && in_progress == 0
      schedule_tasks(token, ['reimage_machine', 'fix_machine'], results['probe_machines'])
      return
    end

    if completed == 3 && pending == 0 && in_progress == 0
      # Figure out which one completed and schedule the other one
      if results.has_key?('fix_machine')
        schedule_tasks(token, ['reimage_machine'], results['probe_machines'])
      else
        schedule_tasks(token, ['fix_machine'], results['probe_machines'])
      end
      return
    end

    if completed == 4 && pending == 0 && in_progress == 0
      schedule_tasks(token, ['enable_machine'], results['probe_machines'])
      return
    end

    if completed == 5
      end_execution(workflow_id)
    end
  end

  def schedule_tasks(token, activities, input = nil)
    puts "Scheduling #{activities.join(',')}"
    decisions = []
    activities.each do |activity|
      decisions.push({
        decision_type: 'ScheduleActivityTask',
        schedule_activity_task_decision_attributes: {
          activity_type: {
            name: activity,
            version: VERSION
          },
          task_list: {
            name: TASK_LIST_NAME
          },
          input: input,
          activity_id: activity,
          # Time the workflow will wait for this task to be assigned to a worker
          schedule_to_start_timeout: '600',
          # Time the workflow will wait for this task to complete
          schedule_to_close_timeout: '3600',
          # Used for heartbeats. This example doesn't use them, so just setting
          # to the same value as schedule_to_close_timeout
          heartbeat_timeout: '3600'
        }
      })
    end
    @swf.respond_decision_task_completed({
      task_token: token,
      decisions: decisions
    })
  end

  def end_execution(id)
    puts "Terminating workflow"
    @swf.terminate_workflow_execution({
      domain: DOMAIN_NAME,
      workflow_id: id
    })
  end
end

RepairsWorkflow.new