Extract Action Collection from Data Collector

See the PR for details on this change.

Signed-off-by: Lamont Granquist <lamont@scriptkiddie.org>

27 files changed, 1128 insertions, 1039 deletions

diff --git a/lib/chef/action_collection.rb b/lib/chef/action_collection.rb
new file mode 100644
index 0000000000..a74ae5c449
--- /dev/null
+++ b/lib/chef/action_collection.rb
@@ -0,0 +1,252 @@
+#
+# Copyright:: Copyright 2018-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+require "chef/event_dispatch/base"
+
+class Chef
+  class ActionCollection < EventDispatch::Base
+    include Enumerable
+    extend Forwardable
+
+    class ActionRecord
+
+      # XXX: this is buggy since we (ab)use this resource for "after" state and it may be
+      # inaccurate and it may be mutated by the user.  A third after_resource should be added
+      # to new_resource + current_resource to properly implement this.
+      #
+      # @return [Chef::Resource] The declared resource state.
+      #
+      attr_accessor :new_resource
+
+      # @return [Chef::Resource] The current_resource object (before-state).  This can be nil
+      # for non-why-run-safe resources in why-run mode, or if load_current_resource itself
+      # threw an exception (which should be considered a bug in that load_current_resource
+      # implementation, but must be handled), or for unprocessed resources.
+      attr_accessor :current_resource
+
+      # @return [Symbol] # The action that was run (or scheduled to run in the case of "unprocessed" resources).
+      attr_accessor :action
+
+      # @return [Exception] The exception that was thrown
+      attr_accessor :exception
+
+      # @return [Numeric] The elapsed time in seconds with machine precision
+      attr_accessor :elapsed_time
+
+      # @return [Chef::Resource::Conditional] The conditional that caused the resource to be skipped
+      attr_accessor :conditional
+
+      # The status of the resource:
+      #   - updated:     ran and converged
+      #   - up_to_date:  skipped due to idempotency
+      #   - skipped:     skipped due to a conditional
+      #   - failed:      failed with an exception
+      #   - unprocessed: resources that were not touched by a run that failed
+      #
+      # @return [Symbol] status
+      #
+      attr_accessor :status
+
+      # The "nesting" level.  Outer resources in recipe context are 0 here, while for every
+      # sub-resource_collection inside of a custom resource this number is incremented by 1.
+      # Resources that are fired via build-resource or manually creating and firing
+      #
+      # @return [Integer]
+      #
+      attr_accessor :nesting_level
+
+      def initialize(new_resource, action, nesting_level)
+        @new_resource = new_resource
+        @action = action
+        @nesting_level = nesting_level
+      end
+
+      # @return [Boolean] true if there was no exception
+      def success?
+        !exception
+      end
+    end
+
+    attr_reader :action_records
+    attr_reader :pending_updates
+    attr_reader :run_context
+    attr_reader :consumers
+    attr_reader :events
+
+    def initialize(events, run_context = nil, action_records = [])
+      @action_records  = action_records
+      @pending_updates = []
+      @consumers       = []
+      @events          = events
+      @run_context     = run_context
+    end
+
+    def_delegators :@action_records, :each, :last
+
+    # Allows getting at the action_records collection filtered by nesting level and status.
+    #
+    # TODO: filtering by resource type+name
+    #
+    # @return [Chef::ActionCollection]
+    #
+    def filtered_collection(max_nesting: nil, up_to_date: true, skipped: true, updated: true, failed: true, unprocessed: true)
+      subrecords = action_records.select do |rec|
+        ( max_nesting.nil? || rec.nesting_level <= max_nesting ) &&
+          ( rec.status == :up_to_date && up_to_date ||
+            rec.status == :skipped && skipped ||
+            rec.status == :updated && updated ||
+            rec.status == :failed && failed ||
+            rec.status == :unprocessed && unprocessed )
+      end
+      self.class.new(events, run_context, subrecords)
+    end
+
+    # This hook gives us the run_context immediately after it is created so that we can wire up this object to it.
+    #
+    # This also causes the action_collection_registration event to fire, all consumers that have not yet registered with the
+    # action_collection must register via this callback.  This is the latest point before resources actually start to get
+    # evaluated.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def cookbook_compilation_start(run_context)
+      run_context.action_collection = self
+      # fire the action_colleciton_registration hook after cookbook_compilation_start -- last chance for consumers to register
+      run_context.events.enqueue(:action_collection_registration, self)
+      @run_context = run_context
+    end
+
+    # Consumers must call register -- either directly or through the action_collection_registration hook.  If
+    # nobody has registered any interest, then no action tracking will be done.
+    #
+    # @params object [Object] callers should call with `self`
+    #
+    def register(object)
+      consumers << object
+    end
+
+    # End of an unsuccessful converge used to fire off detect_unprocessed_resources.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def converge_failed(exception)
+      return if consumers.empty?
+      detect_unprocessed_resources
+    end
+
+    # Hook to start processing a resource.  May be called within processing of an outer resource
+    # so the pending_updates array forms a stack that sub-resources are popped onto and off of.
+    # This is always called.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_action_start(new_resource, action, notification_type = nil, notifier = nil)
+      return if consumers.empty?
+      pending_updates << ActionRecord.new(new_resource, action, pending_updates.length)
+    end
+
+    # Hook called after a resource is loaded.  If load_current_resource fails, this hook will
+    # not be called and current_resource will be nil, and the resource_failed hook will be called.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_current_state_loaded(new_resource, action, current_resource)
+      return if consumers.empty?
+      current_record.current_resource = current_resource
+    end
+
+    # Hook called after an action is determined to be up to date.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_up_to_date(new_resource, action)
+      return if consumers.empty?
+      current_record.status = :up_to_date
+    end
+
+    # Hook called after an action is determined to be skipped due to a conditional.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_skipped(resource, action, conditional)
+      return if consumers.empty?
+      current_record.status = :skipped
+      current_record.conditional = conditional
+    end
+
+    # Hook called after an action modifies the system and is marked updated.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_updated(new_resource, action)
+      return if consumers.empty?
+      current_record.status = :updated
+    end
+
+    # Hook called after an action fails.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_failed(new_resource, action, exception)
+      return if consumers.empty?
+      current_record.status = :failed
+      current_record.exception = exception
+    end
+
+    # Hook called after an action is completed.  This is always called, even if the action fails.
+    #
+    # (see EventDispatch::Base#)
+    #
+    def resource_completed(new_resource)
+      return if consumers.empty?
+      current_record.elapsed_time = new_resource.elapsed_time
+
+      # Verify if the resource has sensitive data and create a new blank resource with only
+      # the name so we can report it back without sensitive data
+      # XXX?: what about sensitive data in the current_resource?
+      # FIXME: this needs to be display-logic
+      if current_record.new_resource.sensitive
+        klass = current_record.new_resource.class
+        resource_name = current_record.new_resource.name
+        current_record.new_resource = klass.new(resource_name)
+      end
+
+      action_records << pending_updates.pop
+    end
+
+    private
+
+    # @return [Chef::ActionCollection::ActionRecord] the current record we are working on at the top of the stack
+    def current_record
+      pending_updates[-1]
+    end
+
+    # If the chef-client run fails in the middle, we are left with a half-completed resource_collection, this
+    # method is responsible for adding all of the resources which have not yet been touched.  They are marked
+    # as being "unprocessed".
+    #
+    def detect_unprocessed_resources
+      run_context.resource_collection.all_resources.select { |resource| resource.executed_by_runner == false }.each do |resource|
+        Array(resource.action).each do |action|
+          record = ActionRecord.new(resource, action, 0)
+          record.status = :unprocessed
+          action_records << record
+        end
+      end
+    end
+  end
+end
diff --git a/lib/chef/client.rb b/lib/chef/client.rb
index 3347536c12..0dd6b2666f 100644
--- a/lib/chef/client.rb
+++ b/lib/chef/client.rb
@@ -3,7 +3,7 @@
 # Author:: Christopher Walters (<cw@chef.io>)
 # Author:: Christopher Brown (<cb@chef.io>)
 # Author:: Tim Hinderliter (<tim@chef.io>)
-# Copyright:: Copyright 2008-2018, Chef Software Inc.
+# Copyright:: Copyright 2008-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -45,6 +45,7 @@ require "chef/formatters/base"
 require "chef/formatters/doc"
 require "chef/formatters/minimal"
 require "chef/version"
+require "chef/action_collection"
 require "chef/resource_reporter"
 require "chef/data_collector"
 require "chef/audit/audit_reporter"
@@ -249,9 +250,14 @@ class Chef
       begin
         runlock.save_pid
 
-        request_id = Chef::RequestID.instance.request_id
+        events.register(Chef::DataCollector::Reporter.new(events))
+        events.register(Chef::ActionCollection.new(events))
+
+        run_status.run_id = request_id = Chef::RequestID.instance.request_id
+
         run_context = nil
-        events.run_start(Chef::VERSION)
+        events.run_start(Chef::VERSION, run_status)
+
         logger.info("*** Chef #{Chef::VERSION} ***")
         logger.info("Platform: #{RUBY_PLATFORM}")
         logger.info "Chef-client pid: #{Process.pid}"
@@ -259,16 +265,12 @@ class Chef
         enforce_path_sanity
         run_ohai
 
-        generate_guid
-
         register unless Chef::Config[:solo_legacy_mode]
-        register_data_collector_reporter
 
         load_node
 
         build_node
 
-        run_status.run_id = request_id
         run_status.start_clock
         logger.info("Starting Chef Run for #{node.name}")
         run_started
@@ -998,33 +1000,6 @@ class Chef
 
       Chef::ReservedNames::Win32::Security.has_admin_privileges?
     end
-
-    # Ensure that we have a GUID for this node
-    # If we've got the proper configuration, we'll simply set that.
-    # If we're registed with the data collector, we'll migrate that UUID into our configuration and use that
-    # Otherwise, we'll create a new GUID and save it
-    def generate_guid
-      Chef::Config[:chef_guid] ||=
-        if File.exists?(Chef::Config[:chef_guid_path])
-          File.read(Chef::Config[:chef_guid_path])
-        else
-          uuid = UUIDFetcher.node_uuid
-          File.open(Chef::Config[:chef_guid_path], "w+") do |fh|
-            fh.write(uuid)
-          end
-          uuid
-        end
-    end
-
-    class UUIDFetcher
-      extend Chef::DataCollector::Messages::Helpers
-    end
-
-    # Register the data collector reporter to send event information to the
-    # data collector server
-    def register_data_collector_reporter
-      events.register(Chef::DataCollector::Reporter.new) if Chef::DataCollector.register_reporter?
-    end
   end
 end
 
diff --git a/lib/chef/data_collector.rb b/lib/chef/data_collector.rb
index a0b4eeb4cb..f445dc2ed9 100644
--- a/lib/chef/data_collector.rb
+++ b/lib/chef/data_collector.rb
@@ -2,7 +2,7 @@
 # Author:: Adam Leff (<adamleff@chef.io>)
 # Author:: Ryan Cragun (<ryan@chef.io>)
 #
-# Copyright:: Copyright 2012-2018, Chef Software Inc.
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,321 +18,173 @@
 # limitations under the License.
 #
 
-require "uri"
 require "chef/server_api"
 require "chef/http/simple_json"
 require "chef/event_dispatch/base"
-require "chef/data_collector/messages"
-require "chef/data_collector/resource_report"
-require "ostruct"
 require "set"
+require "chef/data_collector/run_end_message"
+require "chef/data_collector/run_start_message"
+require "chef/data_collector/config_validation"
+require "chef/data_collector/error_handlers"
 
 class Chef
-
-  # == Chef::DataCollector
-  # Provides methods for determinine whether a reporter should be registered.
   class DataCollector
