Extract Action Collection from Data Collector

See the PR for details on this change.

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

45 files changed, 2686 insertions, 3003 deletions

diff --git a/.gitignore b/.gitignore
index ac14a13e8e..200db3afc5 100644
--- a/.gitignore
+++ b/.gitignore
@@ -47,6 +47,7 @@ kitchen-tests/Berksfile.lock
 # Temporary files present during spec runs
 spec/data/test-dir
 spec/data/nodes
+spec/data/chef_guid
 /config/
 
 vendor/
diff --git a/chef-config/lib/chef-config/config.rb b/chef-config/lib/chef-config/config.rb
index 1c921eac72..29dadf54e0 100644
--- a/chef-config/lib/chef-config/config.rb
+++ b/chef-config/lib/chef-config/config.rb
@@ -378,6 +378,11 @@ module ChefConfig
     default :diff_disabled,           false
     default :diff_filesize_threshold, 10000000
     default :diff_output_threshold,   1000000
+
+    # This is true for "local mode" which uses a chef-zero server listening on
+    # localhost one way or another.  This is true for both `chef-solo` (without
+    # the --legacy-mode flag) or `chef-client -z` methods of starting a client run.
+    #
     default :local_mode, false
 
     # Configures the mode of operation for ChefFS, which is applied to the
@@ -445,9 +450,29 @@ module ChefConfig
     end
 
     default :rest_timeout, 300
+
+    # This solo setting is now almost entirely useless.  It is set to true if chef-solo was
+    # invoked that way from the command-line (i.e. from Application::Solo as opposed to
+    # Application::Client).  The more useful information is contained in the :solo_legacy_mode
+    # vs the :local_mode flags which will be set to true or false depending on how solo was
+    # invoked and actually change more of the behavior.  There might be slight differences in
+    # the behavior of :local_mode due to the behavioral differences in Application::Solo vs.
+    # Application::Client and `chef-solo` vs `chef-client -z`, but checking this value and
+    # switching based on it is almost certainly doing the wrong thing and papering over
+    # bugs that should be fixed in one or the other class, and will be brittle and destined
+    # to break in the future (and not necessarily on a major version bump). Checking this value
+    # is also not sufficent to determine if we are not running against a server since this can
+    # be unset but :local_mode may be set.  It would be accurate to check both :solo and :local_mode
+    # to determine if we're not running against a server, but the more semantically accurate test
+    # is going to be combining :solo_legacy_mode and :local_mode.
+    #
+    # TL;DR: `if Chef::Config[:solo]` is almost certainly buggy code, you should use:
+    #        `if Chef::Config[:local_mode] || Chef::Config[:solo_legacy_mode]`
+    #
+    # @api private
     default :solo, false
 
-    # Are we running in old Chef Solo legacy mode?
+    # This is true for old chef-solo legacy mode without any chef-zero server (chef-solo --legacy-mode)
     default :solo_legacy_mode, false
 
     default :splay, nil
@@ -919,7 +944,7 @@ module ChefConfig
       # data collector will not run.
       # Ex: http://my-data-collector.mycompany.com/ingest
       default(:server_url) do
-        if config_parent.solo || config_parent.local_mode
+        if config_parent.solo_legacy_mode || config_parent.local_mode
           nil
         else
           File.join(config_parent.chef_server_url, "/data-collector")
@@ -950,7 +975,7 @@ module ChefConfig
       # generated by the DataCollector when Chef is run in Solo mode. This
       # allows users to associate their Solo nodes with faux organizations
       # without the nodes being connected to an actual Chef Server.
-      default :organization, nil
+      default :organization, "chef_solo"
     end
 
     configurable(:http_proxy)
diff --git a/chef-config/spec/unit/config_spec.rb b/chef-config/spec/unit/config_spec.rb
index b6e88f0bc8..0e17753185 100644
--- a/chef-config/spec/unit/config_spec.rb
+++ b/chef-config/spec/unit/config_spec.rb
@@ -1,7 +1,7 @@
 #
 # Author:: Adam Jacob (<adam@chef.io>)
 # Author:: Kyle Goodwin (<kgoodwin@primerevenue.com>)
-# Copyright:: Copyright 2008-2016, 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");
@@ -1197,10 +1197,23 @@ RSpec.describe ChefConfig::Config do
 
       end
 
-      context "for Chef Solo" do
+      context "for Chef Solo legacy mode" do
 
         before do
-          ChefConfig::Config[:solo] = true
+          ChefConfig::Config[:solo_legacy_mode] = true
+        end
+
+        it "sets the data collector server URL to nil" do
+          ChefConfig::Config[:chef_server_url] = "https://chef.example/organizations/myorg"
+          expect(ChefConfig::Config[:data_collector][:server_url]).to be_nil
+        end
+
+      end
+
+      context "for local mode" do
+
+        before do
+          ChefConfig::Config[:local_mode] = true
         end
 
         it "sets the data collector server URL to nil" do
diff --git a/docs/dev/action_collection.md b/docs/dev/action_collection.md
new file mode 100644
index 0000000000..a0735f65fb
--- /dev/null
+++ b/docs/dev/action_collection.md
@@ -0,0 +1,106 @@
+---
+title: Action Collection
+---
+
+# Action Collection Design
+
+* Extract common code from the Resource Reporter and Data Collector.
+* Expose a general purpose API for querying a record of all actions taken during the Chef run.
+* Enable utilities like the 'zap' cookbook to be written to interact properly with Custom Resources.
+
+The Action Collection tracks all actions taken by all Chef resources.  The resources can be in recipe code, as sub-resources of custom resources or
+they may be built "by hand".  Since the Action Collection hooks the events which are fired from the `run_action` method on Chef::Resource it does
+not matter how the resources were built (as long as they were correctly passed the Chef `run_context`).
+
+This is complementary, but superior, to the resource collection which has an incomplete picture of what might happen or has happened in the run since there are
+many common ways of invoking resource actions which are not captured by how the resource collection is built.  Replaying the sequence of actions in
+the Action Collection would be closer to replaying the chef-client converge than trying to re-converge the resource collection (although both of
+those models are still flawed in the presence of any imperative code that controls the shape of those objects).
+
+This design extracts common duplicated code from the Data Collection and old Resource Reporter, and is designed to be used by other consumers which
+need to ask questions like "in this run, what file resources had actions fired on them?", which can then be used to answer questions like
+"which files is Chef managing in this directory?".
+
+# Usage
+
+## Action Collection Event Hook Registration
+
+Consumers may register an event handler which hooks the `action_collection_registration` hook.  This event is fired directly before recipes are
+compiled and converged (after library loading, attributes, etc).  This is just before the earliest point in time that a resource should fire an
+action so represents the latest point that a consumer should make a decision about if it needs the Action Collection to be enabled or not.
+
+Consumers can hook this method.  They will be passed the Action Collection instance, which can be saved by the caller to be queried later.  They
+should then register themselves with the Action Collection (since without registering any interest, the Action Collection will disable itself).
+
+```ruby
+  def action_collection_registration(action_collection)
+    @action_collection = action_collection
+    action_collection.register(self)
+  end
+```
+
+## Library Registration
+
+Any cookbook library code may also register itself with the Action Collection.  The Action Collection will be registered with the `run_context` after
+it is created, so registration may be accomplished easily:
+
+```ruby
+  Chef.run_context.action_collection.register(self)
+```
+
+## Action Collection Requires Registration
+
+If one of the prior methods is not used to register for the Action Collection, then the Action Collection will disable itself and will not compile
+the Action Collection in order to not waste the memory overhead of tracking the actions during the run.  The Data Collector takes advantage of this
+since if the run start message from the Data Collector is refused by the server, then the Data Collector disables itself, and then does not register
+with the Action Collection, which would disable the Action Collection.  This makes use of the delayed hooking through the `action_collection_regsitration`
+so that the Data Collector never registers itself after it is disabled.
+
+## Searching
+
+There is a function `filtered_collection` which returns "slices" off of the `ActionCollection` object.  The `max_nesting` argument can be used to prune
+how deep into sub-resources the returned view goes (`max_nesting: 0` will return only resources in recipe context, with any hand created resources, but
+no subresources).  There are also 5 different states of the action:  `up_to_date`, `skipped`, `updated`, `failed`, `unprocessed` which can be filtered
+on.  All of these are true by default, so they must be disabled to remove them from the filtered collection.
+
+The `ActionCollection` object itself implements enumerable and returns `ActionRecord` objects (see the `ActionCollection` code for the fields exposed on
+`ActionRecord`s).
+
+This would return all file resources in any state in the recipe context:
+
+```
+Chef.run_context.action_collection.filtered_collection(max_nesting: 0).select { |rec| rec.new_resource.is_a?(Chef::Resource::File) }
+```
+
+NOTE:
+As the Action Collection API was initially designed around the Resource Reporter and Data Collector use cases, the searching API is currently rudimentary
+and could easily lift some of the searching features on the name of the resource from the resource collection, and could use a more fluent API
+for composing searches.
+
+# Implementation Details
+
+## Resource Event Lifecycle Hooks
+
+Resources actions fire off several events in sequence:
+
+1. `resource_action_start` - this is always fired first
+2. `resource_current_state_loaded` - this is normally always second, but may be skipped in the case of a resource which throws an exception during
+`load_current_resource` (which means that the `current_resource` off the `ActionRecord` may be nil).
+3. `resource_up_to_date` / `resource_skipped` / `resource_updated` / `resource_failed` - one of these is always called which corresponds to the state of the action.
+4. `resource_completed` - this is always fired last
+
+For skipped resources, the conditional will be saved in the `ActionRecord`.  For failed resources the exception is saved in the `ActionRecord`.
+
+## Unprocessed Resources
+
+The unprocessed resource concept is to report on resources which are left in the resource collection after a failure.  A successful Chef run should
+never leave any unprocessed resources (`action :nothing` resources are still inspected by the resource collection and are processed).  There must be
+an exception thrown during the execution of the resource collection, and the unprocessed resources were never visited by the runner that executes
+the resource collection.
+
+This list will be necessarily incomplete of any unprocessed sub-resources in custom resources, since the run was aborted before those resources
+executed actions and built their own sub-resource collections.
+
+This was a design requirement of the Data Collector.
+
+To implement this in a more sane manner the runner that evaluates the resource collection now tracks the resources that it visits.
diff --git a/docs/dev/data_collector.md b/docs/dev/data_collector.md
new file mode 100644
index 0000000000..168bf206e2
--- /dev/null
+++ b/docs/dev/data_collector.md
@@ -0,0 +1,117 @@
+---
+title: Data Collector
+---
+
+# Data Collector Design
+
+The Data Collector design and API is covered in:
+
+https://github.com/chef/chef-rfc/blob/master/rfc077-mode-agnostic-data-collection.md
+
+This document will focus entirely on the nuts and bolts of the Data Collector
+
+## Action Collection Integration
+
+Most of the work is done by a separate Action Collection to track the actions of Chef resources.  If the Data Collector is not enabled, it never registers with the
+Action Collection and no work will be done by the Action Collection to track resources.
+
+## Additional Collected Information
+
+The Data Collector also collects:
+
+- the expanded run list
+- deprecations
+- the node
+- formatted error output for exceptions
+
+Most of this is done through hooking events directly in the Data Collector itself.  The ErrorHandlers module is broken out into a module which is directly mixed into
+the Data Collector to separate that concern out into a different file (it is straightforward with fairly little state, but is just a lot of hooked methods).
+
+## Basic Configuration Modes
+
+### Configured for Automate
+
+Do nothing.  The URL is constructed from the base `Chef::Config[:chef_server_url]`, auth is just Chef Server API authentication, and the default behavior is that it
+is configured.
+
+### Configured to Log to a File
+
+Setup a file output location, no token is necessary:
+
+```
+Chef::Config[:data_collector][:output_locations] = { files:  [ "/Users/lamont/data_collector.out" ] }
+```
+
+Note the fact that you can't assign to `Chef::Config[:data_collector][:output_locations][:files]` and will NoMethodError if you try.
+
+### Configured to Log to a Non-Chef Server Endpoint
+
+Setup a server url, requiring a token:
+
+```
+Chef::Config[:data_collector][:server_url] = "https://chef.acme.local/myendpoint.html"
+Chef::Config[:data_collector][:token] = "mytoken"
+```
+
+This works for chef-clients which are configured to hit a chef server, but use a custom non-Chef-Automate endpoint for reporting, or for chef-solo/zero users.
+
+XXX: There is also the `Chef::Config[:data_collector][:output_locations] = { uri: [ "https://chef.acme.local/myendpoint.html" ] }` method -- which is going to behave
+differently, particularly on non-chef-solo use cases.  In that case the Data Collector `server_url` will still be automatically derived from the `chef_server_url` and
+the Data Collector will attempt to contact that endpoint, but with the token being supplied it will use that and will not use Chef Server authentication, and the
+server should 403 back, and if `raise_on_failure` is left to the default of false then it will simply drop that failure and continue without raising, which will
+appear to work, and output will be send to the configured `output_locations`.  Note that the presence of a token flips all external URIs to using the token so that
+it is **not** possible to use this feature to talk to both a Chef Automate endpoint and a custom URI reporting endpoint (which would seem to be the most useful of an
+incredibly marginally useful feature and it does not work).  But given how hopelessly complicated this is, the recommendation is to use the `server_url` and to avoid
+using any `url` options in the `output_locations` since that feature is fairly poorly designed at this point in time.
+
+## Resiliency to Failures
+
+The Data Collector in Chef >= 15.0 is resilient to failures that occur anywhere in the main loop of the `Chef::Client#run` method.  In order to do this there is a lot
+of defensive coding around internal data structures that may be nil (e.g. failures before the node is loaded will result in the node being nil).  The spec tests for
+the Data Collector now run through a large sequence of events (which must, unfortunately, be manually kept in sync with the events in the Chef::Client if those events
+are ever 'moved' around) which should catch any issues in the Data Collector with early failures.  The specs should also serve as documentation for what the messages
+will look like under different failure conditions.  The goal was to keep the format of the messages to look as near as possible to the same schema as possible even
+in the presence of failures.  But some data structures will be entirely empty.
+
+When the Data Collector fails extraordinarily early it still sends both a start and an end message.  This will happen if it fails so early that it would not normally
+have sent a start message.
+
+## Decision to Be Enabled
+
+This is complicated due to over-design and is encapsulated in the `#should_be_enabled?` method and the ConfigValidation module.  The `#should_be_enabled?` message and
+ConfigValidation should probably be merged into one renamed Config module to isolate the concern of processing the Chef::Config options and doing the correct thing.
+
+## Run Start and Run End Message modules
+
+These are separated out into their own modules, which are very deliberately not mixed into the main Data Collector.  They use the Data Collector and Action Collection
+public interfaces.  They are stateless themselves.  This keeps the collaboration between them and the Data Collector very easy to understand.  The start message is
+relatively simple and straightforwards.  The complication of the end message is mostly due to walking through the Action Collection and all the collected action 
+records from the entire run, along with a lot of defensive programming to deal with early errors.
+
+## Relevant Event Sequence
+
+As it happens in the actual chef-client run:
+
+1. `events.register(data_collector)`
+2. `events.register(action_collection)`
+3. `run_status.run_id = request_id`
+4. `events.run_start(Chef::VERSION, run_status)`
+  * failures during registration will cause `registration_failed(node_name, exception, config)` here and skip to #13
+  * failures during node loading will cause `node_load_failed(node_name, exception, config)` here and skip to #13
+5. `events.node_load_success(node)`
+6. `run_status.node = node`
+  * failures during run list expansion will cause `run_list_expand_failed(node, exception)` here and skip to #13
+7. `events.run_list_expanded(expansion)`
+8. `run_status.start_clock`
+9. `events.run_started(run_status)`
+  * failures during cookbook resolution will cause `events.cookbook_resolution_failed(node, exception)` here and skip to #13
+  * failures during cookbook synch will cause `events.cookbook_sync_failed(node, exception)` and skip to #13
+10. `events.cookbook_compilation_start(run_context)`
+11. < the resource events happen here which hit the Action Collection, may throw any of the other failure events >
+12. `events.converge_complete` or `events.converge_failed(exception)`
+13. `run_status.stop_clock`
+14. `run_status.exception = exception` if it failed
+15. `events.run_completed(node, run_status)` or `events.run_failed(exception, run_status)`
+
+
+
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
diff --git a/spec/functional/event_loggers/windows_eventlog_spec.rb b/spec/functional/event_loggers/windows_eventlog_spec.rb
index 8a9475680d..fa84e96c25 100644
--- a/spec/functional/event_loggers/windows_eventlog_spec.rb
+++ b/spec/functional/event_loggers/windows_eventlog_spec.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.
@@ -46,7 +46,7 @@ describe Chef::EventLoggers::WindowsEventLogger, :windows_only do
   end
 
   it "writes run_start event with event_id 10000 and contains version" do
