Implement live streaming for execute resources

This brings live streaming of execute resource output to the
output formatters. It also adds a mechanism for checking to
see if an output formatter is in use through the event dispatch
system.

It adds a new configuration option, "always_stream_execute", which
does what it says on the tin.

8 files changed, 111 insertions, 27 deletions

diff --git a/chef-config/lib/chef-config/config.rb b/chef-config/lib/chef-config/config.rb
index 069f0ed6c0..6a028e9564 100644
--- a/chef-config/lib/chef-config/config.rb
+++ b/chef-config/lib/chef-config/config.rb
@@ -271,6 +271,9 @@ module ChefConfig
     # Using `force_logger` causes chef to default to logger output when STDOUT is a tty
     default :force_logger, false
 
+    # Using 'always_stream_output' will have Chef always stream the execute output
+    default :always_stream_output, false
+
     default :http_retry_count, 5
     default :http_retry_delay, 5
     default :interval, nil
diff --git a/chef-config/spec/unit/config_spec.rb b/chef-config/spec/unit/config_spec.rb
index d99ff428fb..fc8528ad46 100644
--- a/chef-config/spec/unit/config_spec.rb
+++ b/chef-config/spec/unit/config_spec.rb
@@ -264,6 +264,10 @@ RSpec.describe ChefConfig::Config do
           end
         end
 
+        it "ChefConfig::Config[:always_stream_output] defaults to false" do
+          expect(ChefConfig::Config[:always_stream_output]).to eq(false)
+        end
+
         it "ChefConfig::Config[:file_backup_path] defaults to /var/chef/backup" do
           allow(ChefConfig::Config).to receive(:cache_path).and_return(primary_cache_path)
           backup_path = is_windows ? "#{primary_cache_path}\\backup" : "#{primary_cache_path}/backup"
diff --git a/lib/chef/event_dispatch/dispatcher.rb b/lib/chef/event_dispatch/dispatcher.rb
index f3e55539a9..affac8fb9d 100644
--- a/lib/chef/event_dispatch/dispatcher.rb
+++ b/lib/chef/event_dispatch/dispatcher.rb
@@ -20,6 +20,16 @@ class Chef
         @subscribers << subscriber
       end
 
+      # Check to see if we are dispatching to a formatter
+      def formatter?
+        @subscribers.each do |s|
+          if s.class <= Chef::Formatters::Base && s.class != Chef::Formatters::NullFormatter
+            return true
+          end
+        end
+        false
+      end
+
       ####
       # All messages are unconditionally forwarded to all subscribers, so just
       # define the forwarding in one go:
diff --git a/lib/chef/event_dispatch/events_output_stream.rb b/lib/chef/event_dispatch/events_output_stream.rb
index 8de9b0fed1..d9c21642b7 100644
--- a/lib/chef/event_dispatch/events_output_stream.rb
+++ b/lib/chef/event_dispatch/events_output_stream.rb
@@ -21,6 +21,14 @@ class Chef
         events.stream_output(self, str, options)
       end
 
+      def <<(str)
+        events.stream_output(self, str, options)
+      end
+
+      def write(str)
+        events.stream_output(self, str, options)
+      end
+
       def close
         events.stream_closed(self, options)
       end
diff --git a/lib/chef/formatters/indentable_output_stream.rb b/lib/chef/formatters/indentable_output_stream.rb
index 1beb286e7f..f7f470b190 100644
--- a/lib/chef/formatters/indentable_output_stream.rb
+++ b/lib/chef/formatters/indentable_output_stream.rb
@@ -50,6 +50,11 @@ class Chef
         print(string, from_args(args, :start_line => true, :end_line => true))
       end
 
+      # Print a raw chunk
+      def <<(obj)
+        print(obj)
+      end
+
       # Print a string.
       #
       # == Arguments
diff --git a/lib/chef/provider/execute.rb b/lib/chef/provider/execute.rb
index 30de0d3b9e..1091b82932 100644
--- a/lib/chef/provider/execute.rb
+++ b/lib/chef/provider/execute.rb
@@ -89,7 +89,9 @@ class Chef
         opts[:umask]       = umask if umask
         opts[:log_level]   = :info
         opts[:log_tag]     = new_resource.to_s
