Switching transactions to callback-based events

Events are now queued as they are created, and
the queues are managed through simple interfaces,
rather than collecting events over time and
responding to them inline.

This drastically simplifies event management,
and will make moving it to a separate system
essentially trivial.

Signed-off-by: Luke Kanies <luke@madstop.com>

3 files changed, 468 insertions, 22 deletions

diff --git a/spec/integration/transaction.rb b/spec/integration/transaction.rb
index 7ab852bd2..1b8b11d48 100755
--- a/spec/integration/transaction.rb
+++ b/spec/integration/transaction.rb
@@ -2,9 +2,12 @@
 
 require File.dirname(__FILE__) + '/../spec_helper'
 
+require 'puppet_spec/files'
 require 'puppet/transaction'
 
 describe Puppet::Transaction do
+    include PuppetSpec::Files
+
     it "should not apply generated resources if the parent resource fails" do
         catalog = Puppet::Resource::Catalog.new
         resource = Puppet::Type.type(:file).new :path => "/foo/bar", :backup => false
@@ -63,4 +66,119 @@ describe Puppet::Transaction do
         transaction.evaluate
     end
 
+    it "should refresh resources that subscribe to changed resources" do
+        name = tmpfile("something")
+        file = Puppet::Type.type(:file).new(
+            :name => name,
+            :ensure => "file"
+        )
+        exec = Puppet::Type.type(:exec).new(
+            :name => "echo true",
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :subscribe => Puppet::Resource::Reference.new(file.class.name, file.name)
+        )
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource file, exec
+
+        exec.expects(:refresh)
+
+        catalog.apply
+    end
+
+    it "should not refresh resources that only require changed resources" do
+        name = tmpfile("something")
+        file = Puppet::Type.type(:file).new(
+            :name => name,
+            :ensure => "file"
+        )
+        exec = Puppet::Type.type(:exec).new(
+            :name => "echo true",
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :require => Puppet::Resource::Reference.new(file.class.name, file.name)
+        )
+
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource file
+        catalog.add_resource exec
+
+        exec.expects(:refresh).never
+
+        trans = catalog.apply
+
+        trans.events.length.should == 1
+    end
+
+    it "should cascade events such that multiple refreshes result" do
+        files = []
+
+        4.times { |i|
+            files << Puppet::Type.type(:file).new(
+                :name => tmpfile("something"),
+                :ensure => "file"
+            )
+        }
+
+        fname = tmpfile("something")
+        exec = Puppet::Type.type(:exec).new(
+            :name => "touch %s" % fname,
+            :path => "/usr/bin:/bin",
+            :refreshonly => true
+        )
+
+        exec[:subscribe] = files.collect { |f|
+            Puppet::Resource::Reference.new(:file, f.name)
+        }
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource(exec, *files)
+
+        catalog.apply
+        FileTest.should be_exist(fname)
+    end
+
+    # Make sure refreshing happens mid-transaction, rather than at the end.
+    it "should refresh resources as they're encountered rather than all at the end" do
+        file = tmpfile("something")
+
+        exec1 = Puppet::Type.type(:exec).new(
+            :title => "one",
+            :name => "echo one >> %s" % file,
+            :path => "/usr/bin:/bin"
+        )
+
+        exec2 = Puppet::Type.type(:exec).new(
+            :title => "two",
+            :name => "echo two >> %s" % file,
+            :path => "/usr/bin:/bin",
+            :refreshonly => true,
+            :subscribe => exec1
+        )
+
+        exec3 = Puppet::Type.type(:exec).new(
+            :title => "three",
+            :name => "echo three >> %s" % file,
+            :path => "/usr/bin:/bin",
+            :require => exec2
+        )
+        execs = [exec1, exec2, exec3]
+
+        catalog = Puppet::Resource::Catalog.new
+        catalog.add_resource(exec1,exec2,exec3)
+
+        trans = Puppet::Transaction.new(catalog)
+        execs.each { |e| catalog.should be_vertex(e) }
+        trans.prepare
+        execs.each { |e| catalog.should be_vertex(e) }
+        reverse = trans.relationship_graph.reversal
+        execs.each { |e| reverse.should be_vertex(e) }
+
+        catalog.apply
+
+        FileTest.should be_exist(file)
+        File.read(file).should == "one\ntwo\nthree\n"
+    end
 end