-    logger.run_start(version)
+    logger.run_start(version, run_status)
 
     expect(event_log.read(flags, offset).any? do |e|
              e.source == "Chef" && e.event_id == 10000 &&
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index c493987554..d13453b778 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Adam Jacob (<adam@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");
diff --git a/spec/support/shared/context/client.rb b/spec/support/shared/context/client.rb
deleted file mode 100644
index d6c49c896b..0000000000
--- a/spec/support/shared/context/client.rb
+++ /dev/null
@@ -1,305 +0,0 @@
-
-require "spec_helper"
-
-# Stubs a basic client object
-shared_context "client" do
-  let(:fqdn)             { "hostname.example.org" }
-  let(:hostname)         { "hostname" }
-  let(:machinename)      { "machinename.example.org" }
-  let(:platform)         { "example-platform" }
-  let(:platform_version) { "example-platform-1.0" }
-
-  let(:ohai_data) do
-    {
-      fqdn: fqdn,
-      hostname: hostname,
-      machinename: machinename,
-      platform: platform,
-      platform_version: platform_version,
-    }
-  end
-
-  let(:ohai_system) do
-    ohai = instance_double("Ohai::System", all_plugins: true, data: ohai_data, logger: logger)
-    allow(ohai).to receive(:[]) do |k|
-      ohai_data[k]
-    end
-    ohai
-  end
-
-  let(:node) do
-    Chef::Node.new.tap do |n|
-      n.name(fqdn)
-      n.chef_environment("_default")
-    end
-  end
-
-  let(:json_attribs) { nil }
-  let(:client_opts) { {} }
-
-  let(:stdout) { STDOUT }
-  let(:stderr) { STDERR }
-
-  let(:client) do
-    Chef::Config[:event_loggers] = []
-    allow(Ohai::System).to receive(:new).and_return(ohai_system)
-    opts = client_opts.merge({ logger: logger })
-    Chef::Client.new(json_attribs, opts).tap do |c|
-      c.node = node
-    end
-  end
-
-  let(:logger) { instance_double("Mixlib::Log::Child", trace: nil, debug: nil, warn: nil, info: nil, error: nil, fatal: nil) }
-
-  before do
-    stub_const("Chef::Client::STDOUT_FD", stdout)
-    stub_const("Chef::Client::STDERR_FD", stderr)
-    allow(client).to receive(:logger).and_return(logger)
-  end
-end
-
-# Stubs a client for a client run.
-# Requires a client object be defined in the scope of this included context.
-# e.g.:
-#   describe "some functionality" do
-#     include_context "client"
-#     include_context "a client run"
-#     ...
-#   end
-shared_context "a client run" do
-  let(:stdout) { StringIO.new }
-  let(:stderr) { StringIO.new }
-
-  let(:api_client_exists?) { false }
-  let(:enable_fork)        { false }
-
-  let(:http_data_collector)   { double("Chef::ServerAPI (data collector)") }
-  let(:http_cookbook_sync)    { double("Chef::ServerAPI (cookbook sync)") }
-  let(:http_node_load)        { double("Chef::ServerAPI (node)") }
-  let(:http_node_save)        { double("Chef::ServerAPI (node save)") }
-  let(:reporting_rest_client) { double("Chef::ServerAPI (reporting client)") }
-
-  let(:runner)       { instance_double("Chef::Runner") }
-  let(:audit_runner) { instance_double("Chef::Audit::Runner", failed?: false) }
-
-  def stub_for_register
-    # --Client.register
-    #   Make sure Client#register thinks the client key doesn't
-    #   exist, so it tries to register and create one.
-    allow(File).to receive(:exists?).and_call_original
-    expect(File).to receive(:exists?)
-      .with(Chef::Config[:client_key])
-      .exactly(:once)
-      .and_return(api_client_exists?)
-
-    unless api_client_exists?
-      #   Client.register will register with the validation client name.
-      expect_any_instance_of(Chef::ApiClient::Registration).to receive(:run)
-    end
-  end
-
-  def stub_for_data_collector_init
-    expect(Chef::ServerAPI).to receive(:new)
-      .with(Chef::Config[:data_collector][:server_url], validate_utf8: false)
-      .exactly(:once)
-      .and_return(http_data_collector)
-  end
-
-  def stub_for_node_load
-    #   Client.register will then turn around create another
-    #   Chef::ServerAPI object, this time with the client key it got from the
-    #   previous step.
-    expect(Chef::ServerAPI).to receive(:new)
-      .with(Chef::Config[:chef_server_url], client_name: fqdn,
-                                           signing_key_filename: Chef::Config[:client_key])
-      .exactly(:once)
-      .and_return(http_node_load)
-
-    # --Client#build_node
-    #   looks up the node, which we will return, then later saves it.
-    expect(Chef::Node).to receive(:find_or_create).with(fqdn).and_return(node)
-
-    # --ResourceReporter#node_load_completed
-    #   gets a run id from the server for storing resource history
-    #   (has its own tests, so stubbing it here.)
-    expect_any_instance_of(Chef::ResourceReporter).to receive(:node_load_completed)
-  end
-
-  def stub_rest_clean
-    allow(client).to receive(:rest_clean).and_return(reporting_rest_client)
-  end
-
-  def stub_for_sync_cookbooks
-    # --Client#setup_run_context
-    # ---Client#sync_cookbooks -- downloads the list of cookbooks to sync
-    #
-    expect_any_instance_of(Chef::CookbookSynchronizer).to receive(:sync_cookbooks)
-    expect(Chef::ServerAPI).to receive(:new).with(Chef::Config[:chef_server_url], version_class: Chef::CookbookManifestVersions).and_return(http_cookbook_sync)
-    expect(http_cookbook_sync).to receive(:post)
-      .with("environments/_default/cookbook_versions", { run_list: [] })
-      .and_return({})
-  end
-
-  def stub_for_required_recipe
-    response = Net::HTTPNotFound.new("1.1", "404", "Not Found")
-    exception = Net::HTTPClientException.new('404 "Not Found"', response)
-    expect(http_node_load).to receive(:get).with("required_recipe").and_raise(exception)
-  end
-
-  def stub_for_converge
-    # define me
-  end
-
-  def stub_for_audit
-    # define me
-  end
-
-  def stub_for_node_save
-    # define me
-  end
-
-  def stub_for_run
-    # define me
-  end
-
-  before do
-    Chef::Config[:client_fork] = enable_fork
-    Chef::Config[:cache_path] = windows? ? 'C:\chef' : "/var/chef"
-    Chef::Config[:why_run] = false
-    Chef::Config[:audit_mode] = :enabled
-    Chef::Config[:chef_guid] = "default-guid"
-
-    stub_rest_clean
-    stub_for_register
-    stub_for_data_collector_init
-    stub_for_node_load
-    stub_for_sync_cookbooks
-    stub_for_required_recipe
-    stub_for_converge
-    stub_for_audit
-    stub_for_node_save
-
-    expect_any_instance_of(Chef::RunLock).to receive(:acquire)
-    expect_any_instance_of(Chef::RunLock).to receive(:save_pid)
-    expect_any_instance_of(Chef::RunLock).to receive(:release)
-
-    # Post conditions: check that node has been filled in correctly
-    expect(client).to receive(:run_started)
-
-    stub_for_run
-  end
-end
-
-shared_context "converge completed" do
-  def stub_for_converge
-    # --Client#converge
-    expect(Chef::Runner).to receive(:new).and_return(runner)
-    expect(runner).to receive(:converge).and_return(true)
-  end
-
-  def stub_for_node_save
-    allow(node).to receive(:data_for_save).and_return(node.for_json)
-
-    # --Client#save_updated_node
-    expect(Chef::ServerAPI).to receive(:new).with(Chef::Config[:chef_server_url], client_name: fqdn,
-                                                                                  signing_key_filename: Chef::Config[:client_key], validate_utf8: false).and_return(http_node_save)
-    expect(http_node_save).to receive(:put).with("nodes/#{fqdn}", node.for_json).and_return(true)
-  end
-end
-
-shared_context "converge failed" do
-  let(:converge_error) do
-    err = Chef::Exceptions::UnsupportedAction.new("Action unsupported")
-    err.set_backtrace([ "/path/recipe.rb:15", "/path/recipe.rb:12" ])
-    err
-  end
-
-  def stub_for_converge
-    expect(Chef::Runner).to receive(:new).and_return(runner)
-    expect(runner).to receive(:converge).and_raise(converge_error)
-  end
-
-  def stub_for_node_save
-    expect(client).to_not receive(:save_updated_node)
-  end
-end
-
-shared_context "audit phase completed" do
-  def stub_for_audit
-    # -- Client#run_audits
-    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
-    expect(audit_runner).to receive(:run).and_return(true)
-    expect(client.events).to receive(:audit_phase_complete)
-  end
-end
-
-shared_context "audit phase failed with error" do
-  let(:audit_error) do
-    err = RuntimeError.new("Unexpected audit error")
-    err.set_backtrace([ "/path/recipe.rb:57", "/path/recipe.rb:55" ])
-    err
-  end
-
-  def stub_for_audit
-    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
-    expect(Chef::Audit::Logger).to receive(:read_buffer).and_return("Audit mode output!")
-    expect(audit_runner).to receive(:run).and_raise(audit_error)
-    expect(client.events).to receive(:audit_phase_failed).with(audit_error, "Audit mode output!")
-  end
-end
-
-shared_context "audit phase completed with failed controls" do
-  let(:audit_runner) do
-    instance_double("Chef::Audit::Runner", failed?: true,
-                                           num_failed: 1, num_total: 3) end
-
-  let(:audit_error) do
-    err = Chef::Exceptions::AuditsFailed.new(audit_runner.num_failed, audit_runner.num_total)
-    err.set_backtrace([ "/path/recipe.rb:108", "/path/recipe.rb:103" ])
-    err
-  end
-
-  def stub_for_audit
-    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
-    expect(Chef::Audit::Logger).to receive(:read_buffer).and_return("Audit mode output!")
-    expect(audit_runner).to receive(:run)
-    expect(Chef::Exceptions::AuditsFailed).to receive(:new).with(
-      audit_runner.num_failed, audit_runner.num_total
-    ).and_return(audit_error)
-    expect(client.events).to receive(:audit_phase_failed).with(audit_error, "Audit mode output!")
-  end
-end
-
-shared_context "run completed" do
-  def stub_for_run
-    expect(client).to receive(:run_completed_successfully)
-
-    # --ResourceReporter#run_completed
-    #   updates the server with the resource history
-    #   (has its own tests, so stubbing it here.)
-    expect_any_instance_of(Chef::ResourceReporter).to receive(:run_completed)
-    # --AuditReporter#run_completed
-    #   posts the audit data to server.
-    #   (has its own tests, so stubbing it here.)
-    expect_any_instance_of(Chef::Audit::AuditReporter).to receive(:run_completed)
-  end
-end
-
-shared_context "run failed" do
-  def stub_for_run
-    expect(client).to receive(:run_failed)
-
-    # --ResourceReporter#run_completed
-    #   updates the server with the resource history
-    #   (has its own tests, so stubbing it here.)
-    expect_any_instance_of(Chef::ResourceReporter).to receive(:run_failed)
-    # --AuditReporter#run_completed
-    #   posts the audit data to server.
-    #   (has its own tests, so stubbing it here.)
-    expect_any_instance_of(Chef::Audit::AuditReporter).to receive(:run_failed)
-  end
-
-  before do
-    expect(Chef::Application).to receive(:debug_stacktrace).with an_instance_of(Chef::Exceptions::RunFailedWrappingError)
-  end
-end
diff --git a/spec/support/shared/examples/client.rb b/spec/support/shared/examples/client.rb
deleted file mode 100644
index 6479c9d582..0000000000
--- a/spec/support/shared/examples/client.rb
+++ /dev/null
@@ -1,104 +0,0 @@
-
-require "spec_helper"
-require "spec/support/shared/context/client"
-
-# requires platform and platform_version be defined
-shared_examples "a completed run" do
-  include_context "run completed"
-
-  it "runs ohai, sets up authentication, loads node state, synchronizes policy, converges, and runs audits" do
-    # This is what we're testing.
-    expect(client.run).to be true
-
-    # fork is stubbed, so we can see the outcome of the run
-    expect(node.automatic_attrs[:platform]).to eq(platform)
-    expect(node.automatic_attrs[:platform_version]).to eq(platform_version)
-  end
-
-  describe "setting node GUID" do
-    let(:chef_guid_path) { "/tmp/chef_guid" }
-    let(:chef_guid) { "test-test-test" }
-    let(:metadata_file) { "data_collector_metadata.json" }
-    let(:metadata_path) { Pathname.new(File.join(Chef::Config[:file_cache_path], metadata_file)).cleanpath.to_s }
-    let(:file) { instance_double(File) }
-
-    before do
-      allow(File).to receive(:read).and_call_original
-      Chef::Config[:chef_guid_path] = chef_guid_path
-      Chef::Config[:chef_guid] = nil
-    end
-
-    it "loads from the config" do
-      expect(File).to receive(:exists?).with(chef_guid_path).and_return(true)
-      expect(File).to receive(:read).with(chef_guid_path).and_return(chef_guid)
-      client.run
-      expect(Chef::Config[:chef_guid]).to eql(chef_guid)
-      expect(node.automatic_attrs[:chef_guid]).to eql(chef_guid)
-    end
-
-    it "loads from the data collector config" do
-      expect(File).to receive(:exists?).with(chef_guid_path).and_return(false)
-      expect(Chef::FileCache).to receive(:load).with(metadata_file).and_return("{\"node_uuid\": \"#{chef_guid}\"}")
-
-      expect(File).to receive(:open).with(chef_guid_path, "w+").and_yield(file)
-      expect(file).to receive(:write).with(chef_guid)
-
-      client.run
-      expect(Chef::Config[:chef_guid]).to eql(chef_guid)
-      expect(node.automatic_attrs[:chef_guid]).to eql(chef_guid)
-    end
-
-    it "creates a new one" do
-      expect(File).to receive(:exists?).with(chef_guid_path).and_return(false)
-      expect(File).to receive(:exists?).with(metadata_path).and_return(false)
-
-      expect(SecureRandom).to receive(:uuid).and_return(chef_guid).at_least(:once)
-
-      # we'll try and write the generated UUID to the data collector too, and that's ok
-      allow(File).to receive(:open).with(metadata_path, "w", 420)
-
-      expect(File).to receive(:open).with(chef_guid_path, "w+").and_yield(file)
-      expect(file).to receive(:write).with(chef_guid)
-
-      client.run
-      expect(Chef::Config[:chef_guid]).to eql(chef_guid)
-      expect(node.automatic_attrs[:chef_guid]).to eql(chef_guid)
-    end
-  end
-end
-
-shared_examples "a completed run with audit failure" do
-  include_context "run completed"
-
-  before do
-    expect(Chef::Application).to receive(:debug_stacktrace).with an_instance_of(Chef::Exceptions::RunFailedWrappingError)
-  end
-
-  it "converges, runs audits, saves the node and raises the error in a wrapping error" do
-    expect { client.run }.to raise_error(Chef::Exceptions::RunFailedWrappingError) do |error|
-      expect(error.wrapped_errors.size).to eq(run_errors.size)
-      run_errors.each do |run_error|
-        expect(error.wrapped_errors).to include(run_error)
-        expect(error.backtrace).to include(*run_error.backtrace)
-      end
-    end
-
-    # fork is stubbed, so we can see the outcome of the run
-    expect(node.automatic_attrs[:platform]).to eq(platform)
-    expect(node.automatic_attrs[:platform_version]).to eq(platform_version)
-  end
-end
-
-shared_examples "a failed run" do
-  include_context "run failed"
-
-  it "skips node save and raises the error in a wrapping error" do
-    expect { client.run }.to raise_error(Chef::Exceptions::RunFailedWrappingError) do |error|
-      expect(error.wrapped_errors.size).to eq(run_errors.size)
-      run_errors.each do |run_error|
-        expect(error.wrapped_errors).to include(run_error)
-        expect(error.backtrace).to include(*run_error.backtrace)
-      end
-    end
-  end
-end
diff --git a/spec/unit/action_collection_spec.rb b/spec/unit/action_collection_spec.rb
new file mode 100644
index 0000000000..def3d5d7b4
--- /dev/null
+++ b/spec/unit/action_collection_spec.rb
@@ -0,0 +1,19 @@
+#
+# Copyright:: Copyright 2019-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.
+#
+
+# all the testing done on the action_collection presently is done in the
+# resource reporter and data collector tests
diff --git a/spec/unit/client_spec.rb b/spec/unit/client_spec.rb
index 54bd7eab55..370d5d34e7 100644
--- a/spec/unit/client_spec.rb
+++ b/spec/unit/client_spec.rb
@@ -2,7 +2,7 @@
 # Author:: Adam Jacob (<adam@chef.io>)
 # Author:: Tim Hinderliter (<tim@chef.io>)
 # Author:: Christopher Walters (<cw@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");
@@ -19,9 +19,6 @@
 #
 
 require "spec_helper"
-require "spec/support/shared/context/client"
-require "spec/support/shared/examples/client"
-
 require "chef/run_context"
 require "chef/server_api"
 require "rbconfig"
@@ -29,6 +26,344 @@ require "rbconfig"
 class FooError < RuntimeError
 end
 
+#
+# SHARED CONTEXTS
+#
+
+# Stubs a basic client object
+shared_context "client" do
+  let(:fqdn)             { "hostname.example.org" }
+  let(:hostname)         { "hostname" }
+  let(:machinename)      { "machinename.example.org" }
+  let(:platform)         { "example-platform" }
+  let(:platform_version) { "example-platform-1.0" }
+
+  let(:ohai_data) do
+    {
+      fqdn: fqdn,
+      hostname: hostname,
+      machinename: machinename,
+      platform: platform,
+      platform_version: platform_version,
+    }
+  end
+
+  let(:ohai_system) do
+    ohai = instance_double("Ohai::System", all_plugins: true, data: ohai_data, logger: logger)
+    allow(ohai).to receive(:[]) do |k|
+      ohai_data[k]
+    end
+    ohai
+  end
+
+  let(:node) do
+    Chef::Node.new.tap do |n|
+      n.name(fqdn)
+      n.chef_environment("_default")
+    end
+  end
+
+  let(:json_attribs) { nil }
+  let(:client_opts) { {} }
+
+  let(:stdout) { STDOUT }
+  let(:stderr) { STDERR }
+
+  let(:client) do
+    Chef::Config[:event_loggers] = []
+    allow(Ohai::System).to receive(:new).and_return(ohai_system)
+    opts = client_opts.merge({ logger: logger })
+    Chef::Client.new(json_attribs, opts).tap do |c|
+      c.node = node
+    end
+  end
+
+  let(:logger) { instance_double("Mixlib::Log::Child", trace: nil, debug: nil, warn: nil, info: nil, error: nil, fatal: nil) }
+
+  before do
+    stub_const("Chef::Client::STDOUT_FD", stdout)
+    stub_const("Chef::Client::STDERR_FD", stderr)
+    allow(client).to receive(:logger).and_return(logger)
+  end
+end
+
+# Stubs a client for a client run.
+# Requires a client object be defined in the scope of this included context.
+# e.g.:
+#   describe "some functionality" do
+#     include_context "client"
+#     include_context "a client run"
+#     ...
+#   end
+shared_context "a client run" do
+  let(:stdout) { StringIO.new }
+  let(:stderr) { StringIO.new }
+
+  let(:api_client_exists?) { false }
+  let(:enable_fork)        { false }
+
+  let(:http_data_collector)   { double("Chef::ServerAPI (data collector)") }
+  let(:http_cookbook_sync)    { double("Chef::ServerAPI (cookbook sync)") }
+  let(:http_node_load)        { double("Chef::ServerAPI (node)") }
+  let(:http_node_save)        { double("Chef::ServerAPI (node save)") }
+  let(:reporting_rest_client) { double("Chef::ServerAPI (reporting client)") }
+
+  let(:runner)       { instance_double("Chef::Runner") }
+  let(:audit_runner) { instance_double("Chef::Audit::Runner", failed?: false) }
+
+  def stub_for_register
+    # --Client.register
+    #   Make sure Client#register thinks the client key doesn't
+    #   exist, so it tries to register and create one.
+    allow(File).to receive(:exists?).and_call_original
+    expect(File).to receive(:exists?)
+      .with(Chef::Config[:client_key])
+      .exactly(:once)
+      .and_return(api_client_exists?)
+
+    unless api_client_exists?
+      #   Client.register will register with the validation client name.
+      expect_any_instance_of(Chef::ApiClient::Registration).to receive(:run)
+    end
+  end
+
+  def stub_for_event_registration
+    expect(client.events).to receive(:register).with(instance_of(Chef::DataCollector::Reporter))
+    expect(client.events).to receive(:register).with(instance_of(Chef::ResourceReporter))
+    expect(client.events).to receive(:register).with(instance_of(Chef::ActionCollection))
+    expect(client.events).to receive(:register).with(instance_of(Chef::Audit::AuditReporter))
+  end
+
+  def stub_for_node_load
+    #   Client.register will then turn around create another
+    #   Chef::ServerAPI object, this time with the client key it got from the
+    #   previous step.
+    expect(Chef::ServerAPI).to receive(:new)
+      .with(Chef::Config[:chef_server_url], client_name: fqdn,
+                                           signing_key_filename: Chef::Config[:client_key])
+      .exactly(:once)
+      .and_return(http_node_load)
+
+    # --Client#build_node
+    #   looks up the node, which we will return, then later saves it.
+    expect(Chef::Node).to receive(:find_or_create).with(fqdn).and_return(node)
+  end
+
+  def stub_rest_clean
+    allow(client).to receive(:rest_clean).and_return(reporting_rest_client)
+  end
+
+  def stub_for_sync_cookbooks
+    # --Client#setup_run_context
+    # ---Client#sync_cookbooks -- downloads the list of cookbooks to sync
+    #
+    expect_any_instance_of(Chef::CookbookSynchronizer).to receive(:sync_cookbooks)
+    expect(Chef::ServerAPI).to receive(:new).with(Chef::Config[:chef_server_url], version_class: Chef::CookbookManifestVersions).and_return(http_cookbook_sync)
+    expect(http_cookbook_sync).to receive(:post)
+      .with("environments/_default/cookbook_versions", { run_list: [] })
+      .and_return({})
+  end
+
+  def stub_for_required_recipe
+    response = Net::HTTPNotFound.new("1.1", "404", "Not Found")
+    exception = Net::HTTPClientException.new('404 "Not Found"', response)
+    expect(http_node_load).to receive(:get).with("required_recipe").and_raise(exception)
+  end
+
+  def stub_for_converge
+    # define me
+  end
+
+  def stub_for_audit
+    # define me
+  end
+
+  def stub_for_node_save
+    # define me
+  end
+
+  def stub_for_run
+    # define me
+  end
+
+  before do
+    Chef::Config[:client_fork] = enable_fork
+    Chef::Config[:cache_path] = windows? ? 'C:\chef' : "/var/chef"
+    Chef::Config[:why_run] = false
+    Chef::Config[:audit_mode] = :enabled
+    Chef::Config[:chef_guid] = "default-guid"
+
+    stub_rest_clean
+    stub_for_register
+    stub_for_event_registration
+    stub_for_node_load
+    stub_for_sync_cookbooks
+    stub_for_required_recipe
+    stub_for_converge
+    stub_for_audit
+    stub_for_node_save
+
+    expect_any_instance_of(Chef::RunLock).to receive(:acquire)
+    expect_any_instance_of(Chef::RunLock).to receive(:save_pid)
+    expect_any_instance_of(Chef::RunLock).to receive(:release)
+
+    # Post conditions: check that node has been filled in correctly
+    expect(client).to receive(:run_started)
+
+    stub_for_run
+  end
+end
+
+shared_context "converge completed" do
+  def stub_for_converge
+    # --Client#converge
+    expect(Chef::Runner).to receive(:new).and_return(runner)
+    expect(runner).to receive(:converge).and_return(true)
+  end
+
+  def stub_for_node_save
+    allow(node).to receive(:data_for_save).and_return(node.for_json)
+
+    # --Client#save_updated_node
+    expect(Chef::ServerAPI).to receive(:new).with(Chef::Config[:chef_server_url], client_name: fqdn,
+                                                                                  signing_key_filename: Chef::Config[:client_key], validate_utf8: false).and_return(http_node_save)
+    expect(http_node_save).to receive(:put).with("nodes/#{fqdn}", node.for_json).and_return(true)
+  end
+end
+
+shared_context "converge failed" do
+  let(:converge_error) do
+    err = Chef::Exceptions::UnsupportedAction.new("Action unsupported")
+    err.set_backtrace([ "/path/recipe.rb:15", "/path/recipe.rb:12" ])
+    err
+  end
+
+  def stub_for_converge
+    expect(Chef::Runner).to receive(:new).and_return(runner)
+    expect(runner).to receive(:converge).and_raise(converge_error)
+  end
+
+  def stub_for_node_save
+    expect(client).to_not receive(:save_updated_node)
+  end
+end
+
+shared_context "audit phase completed" do
+  def stub_for_audit
+    # -- Client#run_audits
+    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
+    expect(audit_runner).to receive(:run).and_return(true)
+    expect(client.events).to receive(:audit_phase_complete)
+  end
+end
+
+shared_context "audit phase failed with error" do
+  let(:audit_error) do
+    err = RuntimeError.new("Unexpected audit error")
+    err.set_backtrace([ "/path/recipe.rb:57", "/path/recipe.rb:55" ])
+    err
+  end
+
+  def stub_for_audit
+    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
+    expect(Chef::Audit::Logger).to receive(:read_buffer).and_return("Audit mode output!")
+    expect(audit_runner).to receive(:run).and_raise(audit_error)
+    expect(client.events).to receive(:audit_phase_failed).with(audit_error, "Audit mode output!")
+  end
+end
+
+shared_context "audit phase completed with failed controls" do
+  let(:audit_runner) do
+    instance_double("Chef::Audit::Runner", failed?: true,
+                                           num_failed: 1, num_total: 3) end
+
+  let(:audit_error) do
+    err = Chef::Exceptions::AuditsFailed.new(audit_runner.num_failed, audit_runner.num_total)
+    err.set_backtrace([ "/path/recipe.rb:108", "/path/recipe.rb:103" ])
+    err
+  end
+
+  def stub_for_audit
+    expect(Chef::Audit::Runner).to receive(:new).and_return(audit_runner)
+    expect(Chef::Audit::Logger).to receive(:read_buffer).and_return("Audit mode output!")
+    expect(audit_runner).to receive(:run)
+    expect(Chef::Exceptions::AuditsFailed).to receive(:new).with(
+      audit_runner.num_failed, audit_runner.num_total
+    ).and_return(audit_error)
+    expect(client.events).to receive(:audit_phase_failed).with(audit_error, "Audit mode output!")
+  end
+end
+
+shared_context "run completed" do
+  def stub_for_run
+    expect(client).to receive(:run_completed_successfully)
+  end
+end
+
+shared_context "run failed" do
+  def stub_for_run
+    expect(client).to receive(:run_failed)
+  end
+
+  before do
+    expect(Chef::Application).to receive(:debug_stacktrace).with an_instance_of(Chef::Exceptions::RunFailedWrappingError)
+  end
+end
+
+#
+# SHARED EXAMPLES
+#
+
+# requires platform and platform_version be defined
+shared_examples "a completed run" do
+  include_context "run completed"
+
+  it "runs ohai, sets up authentication, loads node state, synchronizes policy, converges, and runs audits" do
+    # This is what we're testing.
+    expect(client.run).to be true
+
+    # fork is stubbed, so we can see the outcome of the run
+    expect(node.automatic_attrs[:platform]).to eq(platform)
+    expect(node.automatic_attrs[:platform_version]).to eq(platform_version)
+  end
+end
+
+shared_examples "a completed run with audit failure" do
+  include_context "run completed"
+
+  before do
+    expect(Chef::Application).to receive(:debug_stacktrace).with an_instance_of(Chef::Exceptions::RunFailedWrappingError)
+  end
+
+  it "converges, runs audits, saves the node and raises the error in a wrapping error" do
+    expect { client.run }.to raise_error(Chef::Exceptions::RunFailedWrappingError) do |error|
+      expect(error.wrapped_errors.size).to eq(run_errors.size)
+      run_errors.each do |run_error|
+        expect(error.wrapped_errors).to include(run_error)
+        expect(error.backtrace).to include(*run_error.backtrace)
+      end
+    end
+
+    # fork is stubbed, so we can see the outcome of the run
+    expect(node.automatic_attrs[:platform]).to eq(platform)
+    expect(node.automatic_attrs[:platform_version]).to eq(platform_version)
+  end
+end
+
+shared_examples "a failed run" do
+  include_context "run failed"
+
+  it "skips node save and raises the error in a wrapping error" do
+    expect { client.run }.to raise_error(Chef::Exceptions::RunFailedWrappingError) do |error|
+      expect(error.wrapped_errors.size).to eq(run_errors.size)
+      run_errors.each do |run_error|
+        expect(error.wrapped_errors).to include(run_error)
+        expect(error.backtrace).to include(*run_error.backtrace)
+      end
+    end
+  end
+end
+
 describe Chef::Client do
   include_context "client"
 
diff --git a/spec/unit/data_collector/messages/helpers_spec.rb b/spec/unit/data_collector/messages/helpers_spec.rb
deleted file mode 100644
index a2c6753003..0000000000
--- a/spec/unit/data_collector/messages/helpers_spec.rb
+++ /dev/null
@@ -1,202 +0,0 @@
-#
-# Author:: Adam Leff (<adamleff@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 "spec_helper"
-require "chef/data_collector/messages/helpers"
-
-class TestMessage
-  extend Chef::DataCollector::Messages::Helpers
-end
-
-describe Chef::DataCollector::Messages::Helpers do
-  describe "#organization" do
-    context "when the run is a solo run" do
-      it "returns the data collector organization" do
-        allow(TestMessage).to receive(:solo_run?).and_return(true)
-        expect(TestMessage).to receive(:data_collector_organization).and_return("org1")
-        expect(TestMessage.organization).to eq("org1")
-      end
-    end
-
-    context "when the run is not a solo run" do
-      it "returns the data collector organization" do
-        allow(TestMessage).to receive(:solo_run?).and_return(false)
-        expect(TestMessage).to receive(:chef_server_organization).and_return("org2")
-        expect(TestMessage.organization).to eq("org2")
-      end
-    end
-  end
-
-  describe "#data_collector_organization" do
-    context "when the org is specified in the config" do
-      it "returns the org from the config" do
-        Chef::Config[:data_collector][:organization] = "org1"
-        expect(TestMessage.data_collector_organization).to eq("org1")
-      end
-    end
-
-    context "when the org is not specified in the config" do
-      it "returns the default chef_solo org" do
-        expect(TestMessage.data_collector_organization).to eq("chef_solo")
-      end
-    end
-  end
-
-  describe "#chef_server_organization" do
-    context "when the URL is properly formatted" do
-      it "returns the org from the parsed URL" do
-        Chef::Config[:chef_server_url] = "http://mycompany.com/organizations/myorg"
-        expect(TestMessage.chef_server_organization).to eq("myorg")
-      end
-    end
-
-    context "when the URL is not properly formatted" do
-      it "returns unknown_organization" do
-        Chef::Config[:chef_server_url] = "http://mycompany.com/what/url/is/this"
-        expect(TestMessage.chef_server_organization).to eq("unknown_organization")
-      end
-    end
-
-    context "when the organization in the URL contains hyphens" do
-      it "returns the full org name" do
-        Chef::Config[:chef_server_url] = "http://mycompany.com/organizations/myorg-test"
-        expect(TestMessage.chef_server_organization).to eq("myorg-test")
-      end
-    end
-  end
-
-  describe "#collector_source" do
-    context "when the run is a solo run" do
-      it "returns chef_solo" do
-        allow(TestMessage).to receive(:solo_run?).and_return(true)
-        expect(TestMessage.collector_source).to eq("chef_solo")
-      end
-    end
-
-    context "when the run is not a solo run" do
-      it "returns chef_client" do
-        allow(TestMessage).to receive(:solo_run?).and_return(false)
-        expect(TestMessage.collector_source).to eq("chef_client")
-      end
-    end
-  end
-
-  describe "#solo_run?" do
-    context "when :solo is set in Chef::Config" do
-      it "returns true" do
-        Chef::Config[:solo] = true
-        Chef::Config[:local_mode] = nil
-        expect(TestMessage.solo_run?).to be_truthy
-      end
-    end
-
-    context "when :local_mode is set in Chef::Config" do
-      it "returns true" do
-        Chef::Config[:solo] = nil
-        Chef::Config[:local_mode] = true
-        expect(TestMessage.solo_run?).to be_truthy
-      end
-    end
-
-    context "when neither :solo or :local_mode is set in Chef::Config" do
-      it "returns false" do
-        Chef::Config[:solo] = nil
-        Chef::Config[:local_mode] = nil
-        expect(TestMessage.solo_run?).to be_falsey
-      end
-    end
-  end
-
-  describe "#node_uuid" do
-    context "when the node UUID is available in Chef::Config" do
-      it "returns the configured value" do
-        Chef::Config[:chef_guid] = "configured_uuid"
-        expect(TestMessage.node_uuid).to eq("configured_uuid")
-      end
-    end
-
-    context "when the node UUID can be read" do
-      it "returns the read-in node UUID" do
-        Chef::Config[:chef_guid] = nil
-        allow(TestMessage).to receive(:read_node_uuid).and_return("read_uuid")
-        expect(TestMessage.node_uuid).to eq("read_uuid")
-      end
-    end
-
-    context "when the node UUID cannot be read" do
-      it "generated a new node UUID" do
-        Chef::Config[:chef_guid] = nil
-        allow(TestMessage).to receive(:read_node_uuid).and_return(nil)
-        allow(TestMessage).to receive(:generate_node_uuid).and_return("generated_uuid")
-        expect(TestMessage.node_uuid).to eq("generated_uuid")
-      end
-    end
-  end
-
-  describe "#generate_node_uuid" do
-    it "generates a new UUID, stores it, and returns it" do
-      expect(SecureRandom).to receive(:uuid).and_return("generated_uuid")
-      expect(TestMessage).to receive(:update_metadata).with("node_uuid", "generated_uuid")
-      expect(TestMessage.generate_node_uuid).to eq("generated_uuid")
-    end
-  end
-
-  describe "#read_node_uuid" do
-    it "reads the node UUID from metadata" do
-      expect(TestMessage).to receive(:metadata).and_return({ "node_uuid" => "read_uuid" })
-      expect(TestMessage.read_node_uuid).to eq("read_uuid")
-    end
-  end
-
-  describe "metadata" do
-    let(:metadata_filename) { "fake_metadata_file.json" }
-
-    before do
-      allow(TestMessage).to receive(:metadata_filename).and_return(metadata_filename)
-    end
-
-    context "when the metadata file exists" do
-      it "returns the contents of the metadata file" do
-        expect(Chef::FileCache).to receive(:load).with(metadata_filename).and_return('{"foo":"bar"}')
-        expect(TestMessage.metadata["foo"]).to eq("bar")
-      end
-    end
-
-    context "when the metadata file does not exist" do
-      it "returns an empty hash" do
-        expect(Chef::FileCache).to receive(:load).with(metadata_filename).and_raise(Chef::Exceptions::FileNotFound)
-        expect(TestMessage.metadata).to eq({})
-      end
-    end
-  end
-
-  describe "#update_metadata" do
-    it "updates the file" do
-      allow(TestMessage).to receive(:metadata_filename).and_return("fake_metadata_file.json")
-      allow(TestMessage).to receive(:metadata).and_return({ "key" => "current_value" })
-      expect(Chef::FileCache).to receive(:store).with(
-        "fake_metadata_file.json",
-        '{"key":"updated_value"}',
-        0644
-      )
-
-      TestMessage.update_metadata("key", "updated_value")
-    end
-  end
-end
diff --git a/spec/unit/data_collector/messages_spec.rb b/spec/unit/data_collector/messages_spec.rb
deleted file mode 100644
index f5df85a988..0000000000
--- a/spec/unit/data_collector/messages_spec.rb
+++ /dev/null
@@ -1,329 +0,0 @@
-#
-# Author:: Adam Leff (<adamleff@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 "spec_helper"
-require "ffi_yajl"
-require "chef/data_collector/messages/helpers"
-
-describe Chef::DataCollector::Messages do
-  describe "#run_start_message" do
-    let(:run_status) { Chef::RunStatus.new(Chef::Node.new, Chef::EventDispatch::Dispatcher.new) }
-    let(:required_fields) do
-      %w{
-        chef_server_fqdn
-        entity_uuid
-        id
-        message_version
-        message_type
-        node_name
-        organization_name
-        run_id
-        source
-        start_time
-      }
-    end
-    let(:optional_fields) { [] }
-
-    before do
-      allow(run_status).to receive(:start_time).and_return(Time.now)
-    end
-
-    it "is not missing any required fields" do
-      missing_fields = required_fields.select do |key|
-        !Chef::DataCollector::Messages.run_start_message(run_status).key?(key)
-      end
-
-      expect(missing_fields).to eq([])
-    end
-
-    it "does not have any extra fields" do
-      extra_fields = Chef::DataCollector::Messages.run_start_message(run_status).keys.select do |key|
-        !required_fields.include?(key) && !optional_fields.include?(key)
-      end
-
-      expect(extra_fields).to eq([])
-    end
-  end
-
-  describe "#run_end_message" do
-    let(:node)       { Chef::Node.new }
-    let(:run_status) { Chef::RunStatus.new(node, Chef::EventDispatch::Dispatcher.new) }
-    let(:report1)  { double("report1", report_data: { "status" => "updated" }) }
-    let(:report2)  { double("report2", report_data: { "status" => "skipped" }) }
-    let(:reporter_data) do
-      {
-        run_status: run_status,
-        resources: [report1, report2],
-      }
-    end
-
-    before do
-      allow(run_status).to receive(:start_time).and_return(Time.now)
-      allow(run_status).to receive(:end_time).and_return(Time.now)
-    end
-
-    it "includes a valid node object in the payload" do
-      message = Chef::DataCollector::Messages.run_end_message(reporter_data)
-      expect(message["node"]).to be_an_instance_of(Chef::Node)
-    end
-
-    it "returns a sane JSON representation of the node object" do
-      node.chef_environment = "my_test_environment"
-      node.run_list.add("recipe[my_test_cookbook::default]")
-      message = FFI_Yajl::Parser.parse(Chef::DataCollector::Messages.run_end_message(reporter_data).to_json)
-
-      expect(message["node"]["chef_environment"]).to eq("my_test_environment")
-      expect(message["node"]["run_list"]).to eq(["recipe[my_test_cookbook::default]"])
-    end
-
-    context "when the run was successful" do
-      let(:required_fields) do
-        %w{
-          chef_server_fqdn
-          entity_uuid
-          id
-          end_time
-          expanded_run_list
-          message_type
-          message_version
-          node
-          node_name
-          organization_name
-          resources
-          run_id
-          run_list
-          source
-          start_time
-          status
-          total_resource_count
-          updated_resource_count
-          deprecations
-        }
-      end
-      let(:optional_fields) { %w{error policy_group policy_name} }
-
-      before do
-        allow(run_status).to receive(:exception).and_return(nil)
-      end
-
-      it "is not missing any required fields" do
-        missing_fields = required_fields.select do |key|
-          !Chef::DataCollector::Messages.run_end_message(reporter_data).key?(key)
-        end
-        expect(missing_fields).to eq([])
-      end
-
-      it "does not have any extra fields" do
-        extra_fields = Chef::DataCollector::Messages.run_end_message(reporter_data).keys.select do |key|
-          !required_fields.include?(key) && !optional_fields.include?(key)
-        end
-        expect(extra_fields).to eq([])
-      end
-
-      it "only includes updated resources in its count" do
-        message = Chef::DataCollector::Messages.run_end_message(reporter_data)
-        expect(message["total_resource_count"]).to eq(2)
-        expect(message["updated_resource_count"]).to eq(1)
-      end
-    end
-
-    context "when the run was not successful" do
-      let(:required_fields) do
-        %w{
-          chef_server_fqdn
-          entity_uuid
-          id
-          end_time
-          error
-          expanded_run_list
-          message_type
-          message_version
-          node
-          node_name
-          organization_name
-          resources
-          run_id
-          run_list
-          source
-          start_time
-          status
-          total_resource_count
-          updated_resource_count
-          deprecations
-        }
-      end
-      let(:optional_fields) { %w{policy_group policy_name} }
-
-      before do
-        allow(run_status).to receive(:exception).and_return(RuntimeError.new("an error happened"))
-      end
-
-      it "is not missing any required fields" do
-        missing_fields = required_fields.select do |key|
-          !Chef::DataCollector::Messages.run_end_message(reporter_data).key?(key)
-        end
-        expect(missing_fields).to eq([])
-      end
-
-      it "does not have any extra fields" do
-        extra_fields = Chef::DataCollector::Messages.run_end_message(reporter_data).keys.select do |key|
-          !required_fields.include?(key) && !optional_fields.include?(key)
-        end
-        expect(extra_fields).to eq([])
-      end
-    end
-  end
-
-  describe "#run_end_message in policy mode" do
-    let(:node) { Chef::Node.new }
-    let(:run_status) { Chef::RunStatus.new(node, Chef::EventDispatch::Dispatcher.new) }
-    let(:report1)  { double("report1", report_data: { "status" => "updated" }) }
-    let(:report2)  { double("report2", report_data: { "status" => "skipped" }) }
-    let(:reporter_data) do
-      {
-        run_status: run_status,
-        resources: [report1, report2],
-      }
-    end
-
-    before do
-      allow(run_status).to receive(:start_time).and_return(Time.now)
-      allow(run_status).to receive(:end_time).and_return(Time.now)
-      node.policy_group = "test"
-      node.policy_name = "policy-test"
-    end
-
-    it "includes a valid node object in the payload" do
-      message = Chef::DataCollector::Messages.run_end_message(reporter_data)
-      expect(message["node"]).to be_an_instance_of(Chef::Node)
-    end
-
-    it "returns a sane JSON representation of the node object" do
-      node.chef_environment = "my_test_environment"
-      node.run_list.add("recipe[my_test_cookbook::default]")
-      message = FFI_Yajl::Parser.parse(Chef::DataCollector::Messages.run_end_message(reporter_data).to_json)
-
-      expect(message["node"]["chef_environment"]).to eq("my_test_environment")
-      expect(message["node"]["run_list"]).to eq(["recipe[my_test_cookbook::default]"])
-      expect(message["node"]["policy_name"]).to eq("policy-test")
-      expect(message["node"]["policy_group"]).to eq("test")
-    end
-
-    context "when the run was successful" do
-      let(:required_fields) do
-        %w{
-          chef_server_fqdn
-          entity_uuid
-          id
-          end_time
-          expanded_run_list
-          message_type
-          message_version
-          node
-          node_name
-          organization_name
-          resources
-          run_id
-          run_list
-          source
-          start_time
-          status
-          total_resource_count
-          updated_resource_count
-          deprecations
-          policy_name
-          policy_group
-        }
-      end
-      let(:optional_fields) { %w{error} }
-
-      before do
-        allow(run_status).to receive(:exception).and_return(nil)
-      end
-
-      it "is not missing any required fields" do
-        missing_fields = required_fields.select do |key|
-          !Chef::DataCollector::Messages.run_end_message(reporter_data).key?(key)
-        end
-        expect(missing_fields).to eq([])
-      end
-
-      it "does not have any extra fields" do
-        extra_fields = Chef::DataCollector::Messages.run_end_message(reporter_data).keys.select do |key|
-          !required_fields.include?(key) && !optional_fields.include?(key)
-        end
-        expect(extra_fields).to eq([])
-      end
-
-      it "only includes updated resources in its count" do
-        message = Chef::DataCollector::Messages.run_end_message(reporter_data)
-        expect(message["total_resource_count"]).to eq(2)
-        expect(message["updated_resource_count"]).to eq(1)
-      end
-    end
-
-    context "when the run was not successful" do
-      let(:required_fields) do
-        %w{
-          chef_server_fqdn
-          entity_uuid
-          id
-          end_time
-          error
-          expanded_run_list
-          message_type
-          message_version
-          node
-          node_name
-          organization_name
-          resources
-          run_id
-          run_list
-          source
-          start_time
-          status
-          total_resource_count
-          updated_resource_count
-          deprecations
-          policy_name
-          policy_group
-        }
-      end
-      let(:optional_fields) { [] }
-
-      before do
-        allow(run_status).to receive(:exception).and_return(RuntimeError.new("an error happened"))
-      end
-
-      it "is not missing any required fields" do
-        missing_fields = required_fields.select do |key|
-          !Chef::DataCollector::Messages.run_end_message(reporter_data).key?(key)
-        end
-        expect(missing_fields).to eq([])
-      end
-
-      it "does not have any extra fields" do
-        extra_fields = Chef::DataCollector::Messages.run_end_message(reporter_data).keys.select do |key|
-          !required_fields.include?(key) && !optional_fields.include?(key)
-        end
-        expect(extra_fields).to eq([])
-      end
-    end
-  end
-end
diff --git a/spec/unit/data_collector/resource_report_spec.rb b/spec/unit/data_collector/resource_report_spec.rb
deleted file mode 100644
index 1a5eab796e..0000000000
--- a/spec/unit/data_collector/resource_report_spec.rb
+++ /dev/null
@@ -1,145 +0,0 @@
-#
-# Author:: Salim Afiune (<afiune@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 "spec_helper"
-
-describe Chef::DataCollector::ResourceReport do
-  let(:cookbook_repo_path) { File.join(CHEF_SPEC_DATA, "cookbooks") }
-  let(:cookbook_collection) { Chef::CookbookCollection.new(Chef::CookbookLoader.new(cookbook_repo_path)) }
-  let(:node) { Chef::Node.new }
-  let(:events) { Chef::EventDispatch::Dispatcher.new }
-  let(:run_context) { Chef::RunContext.new(node, cookbook_collection, events) }
-  let(:resource) { Chef::Resource.new("zelda", run_context) }
-  let(:report) { described_class.new(resource, :create) }
-
-  describe "#skipped" do
-    let(:conditional) { double("Chef::Resource::Conditional") }
-
-    it "should set status and conditional" do
-      report.skipped(conditional)
-      expect(report.conditional).to eq conditional
-      expect(report.status).to eq "skipped"
-    end
-  end
-
-  describe "#up_to_date" do
-    it "should set status" do
-      report.up_to_date
-      expect(report.status).to eq "up-to-date"
-    end
-  end
-
-  describe "#updated" do
-    it "should set status" do
-      report.updated
-      expect(report.status).to eq "updated"
-    end
-  end
-
-  describe "#elapsed_time_in_milliseconds" do
-
-    context "when elapsed_time is not set" do
-      it "should return nil" do
-        allow(report).to receive(:elapsed_time).and_return(nil)
-        expect(report.elapsed_time_in_milliseconds).to eq nil
-      end
-    end
-
-    context "when elapsed_time is set" do
-      it "should return it in milliseconds" do
-        allow(report).to receive(:elapsed_time).and_return(1)
-        expect(report.elapsed_time_in_milliseconds).to eq 1000
-      end
-    end
-  end
-
-  describe "#failed" do
-    let(:exception) { double("Chef::Exception::Test") }
-
-    it "should set exception and status" do
-      report.failed(exception)
-      expect(report.exception).to eq exception
-      expect(report.status).to eq "failed"
-    end
-  end
-
-  describe "#to_hash" do
-    context "for a simple_resource" do
-      let(:resource) do
-        klass = Class.new(Chef::Resource) do
-          resource_name "zelda"
-        end
-        klass.new("hyrule", run_context)
-      end
-      let(:hash) do
-        {
-          "after" => {},
-          "before" => {},
-          "delta" => "",
-          "duration" => "",
-          "id" => "hyrule",
-          "ignore_failure" => false,
-          "name" => "hyrule",
-          "result" => "create",
-          "status" => "unprocessed",
-          "type" => :zelda,
-        }
-      end
-
-      it "returns a hash containing the expected values" do
-        expect(report.to_hash).to eq hash
-      end
-    end
-
-    context "for a lazy_resource that got skipped" do
-      let(:resource) do
-        klass = Class.new(Chef::Resource) do
-          resource_name "butters"
-          property :sword, String, name_property: true, identity: true
-        end
-        resource = klass.new("hyrule")
-        resource.sword = Chef::DelayedEvaluator.new { nil }
-        resource
-      end
-      let(:hash) do
-        {
-          "after" => {},
-          "before" => {},
-          "delta" => "",
-          "duration" => "",
-          "conditional" => "because",
-          "id" => "unknown identity (due to Chef::Exceptions::ValidationFailed)",
-          "ignore_failure" => false,
-          "name" => "hyrule",
-          "result" => "create",
-          "status" => "skipped",
-          "type" => :butters,
-        }
-      end
-      let(:conditional) do
-        double("Chef::Resource::Conditional", to_text: "because")
-      end
-
-      it "should handle any Exception and throw a helpful message by mocking the identity" do
-        report.skipped(conditional)
-        expect(report.to_hash).to eq hash
-      end
-    end
-  end
-end
diff --git a/spec/unit/data_collector_spec.rb b/spec/unit/data_collector_spec.rb
index 87729d8652..154ab4681c 100644
--- a/spec/unit/data_collector_spec.rb
+++ b/spec/unit/data_collector_spec.rb
@@ -1,8 +1,5 @@
 #
-# Author:: Adam Leff (<adamleff@chef.io)
-# Author:: Ryan Cragun (<ryan@chef.io>)
-#
-# Copyright:: Copyright 2012-2018, Chef Software Inc.
+# Copyright:: Copyright 2019-2019, Chef Software Inc.
 # License:: Apache License, Version 2.0
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,859 +15,860 @@
 # limitations under the License.
 #
 
-require "spec_helper"
+require File.expand_path("../../spec_helper", __FILE__)
 require "chef/data_collector"
-require "chef/resource_builder"
+require "socket"
 
 describe Chef::DataCollector do
+  before(:each) do
+    Chef::Config[:enable_reporting] = true
+  end
 
-  describe ".register_reporter?" do
-    context "when no data collector URL or output locations are configured" do
-      it "returns false" do
-        Chef::Config[:data_collector][:server_url] = nil
-        Chef::Config[:data_collector][:output_locations] = nil
-        expect(Chef::DataCollector.register_reporter?).to be_falsey
-      end
-    end
-
-    context "when a data collector URL is configured" do
-      before do
-        Chef::Config[:data_collector][:server_url] = "http://data_collector"
-      end
-
-      context "when operating in why_run mode" do
-        it "returns false" do
-          Chef::Config[:why_run] = true
-          expect(Chef::DataCollector.register_reporter?).to be_falsey
-        end
-      end
-
-      context "when not operating in why_run mode" do
-
-        before do
-          Chef::Config[:why_run] = false
-          Chef::Config[:data_collector][:token] = token
-        end
-
-        context "when a token is configured" do
-
-          let(:token) { "supersecrettoken" }
+  let(:node) { Chef::Node.new }
 
-          context "when report is enabled for current mode" do
-            it "returns true" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(true)
-              expect(Chef::DataCollector.register_reporter?).to be_truthy
-            end
-          end
+  let(:rest_client) { double("Chef::ServerAPI (mock)") }
 
-          context "when report is disabled for current mode" do
-            it "returns false" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(false)
-              expect(Chef::DataCollector.register_reporter?).to be_falsey
-            end
-          end
+  let(:data_collector) { Chef::DataCollector::Reporter.new(events) }
 
-        end
+  let(:new_resource) { Chef::Resource::File.new("/tmp/a-file.txt") }
 
-        # `Chef::Config[:data_collector][:server_url]` defaults to a URL
-        # relative to the `chef_server_url`, so we use configuration of the
-        # token to infer whether a solo/local mode user intends for data
-        # collection to be enabled.
-        context "when a token is not configured" do
+  let(:current_resource) { Chef::Resource::File.new("/tmp/a-file.txt") }
 
-          let(:token) { nil }
+  let(:events) { Chef::EventDispatch::Dispatcher.new }
 
-          context "when report is enabled for current mode" do
+  let(:run_context) { Chef::RunContext.new(node, {}, events) }
 
-            before do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(true)
-            end
+  let(:run_status) { Chef::RunStatus.new(node, events) }
 
-            context "when the current mode is solo" do
+  let(:start_time) { Time.new }
 
-              before do
-                Chef::Config[:solo] = true
-              end
+  let(:end_time) { Time.new + 20 }
 
-              it "returns true" do
-                expect(Chef::DataCollector.register_reporter?).to be(true)
-              end
+  let(:run_list) { node.run_list }
 
-            end
+  let(:run_id) { run_status.run_id }
 
-            context "when the current mode is local mode" do
+  let(:expansion) { Chef::RunList::RunListExpansion.new("_default", run_list.run_list_items) }
 
-              before do
-                Chef::Config[:local_mode] = true
-              end
+  let(:cookbook_name) { "monkey" }
 
-              it "returns false" do
-                expect(Chef::DataCollector.register_reporter?).to be(true)
-              end
-            end
+  let(:recipe_name) { "atlas" }
 
-            context "when the current mode is client mode" do
+  let(:node_name) { "spitfire" }
 
-              before do
-                Chef::Config[:local_mode] = false
-                Chef::Config[:solo] = false
-              end
+  let(:cookbook_version) { double("Cookbook::Version", version: "1.2.3") }
 
-              it "returns true" do
-                expect(Chef::DataCollector.register_reporter?).to be_truthy
-              end
+  let(:resource_record) { [] }
 
-            end
+  let(:exception) { nil }
 
-          end
+  let(:action_collection) { Chef::ActionCollection.new(events) }
 
-          context "when report is disabled for current mode" do
-            it "returns false" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(false)
-              expect(Chef::DataCollector.register_reporter?).to be_falsey
-            end
-          end
+  let(:expected_node) { node }
 
-        end
-
-      end
-    end
+  let(:expected_expansion) { expansion }
 
-    context "when output_locations are configured" do
-      before do
-        Chef::Config[:data_collector][:output_locations] = ["http://data_collector", "/tmp/data_collector.json"]
-      end
+  let(:expected_run_list) { run_list.for_json }
 
-      context "when operating in why_run mode" do
-        it "returns false" do
-          Chef::Config[:why_run] = true
-          expect(Chef::DataCollector.register_reporter?).to be_falsey
-        end
-      end
+  let(:node_uuid) { "779196c6-f94f-4501-9dae-af8081ab4d3a" }
 
-      context "when not operating in why_run mode" do
+  let(:request_id) { "5db5d686-d18d-4234-a86a-28848c35dfc2" }
 
-        before do
-          Chef::Config[:why_run] = false
-          Chef::Config[:data_collector][:token] = token
-        end
+  before do
+    allow(Time).to receive(:now).and_return(start_time, end_time)
+    allow(Chef::HTTP::SimpleJSON).to receive(:new).and_return(rest_client)
+    allow(Chef::ServerAPI).to receive(:new).and_return(rest_client)
+    node.name(node_name) unless node.is_a?(Hash)
+    new_resource.cookbook_name = cookbook_name
+    new_resource.recipe_name = recipe_name
+    allow(new_resource).to receive(:cookbook_version).and_return(cookbook_version)
+
+    run_list << "recipe[lobster]" << "role[rage]" << "recipe[fist]"
+    events.register(data_collector)
+    events.register(action_collection)
+    run_status.run_id = request_id
+    events.run_start(Chef::VERSION, run_status)
+    Chef::Config[:chef_guid] = node_uuid
+    # we're guaranteed that those events are processed or else the data collector has no hope
+    # all other events could see the chef-client crash before executing them and the data collector
+    # still needs to work in those cases, so must come later, and the failure cases must be tested.
+  end
 
-        context "when a token is configured" do
+  def expect_start_message(keys = nil)
+    keys ||= {
+      "chef_server_fqdn" => "localhost",
+      "entity_uuid" => node_uuid,
+      "id" => request_id,
+      "message_type" => "run_start",
+      "message_version" => "1.0.0",
+      "node_name" => node_name,
+      "organization_name" => "unknown_organization",
+      "run_id" => request_id,
+      "source" => "chef_client",
+      "start_time" => start_time.utc.iso8601,
+    }
+    expect(rest_client).to receive(:post).with(
+      nil,
+      hash_including(keys),
+      { "Content-Type" => "application/json" }
+    )
+  end
 
-          let(:token) { "supersecrettoken" }
+  def expect_converge_message(keys)
+    keys["message_type"] = "run_converge"
+    keys["message_version"] = "1.1.0"
+    expect(rest_client).to receive(:post).with(
+      nil,
+      hash_including(keys),
+      { "Content-Type" => "application/json" }
+    )
+  end
 
-          context "when report is enabled for current mode" do
-            it "returns true" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(true)
-              expect(Chef::DataCollector.register_reporter?).to be_truthy
-            end
-          end
+  def resource_has_diff(new_resource, status)
+    new_resource.respond_to?(:diff) && %w{updated failed}.include?(status)
+  end
 
-          context "when report is disabled for current mode" do
-            it "returns false" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(false)
-              expect(Chef::DataCollector.register_reporter?).to be_falsey
-            end
-          end
+  def resource_record_for(current_resource, new_resource, action, status, duration)
+    {
+      "after" => new_resource.state_for_resource_reporter,
+      "before" => current_resource&.state_for_resource_reporter,
+      "cookbook_name" => cookbook_name,
+      "cookbook_version" => cookbook_version.version,
+      "delta" => resource_has_diff(new_resource, status) ? new_resource.diff : "",
+      "duration" => duration,
+      "id" => new_resource.identity,
+      "ignore_failure" => new_resource.ignore_failure,
+      "name" => new_resource.name,
+      "recipe_name" => recipe_name,
+      "result" => action.to_s,
+      "status" => status,
+      "type" => new_resource.resource_name.to_sym,
+    }
+  end
 
-        end
+  def send_run_failed_or_completed_event
+    status == "success" ? events.run_completed(node, run_status) : events.run_failed(exception, run_status)
+  end
 
-        # `Chef::Config[:data_collector][:server_url]` defaults to a URL
-        # relative to the `chef_server_url`, so we use configuration of the
-        # token to infer whether a solo/local mode user intends for data
-        # collection to be enabled.
-        context "when a token is not configured" do
+  shared_examples_for "sends a converge message" do
+    it "has a chef_server_fqdn" do
+      expect_converge_message("chef_server_fqdn" => "localhost")
+      send_run_failed_or_completed_event
+    end
 
-          let(:token) { nil }
+    it "has a start_time" do
+      expect_converge_message("start_time" => start_time.utc.iso8601)
+      send_run_failed_or_completed_event
+    end
 
-          context "when report is enabled for current mode" do
+    it "has a end_time" do
+      expect_converge_message("end_time" => end_time.utc.iso8601)
+      send_run_failed_or_completed_event
+    end
 
-            before do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(true)
-            end
+    it "has a entity_uuid" do
+      expect_converge_message("entity_uuid" => node_uuid)
+      send_run_failed_or_completed_event
+    end
 
-            context "when the current mode is solo" do
+    it "has a expanded_run_list" do
+      expect_converge_message("expanded_run_list" => expected_expansion)
+      send_run_failed_or_completed_event
+    end
 
-              before do
-                Chef::Config[:solo] = true
-              end
+    it "has a node" do
+      expect_converge_message("node" => expected_node)
+      send_run_failed_or_completed_event
+    end
 
-              it "returns true" do
-                expect(Chef::DataCollector.register_reporter?).to be(true)
-              end
+    it "has a node_name" do
+      expect_converge_message("node_name" => node_name)
+      send_run_failed_or_completed_event
+    end
 
-            end
+    it "has an organization" do
+      expect_converge_message("organization_name" => "unknown_organization")
+      send_run_failed_or_completed_event
+    end
 
-            context "when the current mode is local mode" do
+    it "has a policy_group" do
+      expect_converge_message("policy_group" => nil)
+      send_run_failed_or_completed_event
+    end
 
-              before do
-                Chef::Config[:local_mode] = true
-              end
+    it "has a policy_name" do
+      expect_converge_message("policy_name" => nil)
+      send_run_failed_or_completed_event
+    end
 
-              it "returns false" do
-                expect(Chef::DataCollector.register_reporter?).to be(true)
-              end
-            end
+    it "has a run_id" do
+      expect_converge_message("run_id" => request_id)
+      send_run_failed_or_completed_event
+    end
 
-            context "when the current mode is client mode" do
+    it "has a run_list" do
+      expect_converge_message("run_list" => expected_run_list)
+      send_run_failed_or_completed_event
+    end
 
-              before do
-                Chef::Config[:local_mode] = false
-                Chef::Config[:solo] = false
-              end
+    it "has a source" do
+      expect_converge_message("source" => "chef_client")
+      send_run_failed_or_completed_event
+    end
 
-              it "returns true" do
-                expect(Chef::DataCollector.register_reporter?).to be_truthy
-              end
+    it "has a status" do
+      expect_converge_message("status" => status)
+      send_run_failed_or_completed_event
+    end
 
-            end
+    it "has no deprecations" do
+      expect_converge_message("deprecations" => [])
+      send_run_failed_or_completed_event
+    end
 
-          end
+    it "has an error field" do
+      if exception
+        expect_converge_message(
+          "error" => {
+            "class" => exception.class,
+            "message" => exception.message,
+            "backtrace" => exception.backtrace,
+            "description" => error_description,
+          }
+        )
+      else
+        expect(rest_client).to receive(:post).with(
+          nil,
+          hash_excluding("error"),
+          { "Content-Type" => "application/json" }
+        )
+      end
+      send_run_failed_or_completed_event
+    end
 
-          context "when report is disabled for current mode" do
-            it "returns false" do
-              allow(Chef::DataCollector).to receive(:reporter_enabled_for_current_mode?).and_return(false)
-              expect(Chef::DataCollector.register_reporter?).to be_falsey
-            end
-          end
+    it "has a total resource count of zero" do
+      expect_converge_message("total_resource_count" => total_resource_count)
+      send_run_failed_or_completed_event
+    end
 
-        end
+    it "has a updated resource count of zero" do
+      expect_converge_message("updated_resource_count" => updated_resource_count)
+      send_run_failed_or_completed_event
+    end
 
-      end
+    it "includes the resource record" do
+      expect_converge_message("resources" => resource_record)
+      send_run_failed_or_completed_event
     end
   end
 
-  describe ".reporter_enabled_for_current_mode?" do
-    context "when running in solo mode" do
-      before do
-        Chef::Config[:solo] = true
-        Chef::Config[:local_mode] = false
+  describe "#should_be_enabled?" do
+    shared_examples_for "a solo-like run" do
+      it "is disabled in solo-legacy without a data_collector url and token" do
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
 
-      context "when data_collector_mode is :solo" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :solo
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is disabled in solo-legacy with only a url" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
 
-      context "when data_collector_mode is :client" do
-        it "returns false" do
-          Chef::Config[:data_collector][:mode] = :client
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(false)
-        end
-      end
-
-      context "when data_collector_mode is :both" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :both
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is disabled in solo-legacy with only a token" do
+        Chef::Config[:data_collector][:token] = "admit_one"
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
-    end
 
-    context "when running in local mode" do
-      before do
-        Chef::Config[:solo] = false
-        Chef::Config[:local_mode] = true
+      it "is enabled in solo-legacy with both a token and url" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        Chef::Config[:data_collector][:token] = "no_cash_value"
+        expect(data_collector.send(:should_be_enabled?)).to be true
       end
 
-      context "when data_collector_mode is :solo" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :solo
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is enabled in solo-legacy with only an output location to a file" do
+        Chef::Config[:data_collector][:output_locations] = { files: [ "/always/be/counting/down" ] }
+        expect(data_collector.send(:should_be_enabled?)).to be true
       end
 
-      context "when data_collector_mode is :client" do
-        it "returns false" do
-          Chef::Config[:data_collector][:mode] = :client
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(false)
-        end
+      it "is disabled in solo-legacy with only an output location to a uri" do
+        Chef::Config[:data_collector][:output_locations] = { urls: [ "https://esa.local/ariane5" ] }
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
 
-      context "when data_collector_mode is :both" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :both
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is enabled in solo-legacy with only an output location to a uri with a token" do
+        Chef::Config[:data_collector][:output_locations] = { urls: [ "https://esa.local/ariane5" ] }
+        Chef::Config[:data_collector][:token] = "good_for_one_fare"
+        expect(data_collector.send(:should_be_enabled?)).to be true
       end
-    end
 
-    context "when running in client mode" do
-      before do
-        Chef::Config[:solo] = false
-        Chef::Config[:local_mode] = false
+      it "is enabled in solo-legacy when the mode is :solo" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        Chef::Config[:data_collector][:token] = "non_redeemable"
+        Chef::Config[:data_collector][:mode] = :solo
+        expect(data_collector.send(:should_be_enabled?)).to be true
       end
 
-      context "when data_collector_mode is :solo" do
-        it "returns false" do
-          Chef::Config[:data_collector][:mode] = :solo
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(false)
-        end
+      it "is enabled in solo-legacy when the mode is :both" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        Chef::Config[:data_collector][:token] = "non_negotiable"
+        Chef::Config[:data_collector][:mode] = :both
+        expect(data_collector.send(:should_be_enabled?)).to be true
       end
 
-      context "when data_collector_mode is :client" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :client
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is disabled in solo-legacy when the mode is :client" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        Chef::Config[:data_collector][:token] = "NYCTA"
+        Chef::Config[:data_collector][:mode] = :client
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
 
-      context "when data_collector_mode is :both" do
-        it "returns true" do
-          Chef::Config[:data_collector][:mode] = :both
-          expect(Chef::DataCollector.reporter_enabled_for_current_mode?).to eq(true)
-        end
+      it "is disabled in solo-legacy mode when the mode is :nonsense" do
+        Chef::Config[:data_collector][:server_url] = "https://www.esa.local/ariane5"
+        Chef::Config[:data_collector][:token] = "MTA"
+        Chef::Config[:data_collector][:mode] = :nonsense
+        expect(data_collector.send(:should_be_enabled?)).to be false
       end
     end
-  end
-
-end
-
-describe Chef::DataCollector::Reporter do
-  let(:reporter) { described_class.new }
-  let(:run_status) { Chef::RunStatus.new(Chef::Node.new, Chef::EventDispatch::Dispatcher.new) }
-
-  let(:token) { "supersecrettoken" }
-
-  before do
-    Chef::Config[:data_collector][:server_url] = "http://my-data-collector-server.mycompany.com"
-    Chef::Config[:data_collector][:token] = token
-  end
-
-  describe "selecting token or signed header authentication" do
-
-    context "when the token is set in the config" do
-
-      before do
-        Chef::Config[:client_key] = "/no/key/should/exist/at/this/path.pem"
-      end
-
-      it "configures an HTTP client that doesn't do signed header auth" do
-        # Initializing with the wrong kind of HTTP class should cause Chef::Exceptions::PrivateKeyMissing
-        expect { reporter.http }.to_not raise_error
-      end
 
+    it "by default it is enabled" do
+      expect(data_collector.send(:should_be_enabled?)).to be true
     end
 
-    context "when no token is set in the config" do
-
-      let(:token) { nil }
-
-      let(:client_key) { File.join(CHEF_SPEC_DATA, "ssl", "private_key.pem") }
+    it "is disabled in why-run" do
+      Chef::Config[:why_run] = true
+      expect(data_collector.send(:should_be_enabled?)).to be false
+    end
 
-      before do
-        Chef::Config[:client_key] = client_key
+    describe "a solo legacy run" do
+      before(:each) do
+        Chef::Config[:solo_legacy_mode] = true
       end
 
-      it "configures an HTTP client that does signed header auth" do
-        expect { reporter.http }.to_not raise_error
-        expect(reporter.http.options).to have_key(:signing_key_filename)
-        expect(reporter.http.options[:signing_key_filename]).to eq(client_key)
-      end
+      it_behaves_like "a solo-like run"
     end
 
-  end
-
-  describe "#run_started" do
-    before do
-      allow(reporter).to receive(:update_run_status)
-      allow(reporter).to receive(:send_to_data_collector)
-      allow(Chef::DataCollector::Messages).to receive(:run_start_message)
-    end
+    describe "a local mode run" do
+      before(:each) do
+        Chef::Config[:local_mode] = true
+      end
 
-    it "updates the run status" do
-      expect(reporter).to receive(:update_run_status).with(run_status)
-      reporter.run_started(run_status)
+      it_behaves_like "a solo-like run"
     end
 
-    it "sends the RunStart message output to the Data Collector server" do
-      expect(Chef::DataCollector::Messages)
-        .to receive(:run_start_message)
-        .with(run_status)
-        .and_return(key: "value")
-      expect(reporter).to receive(:send_to_data_collector).with({ key: "value" })
-      reporter.run_started(run_status)
+    it "is enabled in client mode when the mode is :both" do
+      Chef::Config[:data_collector][:mode] = :both
+      expect(data_collector.send(:should_be_enabled?)).to be true
     end
-  end
-
-  describe "when sending a message at chef run completion" do
 
-    let(:node) { Chef::Node.new }
-
-    let(:run_status) do
-      instance_double("Chef::RunStatus",
-                      run_id: "run_id",
-                      node: node,
-                      start_time: Time.new,
-                      end_time: Time.new,
-                      exception: exception)
+    it "is disabled in client mode when the mode is :solo" do
+      Chef::Config[:data_collector][:mode] = :solo
+      expect(data_collector.send(:should_be_enabled?)).to be false
     end
 
-    before do
-      reporter.send(:update_run_status, run_status)
+    it "is disabled in client mode when the mode is :nonsense" do
+      Chef::Config[:data_collector][:mode] = :nonsense
+      expect(data_collector.send(:should_be_enabled?)).to be false
     end
 
-    describe "#run_completed" do
-
-      let(:exception) { nil }
-
-      it "sends the run completion" do
-        expect(reporter).to receive(:send_to_data_collector) do |message|
-          expect(message).to be_a(Hash)
-          expect(message["status"]).to eq("success")
-        end
-        reporter.run_completed(node)
-      end
-    end
-
-    describe "#run_failed" do
-
-      let(:exception) { StandardError.new("oops") }
-
-      it "updates the exception and sends the run completion" do
-        expect(reporter).to receive(:send_to_data_collector) do |message|
-          expect(message).to be_a(Hash)
-          expect(message["status"]).to eq("failure")
-        end
-        reporter.run_failed("test_exception")
-      end
+    it "is still enabled if you set a token in client mode" do
+      Chef::Config[:data_collector][:token] =  "good_for_one_ride"
+      expect(data_collector.send(:should_be_enabled?)).to be true
     end
   end
 
-  describe "#converge_start" do
-    it "stashes the run_context for later use" do
-      reporter.converge_start("test_context")
-      expect(reporter.run_context).to eq("test_context")
-    end
-  end
+  describe "when the run fails during node load" do
+    let(:exception) { Exception.new("imperial to metric conversion error") }
+    let(:error_description) { Chef::Formatters::ErrorMapper.registration_failed(node_name, exception, Chef::Config).for_json }
+    let(:total_resource_count) { 0 }
+    let(:updated_resource_count) { 0 }
+    let(:status) { "failure" }
+    let(:expected_node) { {} } # no node because that failed
+    let(:expected_run_list) { [] } # no run_list without a node
+    let(:expected_expansion) { {} } # no run_list expansion without a run_list
+    let(:resource_record) { [] } # and certainly no resources
 
-  describe "#converge_complete" do
-    it "detects and processes any unprocessed resources" do
-      expect(reporter).to receive(:detect_unprocessed_resources)
-      reporter.converge_complete
+    before do
+      events.registration_failed(node_name, exception, Chef::Config)
+      run_status.stop_clock
+      run_status.exception = exception
+      expect_start_message
     end
-  end
 
-  describe "#converge_failed" do
-    it "detects and processes any unprocessed resources" do
-      expect(reporter).to receive(:detect_unprocessed_resources)
-      reporter.converge_failed("exception")
-    end
+    it_behaves_like "sends a converge message"
   end
 
-  describe "#resource_current_state_loaded" do
-    let(:new_resource)     { double("new_resource") }
-    let(:action)           { double("action") }
-    let(:current_resource) { double("current_resource") }
-    let(:resource_report)  { double("resource_report") }
+  describe "when the run fails during node load" do
+    let(:exception) { Exception.new("imperial to metric conversion error") }
+    let(:error_description) { Chef::Formatters::ErrorMapper.node_load_failed(node_name, exception, Chef::Config).for_json }
+    let(:total_resource_count) { 0 }
+    let(:updated_resource_count) { 0 }
+    let(:status) { "failure" }
+    let(:expected_node) { {} } # no node because that failed
+    let(:expected_run_list) { [] } # no run_list without a node
+    let(:expected_expansion) { {} } # no run_list expansion without a run_list
+    let(:resource_record) { [] } # and certainly no resources
 
-    context "when resource is a nested resource" do
-      it "does not update the resource report" do
-        allow(reporter).to receive(:nested_resource?).and_return(true)
-        expect(reporter).not_to receive(:update_current_resource_report)
-        reporter.resource_current_state_loaded(new_resource, action, current_resource)
-      end
+    before do
+      events.node_load_failed(node_name, exception, Chef::Config)
+      run_status.stop_clock
+      run_status.exception = exception
+      expect_start_message
     end
 
-    context "when resource is not a nested resource" do
-      it "initializes the resource report" do
-        allow(reporter).to receive(:nested_resource?).and_return(false)
-        expect(reporter).to receive(:initialize_resource_report_if_needed)
-          .with(new_resource, action, current_resource)
-        reporter.resource_current_state_loaded(new_resource, action, current_resource)
-      end
-    end
+    it_behaves_like "sends a converge message"
   end
 
-  describe "#resource_up_to_date" do
-    let(:new_resource)    { double("new_resource") }
-    let(:action)          { double("action") }
-    let(:resource_report) { double("resource_report") }
+  describe "when the run fails during run_list_expansion" do
+    let(:exception) { Exception.new("imperial to metric conversion error") }
+    let(:error_description) { Chef::Formatters::ErrorMapper.run_list_expand_failed(node, exception).for_json }
+    let(:total_resource_count) { 0 }
+    let(:updated_resource_count) { 0 }
+    let(:status) { "failure" }
+    let(:expected_expansion) { {} } # no run_list expanasion when it failed
+    let(:resource_record) { [] } # and no resources
 
     before do
-      allow(reporter).to receive(:nested_resource?)
-      allow(reporter).to receive(:current_resource_report).and_return(resource_report)
-      allow(resource_report).to receive(:up_to_date)
-    end
-
-    context "when the resource is a nested resource" do
-      it "does not mark the resource report as up-to-date" do
-        allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(true)
-        expect(resource_report).not_to receive(:up_to_date)
-        reporter.resource_up_to_date(new_resource, action)
-      end
+      events.node_load_success(node)
+      run_status.node = node
+      events.run_list_expand_failed(node, exception)
+      run_status.stop_clock
+      run_status.exception = exception
+      expect_start_message
     end
 
-    context "when the resource is not a nested resource" do
-      it "marks the resource report as up-to-date" do
-        allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(false)
-        expect(resource_report).to receive(:up_to_date)
-        reporter.resource_up_to_date(new_resource, action)
-      end
-    end
+    it_behaves_like "sends a converge message"
   end
 
-  describe "#resource_skipped" do
-    let(:new_resource)    { double("new_resource") }
-    let(:action)          { double("action") }
-    let(:conditional)     { double("conditional") }
-    let(:resource_report) { double("resource_report") }
+  describe "when the run fails during cookbook resolution" do
+    let(:exception) { Exception.new("imperial to metric conversion error") }
+    let(:error_description) { Chef::Formatters::ErrorMapper.cookbook_resolution_failed(node, exception).for_json }
+    let(:total_resource_count) { 0 }
+    let(:updated_resource_count) { 0 }
+    let(:status) { "failure" }
+    let(:resource_record) { [] } # and no resources
 
     before do
-      allow(reporter).to receive(:nested_resource?)
-      allow(resource_report).to receive(:skipped)
+      events.node_load_success(node)
+      run_status.node = node
+      events.run_list_expanded(expansion)
+      run_status.start_clock
+      expect_start_message
+      events.cookbook_resolution_failed(node, exception)
+      run_status.stop_clock
+      run_status.exception = exception
     end
 
-    context "when the resource is a nested resource" do
-      it "does not mark the resource report as skipped" do
-        allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(true)
-        expect(resource_report).not_to receive(:skipped).with(conditional)
-        reporter.resource_skipped(new_resource, action, conditional)
-      end
-    end
-
-    context "when the resource is not a nested resource" do
-      it "initializes the resource report and marks it as skipped" do
-        allow(reporter).to receive(:nested_resource?).and_return(false)
-        allow(reporter).to receive(:current_resource_report).and_return(resource_report)
-        expect(reporter).to receive(:initialize_resource_report_if_needed).with(new_resource, action)
-        expect(resource_report).to receive(:skipped).with(conditional)
-        reporter.resource_skipped(new_resource, action, conditional)
-      end
-    end
+    it_behaves_like "sends a converge message"
   end
 
-  describe "#resource_updated" do
-    let(:resource_report) { double("resource_report") }
+  describe "when the run fails during cookbook synchronization" do
+    let(:exception) { Exception.new("imperial to metric conversion error") }
+    let(:error_description) { Chef::Formatters::ErrorMapper.cookbook_sync_failed(node, exception).for_json }
+    let(:total_resource_count) { 0 }
+    let(:updated_resource_count) { 0 }
+    let(:status) { "failure" }
+    let(:resource_record) { [] } # and no resources
 
     before do
-      allow(reporter).to receive(:current_resource_report).and_return(resource_report)
-      allow(resource_report).to receive(:updated)
+      events.node_load_success(node)
+      run_status.node = node
+      events.run_list_expanded(expansion)
+      run_status.start_clock
+      expect_start_message
+      events.cookbook_sync_failed(node, exception)
+      run_status.stop_clock
+      run_status.exception = exception
     end
 
-    it "marks the resource report as updated" do
-      expect(resource_report).to receive(:updated)
-      reporter.resource_updated("new_resource", "action")
-    end
+    it_behaves_like "sends a converge message"
   end
 
-  describe "#resource_failed" do
-    let(:new_resource)    { double("new_resource") }
-    let(:action)          { double("action") }
-    let(:exception)       { double("exception") }
-    let(:error_mapper)    { double("error_mapper") }
-    let(:resource_report) { double("resource_report") }
-
+  describe "after successfully starting the run" do
     before do
-      allow(reporter).to receive(:update_error_description)
-      allow(reporter).to receive(:current_resource_report).and_return(resource_report)
-      allow(resource_report).to receive(:failed)
-      allow(Chef::Formatters::ErrorMapper).to receive(:resource_failed).and_return(error_mapper)
-      allow(error_mapper).to receive(:for_json)
+      # these events happen in this order in the client
+      events.node_load_success(node)
+      run_status.node = node
+      events.run_list_expanded(expansion)
+      run_status.start_clock
     end
 
-    it "updates the error description" do
-      expect(Chef::Formatters::ErrorMapper).to receive(:resource_failed).with(
-        new_resource,
-        action,
-        exception
-      ).and_return(error_mapper)
-      expect(error_mapper).to receive(:for_json).and_return("error_description")
-      expect(reporter).to receive(:update_error_description).with("error_description")
-      reporter.resource_failed(new_resource, action, exception)
-    end
+    describe "run_start_message" do
+      it "sends a run_start_message" do
+        expect_start_message
+        events.run_started(run_status)
+      end
 
-    context "when the resource is not a nested resource" do
-      it "marks the resource report as failed" do
-        allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(false)
-        expect(resource_report).to receive(:failed).with(exception)
-        reporter.resource_failed(new_resource, action, exception)
+      it "extracts the hostname from the chef_server_url" do
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local"
+        expect_start_message("chef_server_fqdn" => "spacex.rockets.local")
+        events.run_started(run_status)
       end
-    end
 
-    context "when the resource is a nested resource" do
-      it "does not mark the resource report as failed" do
-        allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(true)
-        expect(resource_report).not_to receive(:failed).with(exception)
-        reporter.resource_failed(new_resource, action, exception)
+      it "extracts the organization from the chef_server_url" do
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc"
+        expect_start_message("organization_name" => "gnc")
+        events.run_started(run_status)
       end
-    end
-  end
 
-  describe "#resource_completed" do
-    let(:new_resource)    { double("new_resource") }
-    let(:resource_report) { double("resource_report") }
+      it "extracts the organization from the chef_server_url if there are extra slashes" do
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local///organizations///gnc"
+        expect_start_message("organization_name" => "gnc")
+        events.run_started(run_status)
+      end
 
-    before do
-      allow(reporter).to receive(:update_current_resource_report)
-      allow(reporter).to receive(:add_resource_report)
-      allow(reporter).to receive(:current_resource_report)
-      allow(resource_report).to receive(:finish)
-    end
+      it "extracts the organization from the chef_server_url if there is a trailing slash" do
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/"
+        expect_start_message("organization_name" => "gnc")
+        events.run_started(run_status)
+      end
 
-    context "when there is no current resource report" do
-      it "does not touch the current resource report" do
-        allow(reporter).to receive(:current_resource_report).and_return(nil)
-        expect(reporter).not_to receive(:update_current_resource_report)
-        reporter.resource_completed(new_resource)
+      it "sets 'unknown_organization' if the cher_server_url does not contain one" do
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local"
+        expect_start_message("organization_name" => "unknown_organization")
+        events.run_started(run_status)
       end
-    end
 
-    context "when there is a current resource report" do
-      before do
-        allow(reporter).to receive(:current_resource_report).and_return(resource_report)
+      it "still uses the chef_server_url in non-solo mode even if the data_collector organization is set" do
+        Chef::Config[:data_collector][:organization] = "blue-origin"
+        Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/"
+        expect_start_message("organization_name" => "gnc")
+        events.run_started(run_status)
       end
 
-      context "when the resource is a nested resource" do
-        it "does not mark the resource as finished" do
-          allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(true)
-          expect(resource_report).not_to receive(:finish)
-          reporter.resource_completed(new_resource)
+      describe "in legacy mode" do
+        before do
+          Chef::Config[:solo_legacy_mode] = true
+          Chef::Config[:data_collector][:server_url] = "https://nasa.rockets.local/organizations/sls"
+        end
+
+        it "we get the data collector organization" do
+          Chef::Config[:data_collector][:organization] = "blue-origin"
+          Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/" # should be ignored
+          expect_start_message("organization_name" => "blue-origin")
+          events.run_started(run_status)
+        end
+
+        it "if the data collector org is unset we get 'chef_solo'" do
+          Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/" # should be ignored
+          expect_start_message("organization_name" => "chef_solo")
+          events.run_started(run_status)
+        end
+
+        it "sets the source" do
+          expect_start_message("source" => "chef_solo")
+          events.run_started(run_status)
         end
       end
 
-      context "when the resource is not a nested resource" do
+      describe "in local mode" do
         before do
-          allow(reporter).to receive(:nested_resource?).with(new_resource).and_return(false)
+          Chef::Config[:local_mode] = true
+          Chef::Config[:data_collector][:server_url] = "https://nasa.rockets.local/organizations/sls"
         end
 
-        it "marks the current resource report as finished" do
-          expect(resource_report).to receive(:finish)
-          reporter.resource_completed(new_resource)
+        it "we get the data collector organization" do
+          Chef::Config[:data_collector][:organization] = "blue-origin"
+          Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/" # should be ignored
+          expect_start_message("organization_name" => "blue-origin")
+          events.run_started(run_status)
         end
 
-        it "nils out the current resource report" do
-          expect(reporter).to receive(:clear_current_resource_report)
-          reporter.resource_completed(new_resource)
+        it "if the data collector org is unset we get 'chef_solo'" do
+          Chef::Config[:chef_server_url] = "https://spacex.rockets.local/organizations/gnc/" # should be ignored
+          expect_start_message("organization_name" => "chef_solo")
+          events.run_started(run_status)
+        end
+
+        it "sets the source" do
+          expect_start_message("source" => "chef_solo")
+          events.run_started(run_status)
         end
       end
     end
-  end
 
-  describe "#run_list_expanded" do
-    it "sets the expanded run list" do
-      reporter.run_list_expanded("test_run_list")
-      expect(reporter.expanded_run_list).to eq("test_run_list")
-    end
-  end
+    describe "converge messages" do
+      before do
+        expect_start_message
+        events.run_started(run_status)
+        events.cookbook_compilation_start(run_context)
+      end
 
-  describe "#run_list_expand_failed" do
-    let(:node)         { double("node") }
-    let(:error_mapper) { double("error_mapper") }
-    let(:exception)    { double("exception") }
+      context "with no resources" do
+        let(:total_resource_count) { 0 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) { [ ] }
+        let(:status) { "success" }
 
-    it "updates the error description" do
-      expect(Chef::Formatters::ErrorMapper).to receive(:run_list_expand_failed).with(
-        node,
-        exception
-      ).and_return(error_mapper)
-      expect(error_mapper).to receive(:for_json).and_return("error_description")
-      expect(reporter).to receive(:update_error_description).with("error_description")
-      reporter.run_list_expand_failed(node, exception)
-    end
-  end
+        before do
+          run_status.stop_clock
+        end
 
-  describe "#cookbook_resolution_failed" do
-    let(:error_mapper)      { double("error_mapper") }
-    let(:exception)         { double("exception") }
-    let(:expanded_run_list) { double("expanded_run_list") }
+        it_behaves_like "sends a converge message"
 
-    it "updates the error description" do
-      expect(Chef::Formatters::ErrorMapper).to receive(:cookbook_resolution_failed).with(
-        expanded_run_list,
-        exception
-      ).and_return(error_mapper)
-      expect(error_mapper).to receive(:for_json).and_return("error_description")
-      expect(reporter).to receive(:update_error_description).with("error_description")
-      reporter.cookbook_resolution_failed(expanded_run_list, exception)
-    end
+        it "sets the policy_group" do
+          node.policy_group = "acceptionsal"
+          expect_converge_message("policy_group" => "acceptionsal")
+          send_run_failed_or_completed_event
+        end
 
-  end
+        it "has a policy_name" do
+          node.policy_name = "webappdb"
+          expect_converge_message("policy_name" => "webappdb")
+          send_run_failed_or_completed_event
+        end
 
-  describe "#cookbook_sync_failed" do
-    let(:cookbooks)    { double("cookbooks") }
-    let(:error_mapper) { double("error_mapper") }
-    let(:exception)    { double("exception") }
+        it "collects deprecation messages" do
+          location = Chef::Log.caller_location
+          events.deprecation(Chef::Deprecated.create(:internal_api, "deprecation warning", location))
+          expect_converge_message("deprecations" => [{ location: location, message: "deprecation warning", url: "https://docs.chef.io/deprecations_internal_api.html" }])
+          send_run_failed_or_completed_event
+        end
+      end
 
-    it "updates the error description" do
-      expect(Chef::Formatters::ErrorMapper).to receive(:cookbook_sync_failed).with(
-        cookbooks,
-        exception
-      ).and_return(error_mapper)
-      expect(error_mapper).to receive(:for_json).and_return("error_description")
-      expect(reporter).to receive(:update_error_description).with("error_description")
-      reporter.cookbook_sync_failed(cookbooks, exception)
-    end
-  end
+      context "when the run contains a file resource that is up-to-date" do
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) { [ resource_record_for(current_resource, new_resource, :create, "up-to-date", "1234") ] }
+        let(:status) { "success" }
 
-  describe "#disable_reporter_on_error" do
-    context "when no exception is raise by the block" do
-      it "does not disable the reporter" do
-        expect(reporter).not_to receive(:disable_data_collector_reporter)
-        reporter.send(:disable_reporter_on_error) { true }
-      end
+        before do
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_up_to_date(new_resource, :create)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
+        end
 
-      it "does not raise an exception" do
-        expect { reporter.send(:disable_reporter_on_error) { true } }.not_to raise_error
+        it_behaves_like "sends a converge message"
       end
-    end
 
-    context "when an unexpected exception is raised by the block" do
-      it "re-raises the exception" do
-        expect { reporter.send(:disable_reporter_on_error) { raise "bummer" } }.to raise_error(RuntimeError)
-      end
-    end
+      context "when the run contains a file resource that is updated" do
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 1 }
+        let(:resource_record) { [ resource_record_for(current_resource, new_resource, :create, "updated", "1234") ] }
+        let(:status) { "success" }
 
