Fix #3373 - Client side file streaming

This patch moves file content writing to the content properties and
always write (or read) contents by chunks.

This reduces drastically puppetd memory consumption when handling large
sourced files.

Signed-off-by: Brice Figureau <brice-puppet@daysofwonder.com>

6 files changed, 135 insertions, 43 deletions

diff --git a/lib/puppet/type/file.rb b/lib/puppet/type/file.rb
index 67bba5c0a..e94761503 100644
--- a/lib/puppet/type/file.rb
+++ b/lib/puppet/type/file.rb
@@ -719,35 +719,33 @@ Puppet::Type.newtype(:file) do
         obj
     end
 
-    # Write out the file.  Requires the content to be written,
-    # the property name for logging, and the checksum for validation.
-    def write(content, property, checksum = nil)
+    # Write out the file.  Requires the property name for logging.
+    # Write will be done by the content property, along with checksum computation
+    def write(property)
         remove_existing(:file)
 
-        use_temporary_file = (content.length != 0)
+        use_temporary_file = write_temporary_file?
         if use_temporary_file
             path = "#{self[:path]}.puppettmp_#{rand(10000)}"
             while File.exists?(path) or File.symlink?(path)
                 path = "#{self[:path]}.puppettmp_#{rand(10000)}"
-                end
-          else
+            end
+        else
             path = self[:path]
-          end
+        end
 
         mode = self.should(:mode) # might be nil
         umask = mode ? 000 : 022
 
-        Puppet::Util.withumask(umask) do
-            File.open(path, File::CREAT|File::WRONLY|File::TRUNC, mode) { |f| f.print content }
-        end
+        content_checksum = Puppet::Util.withumask(umask) { File.open(path, 'w', mode) { |f| write_content(f) } }
 
         # And put our new file in place
         if use_temporary_file # This is only not true when our file is empty.
             begin
-                fail_if_checksum_is_wrong(path, content) if validate_checksum?
+                fail_if_checksum_is_wrong(path, content_checksum) if validate_checksum?
                 File.rename(path, self[:path])
             rescue => detail
-                fail "Could not rename temporary file %s to %s : %s" % [path, self[:path], detail]
+                fail "Could not rename temporary file %s to %s: %s" % [path, self[:path], detail]
             ensure
                 # Make sure the created file gets removed
                 File.unlink(path) if FileTest.exists?(path)
@@ -761,19 +759,32 @@ Puppet::Type.newtype(:file) do
 
     private
 
+    # Should we validate the checksum of the file we're writing?
     def validate_checksum?
-        false
+         self[:checksum] !~ /time/
     end
 
     # Make sure the file we wrote out is what we think it is.
-    def fail_if_checksum_is_wrong(path, checksum)
-        # Use the appropriate checksum type -- md5, md5lite, etc.
-        checksum = parameter(:checksum).sum(content)
-
+    def fail_if_checksum_is_wrong(path, content_checksum)
         newsum = parameter(:checksum).sum_file(path)
-        return if [:absent, nil, checksum].include?(newsum)
+        return if [:absent, nil, content_checksum].include?(newsum)
+
+        self.fail "File written to disk did not match checksum; discarding changes (%s vs %s)" % [content_checksum, newsum]
+    end
+
+    # write the current content. Note that if there is no content property
+    # simply opening the file with 'w' as done in write is enough to truncate
+    # or write an empty length file.
+    def write_content(file)
+        (content = property(:content)) && content.write(file)
+    end
+
+    private
 
-        self.fail "File written to disk did not match checksum; discarding changes (%s vs %s)" % [checksum, newsum]
+    def write_temporary_file?
+        # unfortunately we don't know the source file size before fetching it
+        # so let's assume the file won't be empty 
+        (c = property(:content) and c.length) || (s = @parameters[:source] and 1)
     end
 
     # There are some cases where all of the work does not get done on
diff --git a/lib/puppet/type/file/checksum.rb b/lib/puppet/type/file/checksum.rb
index d4c24886c..3e2fdbf09 100755
--- a/lib/puppet/type/file/checksum.rb
+++ b/lib/puppet/type/file/checksum.rb
@@ -23,4 +23,11 @@ Puppet::Type.type(:file).newparam(:checksum) do
         method = type.to_s + "_file"
         "{#{type}}" + send(method, path).to_s
     end
+
+    def sum_stream(&block)
+        type = value || :md5 # same comment as above
+        method = type.to_s + "_stream"
+        checksum = send(method, &block)
+        "{#{type}}#{checksum}"
+    end
 end
diff --git a/lib/puppet/type/file/content.rb b/lib/puppet/type/file/content.rb
index cc2494b00..1817d5646 100755
--- a/lib/puppet/type/file/content.rb
+++ b/lib/puppet/type/file/content.rb
@@ -1,9 +1,16 @@
+require 'net/http'
+require 'uri'
+
 require 'puppet/util/checksums'
+require 'puppet/network/http/api/v1'
 
 module Puppet
     Puppet::Type.type(:file).newproperty(:content) do
         include Puppet::Util::Diff
         include Puppet::Util::Checksums