-        if STDOUT.tty? && !Chef::Config[:daemon] && Chef::Log.info? && !sensitive?
+        if (Chef::Config[:always_stream_execute] || run_context.events.formatter?) && !sensitive?
+          opts[:live_stream] = Chef::EventDispatch::EventsOutputStream.new(run_context.events, :name => :execute)
+        elsif STDOUT.tty? && !Chef::Config[:daemon] && Chef::Log.info? && !sensitive?
           opts[:live_stream] = STDOUT
         end
         opts
diff --git a/spec/unit/provider/execute_spec.rb b/spec/unit/provider/execute_spec.rb
index e7607d9417..69fd7ae655 100644
--- a/spec/unit/provider/execute_spec.rb
+++ b/spec/unit/provider/execute_spec.rb
@@ -25,28 +25,32 @@ describe Chef::Provider::Execute do
   let(:run_context) { Chef::RunContext.new(node, {}, events) }
   let(:provider) { Chef::Provider::Execute.new(new_resource, run_context) }
   let(:current_resource) { Chef::Resource::Ifconfig.new("foo_resource", run_context) }
+  # You will be the same object, I promise.
+  @live_stream = Chef::EventDispatch::EventsOutputStream.new(run_context.events, :name => :execute)
 
   let(:opts) do
     {
       timeout:      3600,
       returns:      0,
       log_level:    :info,
-      log_tag:      new_resource.to_s,
-      live_stream:  STDOUT,
+      log_tag:      new_resource.to_s
     }
   end
 
   let(:new_resource) { Chef::Resource::Execute.new("foo_resource", run_context) }
 
   before do
+    allow(Chef::EventDispatch::EventsOutputStream).to receive(:new) { @live_stream }
     allow(ChefConfig).to receive(:windows?) { false }
     @original_log_level = Chef::Log.level
     Chef::Log.level = :info
-    allow(STDOUT).to receive(:tty?).and_return(true)
+    allow(STDOUT).to receive(:tty?).and_return(false)
   end
 
   after do
     Chef::Log.level = @original_log_level
+    Chef::Config[:always_stream_execute] = false
+    Chef::Config[:daemon] = false
   end
 
   describe "#initialize" do
@@ -142,44 +146,92 @@ describe Chef::Provider::Execute do
       expect(new_resource).not_to be_updated
     end
 
-    it "should unset the live_stream if STDOUT is not a tty" do
-      expect(STDOUT).to receive(:tty?).and_return(false)
+    it "should not include stdout/stderr in failure exception for sensitive resource" do
       opts.delete(:live_stream)
-      expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
-      expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
-      expect(Chef::Log).not_to receive(:warn)
-      provider.run_action(:run)
-      expect(new_resource).to be_updated
+      new_resource.sensitive true
+      expect(provider).to receive(:shell_out!).and_raise(Mixlib::ShellOut::ShellCommandFailed)
+      expect do
+        provider.run_action(:run)
+      end.to raise_error(Mixlib::ShellOut::ShellCommandFailed, /suppressed for sensitive resource/)
     end
 
-    it "should unset the live_stream if chef is running as a daemon" do
-      allow(Chef::Config).to receive(:[]).and_call_original
-      expect(Chef::Config).to receive(:[]).with(:daemon).and_return(true)
-      opts.delete(:live_stream)
-      expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
+    it "should set the live_stream if Chef::Config[:always_stream_execute] is set" do
+      Chef::Config[:always_stream_execute] = true
+      nopts = opts
+      nopts[:live_stream] = @live_stream
+      expect(provider).to receive(:shell_out!).with(new_resource.name, nopts)
       expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
       expect(Chef::Log).not_to receive(:warn)
       provider.run_action(:run)
       expect(new_resource).to be_updated
     end
 