-    [ Timeout::Error, Errno::EINVAL, Errno::ECONNRESET,
-      Errno::ECONNREFUSED, EOFError, Net::HTTPBadResponse,
-      Net::HTTPHeaderSyntaxError, Net::ProtocolError, OpenSSL::SSL::SSLError,
-      Errno::EHOSTDOWN ].each do |exception_class|
-      context "when the block raises a #{exception_class} exception" do
-        it "disables the reporter" do
-          expect(reporter).to receive(:disable_data_collector_reporter)
-          reporter.send(:disable_reporter_on_error) { raise exception_class.new("bummer") }
+        before do
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_updated(new_resource, :create)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
         end
 
-        context "when raise-on-failure is enabled" do
-          it "logs an error and raises" do
-            Chef::Config[:data_collector][:raise_on_failure] = true
-            expect(Chef::Log).to receive(:error)
-            expect { reporter.send(:disable_reporter_on_error) { raise exception_class.new("bummer") } }.to raise_error(exception_class)
-          end
+        it_behaves_like "sends a converge message"
+      end
+
+      context "When there is an embedded resource, it includes the sub-resource in the report" do
+        let(:total_resource_count) { 2 }
+        let(:updated_resource_count) { 2 }
+        let(:implementation_resource) do
+          r = Chef::Resource::CookbookFile.new("/preseed-file.txt")
+          r.cookbook_name = cookbook_name
+          r.recipe_name = recipe_name
+          allow(r).to receive(:cookbook_version).and_return(cookbook_version)
+          r
         end