+    # The DataCollector is mode-agnostic reporting tool which can be used with
+    # server-based and solo-based clients.  It can report to a file, to an
+    # authenticated Chef Automate reporting endpoint, or to a user-supplied
+    # webhook.  It sends two messages:  one at the start of the run and one
+    # at the end of the run.  Most early failures in the actual Chef::Client itself
+    # are reported, but parsing of the client.rb must have succeeded and some code
+    # in Chef::Application could throw so early as to prevent reporting.  If
+    # exceptions are thrown both run-start and run-end messages are still sent in
+    # pairs.
+    #
+    class Reporter < EventDispatch::Base
+      include Chef::DataCollector::ErrorHandlers
 
-    # Whether or not to enable data collection:
-    # * always disabled for why run mode
-    # * disabled when the user sets `Chef::Config[:data_collector][:mode]` to a
-    #   value that excludes the mode (client or solo) that we are running as
-    # * disabled in solo mode if the user did not configure the auth token
-    # * disabled if `Chef::Config[:data_collector][:server_url]` is set to a
-    #   falsey value
-    def self.register_reporter?
-      if why_run?
-        Chef::Log.trace("data collector is disabled for why run mode")
-        return false
-      end
-      unless reporter_enabled_for_current_mode?
-        Chef::Log.trace("data collector is configured to only run in " \
-                        "#{Chef::Config[:data_collector][:mode].inspect} modes, disabling it")
-        return false
-      end
-      unless data_collector_url_configured? || data_collector_output_locations_configured?
-        Chef::Log.trace("Neither data collector URL or output locations have been configured, disabling data collector")
-        return false
-      end
-      if solo? && !token_auth_configured?
-        Chef::Log.trace("Data collector token must be configured to use Chef Automate data collector with Chef Solo")
-      end
-      if !solo? && token_auth_configured?
-        Chef::Log.warn("Data collector token authentication is not recommended for client-server mode" \
-                       "Please upgrade Chef Server to 12.11.0 and remove the token from your config file " \
-                       "to use key based authentication instead")
-      end
-      true
-    end
-
-    def self.data_collector_url_configured?
-      !!Chef::Config[:data_collector][:server_url]
-    end
-
-    def self.data_collector_output_locations_configured?
-      !!Chef::Config[:data_collector][:output_locations]
-    end
+      # @return [Chef::RunList::RunListExpansion] the expanded run list
+      attr_reader :expanded_run_list
 
-    def self.why_run?
-      !!Chef::Config[:why_run]
-    end
+      # @return [Chef::RunStatus] the run status
+      attr_reader :run_status
 
-    def self.token_auth_configured?
-      !!Chef::Config[:data_collector][:token]
-    end
+      # @return [Chef::Node] the chef node
+      attr_reader :node
 
-    def self.solo?
-      !!Chef::Config[:solo] || !!Chef::Config[:local_mode]
-    end
+      # @return [Set<Hash>] the acculumated list of deprecation warnings
+      attr_reader :deprecations
 
-    def self.reporter_enabled_for_current_mode?
-      if Chef::Config[:solo] || Chef::Config[:local_mode]
-        acceptable_modes = [:solo, :both]
-      else
-        acceptable_modes = [:client, :both]
-      end
+      # @return [Chef::ActionCollection] the action collection object
+      attr_reader :action_collection
 
-      acceptable_modes.include?(Chef::Config[:data_collector][:mode])
-    end
+      # @return [Chef::EventDispatch::Dispatcher] the event dispatcher
+      attr_reader :events
 
-    # == Chef::DataCollector::Reporter
-    # Provides an event handler that can be registered to report on Chef
-    # run data. Unlike the existing Chef::ResourceReporter event handler,
-    # the DataCollector handler is not tied to a Chef Server / Chef Reporting
-    # and exports its data through a webhook-like mechanism to a configured
-    # endpoint.
-    class Reporter < EventDispatch::Base
-      attr_reader :all_resource_reports, :status, :exception, :error_descriptions,
-                  :expanded_run_list, :run_context, :run_status, :http,
-                  :current_resource_report, :enabled, :deprecations
-
-      def initialize
-        validate_data_collector_server_url!
-        validate_data_collector_output_locations! if data_collector_output_locations
-        @all_resource_reports    = []
-        @current_resource_loaded = nil
-        @error_descriptions      = {}
+      # @param events [Chef::EventDispatch::Dispatcher] the event dispatcher
+      def initialize(events)
+        @events = events
         @expanded_run_list       = {}
         @deprecations            = Set.new
-        @enabled                 = true
-
-        @http = setup_http_client(data_collector_server_url)
-        if data_collector_output_locations
-          @http_output_locations = setup_http_output_locations if data_collector_output_locations[:urls]
-        end
-      end
-
-      # see EventDispatch::Base#run_started
-      # Upon receipt, we will send our run start message to the
-      # configured DataCollector endpoint. Depending on whether
-      # the user has configured raise_on_failure, if we cannot
-      # send the message, we will either disable the DataCollector
-      # Reporter for the duration of this run, or we'll raise an
-      # exception.
-      def run_started(current_run_status)
-        update_run_status(current_run_status)
-
-        message = Chef::DataCollector::Messages.run_start_message(current_run_status)
-        disable_reporter_on_error do
-          send_to_data_collector(message)
-        end
-        send_to_output_locations(message) if data_collector_output_locations
-      end
-
-      # see EventDispatch::Base#run_completed
-      # Upon receipt, we will send our run completion message to the
-      # configured DataCollector endpoint.
-      def run_completed(node)
-        send_run_completion(status: "success")
-      end
-
-      # see EventDispatch::Base#run_failed
-      def run_failed(exception)
-        send_run_completion(status: "failure")
       end
 
-      # see EventDispatch::Base#converge_start
-      # Upon receipt, we stash the run_context for use at the
-      # end of the run in order to determine what resource+action
-      # combinations have not yet fired so we can report on
-      # unprocessed resources.
-      def converge_start(run_context)
-        @run_context = run_context
-      end
-
-      # see EventDispatch::Base#converge_complete
-      # At the end of the converge, we add any unprocessed resources
-      # to our report list.
-      def converge_complete
-        detect_unprocessed_resources
-      end
-
-      # see EventDispatch::Base#converge_failed
-      # At the end of the converge, we add any unprocessed resources
-      # to our report list
-      def converge_failed(exception)
-        detect_unprocessed_resources
-      end
-
-      # see EventDispatch::Base#resource_current_state_loaded
-      # Create a new ResourceReport instance that we'll use to track
-      # the state of this resource during the run. Nested resources are
-      # ignored as they are assumed to be an inline resource of a custom
-      # resource, and we only care about tracking top-level resources.
-      def resource_current_state_loaded(new_resource, action, current_resource)
-        return if nested_resource?(new_resource)
-        initialize_resource_report_if_needed(new_resource, action, current_resource)
-      end
-
-      # see EventDispatch::Base#resource_up_to_date
-      # Mark our ResourceReport status accordingly
-      def resource_up_to_date(new_resource, action)
-        initialize_resource_report_if_needed(new_resource, action)
-        current_resource_report.up_to_date unless nested_resource?(new_resource)
-      end
-
-      # see EventDispatch::Base#resource_skipped
-      # If this is a top-level resource, we create a ResourceReport
-      # instance (because a skipped resource does not trigger the
-      # resource_current_state_loaded event), and flag it as skipped.
-      def resource_skipped(new_resource, action, conditional)
-        return if nested_resource?(new_resource)
-
-        initialize_resource_report_if_needed(new_resource, action)
-        current_resource_report.skipped(conditional)
-      end
-
-      # see EventDispatch::Base#resource_updated
-      # Flag the current ResourceReport instance as updated (as long as it's
-      # a top-level resource).
-      def resource_updated(new_resource, action)
-        initialize_resource_report_if_needed(new_resource, action)
-        current_resource_report.updated unless nested_resource?(new_resource)
-      end
-
-      # see EventDispatch::Base#resource_failed
-      # Flag the current ResourceReport as failed and supply the exception as
-      # long as it's a top-level resource, and update the run error text
-      # with the proper Formatter.
-      def resource_failed(new_resource, action, exception)
-        initialize_resource_report_if_needed(new_resource, action)
-        current_resource_report.failed(exception) unless nested_resource?(new_resource)
-        update_error_description(
-          Formatters::ErrorMapper.resource_failed(
-            new_resource,
-            action,
-            exception
-          ).for_json
-        )
+      # Hook to grab the run_status.  We also make the decision to run or not run here (our
+      # config has been parsed so we should know if we need to run, we unregister if we do
+      # not want to run).
+      #
+      # (see EventDispatch::Base#run_start)
+      #
+      def run_start(chef_version, run_status)
+        events.unregister(self) unless should_be_enabled?
+        @run_status = run_status
       end
 
-      # see EventDispatch::Base#resource_completed
-      # Mark the ResourceReport instance as finished (for timing details).
-      # This marks the end of this resource during this run.
-      def resource_completed(new_resource)
-        if current_resource_report && !nested_resource?(new_resource)
-          current_resource_report.finish
-          add_resource_report(current_resource_report)
-          clear_current_resource_report
-        end
+      # Hook to grab the node object after it has been successfully loaded
+      #
+      # (see EventDispatch::Base#node_load_success)
+      #
+      def node_load_success(node)
+        @node = node
       end
 
-      # see EventDispatch::Base#run_list_expanded
       # The expanded run list is stored for later use by the run_completed
       # event and message.
+      #
+      # (see EventDispatch::Base#run_list_expanded)
+      #
       def run_list_expanded(run_list_expansion)
         @expanded_run_list = run_list_expansion
       end
 
-      # see EventDispatch::Base#run_list_expand_failed
-      # The run error text is updated with the output of the appropriate
-      # formatter.
-      def run_list_expand_failed(node, exception)
-        update_error_description(
-          Formatters::ErrorMapper.run_list_expand_failed(
-            node,
-            exception
-          ).for_json
-        )
+      # Hook event to register with the action_collection if we are still enabled.
+      #
+      # This is also how we wire up to the action_collection since it passes itself as the argument.
+      #
+      # (see EventDispatch::Base#action_collection_registration)
+      #
+      def action_collection_registration(action_collection)
+        @action_collection = action_collection
+        action_collection.register(self)
       end
 
-      # see EventDispatch::Base#cookbook_resolution_failed
-      # The run error text is updated with the output of the appropriate
-      # formatter.
-      def cookbook_resolution_failed(expanded_run_list, exception)
-        update_error_description(
-          Formatters::ErrorMapper.cookbook_resolution_failed(
-            expanded_run_list,
-            exception
-          ).for_json
-        )
-      end
+      # - Creates and writes our NodeUUID back to the node object
+      # - Sanity checks the data collector
+      # - Sends the run start message
+      # - If the run_start message fails, this may disable the rest of data collection or fail hard
+      #
+      # (see EventDispatch::Base#run_started)
+      #
+      def run_started(run_status)
+        Chef::DataCollector::ConfigValidation.validate_server_url!
+        Chef::DataCollector::ConfigValidation.validate_output_locations!
 
-      # see EventDispatch::Base#cookbook_sync_failed
-      # The run error text is updated with the output of the appropriate
-      # formatter.
-      def cookbook_sync_failed(cookbooks, exception)
-        update_error_description(
-          Formatters::ErrorMapper.cookbook_sync_failed(
-            cookbooks,
-            exception
-          ).for_json
-        )
+        send_run_start
       end
 
-      # see EventDispatch::Base#deprecation
-      # Append a received deprecation to the list of deprecations
+      # Hook event to accumulating deprecation messages
+      #
+      # (see EventDispatch::Base#deprecation)
+      #
       def deprecation(message, location = caller(2..2)[0])
