Merge pull request #4040 from chef/tm/live_execute_streaming

Implement live streaming for execute resources

11 files changed, 153 insertions, 38 deletions

diff --git a/chef-config/lib/chef-config/config.rb b/chef-config/lib/chef-config/config.rb
index 069f0ed6c0..4e9355192a 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 'stream_execute_output' will have Chef always stream the execute output
+    default :stream_execute_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..8e9a499a1a 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[:stream_execute_output] defaults to false" do
+          expect(ChefConfig::Config[:stream_execute_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..ad7df46cd0 100644
--- a/lib/chef/event_dispatch/dispatcher.rb
+++ b/lib/chef/event_dispatch/dispatcher.rb
@@ -20,6 +20,11 @@ class Chef
         @subscribers << subscriber
       end
 
+      # Check to see if we are dispatching to a formatter
+      def formatter?
+        @subscribers.any? { |s| s.respond_to?(:is_formatter?) && s.is_formatter? }
+      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/base.rb b/lib/chef/formatters/base.rb
index d3756ef00c..4c393f7a48 100644
--- a/lib/chef/formatters/base.rb
+++ b/lib/chef/formatters/base.rb
@@ -215,6 +215,10 @@ class Chef
       def deprecation(message, location=caller(2..2)[0])
         Chef::Log.deprecation("#{message} at #{location}")
       end
+
+      def is_formatter?
+        true
+      end
     end
 
 
@@ -225,6 +229,9 @@ class Chef
 
       cli_name(:null)
 
+      def is_formatter?
+        false
+      end
     end
 
   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..200beb02ad 100644
--- a/lib/chef/provider/execute.rb
+++ b/lib/chef/provider/execute.rb
@@ -78,6 +78,14 @@ class Chef
         !!new_resource.sensitive
       end
 
+      def live_stream?
+        Chef::Config[:stream_execute_output] || !!new_resource.live_stream
+      end
+
+      def stream_to_stdout?
+        STDOUT.tty? && !Chef::Config[:daemon]
+      end
+
       def opts
         opts = {}
         opts[:timeout]     = timeout
@@ -89,8 +97,12 @@ 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?
-          opts[:live_stream] = STDOUT
+        if (Chef::Log.info? || live_stream?) && !sensitive?
+          if run_context.events.formatter?
+            opts[:live_stream] = Chef::EventDispatch::EventsOutputStream.new(run_context.events, :name => :execute)
+          elsif stream_to_stdout?
+            opts[:live_stream] = STDOUT
+          end
         end
         opts
       end
diff --git a/lib/chef/resource/execute.rb b/lib/chef/resource/execute.rb
index 11c4ae045c..4f92a7ad35 100644
--- a/lib/chef/resource/execute.rb
+++ b/lib/chef/resource/execute.rb
@@ -49,6 +49,7 @@ class Chef
         @umask = nil
         @default_guard_interpreter = :execute
         @is_guard_interpreter = false
+        @live_stream = false
       end
 
       def umask(arg=nil)
@@ -101,6 +102,13 @@ class Chef
         )
       end
 
+      def live_stream(arg=nil)
+        set_or_return(
+          :live_stream,
+          arg,
+          :kind_of => [ TrueClass, FalseClass ])
+      end
+
       def path(arg=nil)
         Chef::Log.warn "The 'path' attribute of 'execute' is not used by any provider in Chef 11 or Chef 12. Use 'environment' attribute to configure 'PATH'. This attribute will be removed in Chef 13."
 
diff --git a/spec/support/shared/unit/execute_resource.rb b/spec/support/shared/unit/execute_resource.rb
index e969a2ebd5..3a88ff8890 100644
--- a/spec/support/shared/unit/execute_resource.rb
+++ b/spec/support/shared/unit/execute_resource.rb
@@ -111,6 +111,11 @@ shared_examples_for "an execute resource" do
     expect(@resource.creates).to eql("something")
   end
 
+  it "should accept a boolean for live streaming" do
+    @resource.live_stream true
+    expect(@resource.live_stream).to be true
+  end
+
   describe "when it has cwd, environment, group, path, return value, and a user" do
     before do
       @resource.command("grep")
diff --git a/spec/unit/provider/execute_spec.rb b/spec/unit/provider/execute_spec.rb
index e7607d9417..3af35a17ca 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,37 +146,6 @@ 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)
-      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
-    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)
-      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)
-      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
-
     it "should not include stdout/stderr in failure exception for sensitive resource" do
       opts.delete(:live_stream)
       new_resource.sensitive true
@@ -181,5 +154,90 @@ describe Chef::Provider::Execute do
         provider.run_action(:run)
       end.to raise_error(Mixlib::ShellOut::ShellCommandFailed, /suppressed for sensitive resource/)
     end
+
+    describe "streaming output" do
+      it "should not set the live_stream if sensitive is on" 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
+
+      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 }
+
+        before do
+          Chef::Config[:stream_execute_output] = true
+        end
+
+        it "should set the live_stream if the log level is info or above" 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)
+          expect(new_resource).to be_updated
+        end
+
+        it "should set the live_stream if the resource requests live streaming" do
+          Chef::Log.level = :warn
+          new_resource.live_stream 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 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
 end
diff --git a/spec/unit/provider/script_spec.rb b/spec/unit/provider/script_spec.rb
index d1759981aa..7cc5abbd15 100644
--- a/spec/unit/provider/script_spec.rb
+++ b/spec/unit/provider/script_spec.rb
@@ -88,11 +88,11 @@ 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
-        allow(STDOUT).to receive(:tty?).and_return(true)
+        allow(STDOUT).to receive(:tty?).and_return(false)
       end
 
       it 'should set the command to "interpreter"  "tempfile"' do