diff --git a/spec/unit/transaction.rb b/spec/unit/transaction.rb
index 260ed3703..a797d9438 100755
--- a/spec/unit/transaction.rb
+++ b/spec/unit/transaction.rb
@@ -5,17 +5,334 @@ require File.dirname(__FILE__) + '/../spec_helper'
 require 'puppet/transaction'
 
 describe Puppet::Transaction do
-    it "should match resources by name, not title, when prefetching" do
-        @catalog = Puppet::Resource::Catalog.new
-        @transaction = Puppet::Transaction.new(@catalog)
+    describe "when evaluating a resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:eval_children_and_apply_resource)
+            @transaction.stubs(:skip?).returns false
+
+            @resource = stub("resource")
+        end
+
+        it "should check whether the resource should be skipped" do
+            @transaction.expects(:skip?).with(@resource).returns false
+
+            @transaction.eval_resource(@resource)
+        end
+
+        it "should eval and apply children" do
+            @transaction.expects(:eval_children_and_apply_resource).with(@resource)
+
+            @transaction.eval_resource(@resource)
+        end
+
+        it "should process events" do
+            @transaction.expects(:process_events).with(@resource)
+
+            @transaction.eval_resource(@resource)
+        end
+
+        describe "and the resource should be skipped" do
+            before do
+                @transaction.expects(:skip?).with(@resource).returns true
+            end
+
+            it "should increment the 'skipped' count" do
+                @transaction.eval_resource(@resource)
+                @transaction.resourcemetrics[:skipped].should == 1
+            end
+        end
+    end
+
+    describe "when applying changes" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:queue_events)
+
+            @resource = stub 'resource'
+            @property = stub 'property', :is_to_s => "is", :should_to_s => "should"
+            @change = stub 'change', :property => @property, :changed= => nil, :forward => nil, :is => "is", :should => "should"
+        end
+
+        it "should apply each change" do
+            c1 = stub 'c1', :property => @property, :changed= => nil
+            c1.expects(:forward)
+            c2 = stub 'c2', :property => @property, :changed= => nil
+            c2.expects(:forward)
+
+            @transaction.apply_changes(@resource, [c1, c2])
+        end
+
+        it "should queue the events from each change" do
+            c1 = stub 'c1', :forward => "e1", :property => @property, :changed= => nil
+            c2 = stub 'c2', :forward => "e2", :property => @property, :changed= => nil
+
+            @transaction.expects(:queue_events).with(["e1"])
+            @transaction.expects(:queue_events).with(["e2"])
+
+            @transaction.apply_changes(@resource, [c1, c2])
+        end
+
+        it "should mark the change as 'changed'" do
+            @change.expects(:changed=).with true
+            @transaction.apply_changes(@resource, [@change])
+        end
+
+        it "should store the change in the transaction's change list" do
+            @transaction.apply_changes(@resource, [@change])
+            @transaction.changes.should include(@change)
+        end
+
+        it "should increment the number of applied resources" do
+            @transaction.apply_changes(@resource, [@change])
+            @transaction.resourcemetrics[:applied].should == 1
+        end
+
+        describe "and a change fails" do
+            before do
+                @change.expects(:forward).raises "a failure"
+                @change.property.stubs(:err)
+            end
+
+            it "should rescue the exception" do
+                lambda { @transaction.apply_changes(@resource, [@change]) }.should_not raise_error
+            end
+
+            it "should log the failure with the change's property" do
+                @change.property.expects(:err)
+                @transaction.apply_changes(@resource, [@change])
+            end
+
+            it "should increment the failures" do
+                @transaction.apply_changes(@resource, [@change])
+                @transaction.should be_any_failed
+            end
+        end
+    end
+
+    describe "when queueing events" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+
+            @graph = stub 'graph', :matching_edges => []
+            @transaction.stubs(:relationship_graph).returns @graph
+
+            @resource = stub("resource", :self_refresh? => false, :deleting => false)
+            @event = Puppet::Transaction::Event.new(:foo, @resource)
+        end
+
+        it "should store each event in its event list" do
+            @transaction.queue_events(@event)
+
+            @transaction.events.should include(@event)
+        end
+
+        it "should use the resource from the first edge as the source of all of the events" do
+            event1 = Puppet::Transaction::Event.new(:foo, @resource)
+            event2 = Puppet::Transaction::Event.new(:foo, stub('resource2', :self_refresh? => false, :deleting? => false))
+
+            @graph.expects(:matching_edges).with { |events, resource| resource == event1.resource }.returns []
+
+            @transaction.queue_events(event1, event2)
+        end
+
+        it "should queue events for the target and callback of any matching edges" do
+            edge1 = stub("edge1", :callback => :c1, :source => stub("s1"), :target => stub("t1", :c1 => nil))
+            edge2 = stub("edge2", :callback => :c2, :source => stub("s2"), :target => stub("t2", :c2 => nil))
+
+            @graph.expects(:matching_edges).with { |events, resource| events == [@event] }.returns [edge1, edge2]
+
+            @transaction.expects(:queue_events_for_resource).with(@resource, edge1.target, edge1.callback, [@event])
+            @transaction.expects(:queue_events_for_resource).with(@resource, edge2.target, edge2.callback, [@event])
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should queue events for the changed resource if the resource is self-refreshing and not being deleted" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns true
+            @resource.expects(:deleting?).returns false
+            @transaction.expects(:queue_events_for_resource).with(@resource, @resource, :refresh, [@event])
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should not queue events for the changed resource if the resource is not self-refreshing" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns false
+            @resource.stubs(:deleting?).returns false
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should not queue events for the changed resource if the resource is being deleted" do
+            @graph.stubs(:matching_edges).returns []
+
+            @resource.expects(:self_refresh?).returns true
+            @resource.expects(:deleting?).returns true
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+
+        it "should ignore edges that don't have a callback" do
+            edge1 = stub("edge1", :callback => :nil, :source => stub("s1"), :target => stub("t1", :c1 => nil))
+
+            @graph.expects(:matching_edges).returns [edge1]
+
+            @transaction.expects(:queue_events_for_resource).never
 