-        add_deprecation(message.message, message.url, location)
+        @deprecations << { message: message.message, url: message.url, location: message.location }
+      end
+
+      # Hook to send the run completion message with a status of success
+      #
+      # (see EventDispatch::Base#run_completed)
+      #
+      def run_completed(node)
+        send_run_completion("success")
+      end
+
+      # Hook to send the run completion message with a status of failed
+      #
+      # (see EventDispatch::Base#run_failed)
+      #
+      def run_failed(exception)
+        send_run_completion("failure")
       end
 
       private
 
-      # Selects the type of HTTP client to use based on whether we are using
-      # token-based or signed header authentication. Token authentication is
-      # intended to be used primarily for Chef Solo in which case no signing
-      # key will be available (in which case `Chef::ServerAPI.new()` would
-      # raise an exception.
+      # Construct a http client for either the main data collector or for the http output_locations.
+      #
+      # Note that based on the token setting either the main data collector and all the http output_locations
+      # are going to all require chef-server authentication or not.  There is no facility to mix-and-match on
+      # a per-url basis.
+      #
+      # @param url [String] the string url to connect to
+      # @returns [Chef::HTTP] the appropriate Chef::HTTP subclass instance to use
+      #
       def setup_http_client(url)
-        if data_collector_token.nil?
+        if Chef::Config[:data_collector][:token].nil?
           Chef::ServerAPI.new(url, validate_utf8: false)
         else
           Chef::HTTP::SimpleJSON.new(url, validate_utf8: false)
         end
       end
 
-      def setup_http_output_locations
-        Chef::Config[:data_collector][:output_locations][:urls].each_with_object({}) do |location_url, http_output_locations|
-          http_output_locations[location_url] = setup_http_client(location_url)
-        end
-      end
-
+      # Handle POST'ing data to the data collector.  Note that this is a totally separate concern
+      # from the array of URI's in the extra configured output_locations.
       #
-      # Yields to the passed-in block (which is expected to be some interaction
-      # with the DataCollector endpoint). If some communication failure occurs,
-      # either disable any future communications to the DataCollector endpoint, or
-      # raise an exception (if the user has set
-      # Chef::Config.data_collector.raise_on_failure to true.)
+      # On failure this will unregister the data collector (if there are no other configured output_locations)
+      # and optionally will either silently continue or fail hard depending on configuration.
       #
-      # @param block [Proc] A ruby block to run. Ignored if a command is given.
+      # @param message [Hash] message to send
       #
-      def disable_reporter_on_error
-        yield
+      def send_to_data_collector(message)
+        return unless Chef::Config[:data_collector][:server_url]
+        @http ||= setup_http_client(Chef::Config[:data_collector][:server_url])
+        @http.post(nil, message, headers)
       rescue Timeout::Error, Errno::EINVAL, Errno::ECONNRESET,
-             Errno::ECONNREFUSED, EOFError, Net::HTTPBadResponse,
-             Net::HTTPHeaderSyntaxError, Net::ProtocolError, OpenSSL::SSL::SSLError,
-             Errno::EHOSTDOWN => e
+        Errno::ECONNREFUSED, EOFError, Net::HTTPBadResponse,
+        Net::HTTPHeaderSyntaxError, Net::ProtocolError, OpenSSL::SSL::SSLError,
+        Errno::EHOSTDOWN => e
         # Do not disable data collector reporter if additional output_locations have been specified
-        disable_data_collector_reporter unless data_collector_output_locations
-        code = if e.respond_to?(:response) && e.response.code
-                 e.response.code.to_s
-               else
-                 "Exception Code Empty"
-               end
+        events.unregister(self) unless Chef::Config[:data_collector][:output_locations]
+
+        code = e&.response&.code&.to_s || "Exception Code Empty"
 
-        msg = "Error while reporting run start to Data Collector. " \
-              "URL: #{data_collector_server_url} " \
-              "Exception: #{code} -- #{e.message} "
+        msg = "Error while reporting run start to Data Collector. URL: #{Chef::Config[:data_collector][:server_url]} Exception: #{code} -- #{e.message} "
 
         if Chef::Config[:data_collector][:raise_on_failure]
           Chef::Log.error(msg)
@@ -344,222 +196,128 @@ class Chef
         end
       end
 
-      def send_to_data_collector(message)
-        return unless data_collector_accessible?
-        http.post(nil, message, headers) if data_collector_server_url
-      end
-
+      # Process sending the configured message to all the extra output locations.
+      #
+      # @param message [Hash] message to send
+      #
       def send_to_output_locations(message)
-        data_collector_output_locations.each do |type, location_list|
-          location_list.each do |l|
-            handle_output_location(type, l, message)
+        return unless Chef::Config[:data_collector][:output_locations]
+
+        Chef::Config[:data_collector][:output_locations].each do |type, locations|
+          locations.each do |location|
+            send_to_file_location(location, message) if type == :files
+            send_to_http_location(location, message) if type == :urls
           end
         end
       end
 
-      def handle_output_location(type, loc, message)
-        type == :urls ? send_to_http_location(loc, message) : send_to_file_location(loc, message)
-      end
-
+      # Sends a single message to a file, rendered as JSON.
+      #
+      # @param file_name [String] the file to write to
+      # @param message [Hash] the message to render as JSON
+      #
       def send_to_file_location(file_name, message)
-        open(file_name, "a") { |f| f.puts message }
+        File.open(file_name, "a") do |fh|
+          fh.puts Chef::JSONCompat.to_json(message)
+        end
       end
 
+      # Sends a single message to a http uri, rendered as JSON.  Maintains a cache of Chef::HTTP
+      # objects to use on subsequent requests.
+      #
+      # @param http_url [String] the configured http uri string endpoint to send to
+      # @param message [Hash] the message to render as JSON
+      #
       def send_to_http_location(http_url, message)
-        @http_output_locations[http_url].post(nil, message, headers) if @http_output_locations[http_url]
+        @http_output_locations_clients[http_url] ||= setup_http_client(http_url)
+        @http_output_locations_clients[http_url].post(nil, message, headers)
       rescue
-        Chef::Log.trace("Data collector failed to send to URL location #{http_url}. Please check your configured data_collector.output_locations")
+        # FIXME: we do all kinds of complexity to deal with errors in send_to_data_collector and we just don't care here, which feels like
+        # like poor behavior on several different levels, at least its a warn now... (I don't quite understand why it was written this way)
+        Chef::Log.warn("Data collector failed to send to URL location #{http_url}. Please check your configured data_collector.output_locations")
       end
 
+      # @return [Boolean] if we've sent a run_start message yet
+      def sent_run_start?
+        !!@sent_run_start
+      end
+
+      # Send the run start message to the configured server or output locations
       #
-      # Send any messages to the DataCollector endpoint that are necessary to
-      # indicate the run has completed. Currently, two messages are sent:
-      #
-      # - An "action" message with the node object indicating it's been updated
-      # - An "run_converge" (i.e. RunEnd) message with details about the run,
-      #   what resources were modified/up-to-date/skipped, etc.
+      def send_run_start
+        message = Chef::DataCollector::RunStartMessage.construct_message(self)
+        send_to_data_collector(message)
+        send_to_output_locations(message)
+        @sent_run_start = true
+      end
+
+      # Send the run completion message to the configured server or output locations
       #
-      # @param opts [Hash] Additional details about the run, such as its success/failure.
+      # @param status [String] Either "success" or "failed"
       #
-      def send_run_completion(opts)
-        # If run_status is nil we probably failed before the client triggered
-        # the run_started callback. In this case we'll skip updating because
-        # we have nothing to report.
-        return unless run_status
+      def send_run_completion(status)
+        # this is necessary to send a run_start message when we fail before the run_started chef event.
+        # we adhere to a contract that run_start + run_completion events happen in pairs.
+        send_run_start unless sent_run_start?
 
-        message = Chef::DataCollector::Messages.run_end_message(
-                   run_status: run_status,
-                   expanded_run_list: expanded_run_list,
-                   resources: all_resource_reports,
-                   status: opts[:status],
-                   error_descriptions: error_descriptions,
-                   deprecations: deprecations.to_a
-                  )
-        disable_reporter_on_error do
-          send_to_data_collector(message)
-        end
-        send_to_output_locations(message) if data_collector_output_locations
+        message = Chef::DataCollector::RunEndMessage.construct_message(self, status)
+        send_to_data_collector(message)
+        send_to_output_locations(message)
       end
 
+      # @return [Hash] HTTP headers for the data collector endpoint
       def headers
         headers = { "Content-Type" => "application/json" }
 
-        unless data_collector_token.nil?
-          headers["x-data-collector-token"] = data_collector_token
+        unless Chef::Config[:data_collector][:token].nil?
+          headers["x-data-collector-token"] = Chef::Config[:data_collector][:token]
           headers["x-data-collector-auth"]  = "version=1.0"
         end
 
         headers
       end
 