+        let(:resource_record) { [ resource_record_for(implementation_resource, implementation_resource, :create, "updated", "2345"), resource_record_for(current_resource, new_resource, :create, "updated", "1234") ] }
+        let(:status) { "success" }
 
-        context "when raise-on-failure is disabled" do
-          it "logs an info message and does not raise an exception" do
-            Chef::Config[:data_collector][:raise_on_failure] = false
-            expect(Chef::Log).to receive(:info)
-            expect { reporter.send(:disable_reporter_on_error) { raise exception_class.new("bummer") } }.not_to raise_error
-          end
+        before do
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+
+          events.resource_action_start(implementation_resource , :create)
+          events.resource_current_state_loaded(implementation_resource, :create, implementation_resource)
+          events.resource_updated(implementation_resource, :create)
+          implementation_resource.instance_variable_set(:@elapsed_time, 2.3456)
+          events.resource_completed(implementation_resource)
+
+          events.resource_updated(new_resource, :create)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
         end
-      end
-    end
-  end
 
-  describe "#validate_data_collector_server_url!" do
-    context "when server_url is empty" do
-      it "raises an exception" do
-        Chef::Config[:data_collector][:server_url] = ""
-        expect { reporter.send(:validate_data_collector_server_url!) }.to raise_error(Chef::Exceptions::ConfigurationError)
+        it_behaves_like "sends a converge message"
       end