-        # Have both a title and name
-        resource = Puppet::Type.type(:sshkey).create :title => "foo", :name => "bar", :type => :dsa, :key => "eh"
-        @catalog.add_resource resource
+            @transaction.queue_events(@event)
+        end
+
+        it "should ignore targets that don't respond to the callback" do
+            edge1 = stub("edge1", :callback => :c1, :source => stub("s1"), :target => stub("t1"))
+
+            @graph.expects(:matching_edges).returns [edge1]
+
+            @transaction.expects(:queue_events_for_resource).never
+
+            @transaction.queue_events(@event)
+        end
+    end
+
+    describe "when queueing events for a resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+        end
 
-        resource.provider.class.expects(:prefetch).with("bar" => resource)
+        it "should do nothing if no events are queued" do
+            @transaction.queued_events(stub("target")) { |callback, events| raise "should never reach this" }
+        end
+
+        it "should yield the callback and events for each callback" do
+            target = stub("target")
+
+            2.times do |i|
+                @transaction.queue_events_for_resource(stub("source", :info => nil), target, "callback#{i}", ["event#{i}"])
+            end
+
+            @transaction.queued_events(target) { |callback, events| }
+        end
+
+        it "should use the source to log that it's scheduling a refresh of the target" do
+            target = stub("target")
+            source = stub 'source'
+            source.expects(:info)
+
+            @transaction.queue_events_for_resource(source, target, "callback", ["event"])
 