-      def data_collector_server_url
-        Chef::Config[:data_collector][:server_url]
-      end
-
-      def data_collector_output_locations
-        Chef::Config[:data_collector][:output_locations]
-      end
-
-      def data_collector_token
-        Chef::Config[:data_collector][:token]
-      end
-
-      def add_resource_report(resource_report)
-        @all_resource_reports << OpenStruct.new(
-          resource: resource_report.new_resource,
-          action: resource_report.action,
-          report_data: resource_report.to_h
-        )
-      end
-
-      def disable_data_collector_reporter
-        @enabled = false
-      end
-
-      def data_collector_accessible?
-        @enabled
-      end
-
-      def update_run_status(run_status)
-        @run_status = run_status
-      end
-
-      def update_error_description(discription_hash)
-        @error_descriptions = discription_hash
-      end
-
-      def add_deprecation(message, url, location)
-        @deprecations << { message: message, url: url, location: location }
-      end
-
-      def initialize_resource_report_if_needed(new_resource, action, current_resource = nil)
-        return unless current_resource_report.nil?
-        @current_resource_report = create_resource_report(new_resource, action, current_resource)
-      end
-
-      def create_resource_report(new_resource, action, current_resource = nil)
-        Chef::DataCollector::ResourceReport.new(
-          new_resource,
-          action,
-          current_resource
-        )
-      end
-
-      def clear_current_resource_report
-        @current_resource_report = nil
-      end
-
-      def detect_unprocessed_resources
-        # create a Hash (for performance reasons, rather than an Array) containing all
-        # resource+action combinations from the Resource Collection
-        #
-        # We use the object ID instead of the resource itself in the Hash key because
-        # we currently allow users to create a property called "hash" which creates
-        # a #hash instance method on the resource. Ruby expects that to be a Fixnum,
-        # so bad things happen when adding an object to an Array or a Hash if it's not.
-        collection_resources = {}
-        run_context.resource_collection.all_resources.each do |resource|
-          Array(resource.action).each do |action|
-            collection_resources[[resource.__id__, action]] = resource
-          end
-        end
-
-        # Delete from the Hash any resource+action combination we have
-        # already processed.
-        all_resource_reports.each do |report|
-          collection_resources.delete([report.resource.__id__, report.action])
-        end
-
-        # The items remaining in the Hash are unprocessed resource+actions,
-        # so we'll create new resource reports for them which default to
-        # a state of "unprocessed".
-        collection_resources.each do |key, resource|
-          # The Hash key is an array of the Resource's object ID and the action.
-          # We need to pluck out the action.
-          add_resource_report(create_resource_report(resource, key[1]))
-        end
-      end
-
-      # If we are getting messages about a resource while we are in the middle of
-      # another resource's update, we assume that the nested resource is just the
-      # implementation of a provider, and we want to hide it from the reporting
-      # output.
-      def nested_resource?(new_resource)
-        @current_resource_report && @current_resource_report.new_resource != new_resource
-      end
-
-      def validate_and_return_uri(uri)
-        URI(uri)
-      rescue URI::InvalidURIError
-        nil
-      end
-
-      def validate_and_create_file(file)
-        send_to_file_location(file, "")
-        true
-      # Rescue exceptions raised by the file path being non-existent or not writeable and re-raise them to the user
-      # with clearer explanatory text.
-      rescue Errno::ENOENT
-        raise Chef::Exceptions::ConfigurationError,
-              "Chef::Config[:data_collector][:output_locations][:files] contains the location #{file}, which is a non existent file path."
-      rescue Errno::EACCES
-        raise Chef::Exceptions::ConfigurationError,
-              "Chef::Config[:data_collector][:output_locations][:files] contains the location #{file}, which cannnot be written to by Chef."
-      end
-
-      def validate_data_collector_server_url!
-        unless !data_collector_server_url && data_collector_output_locations
-          uri = validate_and_return_uri(data_collector_server_url)
-          unless uri
-            raise Chef::Exceptions::ConfigurationError, "Chef::Config[:data_collector][:server_url] (#{data_collector_server_url}) is not a valid URI."
-          end
-
-          if uri.host.nil?
-            raise Chef::Exceptions::ConfigurationError,
-              "Chef::Config[:data_collector][:server_url] (#{data_collector_server_url}) is a URI with no host. Please supply a valid URL."
-          end
+      # Main logic controlling the data collector being enabled or disabled:
+      #
+      # * disabled in why-run mode
+      # * disabled when `Chef::Config[:data_collector][:mode]` excludes the solo-vs-client mode
+      # * disabled if there is no server_url or no output_locations to log to
+      # * enabled if there is a configured output_location even without a token
+      # * disabled in solo mode if the user did not configure the auth token
+      #
+      # @return [Boolean] true if the data collector should be enabled
+      #
+      def should_be_enabled?
+        running_mode = ( Chef::Config[:solo_legacy_mode] || Chef::Config[:local_mode] ) ? :solo : :client
+        want_mode = Chef::Config[:data_collector][:mode]
+
+        case
+        when Chef::Config[:why_run]
+          Chef::Log.trace("data collector is disabled for why run mode")
+          return false
+        when (want_mode != :both) && running_mode != want_mode
+          Chef::Log.trace("data collector is configured to only run in #{Chef::Config[:data_collector][:mode]} modes, disabling it")
+          return false
+        when !(Chef::Config[:data_collector][:server_url] || Chef::Config[:data_collector][:output_locations])
+          Chef::Log.trace("Neither data collector URL or output locations have been configured, disabling data collector")
+          return false
+        when running_mode == :client && Chef::Config[:data_collector][:token]
+          Chef::Log.warn("Data collector token authentication is not recommended for client-server mode. " \
+                         "Please upgrade Chef Server to 12.11.0 and remove the token from your config file " \
+                         "to use key based authentication instead")
+          return true
+        when Chef::Config[:data_collector][:output_locations] && Chef::Config[:data_collector][:output_locations][:files] && !Chef::Config[:data_collector][:output_locations][:files].empty?
+          # we can run fine to a file without a token, even in solo mode.
+          return true
+        when running_mode == :solo && !Chef::Config[:data_collector][:token]
+          # we are in solo mode and are not logging to a file, so must have a token
+          Chef::Log.trace("Data collector token must be configured to use Chef Automate data collector with Chef Solo")
+          return false
+        else
+          return true
         end
       end
 
-      def handle_type(type, loc)
-        type == :urls ? validate_and_return_uri(loc) : validate_and_create_file(loc)
-      end
-
-      def validate_data_collector_output_locations!
-        if data_collector_output_locations.empty?
-          raise Chef::Exceptions::ConfigurationError,
-                "Chef::Config[:data_collector][:output_locations] is empty. Please supply an hash of valid URLs and / or local file paths."
-        end
-
-        data_collector_output_locations.each do |type, locations|
-          locations.each do |l|
-            unless handle_type(type, l)
-              raise Chef::Exceptions::ConfigurationError,
-                      "Chef::Config[:data_collector][:output_locations] contains the location #{l} which is not valid."
-            end
-          end
-        end
-      end
     end
   end
 end