-    end
 
-    context "when server_url is omitted but output_locations is specified" do
-      it "does not an exception" do
-        Chef::Config[:data_collector][:output_locations] = ["http://data_collector", "/tmp/data_collector.json"]
-        expect { reporter.send(:validate_data_collector_server_url!) }.not_to raise_error(Chef::Exceptions::ConfigurationError)
-      end
-    end
+      context "when the run contains a file resource that is skipped due to a block conditional" do
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) do
+          rec = resource_record_for(current_resource, new_resource, :create, "skipped", "1234")
+          rec["conditional"] = "not_if { #code block }" # FIXME: "#code block" is poor, is there some way to fix this?
+          [ rec ]
+        end
+        let(:status) { "success" }
 
-    context "when server_url is not empty" do
-      context "when server_url is an invalid URI" do
-        it "raises an exception" do
-          Chef::Config[:data_collector][:server_url] = "this is not a URI"
-          expect { reporter.send(:validate_data_collector_server_url!) }.to raise_error(Chef::Exceptions::ConfigurationError)
+        before do
+          conditional = (new_resource.not_if { true }).first
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_skipped(new_resource, :create, conditional)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
         end
-      end
 
-      context "when server_url is a valid URI" do
-        context "when server_url is a URI with no host" do
-          it "raises an exception" do
-            Chef::Config[:data_collector][:server_url] = "/file/uri.txt"
-            expect { reporter.send(:validate_data_collector_server_url!) }.to raise_error(Chef::Exceptions::ConfigurationError)
-          end
+        it_behaves_like "sends a converge message"
+      end
 