-        @transaction.prefetch
+            @transaction.queued_events(target) { |callback, events| }
+        end
+    end
+
+    describe "when processing events for a given resource" do
+        before do
+            @transaction = Puppet::Transaction.new(Puppet::Resource::Catalog.new)
+            @transaction.stubs(:queue_events)
+
+            @resource = stub 'resource', :notice => nil
+            @event = Puppet::Transaction::Event.new(:event, @resource)
+        end
+
+        it "should call the required callback once for each set of associated events" do
+            @transaction.expects(:queued_events).with(@resource).multiple_yields([:callback1, [@event]], [:callback2, [@event]])
+
+            @resource.expects(:callback1)
+            @resource.expects(:callback2)
+
+            @transaction.process_events(@resource)
+        end
+
+        it "should update the 'restarted' metric" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @transaction.process_events(@resource)
+
+            @transaction.resourcemetrics[:restarted].should == 1
+        end
+
+        it "should queue a 'restarted' event with the resource as the source" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @transaction.expects(:queue_events).with do |event|
+                event.name == :restarted and event.resource == @resource
+            end
+
+            @transaction.process_events(@resource)
+        end
+
+        it "should log that it restarted" do
+            @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+
+            @resource.stubs(:callback1)
+
+            @resource.expects(:notice).with { |msg| msg.include?("Triggered 'callback1'") }
+
+            @transaction.process_events(@resource)
+        end
+
+        describe "and the events include a noop event" do
+            before do
+                @event.stubs(:name).returns :noop
+                @transaction.expects(:queued_events).with(@resource).yields(:callback1, [@event])
+            end
+
+            it "should log" do
+                @resource.expects(:notice).with { |msg| msg.include?("Would have triggered 'callback1'") }
+
+                @transaction.process_events(@resource)
+            end
+
+            it "should not call the callback" do
+                @resource.expects(:callback1).never
+
+                @transaction.process_events(@resource)
+            end
+
+            it "should queue a new noop event" do
+                @transaction.expects(:queue_events).with { |event| event.name == :noop and event.resource == @resource }
+
+                @transaction.process_events(@resource)
+            end
+        end
+
+        describe "and the callback fails" do
+            before do
+                @resource.expects(:callback1).raises "a failure"
+                @resource.stubs(:err)
+
+                @transaction.expects(:queued_events).yields(:callback1, [@event])
+            end
+
+            it "should log but not fail" do
+                @resource.expects(:err)
+
+                lambda { @transaction.process_events(@resource) }.should_not raise_error
+            end
+
+            it "should update the 'failed_restarts' metric" do
+                @transaction.process_events(@resource)
+                @transaction.resourcemetrics[:failed_restarts].should == 1
+            end
+
+            it "should not queue a 'restarted' event" do
+                @transaction.expects(:queue_events).never
+                @transaction.process_events(@resource)
+            end
+
+            it "should not increase the restarted resource count" do
+                @transaction.process_events(@resource)
+                @transaction.resourcemetrics[:restarted].should == 0
+            end
+        end
     end
 
     describe "when initializing" do
@@ -136,6 +453,21 @@ describe Puppet::Transaction do
             @transaction.add_metrics_to_report(@report)
         end
     end
+
+    describe "when prefetching" do
+        it "should match resources by name, not title" do
+            @catalog = Puppet::Resource::Catalog.new
+            @transaction = Puppet::Transaction.new(@catalog)
+
+            # Have both a title and name
+            resource = Puppet::Type.type(:sshkey).create :title => "foo", :name => "bar", :type => :dsa, :key => "eh"
+            @catalog.add_resource resource
+
+            resource.provider.class.expects(:prefetch).with("bar" => resource)
+
+            @transaction.prefetch
+        end
+    end
 end
 
 describe Puppet::Transaction, " when determining tags" do
diff --git a/spec/unit/transaction/event.rb b/spec/unit/transaction/event.rb
index 9fd71aae1..b7bd179f6 100755
--- a/spec/unit/transaction/event.rb
+++ b/spec/unit/transaction/event.rb
@@ -5,21 +5,17 @@ require File.dirname(__FILE__) + '/../../spec_helper'
 require 'puppet/transaction/event'
 
 describe Puppet::Transaction::Event do
-    Event = Puppet::Transaction::Event
-
-    it "should require a name and a source" do
-        lambda { Event.new }.should raise_error(ArgumentError)
-    end
-
-    it "should have a name getter" do
-        Event.new(:foo, "bar").name.should == :foo
-    end
-
-    it "should have a source accessor" do
-        Event.new(:foo, "bar").source.should == "bar"
+    [:log, :previous_value, :desired_value, :property, :resource, :name, :result].each do |attr|
+        it "should support #{attr}" do
+            event = Puppet::Transaction::Event.new
+            event.send(attr.to_s + "=", "foo")
+            event.send(attr).should == "foo"
+        end
     end
 
-    it "should be able to produce a string containing the event name and the source" do
-        Event.new(:event, :source).to_s.should == "source -> event"
+    it "should produce the log when converted to a string" do
+        event = Puppet::Transaction::Event.new
+        event.expects(:log).returns "my log"
+        event.to_s.should == "my log"
     end
 end