Robust error handling on CloudFormationProvisioner using event polling

Author: wpc

lib/eb_deployer.rb

@@ -21,6 +21,7 @@
 require 'eb_deployer/default_component'
 require 'eb_deployer/component'
 require 'eb_deployer/eb_event_source'
+require 'eb_deployer/cf_event_source'
 require 'eb_deployer/event_poller'
 require 'eb_deployer/package'
 require 'eb_deployer/config_loader'

lib/eb_deployer/aws_driver/cloud_formation_driver.rb

@@ -20,28 +20,21 @@ def create_stack(name, template, opts)
       end
 
       def update_stack(name, template, opts)
-        begin
-          @client.update_stack(opts.merge(:stack_name => name,
-                                          :template_body => template,
-                                          :parameters => convert_parameters(opts[:parameters])))
-        rescue Aws::CloudFormation::Errors::ValidationError => e
-          if e.message =~ /No updates are to be performed/
-            log(e.message)
-          else
-            raise
-          end
-        end
-      end
-
-      def stack_status(name)
-        describe_stack(name)[:stack_status].downcase.to_sym
+        @client.update_stack(opts.merge(:stack_name => name,
+                                        :template_body => template,
+                                        :parameters => convert_parameters(opts[:parameters])))
       end
 
       def query_output(name, key)
         output = describe_stack(name)[:outputs].find { |o| o[:output_key] == key }
         output && output[:output_value]
       end
 
+      def fetch_events(name, options={})
+        response = @client.describe_stack_events(options.merge(:stack_name => name))
+        return response.stack_events, response.next_token
+      end
+
       private
 
       def describe_stack(name)

lib/eb_deployer/cf_event_source.rb

@@ -0,0 +1,26 @@
+module EbDeployer
+  class CfEventSource
+    def initialize(stack_name, cf_driver)
+      @stack_name = stack_name
+      @cf_driver = cf_driver
+    end
+
+    def get_anchor
+      events, _ = @cf_driver.fetch_events(@stack_name)
+      events.first
+    end
+
+    def fetch_events(from_anchor, &block)
+      events, next_token = @cf_driver.fetch_events(@stack_name)
+      should_continue = yield(events)
+      fetch_next(next_token, &block) if next_token && should_continue
+    end
+
+    private
+    def fetch_next(next_token, &block)
+      events, next_token = @cf_driver.fetch_events(@stack_name, :next_token => next_token)
+      should_continue = yield(events)
+      fetch_next(next_token, &block) if next_token && should_continue
+    end
+  end
+end

lib/eb_deployer/cloud_formation_provisioner.rb

@@ -3,21 +3,38 @@ class ResourceNotInReadyState < StandardError
   end
 
   class CloudFormationProvisioner
-    SUCCESS_STATS = [:create_complete, :update_complete, :update_rollback_complete]
-    FAILED_STATS = [:create_failed, :update_failed]
+    SUCCESS_STATS = ["CREATE_COMPLETE", "UPDATE_COMPLETE"]
+    FAILED_STATS = ["CREATE_FAILED", "UPDATE_FAILED"]
 
     def initialize(stack_name, cf_driver)
       @stack_name = stack_name
       @cf_driver = cf_driver
+      @poller = EventPoller.new(CfEventSource.new(@stack_name, @cf_driver))
     end
 
     def provision(resources)
       resources = symbolize_keys(resources)
       template = File.read(resources[:template])
       capabilities = resources[:capabilities] || []
       params = resources[:inputs] || resources[:parameters] || {}
-      stack_exists? ? update_stack(template, params, capabilities) : create_stack(template, params, capabilities)
-      wait_for_stack_op_terminate
+      anchor = nil
+      begin
+        if stack_exists?
+          anchor = @poller.get_anchor
+          update_stack(template, params, capabilities)
+        else
+          create_stack(template, params, capabilities)
+        end
+      rescue Aws::CloudFormation::Errors::ValidationError => e
+        if e.message =~ /No updates are to be performed/
+          log(e.message)
+          return
+        else
+          raise
+        end
+      end
+      wait_for_stack_op_terminate(anchor)
+      log("Resource stack provisioned successfully")
     end
 
     def transform_outputs(resources)
@@ -65,10 +82,6 @@ def create_stack(template, params, capabilities)
                               :parameters => params)
     end
 
-    def stack_status
-      @cf_driver.stack_status(@stack_name)
-    end
-
     def transform_output_to_settings(transforms)
       (transforms || []).inject([]) do |settings, pair|
         key, transform = pair
@@ -77,16 +90,22 @@ def transform_output_to_settings(transforms)
       end.flatten
     end
 