diff --git a/lib/chef/data_collector/config_validation.rb b/lib/chef/data_collector/config_validation.rb
new file mode 100644
index 0000000000..670b59dd80
--- /dev/null
+++ b/lib/chef/data_collector/config_validation.rb
@@ -0,0 +1,88 @@
+#
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+require "uri"
+
+class Chef
+  class DataCollector
+    module ConfigValidation
+      class << self
+        def validate_server_url!
+          # if we have a server_url set we ALWAYS validate it, and we MUST have an output_location set to skip server_url validation
+          # (having output_locations set and no server_url is valid, but both of them unset blows up in here)
+          return if !Chef::Config[:data_collector][:server_url] && Chef::Config[:data_collector][:output_locations]
+
+          begin
+            uri = URI(Chef::Config[:data_collector][:server_url])
+          rescue
+            raise Chef::Exceptions::ConfigurationError, "Chef::Config[:data_collector][:server_url] (#{Chef::Config[:data_collector][:server_url]}) is not a valid URI."
+          end
+
+          if uri.host.nil?
+            raise Chef::Exceptions::ConfigurationError,
+              "Chef::Config[:data_collector][:server_url] (#{Chef::Config[:data_collector][:server_url]}) is a URI with no host. Please supply a valid URL."
+          end
+        end
+
+        def validate_output_locations!
+          # not having an output_location set at all is fine, we just skip it then
+          output_locations = Chef::Config[:data_collector][:output_locations]
+          return unless output_locations
+
+          # but deliberately setting an empty output_location we consider to be an error (XXX: but should we?)
+          if output_locations.empty?
+            raise Chef::Exceptions::ConfigurationError,
+              "Chef::Config[:data_collector][:output_locations] is empty. Please supply an hash of valid URLs and / or local file paths."
+          end
+
+          # loop through all the types and locations and validate each one-by-one
+          output_locations.each do |type, locations|
+            locations.each do |location|
+              validate_url!(location) if type == :urls
+              validate_file!(location) if type == :files
+            end
+          end
+        end
+
+        private
+
+        # validate an output_location file
+        def validate_file!(file)
+          open(file, "a") {}
+        rescue Errno::ENOENT
+          raise Chef::Exceptions::ConfigurationError,
+            "Chef::Config[:data_collector][:output_locations][:files] contains the location #{file}, which is a non existent file path."
+        rescue Errno::EACCES
+          raise Chef::Exceptions::ConfigurationError,
+            "Chef::Config[:data_collector][:output_locations][:files] contains the location #{file}, which cannnot be written to by Chef."
+        rescue Exception => e
+          raise Chef::Exceptions::ConfigurationError,
+            "Chef::Config[:data_collector][:output_locations][:files] contains the location #{file}, which is invalid: #{e.message}."
+        end
+
+        # validate an output_location url
+        def validate_url!(url)
+          URI(url)
+        rescue
+          raise Chef::Exceptions::ConfigurationError,
+            "Chef::Config[:data_collector][:output_locations][:urls] contains the url #{url} which is not valid."
+        end
+
+      end
+    end
+  end
+end
diff --git a/lib/chef/data_collector/error_handlers.rb b/lib/chef/data_collector/error_handlers.rb
new file mode 100644
index 0000000000..bb892c457c
--- /dev/null
+++ b/lib/chef/data_collector/error_handlers.rb
@@ -0,0 +1,116 @@
+#
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+class Chef
+  class DataCollector
+
+    # This module isolates the handling of collecting error descriptions to insert into the data_colletor
+    # report output.  For very early errors it is repsonsible for collecting the node_name for the report
+    # to use.  For all failure conditions that have an ErrorMapper it collects the output.
+    #
+    # No external code should call anything in this module directly.
+    #
+    # @api private
+    #
+    module ErrorHandlers
+
+      # @return [String] the fallback node name if we do NOT have a node due to early failures
+      attr_reader :node_name
+
+      # @return [Hash] JSON-formatted error description from the Chef::Formatters::ErrorMapper
+      def error_description
+        @error_description ||= {}
+      end
+
+      # This is an exceptionally "early" failure that results in not having a valid Chef::Node object,
+      # so it must capture the node_name from the config.rb
+      #
+      # (see EventDispatch::Base#registration_failed)
+      #
+      def registration_failed(node_name, exception, config)
+        description = Formatters::ErrorMapper.registration_failed(node_name, exception, config)
+        @node_name = node_name
+        @error_description = description.for_json
+      end
+
+      # This is an exceptionally "early" failure that results in not having a valid Chef::Node object,
+      # so it must capture the node_name from the config.rb
+      #
+      # (see EventDispatch::Base#node_load_failed)
+      #
+      def node_load_failed(node_name, exception, config)
+        description = Formatters::ErrorMapper.node_load_failed(node_name, exception, config)
+        @node_name = node_name
+        @error_description = description.for_json
+      end
+
+      # This is an "early" failure during run_list expansion
+      #
+      # (see EventDispatch::Base#run_list_expand_failed)
+      #
+      def run_list_expand_failed(node, exception)
+        description = Formatters::ErrorMapper.run_list_expand_failed(node, exception)
+        @error_description = description.for_json
+      end
+
+      # This is an "early" failure during cookbook resolution / depsolving / talking to cookbook_version endpoint on a server
+      #
+      # (see EventDispatch::Base#cookbook_resolution_failed)
+      #
+      def cookbook_resolution_failed(expanded_run_list, exception)
+        description = Formatters::ErrorMapper.cookbook_resolution_failed(expanded_run_list, exception)
+        @error_description = description.for_json
+      end
+
+      # This is an "early" failure during cookbook synchronization
+      #
+      # (see EventDispatch::Base#cookbook_sync_failed)
+      #
+      def cookbook_sync_failed(cookbooks, exception)
+        description = Formatters::ErrorMapper.cookbook_sync_failed(cookbooks, exception)
+        @error_description = description.for_json
+      end
+
+      # This failure happens during library loading / attribute file parsing, etc.
+      #
+      # (see EventDispatch::Base#file_load_failed)
+      #
+      def file_load_failed(path, exception)
+        description = Formatters::ErrorMapper.file_load_failed(path, exception)
+        @error_description = description.for_json
+      end
+
+      # This failure happens at converge time during recipe parsing
+      #
+      # (see EventDispatch::Base#recipe_not_failed)
+      #
+      def recipe_not_found(exception)
+        description = Formatters::ErrorMapper.file_load_failed(nil, exception)
+        @error_description = description.for_json
+      end
+
+      # This is a normal resource failure event during compile/converge phases
+      #
+      # (see EventDispatch::Base#resource_failed)
+      #
+      def resource_failed(new_resource, action, exception)
+        description = Formatters::ErrorMapper.resource_failed(new_resource, action, exception)
+        @error_description = description.for_json
+      end
+    end
+  end
+end
diff --git a/lib/chef/data_collector/message_helpers.rb b/lib/chef/data_collector/message_helpers.rb
new file mode 100644
index 0000000000..75783d8fd0
--- /dev/null
+++ b/lib/chef/data_collector/message_helpers.rb
@@ -0,0 +1,50 @@
+#
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+class Chef
+  class DataCollector
+
+    # This is for shared code between the run_start_message and run_end_message modules.
+    #
+    # No external code should call this module directly
+    #
+    # @api private
+    #
+    module MessageHelpers
+      private
+
+      # The organization name the node is associated with. For Chef Solo runs the default
+      # is "chef_solo" which can be overridden by the user.
+      #
+      # @return [String] Chef organization associated with the node
+      #
+      def organization
+        if solo_run?
+          # configurable fake organization name for chef-solo users
+          Chef::Config[:data_collector][:organization]
+        else
+          Chef::Config[:chef_server_url].match(%r{/+organizations/+([^\s/]+)}).nil? ? "unknown_organization" : $1
+        end
+      end
+
+      # @return [Boolean] True if we're in a chef-solo/chef-zero or legacy chef-solo run
+      def solo_run?
+        Chef::Config[:solo_legacy_mode] || Chef::Config[:local_mode]
+      end
+    end
+  end
+end
diff --git a/lib/chef/data_collector/messages.rb b/lib/chef/data_collector/messages.rb
deleted file mode 100644
index c375475c72..0000000000
--- a/lib/chef/data_collector/messages.rb
+++ /dev/null
@@ -1,100 +0,0 @@
-#
-# Author:: Adam Leff (<adamleff@chef.io)
-# Author:: Ryan Cragun (<ryan@chef.io>)
-#
-# Copyright:: Copyright 2012-2016, Chef Software Inc.
-# License:: Apache License, Version 2.0
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-require "securerandom"
-require_relative "messages/helpers"
-
-class Chef
-  class DataCollector
-    module Messages
-      extend Helpers
-
-      #
-      # Message payload that is sent to the DataCollector server at the
-      # start of a Chef run.
-      #
-      # @param run_status [Chef::RunStatus] The RunStatus instance for this node/run.
-      #
-      # @return [Hash] A hash containing the run start message data.
-      #
-      def self.run_start_message(run_status)
-        {
-          "chef_server_fqdn" => chef_server_fqdn,
-          "entity_uuid" => node_uuid,
-          "id" => run_status.run_id,
-          "message_version" => "1.0.0",
-          "message_type" => "run_start",
-          "node_name" => run_status.node.name,
-          "organization_name" => organization,
-          "run_id" => run_status.run_id,
-          "source" => collector_source,
-          "start_time" => run_status.start_time.utc.iso8601,
-        }
-      end
-
-      #
-      # Message payload that is sent to the DataCollector server at the
-      # end of a Chef run.
-      #
-      # @param reporter_data [Hash] Data supplied by the Reporter, such as run_status, resource counts, etc.
-      #
-      # @return [Hash] A hash containing the run end message data.
-      #
-      def self.run_end_message(reporter_data)
-        run_status = reporter_data[:run_status]
-
-        message = {
-          "chef_server_fqdn" => chef_server_fqdn,
-          "entity_uuid" => node_uuid,
-          "expanded_run_list" => reporter_data[:expanded_run_list],
-          "id" => run_status.run_id,
-          "message_version" => "1.1.0",
-          "message_type" => "run_converge",
-          "node" => run_status.node,
-          "node_name" => run_status.node.name,
-          "organization_name" => organization,
-          "resources" => reporter_data[:resources].map(&:report_data),
-          "run_id" => run_status.run_id,
-          "run_list" => run_status.node.run_list.for_json,
-          "policy_name" => run_status.node.policy_name,
-          "policy_group" => run_status.node.policy_group,
-          "start_time" => run_status.start_time.utc.iso8601,
-          "end_time" => run_status.end_time.utc.iso8601,
-          "source" => collector_source,
-          "status" => reporter_data[:status],
-          "total_resource_count" => reporter_data[:resources].count,
-          "updated_resource_count" => reporter_data[:resources].select { |r| r.report_data["status"] == "updated" }.count,
-          "deprecations" => reporter_data[:deprecations],
-        }
-
-        if run_status.exception
-          message["error"] = {
-            "class" => run_status.exception.class,
-            "message" => run_status.exception.message,
-            "backtrace" => run_status.exception.backtrace,
-            "description" => reporter_data[:error_descriptions],
-          }
-        end
-
-        message
-      end
-    end
-  end
-end
diff --git a/lib/chef/data_collector/messages/helpers.rb b/lib/chef/data_collector/messages/helpers.rb
deleted file mode 100644
index e4eda5ebb2..0000000000
--- a/lib/chef/data_collector/messages/helpers.rb
+++ /dev/null
@@ -1,159 +0,0 @@
-#
-# Author:: Adam Leff (<adamleff@chef.io)
-# Author:: Ryan Cragun (<ryan@chef.io>)
-#
-# Copyright:: Copyright 2012-2016, Chef Software Inc.
-# License:: Apache License, Version 2.0
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-class Chef
-  class DataCollector
-    module Messages
-      module Helpers
-        #
-        # Fully-qualified domain name of the Chef Server configured in Chef::Config
-        # If the chef_server_url cannot be parsed as a URI, the node["fqdn"] attribute
-        # will be returned, or "localhost" if the run_status is unavailable to us.
-        #
-        # @return [String] FQDN of the configured Chef Server, or node/localhost if not found.
-        #
-        def chef_server_fqdn
-          if !Chef::Config[:chef_server_url].nil?
-            URI(Chef::Config[:chef_server_url]).host
-          elsif !Chef::Config[:node_name].nil?
-            Chef::Config[:node_name]
-          else
-            "localhost"
-          end
-        end
-
-        #
-        # The organization name the node is associated with. For Chef Solo runs, a
-        # user-configured organization string is returned, or the string "chef_solo"
-        # if such a string is not configured.
-        #
-        # @return [String] Organization to which the node is associated
-        #
-        def organization
-          solo_run? ? data_collector_organization : chef_server_organization
-        end
-
-        #
-        # Returns the user-configured organization, or "chef_solo" if none is configured.
-        #
-        # This is only used when Chef is run in Solo mode.
-        #
-        # @return [String] Data-collector-specific organization used when running in Chef Solo
-        #
-        def data_collector_organization
-          Chef::Config[:data_collector][:organization] || "chef_solo"
-        end
-
-        #
-        # Return the organization assumed by the configured chef_server_url.
-        #
-        # We must parse this from the Chef::Config[:chef_server_url] because a node
-        # has no knowledge of an organization or to which organization is belongs.
-        #
-        # If we cannot determine the organization, we return "unknown_organization"
-        #
-        # @return [String] shortname of the Chef Server organization
-        #
-        def chef_server_organization
-          return "unknown_organization" unless Chef::Config[:chef_server_url]
-
-          Chef::Config[:chef_server_url].match(%r{/+organizations/+([a-z0-9][a-z0-9_-]{0,254})}).nil? ? "unknown_organization" : $1
-        end
-
-        #
-        # The source of the data collecting during this run, used by the
-        # DataCollector endpoint to determine if Chef was in Solo mode or not.
-        #
-        # @return [String] "chef_solo" if in Solo mode, "chef_client" if in Client mode
-        #
-        def collector_source
-          solo_run? ? "chef_solo" : "chef_client"
-        end
-
-        #
-        # If we're running in Solo (legacy) mode, or in Solo (formerly
-        # "Chef Client Local Mode"), we're considered to be in a "solo run".
-        #
-        # @return [Boolean] Whether we're in a solo run or not
-        #
-        def solo_run?
-          Chef::Config[:solo] || Chef::Config[:local_mode]
-        end
-
-        #
-        # Returns a UUID that uniquely identifies this node for reporting reasons.
-        #
-        # The node is read in from disk if it exists, or it's generated if it does
-        # does not exist.
-        #
-        # @return [String] UUID for the node
-        #
-        def node_uuid
-          Chef::Config[:chef_guid] || read_node_uuid || generate_node_uuid
-        end
-
-        #
-        # Generates a UUID for the node via SecureRandom.uuid and writes out
-        # metadata file so the UUID persists between runs.
-        #
-        # @return [String] UUID for the node
-        #
-        def generate_node_uuid
-          uuid = SecureRandom.uuid
-          update_metadata("node_uuid", uuid)
-
-          uuid
-        end
-
-        #
-        # Reads in the node UUID from the node metadata file
-        #
-        # @return [String] UUID for the node
-        #
-        def read_node_uuid
-          metadata["node_uuid"]
-        end
-
-        #
-        # Returns the DataCollector metadata for this node
-        #
-        # If the metadata file does not exist in the file cache path,
-        # an empty hash will be returned.
-        #
-        # @return [Hash] DataCollector metadata for this node
-        #
-        def metadata
-          Chef::JSONCompat.parse(Chef::FileCache.load(metadata_filename))
-        rescue Chef::Exceptions::FileNotFound
-          {}
-        end
-
-        def update_metadata(key, value)
-          updated_metadata = metadata.tap { |x| x[key] = value }
-          Chef::FileCache.store(metadata_filename, Chef::JSONCompat.to_json(updated_metadata), 0644)
-        end
-
-        def metadata_filename
-          "data_collector_metadata.json"
-        end
-      end
-    end
-  end
-end
diff --git a/lib/chef/data_collector/resource_report.rb b/lib/chef/data_collector/resource_report.rb
deleted file mode 100644
index 9a99747f7f..0000000000
--- a/lib/chef/data_collector/resource_report.rb
+++ /dev/null
@@ -1,123 +0,0 @@
-#
-# Author:: Adam Leff (<adamleff@chef.io>)
-# Author:: Ryan Cragun (<ryan@chef.io>)
-#
-# Copyright:: Copyright 2012-2018, Chef Software Inc.
-# License:: Apache License, Version 2.0
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-require "chef/exceptions"
-
-class Chef
-  class DataCollector
-    class ResourceReport
-
-      attr_reader :action, :elapsed_time, :new_resource, :status
-      attr_accessor :conditional, :current_resource, :exception
-
-      def initialize(new_resource, action, current_resource = nil)
-        @new_resource     = new_resource
-        @action           = action
-        @current_resource = current_resource
-        @status           = "unprocessed"
-      end
-
-      def skipped(conditional)
-        @status      = "skipped"
-        @conditional = conditional
-      end
-
-      def updated
-        @status = "updated"
-      end
-
-      def failed(exception)
-        @current_resource = nil
-        @status           = "failed"
-        @exception        = exception
-      end
-
-      def up_to_date
-        @status = "up-to-date"
-      end
-
-      def finish
-        @elapsed_time = new_resource.elapsed_time
-      end
-
-      def elapsed_time_in_milliseconds
-        elapsed_time.nil? ? nil : (elapsed_time * 1000).to_i
-      end
-
-      def potentially_changed?
-        %w{updated failed}.include?(status)
-      end
-
-      def to_h
-        hash = {
-          "type" => new_resource.resource_name.to_sym,
-          "name" => new_resource.name.to_s,
-          "id" => resource_identity,
-          "after" => new_resource_state_reporter,
-          "before" => current_resource_state_reporter,
-          "duration" => elapsed_time_in_milliseconds.to_s,
-          "delta" => new_resource.respond_to?(:diff) && potentially_changed? ? new_resource.diff : "",
-          "ignore_failure" => new_resource.ignore_failure,
-          "result" => action.to_s,
-          "status" => status,
-        }
-
-        if new_resource.cookbook_name
-          hash["cookbook_name"]    = new_resource.cookbook_name
-          hash["cookbook_version"] = new_resource.cookbook_version.version
-          hash["recipe_name"]      = new_resource.recipe_name
-        end
-
-        hash["conditional"] = conditional.to_text if status == "skipped"
-        hash["error_message"] = exception.message unless exception.nil?
-
-        hash
-      end
-      alias_method :to_hash, :to_h
-      alias_method :for_json, :to_h
-
-      # We should be able to call the identity of a resource safely, but there
-      # is an edge case where resources that have a lazy property that is both
-      # the name_property and the identity property, it will thow a validation
-      # exception causing the chef-client run to fail. We are not fixing this
-      # case since Chef is actually doing the right thing but we are making the
-      # ResourceReporter smarter so that it detects the failure and sends a
-      # message to the data collector containing a static resource identity
-      # since we were unable to generate a proper one.
-      def resource_identity
-        new_resource.identity.to_s
-      rescue => e
-        "unknown identity (due to #{e.class})"
-      end
-
-      def new_resource_state_reporter
-        new_resource.state_for_resource_reporter
-      rescue
-        {}
-      end
-
-      def current_resource_state_reporter
-        current_resource ? current_resource.state_for_resource_reporter : {}
-      rescue
-        {}
-      end
-    end
-  end
-end
diff --git a/lib/chef/data_collector/run_end_message.rb b/lib/chef/data_collector/run_end_message.rb
new file mode 100644
index 0000000000..1aba651d96
--- /dev/null
+++ b/lib/chef/data_collector/run_end_message.rb
@@ -0,0 +1,172 @@
+#
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+require "chef/data_collector/message_helpers"
+
+class Chef
+  class DataCollector
+    module RunEndMessage
+      extend Chef::DataCollector::MessageHelpers
+
+      # This module encapsulates rendering the run_end_message given the state gathered in the data_collector
+      # and the action_collection.  It is deliberately a stateless module and is deliberately not mixed into
+      # the data_collector and only uses the public api methods of the data_collector and action_collection.
+      #
+      # No external code should call this module directly.
+      #
+      # @api private
+      class << self
+
+        # Construct the message payload that is sent to the DataCollector server at the
+        # end of a Chef run.
+        #
+        # @param data_collector [Chef::DataCollector::Reporter] the calling data_collector instance
+        # @param status [String] the overall status of the run, either "success" or "failure"
+        #
+        # @return [Hash] A hash containing the run end message data.
+        #
+        def construct_message(data_collector, status)
+          action_collection = data_collector.action_collection
+          run_status = data_collector.run_status
+          node = data_collector.node
+
+          message = {
+            "chef_server_fqdn" => URI(Chef::Config[:chef_server_url]).host,
+            "entity_uuid" => Chef::Config[:chef_guid],
+            "expanded_run_list" => data_collector.expanded_run_list,
+            "id" => run_status&.run_id,
+            "message_version" => "1.1.0",
+            "message_type" => "run_converge",
+            "node" => node || {},
+            "node_name" => node&.name || data_collector.node_name,
+            "organization_name" => organization,
+            "resources" => all_action_records(action_collection),
+            "run_id" => run_status&.run_id,
+            "run_list" => node&.run_list&.for_json || [],
+            "policy_name" => node&.policy_name,
+            "policy_group" => node&.policy_group,
+            "start_time" => run_status.start_time.utc.iso8601,
+            "end_time" => run_status.end_time.utc.iso8601,
+            "source" => solo_run? ? "chef_solo" : "chef_client",
+            "status" => status,
+            "total_resource_count" => all_action_records(action_collection).count,
+            "updated_resource_count" => updated_resource_count(action_collection),
+            "deprecations" => data_collector.deprecations.to_a,
+          }
+
+          if run_status&.exception
+            message["error"] = {
+              "class" => run_status.exception.class,
+              "message" => run_status.exception.message,
+              "backtrace" => run_status.exception.backtrace,
+              "description" => data_collector.error_description,
+            }
+          end
+
+          message
+        end
+
+        private
+
+        # @return [Integer] the number of resources successfully updated in the chef-client run
+        def updated_resource_count(action_collection)
+          return 0 if action_collection.nil?
+          action_collection.filtered_collection(up_to_date: false, skipped: false, unprocessed: false, failed: false).count
+        end
+
+        # @return [Array<Chef::ActionCollection::ActionRecord>] list of all action_records for all resources
+        def action_records(action_collection)
+          return [] if action_collection.nil?
+          action_collection.action_records
+        end
+
+        # @return [Array<Hash>] list of all action_records rendered as a Hash for sending to JSON
+        def all_action_records(action_collection)
+          action_records(action_collection).map { |rec| action_record_for_json(rec) }
+        end
+
+        # @return [Hash] the Hash representation of the action_record for sending as JSON
+        def action_record_for_json(action_record)
+          new_resource = action_record.new_resource
+          current_resource = action_record.current_resource
+
+          hash = {
+            "type" => new_resource.resource_name.to_sym,
+            "name" => new_resource.name.to_s,
+            "id" => safe_resource_identity(new_resource),
+            "after" => safe_state_for_resource_reporter(new_resource),
+            "before" => safe_state_for_resource_reporter(current_resource),
+            "duration" => action_record.elapsed_time.nil? ? "" : (action_record.elapsed_time * 1000).to_i.to_s,
+            "delta" => new_resource.respond_to?(:diff) && updated_or_failed?(action_record) ? new_resource.diff : "",
+            "ignore_failure" => new_resource.ignore_failure,
+            "result" => action_record.action.to_s,
+            "status" => action_record_status_for_json(action_record),
+          }
+
+          if new_resource.cookbook_name
+            hash["cookbook_name"]    = new_resource.cookbook_name
+            hash["cookbook_version"] = new_resource.cookbook_version.version
+            hash["recipe_name"]      = new_resource.recipe_name
+          end
+
+          hash["conditional"] = action_record.conditional.to_text if action_record.status == :skipped
+          hash["error_message"] = action_record.exception.message unless action_record.exception.nil?
+
+          hash
+        end
+
+        # If the identity property of a resource has been lazied (via a lazy name resource) evaluating it
+        # for an unprocessed resource (where the preconditions have not been met) may cause the lazy
+        # evaluator to throw -- and would otherwise crash the data collector.
+        #
+        # @return [String] the resource's identity property
+        #
+        def safe_resource_identity(new_resource)
+          new_resource.identity.to_s
+        rescue => e
+          "unknown identity (due to #{e.class})"
+        end
+
+        # FIXME: This is likely necessary due to the same lazy issue with properties and failing resources?
+        #
+        # @return [Hash] the resource's reported state properties
+        #
+        def safe_state_for_resource_reporter(resource)
+          resource ? resource.state_for_resource_reporter : {}
+        rescue
+          {}
+        end
+
+        # Helper to convert action record status (symbols) to strings for the Data Collector server.
+        # Does a bit of necessary underscores-to-dashes conversion to comply with the Data Collector API.
+        #
+        # @return [String] resource status (
+        #
+        def action_record_status_for_json(action_record)
+          action = action_record.status.to_s
+          action = "up-to-date" if action == "up_to_date"
+          action
+        end
+
+        # @return [Boolean] True if the resource was updated or failed
+        def updated_or_failed?(action_record)
+          action_record.status == :updated || action_record.status == :failed
+        end
+      end
+    end
+  end
+end
diff --git a/lib/chef/data_collector/run_start_message.rb b/lib/chef/data_collector/run_start_message.rb
new file mode 100644
index 0000000000..448043a48d
--- /dev/null
+++ b/lib/chef/data_collector/run_start_message.rb
@@ -0,0 +1,60 @@
+#
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
+# License:: Apache License, Version 2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+require "chef/data_collector/message_helpers"
+
+class Chef
+  class DataCollector
+    module RunStartMessage
+      extend Chef::DataCollector::MessageHelpers
+
+      # This module encapsulates rendering the run_start_message given the state gathered in the data_collector.
+      # It is deliberately a stateless module and is deliberately not mixed into the data_collector and only
+      # uses the public api methods of the data_collector.
+      #
+      # No external code should call this module directly.
+      #
+      # @api private
+      class << self
+
+        # Construct the message payload that is sent to the DataCollector server at the
+        # start of a Chef run.
+        #
+        # @param data_collector [Chef::DataCollector::Reporter] the calling data_collector instance
+        #
+        # @return [Hash] A hash containing the run start message data.
+        #
+        def construct_message(data_collector)
+          run_status = data_collector.run_status
+          node = data_collector.node
+          {
+            "chef_server_fqdn" => URI(Chef::Config[:chef_server_url]).host,
+            "entity_uuid" => Chef::Config[:chef_guid],
+            "id" => run_status&.run_id,
+            "message_version" => "1.0.0",
+            "message_type" => "run_start",
+            "node_name" => node&.name || data_collector.node_name,
+            "organization_name" => organization,
+            "run_id" => run_status&.run_id,
+            "source" => solo_run? ? "chef_solo" : "chef_client",
+            "start_time" => run_status.start_time.utc.iso8601,
+          }
+        end
+      end
+    end
+  end
+end
diff --git a/lib/chef/event_dispatch/base.rb b/lib/chef/event_dispatch/base.rb
index 0886d63152..3b0b70c9b9 100644
--- a/lib/chef/event_dispatch/base.rb
+++ b/lib/chef/event_dispatch/base.rb
@@ -29,7 +29,7 @@ class Chef
     class Base
 
       # Called at the very start of a Chef Run