+      context "when the run contains a file resource that is skipped due to a string conditional" do
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) do
+          rec = resource_record_for(current_resource, new_resource, :create, "skipped", "1234")
+          rec["conditional"] = 'not_if "true"'
+          [ rec ]
         end
+        let(:status) { "success" }
 
-        context "when server_url is a URI with a valid host" do
-          it "does not an exception" do
-            Chef::Config[:data_collector][:server_url] = "http://www.google.com/data-collector"
-            expect { reporter.send(:validate_data_collector_server_url!) }.not_to raise_error
-          end
+        before do
+          conditional = (new_resource.not_if "true").first
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_skipped(new_resource, :create, conditional)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
         end
-      end
-    end
-  end
 
-  describe "#validate_data_collector_output_locations!" do
-    context "when output_locations is empty" do
-      it "raises an exception" do
-        Chef::Config[:data_collector][:output_locations] = {}
-        expect { reporter.send(:validate_data_collector_output_locations!) }.to raise_error(Chef::Exceptions::ConfigurationError)
+        it_behaves_like "sends a converge message"
       end
-    end
 
-    context "when valid output_locations are provided" do
-      it "does not raise an exception" do
-        expect(reporter).to receive(:open).with("data_collection.json", "a")
-        Chef::Config[:data_collector][:output_locations] = { urls: ["http://data_collector"], files: ["data_collection.json"] }
-        expect { reporter.send(:validate_data_collector_output_locations!) }.not_to raise_error(Chef::Exceptions::ConfigurationError)
-      end
-    end
+      context "when the run contains a file resource that threw an exception" do
+        let(:exception) { Exception.new("imperial to metric conversion error") }
+        let(:error_description) { Chef::Formatters::ErrorMapper.resource_failed(new_resource, :create, exception).for_json }
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) do
+          rec = resource_record_for(current_resource, new_resource, :create, "failed", "1234")
+          rec["error_message"] = "imperial to metric conversion error"
+          [ rec ]
+        end
+        let(:status) { "failure" }
+
+        before do
+          exception.set_backtrace(caller)
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_failed(new_resource, :create, exception)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_complete
+          run_status.stop_clock
+          run_status.exception = exception
+        end
 
-    context "when output_locations contains an invalid URI" do
-      it "raises an exception" do
-        Chef::Config[:data_collector][:output_locations] = { urls: ["this is not a url"], files: ["/tmp/data_collection.json"] }
-        expect { reporter.send(:validate_data_collector_output_locations!) }.to raise_error(Chef::Exceptions::ConfigurationError)
+        it_behaves_like "sends a converge message"
       end
-    end
-  end
 
-  describe "#detect_unprocessed_resources" do
-    context "when resources do not override core methods" do
-      it "adds resource reports for any resources that have not yet been processed" do
-        resource_a  = Chef::Resource::Service.new("processed service")
-        resource_b  = Chef::Resource::Service.new("unprocessed service")
+      context "when the run contains a file resource that threw an exception in load_current_resource" do
+        let(:exception) { Exception.new("imperial to metric conversion error") }
+        let(:error_description) { Chef::Formatters::ErrorMapper.resource_failed(new_resource, :create, exception).for_json }
+        let(:total_resource_count) { 1 }
+        let(:updated_resource_count) { 0 }
+        let(:resource_record) do
+          rec = resource_record_for(current_resource, new_resource, :create, "failed", "1234")
+          rec["before"] = {}
+          rec["error_message"] = "imperial to metric conversion error"
+          [ rec ]
+        end
+        let(:status) { "failure" }
 
-        resource_a.action = [ :enable, :start ]
-        resource_b.action = :start
+        before do
+          exception.set_backtrace(caller)
+          events.resource_action_start(new_resource, :create)
+          # resource_current_state_loaded is skipped
+          events.resource_failed(new_resource, :create, exception)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          events.converge_failed(exception)
+          run_status.stop_clock
+          run_status.exception = exception
+        end
 
-        run_context = Chef::RunContext.new(Chef::Node.new, Chef::CookbookCollection.new, nil)
-        run_context.resource_collection.insert(resource_a)
-        run_context.resource_collection.insert(resource_b)
+        it_behaves_like "sends a converge message"
+      end
 
-        allow(reporter).to receive(:run_context).and_return(run_context)
+      context "when the resource collection contains a resource that was unproccesed due to prior errors" do
+        let(:exception) { Exception.new("imperial to metric conversion error") }
+        let(:error_description) { Chef::Formatters::ErrorMapper.resource_failed(new_resource, :create, exception).for_json }
+        let(:total_resource_count) { 2 }
+        let(:updated_resource_count) { 0 }
+        let(:unprocessed_resource) do
+          res = Chef::Resource::Service.new("unprocessed service")
+          res.cookbook_name = cookbook_name
+          res.recipe_name = recipe_name
+          allow(res).to receive(:cookbook_version).and_return(cookbook_version)
+          res
+        end
+        let(:resource_record) do
+          rec1 = resource_record_for(current_resource, new_resource, :create, "failed", "1234")
+          rec1["error_message"] = "imperial to metric conversion error"
+          rec2 = resource_record_for(nil, unprocessed_resource, :nothing, "unprocessed", "")
+          rec2["before"] = {}
+          [ rec1, rec2 ]
+        end
+        let(:status) { "failure" }
 
-        # process the actions for resource_a, but not resource_b
-        reporter.resource_up_to_date(resource_a, :enable)
-        reporter.resource_completed(resource_a)
-        reporter.resource_up_to_date(resource_a, :start)
-        reporter.resource_completed(resource_a)
-        expect(reporter.all_resource_reports.size).to eq(2)
+        before do
+          run_context.resource_collection << new_resource
+          run_context.resource_collection << unprocessed_resource
+          exception.set_backtrace(caller)
+          events.resource_action_start(new_resource, :create)
+          events.resource_current_state_loaded(new_resource, :create, current_resource)
+          events.resource_failed(new_resource, :create, exception)
+          new_resource.instance_variable_set(:@elapsed_time, 1.2345)
+          events.resource_completed(new_resource)
+          new_resource.executed_by_runner = true
+          events.converge_failed(exception)
+          run_status.stop_clock
+          run_status.exception = exception
+        end
 
-        # detect unprocessed resources, which should find that resource_b has not yet been processed
-        reporter.send(:detect_unprocessed_resources)
-        expect(reporter.all_resource_reports.size).to eq(3)
+        it_behaves_like "sends a converge message"
       end
-    end
-
-    context "when a resource overrides a core method, such as #hash" do
-      it "does not raise an exception" do
-        resource_a  = Chef::Resource::Service.new("processed service")
-        resource_b  = Chef::Resource::Service.new("unprocessed service")
 