-    def wait_for_stack_op_terminate
-      stats = stack_status
-      while !SUCCESS_STATS.include?(stats)
-        sleep 15
-        stats = stack_status
-        raise "Resource stack update failed!" if FAILED_STATS.include?(stats)
-        log "current status: #{stack_status}"
+    def wait_for_stack_op_terminate(anchor)
+      @poller.poll(anchor) do |event|
+        log_event(event)
+        if FAILED_STATS.include?(event.resource_status)
+          raise "Resource stack update failed!"
+        end
+
+        break if event.logical_resource_id == @stack_name && SUCCESS_STATS.include?(event.resource_status)
       end
     end
 
+    def log_event(event)
+      puts "[#{event.timestamp}][cloud_formation_provisioner] #{event.resource_type}(#{event.logical_resource_id}) #{event.resource_status} \"#{event.resource_status_reason}\""
+    end
+
+
     def log(msg)
       puts "[#{Time.now.utc}][cloud_formation_provisioner] #{msg}"
     end

lib/eb_deployer/event_poller.rb

@@ -1,6 +1,7 @@
 module EbDeployer
   class EventPoller
     include Utils
+    POLL_INTERVAL = 15
 
     def initialize(event_source)
       @event_source = event_source
@@ -14,7 +15,7 @@ def poll(from_anchor, &block)
       handled = Set.new
       loop do
         @event_source.fetch_events(from_anchor) do |events|
-          # events from api is latest first order
+          # events from event source is latest first order
           to_be_handled = []
           reached_anchor = false
 
@@ -35,7 +36,7 @@ def poll(from_anchor, &block)
 
           !reached_anchor
         end
-        sleep 15
+        sleep POLL_INTERVAL
       end
     end
 

test/aws_driver_stubs.rb

@@ -271,6 +271,8 @@ def upload_file(bucket_name, obj_name, file)
 class CFStub
   def initialize
     @stacks = {}
+    @events = {}
+    @event_fetch_counter = 0;
   end
 
   def create_stack(name, template, opts)
@@ -297,4 +299,36 @@ def query_output(name, key)
   def stack_config(name)
     @stacks[name][:opts]
   end
+
+  def set_events(name, *messages)
+    events_seq = []
+    messages.each do |messages_for_call_seq|
+      if old_events = events_seq.last
+        events_seq << generate_event_from_messages(name, messages_for_call_seq) + old_events
+      else
+        events_seq << generate_event_from_messages(name, messages_for_call_seq)
+      end
+    end
+    @events[name] = events_seq
+  end
+
+  def fetch_events(name, opts={})
+    @event_fetch_counter += 1
+    if es = @events[name]
+      return es[@event_fetch_counter - 1], nil
+    else
+      return generate_event_from_messages(name, ["UPDATE_COMPLETE"]), nil
+    end
+  end
+
+  private
+
+  def generate_event_from_messages(stack, messages)
+    messages.map do |message|
+      event = OpenStruct.new(timestamp: Time.now,
+                             resource_type: 'AWS::CloudFormation::Stack',
+                             logical_resource_id: stack,
+                             resource_status: message)
+    end.reverse
+  end
 end

test/cf_event_poller_test.rb

@@ -0,0 +1,32 @@
+require 'test_helper'
+
+class CfEventPollerTest < Test::Unit::TestCase
+  def setup
+    @cf = CFStub.new
+    @poller = EbDeployer::EventPoller.new(EbDeployer::CfEventSource.new("mystack", @cf))
+  end
+
+  def test_run_handle_block_through_all_events_when_there_is_no_from_anchor
+    messages_handled = []
+    @cf.set_events('mystack', ['a', 'b', nil])
+    @poller.poll(nil) do |event|
+      break if event.resource_status.nil?
+      messages_handled << event.resource_status
+    end
+
+    assert_equal ['a', 'b'], messages_handled
+  end
+
+
+  def test_can_poll_all_events_after_an_anchor
+    @cf.set_events('mystack', ['a', 'b'], ['c', 'd', nil])
+    anchor = @poller.get_anchor
+    messages_handled = []
+    @poller.poll(anchor) do |event|
+      break if event.resource_status.nil?
+      messages_handled << event.resource_status
+    end
+
+    assert_equal ['c', 'd'], messages_handled
+  end
+end

test/event_poller_test.rb renamed to test/eb_event_poller_test.rb

@@ -1,6 +1,6 @@
 require 'test_helper'
 
-class EventPollerTest < Test::Unit::TestCase
+class EbEventPollerTest < Test::Unit::TestCase
   def setup
     @eb = EBStub.new
     @poller = EbDeployer::EventPoller.new(EbDeployer::EbEventSource.new("myapp", "test", @eb))

test/test_helper.rb

@@ -13,6 +13,7 @@ def silence_warnings(&block)
 end
 
 silence_warnings { EbDeployer::Utils::BACKOFF_INITIAL_SLEEP = 0 }
+silence_warnings { EbDeployer::EventPoller::POLL_INTERVAL = 0 }
 
 class ErrorRaisingWrapper < SimpleDelegator
   def initialize(stub)