-      def run_start(version)
+      def run_start(version, run_status)
       end
 
       def run_started(run_status)
@@ -44,6 +44,7 @@ class Chef
       end
 
       # Called right after ohai runs.
+      # NOTE: the node object here is always nil because of when it is called
       def ohai_completed(node)
       end
 
@@ -73,6 +74,10 @@ class Chef
 
       # TODO: def node_run_list_overridden(*args)
 
+      # Called once the node is loaded by the policy builder
+      def node_load_success(node)
+      end
+
       # Failed to load node data from the server
       def node_load_failed(node_name, exception, config)
       end
@@ -163,6 +168,10 @@ class Chef
 
       ## TODO: add callbacks for overall cookbook eval start and complete.
 
+      # Called immediately after creating the run_context and before any cookbook compilation happens
+      def cookbook_compilation_start(run_context)
+      end
+
       # Called when library file loading starts
       def library_load_start(file_count)
       end
@@ -263,10 +272,18 @@ class Chef
       def recipe_load_complete
       end
 
+      # This is called after all cookbook compilation phases are completed.
+      def cookbook_compilation_complete(run_context)
+      end
+
       # Called before convergence starts
       def converge_start(run_context)
       end
 
+      # Callback hook for handlers to register their interest in the action_collection
+      def action_collection_registration(action_collection)
+      end
+
       # Called when the converge phase is finished.
       def converge_complete
       end
diff --git a/lib/chef/event_dispatch/dispatcher.rb b/lib/chef/event_dispatch/dispatcher.rb
index 69419a393b..d69a7c76e8 100644
--- a/lib/chef/event_dispatch/dispatcher.rb
+++ b/lib/chef/event_dispatch/dispatcher.rb
@@ -10,19 +10,44 @@ class Chef
     class Dispatcher < Base
 
       attr_reader :subscribers
+      attr_reader :event_list
 
       def initialize(*subscribers)
         @subscribers = subscribers
+        @event_list = []
       end
 
       # Add a new subscriber to the list of registered subscribers
       def register(subscriber)