-        resource_a.action = :start
-        resource_b.action = :start
+      context "when cookbook resolution fails" do
+        let(:exception) { Exception.new("imperial to metric conversion error") }
+        let(:error_description) { Chef::Formatters::ErrorMapper.cookbook_resolution_failed(expansion, exception).for_json }
+        let(:total_resource_count) { 0 }
+        let(:updated_resource_count) { 0 }
+        let(:status) { "failure" }
 
-        run_context = Chef::RunContext.new(Chef::Node.new, Chef::CookbookCollection.new, nil)
-        run_context.resource_collection.insert(resource_a)
-        run_context.resource_collection.insert(resource_b)
+        before do
+          events.cookbook_resolution_failed(expansion, exception)
+          run_status.stop_clock
+          run_status.exception = exception
+        end
 
-        allow(reporter).to receive(:run_context).and_return(run_context)
+        it_behaves_like "sends a converge message"
+      end
 
-        # override the #hash method on resource_a to return a String instead of
-        # a Fixnum. Without the fix in chef/chef#5604, this would raise an
-        # exception when getting added to the Set/Hash.
-        resource_a.define_singleton_method(:hash) { "a string" }
+      context "When cookbook synchronization fails" do
+        let(:exception) { Exception.new("imperial to metric conversion error") }
+        let(:error_description) { Chef::Formatters::ErrorMapper.cookbook_sync_failed({}, exception).for_json }
+        let(:total_resource_count) { 0 }
+        let(:updated_resource_count) { 0 }
+        let(:status) { "failure" }
 
-        # process the actions for resource_a, but not resource_b
-        reporter.resource_up_to_date(resource_a, :start)
-        reporter.resource_completed(resource_a)
+        before do
+          events.cookbook_sync_failed(expansion, exception)
+          run_status.stop_clock
+          run_status.exception = exception
+        end
 
-        expect { reporter.send(:detect_unprocessed_resources) }.not_to raise_error
+        it_behaves_like "sends a converge message"
       end
+
     end
   end
 end
diff --git a/spec/unit/event_dispatch/dispatcher_spec.rb b/spec/unit/event_dispatch/dispatcher_spec.rb
index 5061a9845f..1db43ad740 100644
--- a/spec/unit/event_dispatch/dispatcher_spec.rb
+++ b/spec/unit/event_dispatch/dispatcher_spec.rb
@@ -1,7 +1,7 @@
 #
 # 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");
@@ -43,9 +43,9 @@ describe Chef::EventDispatch::Dispatcher do
 
     it "forwards events to the subscribed event sink" do
       # the events all have different arity and such so we just hit a few different events:
-
-      expect(event_sink).to receive(:run_start).with("12.4.0")
-      dispatcher.run_start("12.4.0")
+      run_status = Chef::RunStatus.new({}, {})
+      expect(event_sink).to receive(:run_start).with("12.4.0", run_status)
+      dispatcher.run_start("12.4.0", run_status)
 
       cookbook_version = double("cookbook_version")
       expect(event_sink).to receive(:synchronized_cookbook).with("apache2", cookbook_version)
@@ -119,4 +119,51 @@ describe Chef::EventDispatch::Dispatcher do
       end
     end
   end
+
+  context "events that queue events" do
+    class Accumulator
+      def self.sequence
+        @secuence ||= []
+      end
+    end
+
+    let(:event_sink_1) do
+      Class.new(Chef::EventDispatch::Base) do
+        def synchronized_cookbook(dispatcher, arg)
+          dispatcher.enqueue(:event_two, arg)
+          Accumulator.sequence << [ :sink_1_event_1, arg ]
+        end
+
+        def event_two(arg)
+          Accumulator.sequence << [ :sink_1_event_2, arg ]
+        end
+      end.new
+    end
+    let(:event_sink_2) do
+      Class.new(Chef::EventDispatch::Base) do
+        def synchronized_cookbook(dispatcher, arg)
+          Accumulator.sequence << [ :sink_2_event_1, arg ]
+        end
+
+        def event_two(arg)
+          Accumulator.sequence << [ :sink_2_event_2, arg ]
+        end
+      end.new
+    end
+
+    before do
+      dispatcher.register(event_sink_1)
+      dispatcher.register(event_sink_2)
+    end
+
+    it "runs the events in the correct order without interleaving the enqueued event" do
+      dispatcher.synchronized_cookbook(dispatcher, "two")
+      expect(Accumulator.sequence).to eql([
+        [:sink_1_event_1, "two"], # the call to enqueue the event happens here
+        [:sink_2_event_1, "two"], # event 1 fully finishes
+        [:sink_1_event_2, "two"],
+        [:sink_2_event_2, "two"], # then event 2 runs and finishes
+      ])
+    end
+  end
 end
diff --git a/spec/unit/node_spec.rb b/spec/unit/node_spec.rb
index 37dc27ec0f..752871e2e9 100644
--- a/spec/unit/node_spec.rb
+++ b/spec/unit/node_spec.rb
@@ -1,6 +1,6 @@
 #
 # Author:: Adam Jacob (<adam@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");
@@ -953,14 +953,6 @@ describe Chef::Node do
       expect(node.automatic_attrs[:platform_version]).to eq("23.42")
     end
 
-    it "sets the chef guid attribute correctly" do
-      guid = Chef::Config[:chef_guid]
-      Chef::Config[:chef_guid] = "test-guid-guid"
-      node.consume_external_attrs(@ohai_data, {})
-      expect(node.automatic_attrs[:chef_guid]).to eq("test-guid-guid")
-      Chef::Config[:chef_guid] = guid
-    end
-
     it "consumes the run list from provided json attributes" do
       node.consume_external_attrs(@ohai_data, { "run_list" => ["recipe[unicorn]"] })
       expect(node.run_list).to eq(["recipe[unicorn]"])
diff --git a/spec/unit/resource_reporter_spec.rb b/spec/unit/resource_reporter_spec.rb
index cec931dd70..2272c83967 100644
--- a/spec/unit/resource_reporter_spec.rb
+++ b/spec/unit/resource_reporter_spec.rb
@@ -1,9 +1,9 @@
-#
+
 # Author:: Daniel DeLeo (<dan@chef.io>)
 # Author:: Prajakta Purohit (<prajakta@chef.io>)
 # Author:: Tyler Cloke (<tyler@chef.io>)
 #
-# Copyright:: Copyright 2012-2017, 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");
@@ -24,57 +24,70 @@ require "chef/resource_reporter"
 require "socket"
 
 describe Chef::ResourceReporter do
-  before(:all) do
-    @reporting_toggle_default = Chef::Config[:enable_reporting]
+  before(:each) do
     Chef::Config[:enable_reporting] = true
   end
 
-  after(:all) do
-    Chef::Config[:enable_reporting] = @reporting_toggle_default
-  end
+  let(:node) { Chef::Node.new }
+
+  let(:rest_client) { double("Chef::ServerAPI (mock)") }
+
+  let(:resource_reporter) { Chef::ResourceReporter.new(rest_client) }
+
+  let(:new_resource) { Chef::Resource::File.new("/tmp/a-file.txt") }
+
+  let(:current_resource) { Chef::Resource::File.new("/tmp/a-file.txt") }
+
+  let(:events) { Chef::EventDispatch::Dispatcher.new }
+
+  let(:run_context) { Chef::RunContext.new(node, {}, events) }
+
+  let(:run_status) { Chef::RunStatus.new(node, events) }
+
+  let(:start_time) { Time.new }
+
+  let(:end_time) { Time.new + 20 }
+
+  let(:run_list) { Chef::RunList.new }
+
+  let(:run_id) { run_status.run_id }
+
+  let(:expansion) { Chef::RunList::RunListExpansion.new("_default", run_list.run_list_items) }
+
+  let(:cookbook_name) { "monkey" }
+
+  let(:cookbook_version) { double("Cookbook::Version", version: "1.2.3") }
+
+  let(:action_collection) { Chef::ActionCollection.new(events) }
 
   before do
-    @node = Chef::Node.new
-    @node.name("spitfire")
-    @rest_client = double("Chef::ServerAPI (mock)")
-    allow(@rest_client).to receive(:post).and_return(true)
-    @resource_reporter = Chef::ResourceReporter.new(@rest_client)
-    @new_resource      = Chef::Resource::File.new("/tmp/a-file.txt")
-    @cookbook_name = "monkey"
-    @new_resource.cookbook_name = @cookbook_name
-    @cookbook_version = double("Cookbook::Version", version: "1.2.3")
-    allow(@new_resource).to receive(:cookbook_version).and_return(@cookbook_version)
-    @current_resource = Chef::Resource::File.new("/tmp/a-file.txt")
-    @start_time = Time.new
-    @end_time = Time.new + 20
-    @events = Chef::EventDispatch::Dispatcher.new
-    @run_context = Chef::RunContext.new(@node, {}, @events)
-    @run_status = Chef::RunStatus.new(@node, @events)
-    @run_list = Chef::RunList.new
-    @run_list << "recipe[lobster]" << "role[rage]" << "recipe[fist]"
-    @expansion = Chef::RunList::RunListExpansion.new("_default", @run_list.run_list_items)
-    @run_id = @run_status.run_id
-    allow(Time).to receive(:now).and_return(@start_time, @end_time)
+    node.name("spitfire")
+    allow(rest_client).to receive(:post).and_return(true)
+    new_resource.cookbook_name = cookbook_name
+    allow(new_resource).to receive(:cookbook_version).and_return(cookbook_version)
+    run_list << "recipe[lobster]" << "role[rage]" << "recipe[fist]"
+    allow(Time).to receive(:now).and_return(start_time, end_time)
+    events.register(action_collection)
+    events.register(resource_reporter)
+    events.cookbook_compilation_start(run_context)
   end
 
   context "when first created" do
 
     it "has no updated resources" do
-      expect(@resource_reporter.updated_resources.size).to eq(0)
+      expect(resource_reporter.updated_resources.count).to eq(0)
     end
 
     it "reports a successful run" do
-      expect(@resource_reporter.status).to eq("success")
+      expect(resource_reporter.status).to eq("success")
     end
 
     it "assumes the resource history feature is supported" do
-      expect(@resource_reporter.reporting_enabled?).to be_truthy
+      expect(resource_reporter.send(:reporting_enabled?)).to be_truthy
     end
 
     it "should have no error_descriptions" do
-      expect(@resource_reporter.error_descriptions).to eq({})
-      # @resource_reporter.error_descriptions.should be_empty
-      # @resource_reporter.should have(0).error_descriptions
+      expect(resource_reporter.error_descriptions).to eq({})
     end
 
   end
@@ -86,49 +99,50 @@ describe Chef::ResourceReporter do
 
     it "reports a successful run" do
       skip "refactor how node gets set."
-      expect(@resource_reporter.status).to eq("success")
+      expect(resource_reporter.status).to eq("success")
     end
   end
 
   context "when chef fails" do
     before do
-      allow(@rest_client).to receive(:raw_request).and_return({ "result" => "ok" })
-      allow(@rest_client).to receive(:post).and_return({ "uri" => "https://example.com/reports/nodes/spitfire/runs/#{@run_id}" })
+      allow(rest_client).to receive(:raw_request).and_return({ "result" => "ok" })
+      allow(rest_client).to receive(:post).and_return({ "uri" => "https://example.com/reports/nodes/spitfire/runs/#{@run_id}" })
 
     end
 
     context "before converging any resources" do
+      let(:exception) { Exception.new }
+
       before do
-        @resource_reporter.run_started(@run_status)
-        @exception = Exception.new
-        @resource_reporter.run_failed(@exception)
+        resource_reporter.run_started(run_status)
+        resource_reporter.run_failed(exception)
       end
 
       it "sets the run status to 'failure'" do
-        expect(@resource_reporter.status).to eq("failure")
+        expect(resource_reporter.status).to eq("failure")
       end
 
       it "keeps the exception data" do
-        expect(@resource_reporter.exception).to eq(@exception)
+        expect(resource_reporter.exception).to eq(exception)
       end
     end
 
     context "when a resource fails before loading current state" do
       before do
-        @exception = Exception.new
-        @exception.set_backtrace(caller)
-        @resource_reporter.resource_action_start(@new_resource, :create)
-        @resource_reporter.resource_failed(@new_resource, :create, @exception)
-        @resource_reporter.resource_completed(@new_resource)
+        exception = Exception.new
+        exception.set_backtrace(caller)
+        events.resource_action_start(new_resource, :create)
+        events.resource_failed(new_resource, :create, exception)
+        events.resource_completed(new_resource)
       end
 
       it "collects the resource as an updated resource" do
-        expect(@resource_reporter.updated_resources.size).to eq(1)
+        expect(resource_reporter.updated_resources.count).to eq(1)
       end
 
       it "collects the desired state of the resource" do
-        update_record = @resource_reporter.updated_resources.first
-        expect(update_record.new_resource).to eq(@new_resource)
+        update_record = resource_reporter.updated_resources.first
+        expect(update_record.new_resource).to eq(new_resource)
       end
     end
 
@@ -137,58 +151,73 @@ describe Chef::ResourceReporter do
 
     context "once the a resource's current state is loaded" do
       before do
-        @resource_reporter.resource_action_start(@new_resource, :create)
-        @resource_reporter.resource_current_state_loaded(@new_resource, :create, @current_resource)
+        events.resource_action_start(new_resource, :create)
+        events.resource_current_state_loaded(new_resource, :create, current_resource)
       end
 
       context "and the resource was not updated" do
         before do
-          @resource_reporter.resource_up_to_date(@new_resource, :create)
+          events.resource_up_to_date(new_resource, :create)
+          events.resource_completed(new_resource)
+        end
+
+        it "has no updated resources" do
+          expect(resource_reporter.updated_resources.count).to eq(0)
+        end
+      end
+
+      context "and the resource was skipped" do
+        before do
+          conditional = nil
+          events.resource_skipped(new_resource, :create, conditional)
+          events.resource_completed(new_resource)
         end
 
         it "has no updated resources" do
-          expect(@resource_reporter.updated_resources.size).to eq(0)
+          expect(resource_reporter.updated_resources.count).to eq(0)
         end
       end
 
       context "and the resource was updated" do
         before do
-          @new_resource.content("this is the old content")
-          @current_resource.content("this is the new hotness")
-          @resource_reporter.resource_updated(@new_resource, :create)
-          @resource_reporter.resource_completed(@new_resource)
+          new_resource.content("this is the old content")
+          current_resource.content("this is the new hotness")
+          events.resource_updated(new_resource, :create)
+          events.resource_completed(new_resource)
         end
 
         it "collects the updated resource" do
-          expect(@resource_reporter.updated_resources.size).to eq(1)
+          expect(resource_reporter.updated_resources.count).to eq(1)
         end
 
         it "collects the old state of the resource" do
-          update_record = @resource_reporter.updated_resources.first
-          expect(update_record.current_resource).to eq(@current_resource)
+          update_record = resource_reporter.updated_resources.first
+          expect(update_record.current_resource).to eq(current_resource)
         end
 
         it "collects the new state of the resource" do
-          update_record = @resource_reporter.updated_resources.first
-          expect(update_record.new_resource).to eq(@new_resource)
+          update_record = resource_reporter.updated_resources.first
+          expect(update_record.new_resource).to eq(new_resource)
         end
 
         context "and a subsequent resource fails before loading current resource" do
+          let(:next_new_resource) { Chef::Resource::Service.new("apache2") }
+
           before do
-            @next_new_resource = Chef::Resource::Service.new("apache2")
-            @exception = Exception.new
-            @exception.set_backtrace(caller)
-            @resource_reporter.resource_failed(@next_new_resource, :create, @exception)
-            @resource_reporter.resource_completed(@next_new_resource)
+            exception = Exception.new
+            exception.set_backtrace(caller)
+            events.resource_action_start(next_new_resource, :create)
+            events.resource_failed(next_new_resource, :create, exception)
+            events.resource_completed(next_new_resource)
           end
 
           it "collects the desired state of the failed resource" do
-            failed_resource_update = @resource_reporter.updated_resources.last
-            expect(failed_resource_update.new_resource).to eq(@next_new_resource)
+            failed_resource_update = resource_reporter.updated_resources.last
+            expect(failed_resource_update.new_resource).to eq(next_new_resource)
           end
 
           it "does not have the current state of the failed resource" do
-            failed_resource_update = @resource_reporter.updated_resources.last
+            failed_resource_update = resource_reporter.updated_resources.last
             expect(failed_resource_update.current_resource).to be_nil
           end
         end
@@ -200,56 +229,56 @@ describe Chef::ResourceReporter do
       # used for implementation.
       context "and a nested resource is updated" do
         before do
-          @implementation_resource = Chef::Resource::CookbookFile.new("/preseed-file.txt")
-          @resource_reporter.resource_action_start(@implementation_resource , :create)
-          @resource_reporter.resource_current_state_loaded(@implementation_resource, :create, @implementation_resource)
-          @resource_reporter.resource_updated(@implementation_resource, :create)
-          @resource_reporter.resource_completed(@implementation_resource)
-          @resource_reporter.resource_updated(@new_resource, :create)
-          @resource_reporter.resource_completed(@new_resource)
+          implementation_resource = Chef::Resource::CookbookFile.new("/preseed-file.txt")
+          events.resource_action_start(implementation_resource , :create)
+          events.resource_current_state_loaded(implementation_resource, :create, implementation_resource)
+          events.resource_updated(implementation_resource, :create)
+          events.resource_completed(implementation_resource)
+          events.resource_updated(new_resource, :create)
+          events.resource_completed(new_resource)
         end
 
         it "does not collect data about the nested resource" do
-          expect(@resource_reporter.updated_resources.size).to eq(1)
+          expect(resource_reporter.updated_resources.count).to eq(1)
         end
       end
 
       context "and a nested resource runs but is not updated" do
         before do
-          @implementation_resource = Chef::Resource::CookbookFile.new("/preseed-file.txt")
-          @resource_reporter.resource_action_start(@implementation_resource , :create)
-          @resource_reporter.resource_current_state_loaded(@implementation_resource, :create, @implementation_resource)
-          @resource_reporter.resource_up_to_date(@implementation_resource, :create)
-          @resource_reporter.resource_completed(@implementation_resource)
-          @resource_reporter.resource_updated(@new_resource, :create)
-          @resource_reporter.resource_completed(@new_resource)
+          implementation_resource = Chef::Resource::CookbookFile.new("/preseed-file.txt")
+          events.resource_action_start(implementation_resource , :create)
+          events.resource_current_state_loaded(implementation_resource, :create, implementation_resource)
+          events.resource_up_to_date(implementation_resource, :create)
+          events.resource_completed(implementation_resource)
+          events.resource_updated(new_resource, :create)
+          events.resource_completed(new_resource)
         end
 
         it "does not collect data about the nested resource" do
-          expect(@resource_reporter.updated_resources.size).to eq(1)
+          expect(resource_reporter.updated_resources.count).to eq(1)
         end
       end
 
       context "and the resource failed to converge" do
         before do
-          @exception = Exception.new
-          @exception.set_backtrace(caller)
-          @resource_reporter.resource_failed(@new_resource, :create, @exception)
-          @resource_reporter.resource_completed(@new_resource)
+          exception = Exception.new
+          exception.set_backtrace(caller)
+          events.resource_failed(new_resource, :create, exception)
+          events.resource_completed(new_resource)
         end
 
         it "collects the resource as an updated resource" do