-    it "should unset the live_stream if we are not running with a log level of at least :info" do
-      expect(Chef::Log).to receive(:info?).and_return(false)
-      opts.delete(:live_stream)
+    it "should not set the live_stream if Chef::Config[:always_stream_execute] is set but sensitive is on" do
+      Chef::Config[:always_stream_execute] = true
+      new_resource.sensitive true
       expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
-      expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
+      expect(provider).to receive(:converge_by).with("execute sensitive resource").and_call_original
       expect(Chef::Log).not_to receive(:warn)
       provider.run_action(:run)
       expect(new_resource).to be_updated
     end
 
-    it "should not include stdout/stderr in failure exception for sensitive resource" do
-      opts.delete(:live_stream)
-      new_resource.sensitive true
-      expect(provider).to receive(:shell_out!).and_raise(Mixlib::ShellOut::ShellCommandFailed)
-      expect do
+    describe "with an output formatter listening" do
+      let(:events) { d = Chef::EventDispatch::Dispatcher.new; d.register(Chef::Formatters::Doc.new(StringIO.new, StringIO.new)); d }
+
+      it "should set the live_stream" do
+        nopts = opts
+        nopts[:live_stream] = @live_stream
+        expect(provider).to receive(:shell_out!).with(new_resource.name, nopts)
+        expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
+        expect(Chef::Log).not_to receive(:warn)
         provider.run_action(:run)
-      end.to raise_error(Mixlib::ShellOut::ShellCommandFailed, /suppressed for sensitive resource/)
+        expect(new_resource).to be_updated
+      end
+
+      it "should not set the live_stream if the resource is sensitive" do
+        new_resource.sensitive true
+        expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
+        expect(provider).to receive(:converge_by).with("execute sensitive resource").and_call_original
+        expect(Chef::Log).not_to receive(:warn)
+        provider.run_action(:run)
+        expect(new_resource).to be_updated
+      end
+    end
+
+    describe "with only logging enabled" do
+      it "should set the live_stream to STDOUT if we are a TTY, not daemonized, not sensitive, and info is enabled" do
+        nopts = opts
+        nopts[:live_stream] = STDOUT
+        allow(STDOUT).to receive(:tty?).and_return(true)
+        expect(provider).to receive(:shell_out!).with(new_resource.name, nopts)
+        expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
+        expect(Chef::Log).not_to receive(:warn)
+        provider.run_action(:run)
+        expect(new_resource).to be_updated
+      end
+
+      it "should not set the live_stream to STDOUT if we are a TTY, not daemonized, but sensitive" do
+        new_resource.sensitive true
+        allow(STDOUT).to receive(:tty?).and_return(true)
+        expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
+        expect(provider).to receive(:converge_by).with("execute sensitive resource").and_call_original
+        expect(Chef::Log).not_to receive(:warn)
+        provider.run_action(:run)
+        expect(new_resource).to be_updated
+      end
+
+      it "should not set the live_stream to STDOUT if we are a TTY, but daemonized" do
+        Chef::Config[:daemon] = true
+        allow(STDOUT).to receive(:tty?).and_return(true)
+        expect(provider).to receive(:shell_out!).with(new_resource.name, opts)
+        expect(provider).to receive(:converge_by).with("execute foo_resource").and_call_original
+        expect(Chef::Log).not_to receive(:warn)
+        provider.run_action(:run)
+        expect(new_resource).to be_updated
+      end
+
     end
+
   end
 end
diff --git a/spec/unit/provider/script_spec.rb b/spec/unit/provider/script_spec.rb
index d1759981aa..423c1f51a1 100644
--- a/spec/unit/provider/script_spec.rb
+++ b/spec/unit/provider/script_spec.rb
@@ -88,7 +88,7 @@ describe Chef::Provider::Script, "action_run" do
 
     describe "when running the script" do
       let (:default_opts) {
-        {timeout: 3600, returns: 0, log_level: :info, log_tag: "script[run some perl code]", live_stream: STDOUT}
+        {timeout: 3600, returns: 0, log_level: :info, log_tag: "script[run some perl code]",}
       }
 
       before do