-        @subscribers << subscriber
+        subscribers << subscriber
+      end
+
+      def unregister(subscriber)
+        subscribers.reject! { |x| x == subscriber }
+      end
+
+      def enqueue(method_name, *args)
+        event_list << [ method_name, *args ]
+        process_events_until_done unless @in_call
+      end
+
+      (Base.instance_methods - Object.instance_methods).each do |method_name|
+        class_eval <<-EOM
+          def #{method_name}(*args)
+            enqueue(#{method_name.inspect}, *args)
+          end
+        EOM
+      end
+
+      # Special case deprecation, since it needs to know its caller
+      def deprecation(message, location = caller(2..2)[0])
+        enqueue(:deprecation, message, location)
       end
 
       # Check to see if we are dispatching to a formatter
+      # @api private
       def formatter?
-        @subscribers.any? { |s| s.respond_to?(:is_formatter?) && s.is_formatter? }
+        subscribers.any? { |s| s.respond_to?(:is_formatter?) && s.is_formatter? }
       end
 
       ####
@@ -30,9 +55,11 @@ class Chef
       # define the forwarding in one go:
       #
 
+      # @api private
       def call_subscribers(method_name, *args)
-        @subscribers.each do |s|
-          # Skip new/unsupported event names.
+        @in_call = true
+        subscribers.each do |s|
+          # Skip new/unsupported event names
           next if !s.respond_to?(method_name)
           mth = s.method(method_name)
           # Trim arguments to match what the subscriber expects to allow
@@ -43,20 +70,19 @@ class Chef
             mth.call(*args)
           end
         end
+      ensure
+        @in_call = false
       end
 
-      (Base.instance_methods - Object.instance_methods).each do |method_name|
-        class_eval <<-EOM
-          def #{method_name}(*args)
-            call_subscribers(#{method_name.inspect}, *args)
-          end
-        EOM
-      end
+      private
 
-      # Special case deprecation, since it needs to know its caller
-      def deprecation(message, location = caller(2..2)[0])
-        call_subscribers(:deprecation, message, location)
+      # events are allowed to enqueue chained events, so pop them off until
+      # empty, rather than iterating over the list.
+      #
+      def process_events_until_done
+        call_subscribers(*event_list.shift) until event_list.empty?
       end
+
     end
   end
 end
diff --git a/lib/chef/event_loggers/windows_eventlog.rb b/lib/chef/event_loggers/windows_eventlog.rb
index 950477556a..e084fac684 100644
--- a/lib/chef/event_loggers/windows_eventlog.rb
+++ b/lib/chef/event_loggers/windows_eventlog.rb
@@ -1,7 +1,7 @@
 #
 # Author:: Jay Mundrawala (<jdm@chef.io>)
 #
-# Copyright:: Copyright 2014-2016, Chef Software, Inc.
+# Copyright:: Copyright 2014-2019, Chef Software Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -45,7 +45,7 @@ class Chef
         @eventlog = ::Win32::EventLog.open("Application")
       end
 
-      def run_start(version)
+      def run_start(version, run_status)
         @eventlog.report_event(
           event_type: ::Win32::EventLog::INFO_TYPE,
           source: SOURCE,
diff --git a/lib/chef/formatters/doc.rb b/lib/chef/formatters/doc.rb
index 19a6fafae6..936738f147 100644
--- a/lib/chef/formatters/doc.rb
+++ b/lib/chef/formatters/doc.rb
@@ -42,7 +42,7 @@ class Chef
         message
       end
 
-      def run_start(version)
+      def run_start(version, run_status)
         puts_line "Starting Chef Client, version #{version}"
         puts_line "OpenSSL FIPS 140 mode enabled" if Chef::Config[:fips]
       end
diff --git a/lib/chef/formatters/minimal.rb b/lib/chef/formatters/minimal.rb
index c8fc504eb0..e182189f7d 100644
--- a/lib/chef/formatters/minimal.rb
+++ b/lib/chef/formatters/minimal.rb
@@ -26,7 +26,7 @@ class Chef
       end
 
       # Called at the very start of a Chef Run
-      def run_start(version)
+      def run_start(version, run_status)
         puts_line "Starting Chef Client, version #{version}"
         puts_line "OpenSSL FIPS 140 mode enabled" if Chef::Config[:fips]
       end
diff --git a/lib/chef/node.rb b/lib/chef/node.rb
index 87418b5732..9b67f27f14 100644
--- a/lib/chef/node.rb
+++ b/lib/chef/node.rb
@@ -2,7 +2,7 @@
 # Author:: Christopher Brown (<cb@chef.io>)
 # Author:: Christopher Walters (<cw@chef.io>)
 # Author:: Tim Hinderliter (<tim@chef.io>)
-# Copyright:: Copyright 2008-2018, Chef Software Inc.
+# Copyright:: Copyright 2008-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -19,6 +19,7 @@
 #
 
 require "forwardable"
+require "securerandom"
 require "chef/config"
 require "chef/nil_argument"
 require "chef/mixin/params_validate"
@@ -338,7 +339,7 @@ class Chef
       logger.debug("Platform is #{platform} version #{version}")
       automatic[:platform] = platform
       automatic[:platform_version] = version
-      automatic[:chef_guid] = Chef::Config[:chef_guid]
+      automatic[:chef_guid] = Chef::Config[:chef_guid] || ( Chef::Config[:chef_guid] = node_uuid )
       automatic[:name] = name
       automatic[:chef_environment] = chef_environment
     end
@@ -687,5 +688,24 @@ class Chef
       data
     end
 
+    # Returns a UUID that uniquely identifies this node for reporting reasons.
+    #
+    # The node is read in from disk if it exists, or it's generated if it does
+    # does not exist.
+    #
+    # @return [String] UUID for the node
+    #
+    def node_uuid
+      path = File.expand_path(Chef::Config[:chef_guid_path])
+      dir = File.dirname(path)
+
+      unless File.exists?(path)
+        FileUtils.mkdir_p(dir)
+        File.write(path, SecureRandom.uuid)
+      end
+
+      File.open(path).first.chomp
+    end
+
   end
 end
diff --git a/lib/chef/policy_builder/dynamic.rb b/lib/chef/policy_builder/dynamic.rb
index 8ce4f25bfa..2c49f41656 100644
--- a/lib/chef/policy_builder/dynamic.rb
+++ b/lib/chef/policy_builder/dynamic.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Daniel DeLeo (<dan@chef.io>)
-# Copyright:: Copyright 2015-2016, Chef Software, Inc.
+# Copyright:: Copyright 2015-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -74,6 +74,7 @@ class Chef
         select_implementation(node)
         implementation.finish_load_node(node)
         node
+        events.node_load_success(node)
       rescue Exception => e
         events.node_load_failed(node_name, e, config)
         raise
diff --git a/lib/chef/policy_builder/expand_node_object.rb b/lib/chef/policy_builder/expand_node_object.rb
index c91c74b047..839c3bb526 100644
--- a/lib/chef/policy_builder/expand_node_object.rb
+++ b/lib/chef/policy_builder/expand_node_object.rb
@@ -3,7 +3,7 @@
 # Author:: Tim Hinderliter (<tim@chef.io>)
 # Author:: Christopher Walters (<cw@chef.io>)
 # Author:: Daniel DeLeo (<dan@chef.io>)
-# Copyright:: Copyright 2008-2017, Chef Software Inc.
+# Copyright:: Copyright 2008-2018, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -68,6 +68,11 @@ class Chef
         Chef.set_run_context(run_context)
       end
 
+      # This not only creates the run_context but this is where we kick off
+      # compiling the entire expanded run_list, loading all the libraries, resources,
+      # attribute files and recipes, and constructing the entire resource collection.
+      # (FIXME: break up creating the run_context and compiling the cookbooks)
+      #
       def setup_run_context(specific_recipes = nil)
         if Chef::Config[:solo_legacy_mode]
           Chef::Cookbook::FileVendor.fetch_from_disk(Chef::Config[:cookbook_path])
@@ -88,18 +93,20 @@ class Chef
           run_context = Chef::RunContext.new(node, cookbook_collection, @events)
         end
 
-        # TODO: this is really obviously not the place for this
-        # FIXME: need same edits
+        # TODO: move this into the cookbook_compilation_start hook
         setup_chef_class(run_context)
 
-        # TODO: this is not the place for this. It should be in Runner or
-        # CookbookCompiler or something.
+        events.cookbook_compilation_start(run_context)
+
         run_context.load(@run_list_expansion)
         if specific_recipes
           specific_recipes.each do |recipe_file|
             run_context.load_recipe_file(recipe_file)
           end
         end
+
+        events.cookbook_compilation_complete(run_context)
+
         run_context
       end
 
diff --git a/lib/chef/policy_builder/policyfile.rb b/lib/chef/policy_builder/policyfile.rb
index 3e7462f0ed..bedbe651a1 100644
--- a/lib/chef/policy_builder/policyfile.rb
+++ b/lib/chef/policy_builder/policyfile.rb
@@ -192,8 +192,12 @@ class Chef
 
         setup_chef_class(run_context)
 
+        events.cookbook_compilation_start(run_context)
+
         run_context.load(run_list_expansion_ish)
 
+        events.cookbook_compilation_complete(run_context)
+
         setup_chef_class(run_context)
         run_context
       end
diff --git a/lib/chef/resource.rb b/lib/chef/resource.rb
index c857d76c02..e08b2b0bc3 100644
--- a/lib/chef/resource.rb
+++ b/lib/chef/resource.rb
@@ -140,6 +140,7 @@ class Chef
       @guard_interpreter = nil
       @default_guard_interpreter = :default
       @elapsed_time = 0
+      @executed_by_runner = false
     end
 
     #
@@ -453,6 +454,11 @@ class Chef
     attr_reader :elapsed_time
 
     #
+    # @return [Boolean] If the resource was executed by the runner
+    #
+    attr_accessor :executed_by_runner
+
+    #
     # The guard interpreter that will be used to process `only_if` and `not_if`
     # statements.  If left unset, the #default_guard_interpreter will be used.
     #
@@ -1184,8 +1190,10 @@ class Chef
     # Internal Resource Interface (for Chef)
     #
 
-    FORBIDDEN_IVARS = [:@run_context, :@logger, :@not_if, :@only_if, :@enclosing_provider, :@description, :@introduced, :@examples, :@validation_message, :@deprecated, :@default_description, :@skip_docs].freeze
-    HIDDEN_IVARS = [:@allowed_actions, :@resource_name, :@source_line, :@run_context, :@logger, :@name, :@not_if, :@only_if, :@elapsed_time, :@enclosing_provider, :@description, :@introduced, :@examples, :@validation_message, :@deprecated, :@default_description, :@skip_docs].freeze
+    # FORBIDDEN_IVARS do not show up when the resource is converted to JSON (ie. hidden from data_collector and sending to the chef server via #to_json/to_h/as_json/inspect)
+    FORBIDDEN_IVARS = [:@run_context, :@logger, :@not_if, :@only_if, :@enclosing_provider, :@description, :@introduced, :@examples, :@validation_message, :@deprecated, :@default_description, :@skip_docs, :@executed_by_runner].freeze
+    # HIDDEN_IVARS do not show up when the resource is displayed to the user as text (ie. in the error inspector output via #to_text)
+    HIDDEN_IVARS = [:@allowed_actions, :@resource_name, :@source_line, :@run_context, :@logger, :@name, :@not_if, :@only_if, :@elapsed_time, :@enclosing_provider, :@description, :@introduced, :@examples, :@validation_message, :@deprecated, :@default_description, :@skip_docs, :@executed_by_runner].freeze
 
     include Chef::Mixin::ConvertToClassName
     extend Chef::Mixin::ConvertToClassName
diff --git a/lib/chef/resource_collection/resource_list.rb b/lib/chef/resource_collection/resource_list.rb
index 75305f63a6..3c993d6e55 100644
--- a/lib/chef/resource_collection/resource_list.rb
+++ b/lib/chef/resource_collection/resource_list.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Tyler Ball (<tball@chef.io>)
-# Copyright:: Copyright 2014-2016, Chef Software, Inc.
+# Copyright:: Copyright 2014-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/lib/chef/resource_reporter.rb b/lib/chef/resource_reporter.rb
index 3198f84d12..01e4073549 100644
--- a/lib/chef/resource_reporter.rb
+++ b/lib/chef/resource_reporter.rb
@@ -3,7 +3,7 @@
 # Author:: Prajakta Purohit (prajakta@chef.io>)
 # Auther:: Tyler Cloke (<tyler@opscode.com>)
 #
-# Copyright:: Copyright 2012-2018, Chef Software Inc.
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -25,88 +25,40 @@ require "chef/event_dispatch/base"
 
 class Chef
   class ResourceReporter < EventDispatch::Base
+    def for_json(action_record)
+      new_resource = action_record.new_resource
+      current_resource = action_record.current_resource
 
-    ResourceReport = Struct.new(:new_resource,
-                                :current_resource,
-                                :action,
-                                :exception,
-                                :elapsed_time) do
-
-      def self.new_with_current_state(new_resource, action, current_resource)
-        report = new
-        report.new_resource = new_resource
-        report.action = action
-        report.current_resource = current_resource
-        report
-      end
+      as_hash = {}
+      as_hash["type"]     = new_resource.resource_name.to_sym
+      as_hash["name"]     = new_resource.name.to_s
+      as_hash["id"]       = new_resource.identity.to_s
+      as_hash["after"]    = new_resource.state_for_resource_reporter
+      as_hash["before"]   = current_resource ? current_resource.state_for_resource_reporter : {}
+      as_hash["duration"] = ( action_record.elapsed_time * 1000 ).to_i
+      as_hash["delta"]    = new_resource.diff if new_resource.respond_to?("diff")
+      as_hash["delta"]    = "" if as_hash["delta"].nil?
 
-      def self.new_for_exception(new_resource, action)
-        report = new
-        report.new_resource = new_resource
-        report.action = action
-        report
+      # TODO: rename as "action"
+      as_hash["result"] = action_record.action.to_s
+      if new_resource.cookbook_name
+        as_hash["cookbook_name"] = new_resource.cookbook_name
+        as_hash["cookbook_version"] = new_resource.cookbook_version.version
       end
 
-      # Future: Some resources store state information that does not convert nicely
-      # to json. We can't call a resource's state method here, since there are conflicts
-      # with some LWRPs, so we can't override a resource's state method to return
-      # json-friendly state data.
-      #
-      # The registry key resource returns json-friendly state data through its state
-      # attribute, and uses a read-only variable for fetching true state data. If
-      # we have conflicts with other resources reporting json incompatible state, we
-      # may want to extend the state_attrs API with the ability to rename POST'd
-      # attrs.
-      def for_json
-        as_hash = {}
-        as_hash["type"]   = new_resource.resource_name.to_sym
-        as_hash["name"]   = new_resource.name.to_s
-        as_hash["id"]     = new_resource.identity.to_s
-        as_hash["after"]  = new_resource.state_for_resource_reporter
-        as_hash["before"] = current_resource ? current_resource.state_for_resource_reporter : {}
-        as_hash["duration"] = (elapsed_time * 1000).to_i.to_s
-        as_hash["delta"]  = new_resource.diff if new_resource.respond_to?("diff")
-        as_hash["delta"]  = "" if as_hash["delta"].nil?
-
-        # TODO: rename as "action"
-        as_hash["result"] = action.to_s
-        if success?
-        else
-          # as_hash["result"] = "failed"
-        end
-        if new_resource.cookbook_name
-          as_hash["cookbook_name"] = new_resource.cookbook_name
-          as_hash["cookbook_version"] = new_resource.cookbook_version.version
-        end
-
-        as_hash
-      end
-
-      def finish
-        self.elapsed_time = new_resource.elapsed_time
-      end
-
-      def success?
-        !exception
-      end
-    end # End class ResouceReport
+      as_hash
+    end
 
-    attr_reader :updated_resources
     attr_reader :status
     attr_reader :exception
     attr_reader :error_descriptions
+    attr_reader :action_collection
+    attr_reader :rest_client
 
     PROTOCOL_VERSION = "0.1.0".freeze
 
     def initialize(rest_client)
-      if Chef::Config[:enable_reporting] && !Chef::Config[:why_run]
-        @reporting_enabled = true
-      else
-        @reporting_enabled = false
-      end
-      @updated_resources = []
-      @total_res_count = 0
-      @pending_update  = nil
+      @pending_update = nil
       @status = "success"
       @exception = nil
       @rest_client = rest_client
@@ -120,8 +72,8 @@ class Chef
       if reporting_enabled?
         begin
           resource_history_url = "reports/nodes/#{node_name}/runs"
-          server_response = @rest_client.post(resource_history_url, { action: :start, run_id: run_id,
-                                                                      start_time: start_time.to_s }, headers)
+          server_response = rest_client.post(resource_history_url, { action: :start, run_id: run_id,
+                                                                     start_time: start_time.to_s }, headers)
         rescue Timeout::Error, Errno::EINVAL, Errno::ECONNRESET, EOFError, Net::HTTPBadResponse, Net::HTTPHeaderSyntaxError, Net::ProtocolError => e
           handle_error_starting_run(e, resource_history_url)
         end
@@ -157,62 +109,13 @@ class Chef
         end
       end
 
-      @reporting_enabled = false
+      @runs_endpoint_failed = true
     end
 
     def run_id
       @run_status.run_id
     end
 
-    def resource_current_state_loaded(new_resource, action, current_resource)
-      unless nested_resource?(new_resource)
-        @pending_update = ResourceReport.new_with_current_state(new_resource, action, current_resource)
-      end
-    end
-
-    def resource_up_to_date(new_resource, action)
-      @total_res_count += 1
-      @pending_update = nil unless nested_resource?(new_resource)
-    end
-
-    def resource_skipped(resource, action, conditional)
-      @total_res_count += 1
-      @pending_update = nil unless nested_resource?(resource)
-    end
-
-    def resource_updated(new_resource, action)
-      @total_res_count += 1
-    end
-
-    def resource_failed(new_resource, action, exception)
-      @total_res_count += 1
-      unless nested_resource?(new_resource)
-        @pending_update ||= ResourceReport.new_for_exception(new_resource, action)
-        @pending_update.exception = exception
-      end
-      description = Formatters::ErrorMapper.resource_failed(new_resource, action, exception)
-      @error_descriptions = description.for_json
-    end
-
-    def resource_completed(new_resource)
-      if @pending_update && !nested_resource?(new_resource)
-        @pending_update.finish
-
-        # Verify if the resource has sensitive data
-        # and create a new blank resource with only
-        # the name so we can report it back without
-        # sensitive data
-        if @pending_update.new_resource.sensitive
-          klass = @pending_update.new_resource.class
-          resource_name = @pending_update.new_resource.name
-          @pending_update.new_resource = klass.new(resource_name)
-        end
-
-        @updated_resources << @pending_update
-        @pending_update = nil
-      end
-    end
-
     def run_completed(node)
       @status = "success"
       post_reporting_data
@@ -232,6 +135,11 @@ class Chef
       @expanded_run_list = run_list_expansion
     end
 
+    def action_collection_registration(action_collection)
+      @action_collection = action_collection
+      action_collection.register(self) if reporting_enabled?
+    end
+
     def post_reporting_data
       if reporting_enabled?
         run_data = prepare_run_data
@@ -242,7 +150,7 @@ class Chef
         Chef::Log.trace("Sending compressed run data...")
         # Since we're posting compressed data we can not directly call post which expects JSON
         begin
-          @rest_client.raw_request(:POST, resource_history_url, headers({ "Content-Encoding" => "gzip" }), compressed_data)
+          rest_client.raw_request(:POST, resource_history_url, headers({ "Content-Encoding" => "gzip" }), compressed_data)
         rescue StandardError => e
           if e.respond_to? :response
             Chef::FileCache.store("failed-reporting-data.json", Chef::JSONCompat.to_json_pretty(run_data), 0640)
@@ -273,15 +181,24 @@ class Chef
       @run_status.end_time
     end
 
+    # get only the top level resources and strip out the subcollections
+    def updated_resources
+      @updated_resources ||= action_collection.filtered_collection(max_nesting: 0, up_to_date: false, skipped: false, unprocessed: false)
+    end
+
+    def total_res_count
+      updated_resources.count
+    end
+
     def prepare_run_data
       run_data = {}
       run_data["action"] = "end"
-      run_data["resources"] = updated_resources.map do |resource_record|
-        resource_record.for_json
+      run_data["resources"] = updated_resources.map do |action_record|
+        for_json(action_record)
       end
       run_data["status"] = @status
       run_data["run_list"] = Chef::JSONCompat.to_json(@run_status.node.run_list)
-      run_data["total_res_count"] = @total_res_count.to_s
+      run_data["total_res_count"] = total_res_count.to_s
       run_data["data"] = {}
       run_data["start_time"] = start_time.to_s
       run_data["end_time"] = end_time.to_s
@@ -313,18 +230,10 @@ class Chef
       @error_descriptions = description.for_json
     end
 
-    def reporting_enabled?
-      @reporting_enabled
-    end
-
     private
 
-    # If we are getting messages about a resource while we are in the middle of
-    # another resource's update, we assume that the nested resource is just the
-    # implementation of a provider, and we want to hide it from the reporting
-    # output.
-    def nested_resource?(new_resource)
-      @pending_update && @pending_update.new_resource != new_resource
+    def reporting_enabled?
+      Chef::Config[:enable_reporting] && !Chef::Config[:why_run] && !@runs_endpoint_failed
     end
 
     def encode_gzip(data)
diff --git a/lib/chef/run_context.rb b/lib/chef/run_context.rb
index b940bfecbd..e407a0e7be 100644
--- a/lib/chef/run_context.rb
+++ b/lib/chef/run_context.rb
@@ -2,7 +2,7 @@
 # Author:: Adam Jacob (<adam@chef.io>)
 # Author:: Christopher Walters (<cw@chef.io>)
 # Author:: Tim Hinderliter (<tim@chef.io>)
-# Copyright:: Copyright 2008-2017, Chef Software Inc.
+# Copyright:: Copyright 2008-2018, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -104,6 +104,7 @@ class Chef
     #
     attr_reader :resource_collection
 
+    attr_accessor :action_collection
     #
     # The list of control groups to execute during the audit phase
     #
@@ -603,6 +604,8 @@ class Chef
     class ChildRunContext < RunContext
       extend Forwardable
       def_delegators :parent_run_context, *%w{
+        action_collection
+        action_collection=
         cancel_reboot
         config
         cookbook_collection
diff --git a/lib/chef/run_context/cookbook_compiler.rb b/lib/chef/run_context/cookbook_compiler.rb
index c3cee5841f..1bd612e721 100644
--- a/lib/chef/run_context/cookbook_compiler.rb
+++ b/lib/chef/run_context/cookbook_compiler.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Daniel DeLeo (<dan@chef.io>)
-# Copyright:: Copyright 2012-2018, Chef Software Inc.
+# Copyright:: Copyright 2012-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
diff --git a/lib/chef/run_status.rb b/lib/chef/run_status.rb
index 37b10fb9be..575d31159b 100644
--- a/lib/chef/run_status.rb
+++ b/lib/chef/run_status.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Daniel DeLeo (<dan@chef.io>)
-# Copyright:: Copyright 2010-2018, Chef Software Inc.
+# Copyright:: Copyright 2010-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -53,6 +53,7 @@ class Chef::RunStatus
 
   # sets +end_time+ to the current time
   def stop_clock
+    @start_time ||= Time.now # if we failed so early we didn't get a start time
     @end_time = Time.now
   end
 
diff --git a/lib/chef/runner.rb b/lib/chef/runner.rb
index 1c82439b57..d5dab75dfd 100644
--- a/lib/chef/runner.rb
+++ b/lib/chef/runner.rb
@@ -2,7 +2,7 @@
 # Author:: Adam Jacob (<adam@chef.io>)
 # Author:: Christopher Walters (<cw@chef.io>)
 # Author:: Tim Hinderliter (<tim@chef.io>)
-# Copyright:: Copyright 2008-2017, Chef Software Inc.
+# Copyright:: Copyright 2008-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -95,7 +95,11 @@ class Chef
 
       # Execute each resource.
       run_context.resource_collection.execute_each_resource do |resource|
-        Array(resource.action).each { |action| run_action(resource, action) }
+        begin
+          Array(resource.action).each { |action| run_action(resource, action) }
+        ensure
+          resource.executed_by_runner = true
+        end
       end
 
     rescue Exception => e