-          expect(@resource_reporter.updated_resources.size).to eq(1)
+          expect(resource_reporter.updated_resources.count).to eq(1)
         end
 
         it "collects the desired state of the resource" do
-          update_record = @resource_reporter.updated_resources.first
-          expect(update_record.new_resource).to eq(@new_resource)
+          update_record = resource_reporter.updated_resources.first
+          expect(update_record.new_resource).to eq(new_resource)
         end
 
         it "collects the current state of the resource" do
-          update_record = @resource_reporter.updated_resources.first
-          expect(update_record.current_resource).to eq(@current_resource)
+          update_record = resource_reporter.updated_resources.first
+          expect(update_record.current_resource).to eq(current_resource)
         end
       end
 
@@ -259,10 +288,10 @@ describe Chef::ResourceReporter do
   describe "when generating a report for the server" do
 
     before do
-      allow(@rest_client).to receive(:raw_request).and_return({ "result" => "ok" })
-      allow(@rest_client).to receive(:post).and_return({ "uri" => "https://example.com/reports/nodes/spitfire/runs/#{@run_id}" })
+      allow(rest_client).to receive(:raw_request).and_return({ "result" => "ok" })
+      allow(rest_client).to receive(:post).and_return({ "uri" => "https://example.com/reports/nodes/spitfire/runs/#{@run_id}" })
 
-      @resource_reporter.run_started(@run_status)
+      resource_reporter.run_started(run_status)
     end
 
     context "when the new_resource is sensitive" do
@@ -271,12 +300,12 @@ describe Chef::ResourceReporter do
         @execute_resource.name("sensitive-resource")
         @execute_resource.command('echo "password: SECRET"')
         @execute_resource.sensitive(true)
-        @resource_reporter.resource_action_start(@execute_resource, :run)
-        @resource_reporter.resource_current_state_loaded(@execute_resource, :run, @current_resource)
-        @resource_reporter.resource_updated(@execute_resource, :run)
-        @resource_reporter.resource_completed(@execute_resource)
-        @run_status.stop_clock
-        @report = @resource_reporter.prepare_run_data
+        events.resource_action_start(@execute_resource, :run)
+        events.resource_current_state_loaded(@execute_resource, :run, current_resource)
+        events.resource_updated(@execute_resource, :run)
+        events.resource_completed(@execute_resource)
+        run_status.stop_clock
+        @report = resource_reporter.prepare_run_data
         @first_update_report = @report["resources"].first
       end
 
@@ -296,12 +325,12 @@ describe Chef::ResourceReporter do
           allow(@bad_resource).to receive(:name).and_return(nil)
           allow(@bad_resource).to receive(:identity).and_return(nil)
           allow(@bad_resource).to receive(:path).and_return(nil)
-          @resource_reporter.resource_action_start(@bad_resource, :create)
-          @resource_reporter.resource_current_state_loaded(@bad_resource, :create, @current_resource)
-          @resource_reporter.resource_updated(@bad_resource, :create)
-          @resource_reporter.resource_completed(@bad_resource)
-          @run_status.stop_clock
-          @report = @resource_reporter.prepare_run_data
+          events.resource_action_start(@bad_resource, :create)
+          events.resource_current_state_loaded(@bad_resource, :create, current_resource)
+          events.resource_updated(@bad_resource, :create)
+          events.resource_completed(@bad_resource)
+          run_status.stop_clock
+          @report = resource_reporter.prepare_run_data
           @first_update_report = @report["resources"].first
         end
 
@@ -320,12 +349,12 @@ describe Chef::ResourceReporter do
           allow(@bad_resource).to receive(:name).and_return({ foo: :bar })
           allow(@bad_resource).to receive(:identity).and_return({ foo: :bar })
           allow(@bad_resource).to receive(:path).and_return({ foo: :bar })
-          @resource_reporter.resource_action_start(@bad_resource, :create)
-          @resource_reporter.resource_current_state_loaded(@bad_resource, :create, @current_resource)
-          @resource_reporter.resource_updated(@bad_resource, :create)
-          @resource_reporter.resource_completed(@bad_resource)
-          @run_status.stop_clock
-          @report = @resource_reporter.prepare_run_data
+          events.resource_action_start(@bad_resource, :create)
+          events.resource_current_state_loaded(@bad_resource, :create, current_resource)
+          events.resource_updated(@bad_resource, :create)
+          events.resource_completed(@bad_resource)
+          run_status.stop_clock
+          @report = resource_reporter.prepare_run_data
           @first_update_report = @report["resources"].first
         end
         # Ruby 1.8.7 flattens out hash to string using join instead of inspect, resulting in
@@ -377,12 +406,12 @@ describe Chef::ResourceReporter do
         #    "status" : "success"
         #    "data" : ""
         # }
-        @resource_reporter.resource_action_start(new_resource, :create)
-        @resource_reporter.resource_current_state_loaded(new_resource, :create, current_resource)
-        @resource_reporter.resource_updated(new_resource, :create)
-        @resource_reporter.resource_completed(new_resource)
-        @run_status.stop_clock
-        @report = @resource_reporter.prepare_run_data
+        events.resource_action_start(new_resource, :create)
+        events.resource_current_state_loaded(new_resource, :create, current_resource)
+        events.resource_updated(new_resource, :create)
+        events.resource_completed(new_resource)
+        run_status.stop_clock
+        @report = resource_reporter.prepare_run_data
         @first_update_report = @report["resources"].first
       end
 
@@ -428,7 +457,7 @@ describe Chef::ResourceReporter do
 
       it "includes the cookbook name of the resource" do
         expect(@first_update_report).to have_key("cookbook_name")
-        expect(@first_update_report["cookbook_name"]).to eq(@cookbook_name)
+        expect(@first_update_report["cookbook_name"]).to eq(cookbook_name)
       end
 
       it "includes the cookbook version of the resource" do
@@ -448,7 +477,7 @@ describe Chef::ResourceReporter do
 
       it "includes the run_list" do
         expect(@report).to have_key("run_list")
-        expect(@report["run_list"]).to eq(Chef::JSONCompat.to_json(@run_status.node.run_list))
+        expect(@report["run_list"]).to eq(Chef::JSONCompat.to_json(run_status.node.run_list))
       end
 
       it "includes the expanded_run_list" do
@@ -457,15 +486,12 @@ describe Chef::ResourceReporter do
 
       it "includes the end_time" do
         expect(@report).to have_key("end_time")
-        expect(@report["end_time"]).to eq(@run_status.end_time.to_s)
+        expect(@report["end_time"]).to eq(run_status.end_time.to_s)
       end
 
     end
 
     context "when the resource is a File" do
-      let(:new_resource) { @new_resource }
-      let(:current_resource) { @current_resource }
-
       it_should_behave_like "a successful client run"
     end
 
@@ -473,8 +499,8 @@ describe Chef::ResourceReporter do
       let(:new_resource) do
         resource = Chef::Resource::RegistryKey.new('Wubba\Lubba\Dub\Dubs')
         resource.values([ { name: "rick", type: :binary, data: 255.chr * 1 } ])
-        allow(resource).to receive(:cookbook_name).and_return(@cookbook_name)
-        allow(resource).to receive(:cookbook_version).and_return(@cookbook_version)
+        allow(resource).to receive(:cookbook_name).and_return(cookbook_name)
+        allow(resource).to receive(:cookbook_version).and_return(cookbook_version)
         resource
       end
 
@@ -491,15 +517,15 @@ describe Chef::ResourceReporter do
 
       before do
         @backtrace = ["foo.rb:1 in `foo!'", "bar.rb:2 in `bar!", "'baz.rb:3 in `baz!'"]
-        @node = Chef::Node.new
-        @node.name("spitfire")
+        node = Chef::Node.new
+        node.name("spitfire")
         @exception = ArgumentError.new
         allow(@exception).to receive(:inspect).and_return("Net::HTTPClientException")
         allow(@exception).to receive(:message).and_return("Object not found")
         allow(@exception).to receive(:backtrace).and_return(@backtrace)
-        @resource_reporter.run_list_expand_failed(@node, @exception)
-        @resource_reporter.run_failed(@exception)
-        @report = @resource_reporter.prepare_run_data
+        resource_reporter.run_list_expand_failed(node, @exception)
+        resource_reporter.run_failed(@exception)
+        @report = resource_reporter.prepare_run_data
       end
 
       it "includes the exception type in the event data" do
@@ -530,29 +556,29 @@ describe Chef::ResourceReporter do
         @bad_resource = Chef::Resource::File.new("/tmp/a-file.txt")
         @bad_resource.cookbook_name = nil
 
-        @resource_reporter.resource_action_start(@bad_resource, :create)
-        @resource_reporter.resource_current_state_loaded(@bad_resource, :create, @current_resource)
-        @resource_reporter.resource_updated(@bad_resource, :create)
-        @resource_reporter.resource_completed(@bad_resource)
-        @run_status.stop_clock
-        @report = @resource_reporter.prepare_run_data
+        events.resource_action_start(@bad_resource, :create)
+        events.resource_current_state_loaded(@bad_resource, :create, current_resource)
+        events.resource_updated(@bad_resource, :create)
+        events.resource_completed(@bad_resource)
+        run_status.stop_clock
+        @report = resource_reporter.prepare_run_data
         @first_update_report = @report["resources"].first
       end
 
       it "includes an updated resource's initial state" do
-        expect(@first_update_report["before"]).to eq(@current_resource.state_for_resource_reporter)
+        expect(@first_update_report["before"]).to eq(current_resource.state_for_resource_reporter)
       end
 
       it "includes an updated resource's final state" do
-        expect(@first_update_report["after"]).to eq(@new_resource.state_for_resource_reporter)
+        expect(@first_update_report["after"]).to eq(new_resource.state_for_resource_reporter)
       end
 
       it "includes the resource's name" do
-        expect(@first_update_report["name"]).to eq(@new_resource.name)
+        expect(@first_update_report["name"]).to eq(new_resource.name)
       end
 
       it "includes the resource's id property" do
-        expect(@first_update_report["id"]).to eq(@new_resource.identity)
+        expect(@first_update_report["id"]).to eq(new_resource.identity)
       end
 
       it "includes the elapsed time for the resource to converge" do
@@ -578,17 +604,17 @@ describe Chef::ResourceReporter do
 
     context "when including a resource that overrides Resource#state" do
       before do
-        @current_state_resource = Chef::Resource::WithState.new("Stateful", @run_context)
+        @current_state_resource = Chef::Resource::WithState.new("Stateful", run_context)
         @current_state_resource.state = nil
 
-        @new_state_resource = Chef::Resource::WithState.new("Stateful", @run_context)
+        @new_state_resource = Chef::Resource::WithState.new("Stateful", run_context)
         @new_state_resource.state = "Running"
-        @resource_reporter.resource_action_start(@new_state_resource, :create)
-        @resource_reporter.resource_current_state_loaded(@new_state_resource, :create, @current_state_resource)
-        @resource_reporter.resource_updated(@new_state_resource, :create)
-        @resource_reporter.resource_completed(@new_state_resource)
-        @run_status.stop_clock
-        @report = @resource_reporter.prepare_run_data
+        events.resource_action_start(@new_state_resource, :create)
+        events.resource_current_state_loaded(@new_state_resource, :create, @current_state_resource)
+        events.resource_updated(@new_state_resource, :create)
+        events.resource_completed(@new_state_resource)
+        run_status.stop_clock
+        @report = resource_reporter.prepare_run_data
         @first_update_report = @report["resources"].first
       end
 
@@ -607,8 +633,8 @@ describe Chef::ResourceReporter do
 
   describe "when updating resource history on the server" do
     before do
-      @resource_reporter.run_started(@run_status)
-      @run_status.start_clock
+      resource_reporter.run_started(run_status)
+      run_status.start_clock
     end
 
     context "when the server does not support storing resource history" do
@@ -616,27 +642,27 @@ describe Chef::ResourceReporter do
         # 404 getting the run_id
         @response = Net::HTTPNotFound.new("a response body", "404", "Not Found")
         @error = Net::HTTPClientException.new("404 message", @response)
-        expect(@rest_client).to receive(:post)
+        expect(rest_client).to receive(:post)
           .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id,
-                                                 start_time: @start_time.to_s },
+                                                 start_time: start_time.to_s },
                { "X-Ops-Reporting-Protocol-Version" => Chef::ResourceReporter::PROTOCOL_VERSION })
           .and_raise(@error)
       end
 
       it "assumes the feature is not enabled" do
-        @resource_reporter.run_started(@run_status)
-        expect(@resource_reporter.reporting_enabled?).to be_falsey
+        resource_reporter.run_started(run_status)
+        expect(resource_reporter.send(:reporting_enabled?)).to be_falsey
       end
 
       it "does not send a resource report to the server" do
-        @resource_reporter.run_started(@run_status)
-        expect(@rest_client).not_to receive(:post)
-        @resource_reporter.run_completed(@node)
+        resource_reporter.run_started(run_status)
+        expect(rest_client).not_to receive(:post)
+        resource_reporter.run_completed(node)
       end
 
       it "prints an error about the 404" do
         expect(Chef::Log).to receive(:trace).with(/404/)
-        @resource_reporter.run_started(@run_status)
+        resource_reporter.run_started(run_status)
       end
 
     end
@@ -646,26 +672,26 @@ describe Chef::ResourceReporter do
         # 500 getting the run_id
         @response = Net::HTTPInternalServerError.new("a response body", "500", "Internal Server Error")
         @error = Net::HTTPClientException.new("500 message", @response)
-        expect(@rest_client).to receive(:post)
-          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: @start_time.to_s },
+        expect(rest_client).to receive(:post)
+          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: start_time.to_s },
                { "X-Ops-Reporting-Protocol-Version" => Chef::ResourceReporter::PROTOCOL_VERSION })
           .and_raise(@error)
       end
 
       it "assumes the feature is not enabled" do
-        @resource_reporter.run_started(@run_status)
-        expect(@resource_reporter.reporting_enabled?).to be_falsey
+        resource_reporter.run_started(run_status)
+        expect(resource_reporter.send(:reporting_enabled?)).to be_falsey
       end
 
       it "does not send a resource report to the server" do
-        @resource_reporter.run_started(@run_status)
-        expect(@rest_client).not_to receive(:post)
-        @resource_reporter.run_completed(@node)
+        resource_reporter.run_started(run_status)
+        expect(rest_client).not_to receive(:post)
+        resource_reporter.run_completed(node)
       end
 
       it "prints an error about the error" do
         expect(Chef::Log).to receive(:info).with(/500/)
-        @resource_reporter.run_started(@run_status)
+        resource_reporter.run_started(run_status)
       end
     end
 
@@ -676,8 +702,8 @@ describe Chef::ResourceReporter do
         # 500 getting the run_id
         @response = Net::HTTPInternalServerError.new("a response body", "500", "Internal Server Error")
         @error = Net::HTTPClientException.new("500 message", @response)
-        expect(@rest_client).to receive(:post)
-          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: @start_time.to_s },
+        expect(rest_client).to receive(:post)
+          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: start_time.to_s },
                { "X-Ops-Reporting-Protocol-Version" => Chef::ResourceReporter::PROTOCOL_VERSION })
           .and_raise(@error)
       end
@@ -689,7 +715,7 @@ describe Chef::ResourceReporter do
       it "fails the run and prints an message about the error" do
         expect(Chef::Log).to receive(:error).with(/500/)
         expect do
-          @resource_reporter.run_started(@run_status)
+          resource_reporter.run_started(run_status)
         end.to raise_error(Net::HTTPClientException)
       end
     end
@@ -697,29 +723,29 @@ describe Chef::ResourceReporter do
     context "after creating the run history document" do
       before do
         response = { "uri" => "https://example.com/reports/nodes/spitfire/runs/@run_id" }
-        expect(@rest_client).to receive(:post)
-          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: @start_time.to_s },
+        expect(rest_client).to receive(:post)
+          .with("reports/nodes/spitfire/runs", { action: :start, run_id: @run_id, start_time: start_time.to_s },
                { "X-Ops-Reporting-Protocol-Version" => Chef::ResourceReporter::PROTOCOL_VERSION })
           .and_return(response)
-        @resource_reporter.run_started(@run_status)
+        resource_reporter.run_started(run_status)
       end
 
       it "creates a run document on the server at the start of the run" do
-        expect(@resource_reporter.run_id).to eq(@run_id)
+        expect(resource_reporter.run_id).to eq(@run_id)
       end
 
       it "updates the run document with resource updates at the end of the run" do
         # update some resources...
-        @resource_reporter.resource_action_start(@new_resource, :create)
-        @resource_reporter.resource_current_state_loaded(@new_resource, :create, @current_resource)
-        @resource_reporter.resource_updated(@new_resource, :create)
+        events.resource_action_start(new_resource, :create)
+        events.resource_current_state_loaded(new_resource, :create, current_resource)
+        events.resource_updated(new_resource, :create)
 
-        allow(@resource_reporter).to receive(:end_time).and_return(@end_time)
-        @expected_data = @resource_reporter.prepare_run_data
+        allow(resource_reporter).to receive(:end_time).and_return(end_time)
+        @expected_data = resource_reporter.prepare_run_data
 
         response = { "result" => "ok" }
 
-        expect(@rest_client).to receive(:raw_request).ordered do |method, url, headers, data|
+        expect(rest_client).to receive(:raw_request).ordered do |method, url, headers, data|
           expect(method).to eq(:POST)
           expect(headers).to eq({ "Content-Encoding" => "gzip",
                                   "X-Ops-Reporting-Protocol-Version" => Chef::ResourceReporter::PROTOCOL_VERSION,
@@ -730,7 +756,7 @@ describe Chef::ResourceReporter do
           response
         end
 
-        @resource_reporter.run_completed(@node)
+        resource_reporter.run_completed(node)
       end
     end
 
@@ -747,34 +773,34 @@ describe Chef::ResourceReporter do
       it "should log 4xx errors" do
         response = Net::HTTPClientError.new("forbidden", "403", "Forbidden")
         error = Net::HTTPClientException.new("403 message", response)
-        allow(@rest_client).to receive(:raw_request).and_raise(error)
+        allow(rest_client).to receive(:raw_request).and_raise(error)
         expect(Chef::Log).to receive(:error).with(/403/)
 
-        @resource_reporter.post_reporting_data
+        resource_reporter.post_reporting_data
       end
 
       it "should log error 5xx errors" do
         response = Net::HTTPServerError.new("internal error", "500", "Internal Server Error")
         error = Net::HTTPFatalError.new("500 message", response)
-        allow(@rest_client).to receive(:raw_request).and_raise(error)
+        allow(rest_client).to receive(:raw_request).and_raise(error)
         expect(Chef::Log).to receive(:error).with(/500/)
 
-        @resource_reporter.post_reporting_data
+        resource_reporter.post_reporting_data
       end
 
       it "should log if a socket error happens" do
-        allow(@rest_client).to receive(:raw_request).and_raise(SocketError.new("test socket error"))
+        allow(rest_client).to receive(:raw_request).and_raise(SocketError.new("test socket error"))
         expect(Chef::Log).to receive(:error).with(/test socket error/)
 
-        @resource_reporter.post_reporting_data
+        resource_reporter.post_reporting_data
 
       end
 
       it "should raise if an unkwown error happens" do
-        allow(@rest_client).to receive(:raw_request).and_raise(Exception.new)
+        allow(rest_client).to receive(:raw_request).and_raise(Exception.new)
 
         expect do
-          @resource_reporter.post_reporting_data
+          resource_reporter.post_reporting_data
         end.to raise_error(Exception)
       end
     end