+        include Puppet::Network::HTTP::API::V1
+
+        attr_reader :actual_content
 
         desc "Specify the contents of a file as a string.  Newlines, tabs, and
             spaces can be specified using the escaped syntax (e.g., \\n for a
@@ -68,21 +75,12 @@ module Puppet
             end
         end
 
-        # If content was specified, return that; else try to return the source content;
-        # else, return nil.
-        def actual_content
-            if defined?(@actual_content) and @actual_content
-                return @actual_content
-            end
-
-            if s = resource.parameter(:source)
-                return s.content
-            end
-            fail "Could not find actual content from checksum"
+        def length
+            (actual_content and actual_content.length) || 0
         end
 
         def content
-            self.should || (s = resource.parameter(:source) and s.content)
+            self.should
         end
 
         # Override this method to provide diffs if asked for.
@@ -132,9 +130,46 @@ module Puppet
             # We're safe not testing for the 'source' if there's no 'should'
             # because we wouldn't have gotten this far if there weren't at least
             # one valid value somewhere.
-            @resource.write(actual_content, :content)
+            @resource.write(:content)
 
             return return_event
         end
+
+        def write(file)
+            self.fail "Writing content that wasn't provided" unless actual_content || resource.parameter(:source)
+            resource.parameter(:checksum).sum_stream { |sum|
+                each_chunk_from(actual_content || resource.parameter(:source)) { |chunk|
+                    sum << chunk
+                    file.print chunk
+                }
+            }
+        end
+
+        def each_chunk_from(source_or_content)
+            if source_or_content.is_a?(String)
+                yield source_or_content
+            elsif source_or_content.nil?
+                nil
+            elsif source_or_content.local?
+                File.open(source_or_content.full_path, "r") do |src|
+                    while chunk = src.read(8192)
+                        yield chunk
+                    end
+                end
+            else
+                request = Puppet::Indirector::Request.new(:file_content, :find, source_or_content.full_path)
+                connection = Puppet::Network::HttpPool.http_instance(source_or_content.server, source_or_content.port)
+                connection.request_get(indirection2uri(request), {"Accept" => "raw"}) do |response|
+                    case response.code
+                    when "404"; nil
+                    when /^2/;  response.read_body { |chunk| yield chunk }
+                    else
+                        # Raise the http error if we didn't get a 'success' of some kind.
+                        message = "Error %s on SERVER: %s" % [response.code, (response.body||'').empty? ? response.message : response.body]
+                        raise Net::HTTPError.new(message, response)
+                    end
+                end
+            end
+        end
     end
 end
diff --git a/lib/puppet/type/file/ensure.rb b/lib/puppet/type/file/ensure.rb
index b2dd39459..83f3d3e6a 100755
--- a/lib/puppet/type/file/ensure.rb
+++ b/lib/puppet/type/file/ensure.rb
@@ -45,7 +45,7 @@ module Puppet
             if property = @resource.property(:content)
                 property.sync
             else
-                @resource.write("", :ensure)
+                @resource.write(:ensure)
                 mode = @resource.should(:mode)
             end
         end
diff --git a/lib/puppet/type/file/source.rb b/lib/puppet/type/file/source.rb
index b3d9b3ec2..1eb418e90 100755
--- a/lib/puppet/type/file/source.rb
+++ b/lib/puppet/type/file/source.rb
@@ -96,16 +96,6 @@ module Puppet
             metadata && metadata.checksum
         end
 
-        # Look up (if necessary) and return remote content.
-        cached_attr(:content) do
-            raise Puppet::DevError, "No source for content was stored with the metadata" unless metadata.source
-
-            unless tmp = Puppet::FileServing::Content.find(metadata.source)
-                fail "Could not find any content at %s" % metadata.source
-            end
-            tmp.content
-        end
-
         # Copy the values from the source to the resource.  Yay.
         def copy_source_values
             devfail "Somehow got asked to copy source values without any metadata" unless metadata
@@ -175,5 +165,29 @@ module Puppet
             resource[:check] = checks
             resource[:checksum] = :md5 unless resource.property(:checksum)
         end
+
+        def local?
+            found? and uri and (uri.scheme || "file") == "file"
+        end
+
+        def full_path
+            if found? and uri
+                return URI.unescape(uri.path)
+            end
+        end
+
+        def server
+            (uri and uri.host) or Puppet.settings[:server]
+        end
+
+        def port
+            (uri and uri.port) or Puppet.settings[:masterport]
+        end
+
+        private
+
+        def uri
+            @uri ||= URI.parse(URI.escape(metadata.source))
+        end
     end
 end
diff --git a/lib/puppet/util/checksums.rb b/lib/puppet/util/checksums.rb
index f68f77624..e534fb0fb 100644
--- a/lib/puppet/util/checksums.rb
+++ b/lib/puppet/util/checksums.rb
@@ -39,11 +39,27 @@ module Puppet::Util::Checksums
         md5_file(filename, true)
     end
 
+    def md5_stream(&block)
+        require 'digest/md5'
+        digest = Digest::MD5.new()
+        yield digest
+        return digest.hexdigest
+    end
+
+    alias :md5lite_stream :md5_stream
+
     # Return the :mtime timestamp of a file.
     def mtime_file(filename)
         File.stat(filename).send(:mtime)
     end
 
+    # by definition this doesn't exist
+    def mtime_stream
+        nil
+    end
+
+    alias :ctime_stream :mtime_stream
+
     # Calculate a checksum using Digest::SHA1.
     def sha1(content)
         require 'digest/sha1'
@@ -68,6 +84,15 @@ module Puppet::Util::Checksums
         sha1_file(filename, true)
     end
 
+    def sha1_stream
+        require 'digest/sha1'
+        digest = Digest::SHA1.new()
+        yield digest
+        return digest.hexdigest
+    end
+
+    alias :sha1lite_stream :sha1_stream
+
     # Return the :ctime of a file.
     def ctime_file(filename)
         File.stat(filename).send(:ctime)