Merge pull request #236 from glennsarti/make-check-pool-more-responsive

(GH-226) Respond quickly to VMs being consumed

Gemfile

@@ -12,15 +12,34 @@ gem 'puma', '>= 3.6.0'
 gem 'rack', '~> 1.6'
 gem 'rake', '>= 10.4'
 gem 'rbvmomi', '>= 1.8'
-if RUBY_VERSION =~ /^1\.9\./
-  gem 'nokogiri', '< 1.7.0'
-end
-gem 'redis', '>= 3.2'
 gem 'sinatra', '>= 1.4'
 gem 'net-ldap', '<= 0.12.1' # keep compatibility w/ jruby & mri-1.9.3
 gem 'statsd-ruby', '>= 1.3.0', :require => 'statsd'
 gem 'connection_pool', '>= 2.2.1'
 
+# Pin gems against Ruby version
+# Note we can't use platform restrictions easily so use
+# lowest version range any platform
+# ----
+# nokogiri
+# redis
+if RUBY_VERSION =~ /^1\.9\./
+  gem 'nokogiri', '~> 1.6.0'
+  gem 'redis', '~> 3.0'
+elsif RUBY_VERSION =~ /^2\.[0]/
+  gem 'nokogiri', '~> 1.6.0'
+  gem 'redis', '~> 3.0'
+elsif RUBY_VERSION =~ /^2\.[1]/
+  gem 'nokogiri', '~> 1.7.0'
+  gem 'redis', '~> 3.0'
+elsif RUBY_VERSION =~ /^2\.2\.[01]/
+  gem 'nokogiri', "~> 1.7"
+  gem 'redis', '~> 3.0'
+else
+  gem 'nokogiri', "~> 1.7"
+  gem 'redis', '>= 3.2'
+end
+
 # Test deps
 group :test do
   gem 'mock_redis', '>= 0.17.0'

lib/vmpooler/pool_manager.rb

@@ -521,6 +521,45 @@ module Vmpooler
       finish
     end
 
+    # Helper method mainly used for unit testing
+    def time_passed?(_event, time)
+      Time.now > time
+    end
+
+    # Possible wakeup events
+    # :pool_size_change
+    #   - Fires when the number of ready VMs changes due to being consumed.
+    #   - Additional options
+    #       :poolname
+    #
+    def sleep_with_wakeup_events(loop_delay, wakeup_period = 5, options = {})
+      exit_by = Time.now + loop_delay
+      wakeup_by = Time.now + wakeup_period
+      return if time_passed?(:exit_by, exit_by)
+
+      if options[:pool_size_change]
+        initial_ready_size = $redis.scard("vmpooler__ready__#{options[:poolname]}")
+      end
+
+      loop do
+        sleep(1)
+        break if time_passed?(:exit_by, exit_by)
+
+        # Check for wakeup events
+        if time_passed?(:wakeup_by, wakeup_by)
+          wakeup_by = Time.now + wakeup_period
+
+          # Wakeup if the number of ready VMs has changed
+          if options[:pool_size_change]
+            ready_size = $redis.scard("vmpooler__ready__#{options[:poolname]}")
+            break unless ready_size == initial_ready_size
+          end
+        end
+
+        break if time_passed?(:exit_by, exit_by)
+      end
+    end
+
     def check_pool(pool,
                    maxloop = 0,
                    loop_delay_min = CHECK_LOOP_DELAY_MIN_DEFAULT,
@@ -551,7 +590,7 @@ module Vmpooler
               loop_delay = (loop_delay * loop_delay_decay).to_i
               loop_delay = loop_delay_max if loop_delay > loop_delay_max
             end
-            sleep(loop_delay)
+            sleep_with_wakeup_events(loop_delay, loop_delay_min, pool_size_change: true, poolname: pool['name'])
 
             unless maxloop.zero?
               break if loop_count >= maxloop

spec/unit/pool_manager_spec.rb

@@ -2024,7 +2024,61 @@ EOT
     end
   end
 
- describe "#check_pool" do
+  describe "#sleep_with_wakeup_events" do
+    let(:loop_delay) { 10 }
+    before(:each) do
+      allow(Kernel).to receive(:sleep).and_raise("sleep should not be called")
+      allow(subject).to receive(:time_passed?).with(:wakeup_by, Time).and_call_original
+      allow(subject).to receive(:time_passed?).with(:exit_by, Time).and_call_original
+    end
+
+    it 'should not sleep if the loop delay is negative' do
+      expect(subject).to receive(:sleep).exactly(0).times
+
+      subject.sleep_with_wakeup_events(-1)
+    end
+
+    it 'should sleep until the loop delay is exceeded' do
+      # This test is a little brittle as it requires knowledge of the implementation
+      # Basically the number of sleep delays will dictate how often the time_passed? method is called
+
+      expect(subject).to receive(:sleep).exactly(2).times
+      expect(subject).to receive(:time_passed?).with(:exit_by, Time).and_return(false, false, false, true)
+      
+      subject.sleep_with_wakeup_events(loop_delay)
+    end
+
+    describe 'with the pool_size_change wakeup option' do
+      let(:wakeup_option) {{
+        :pool_size_change => true,
+        :poolname => pool,
+      }}
+      let(:wakeup_period) { -1 } # A negative number forces the wakeup evaluation to always occur
+
+      it 'should check the number of VMs ready in Redis' do
+        expect(subject).to receive(:time_passed?).with(:exit_by, Time).and_return(false, true)
+        expect(redis).to receive(:scard).with("vmpooler__ready__#{pool}").once
+
+        subject.sleep_with_wakeup_events(loop_delay, wakeup_period, wakeup_option)
+      end
+
+      it 'should sleep until the number of VMs ready in Redis increases' do
+        expect(subject).to receive(:sleep).exactly(3).times
+        expect(redis).to receive(:scard).with("vmpooler__ready__#{pool}").and_return(1,1,1,2)
+
+        subject.sleep_with_wakeup_events(loop_delay, wakeup_period, wakeup_option)
+      end
+
+      it 'should sleep until the number of VMs ready in Redis decreases' do
+        expect(subject).to receive(:sleep).exactly(3).times
+        expect(redis).to receive(:scard).with("vmpooler__ready__#{pool}").and_return(2,2,2,1)
+
+        subject.sleep_with_wakeup_events(loop_delay, wakeup_period, wakeup_option)
+      end
+    end
+  end
+  
+  describe "#check_pool" do
     let(:threads) {{}}
     let(:provider_name) { 'mock_provider' }
     let(:config) {
@@ -2093,7 +2147,19 @@ EOT
       end
 
       it 'when a non-default loop delay is specified' do
-        expect(subject).to receive(:sleep).with(loop_delay).exactly(maxloop).times
+        expect(subject).to receive(:sleep_with_wakeup_events).with(loop_delay, Numeric, Hash).exactly(maxloop).times
+
+        subject.check_pool(pool_object,maxloop,loop_delay,loop_delay)
+      end
+
+      it 'specifies a wakeup option for pool size change' do
+        expect(subject).to receive(:sleep_with_wakeup_events).with(loop_delay, Numeric, hash_including(:pool_size_change => true)).exactly(maxloop).times
+
+        subject.check_pool(pool_object,maxloop,loop_delay,loop_delay)
+      end
+
+      it 'specifies a wakeup option for poolname' do
+        expect(subject).to receive(:sleep_with_wakeup_events).with(loop_delay, Numeric, hash_including(:poolname => pool)).exactly(maxloop).times
 
         subject.check_pool(pool_object,maxloop,loop_delay,loop_delay)
       end
@@ -2126,7 +2192,7 @@ EOT
       end
 
       it 'when a non-default loop delay is specified' do
-        expect(subject).to receive(:sleep).with(pool_loop_delay).exactly(maxloop).times
+        expect(subject).to receive(:sleep_with_wakeup_events).with(pool_loop_delay, pool_loop_delay, Hash).exactly(maxloop).times
 
         subject.check_pool(pool_object,maxloop,loop_delay)
       end
@@ -2151,7 +2217,7 @@ EOT
       [:checked_pending_vms, :discovered_vms, :cloned_vms].each do |testcase|
         describe "when #{testcase} is greater than zero" do
           it "delays the minimum delay time" do
-            expect(subject).to receive(:sleep).with(loop_delay_min).exactly(maxloop).times
+            expect(subject).to receive(:sleep_with_wakeup_events).with(loop_delay_min, loop_delay_min, Hash).exactly(maxloop).times
             check_pool_response[testcase] = 1
 
             subject.check_pool(pool_object,maxloop,loop_delay_min,loop_delay_max)
@@ -2163,10 +2229,10 @@ EOT
         describe "when #{testcase} is greater than zero" do
           let(:loop_decay) { 3.0 }
           it "delays increases with a decay" do
-            expect(subject).to receive(:sleep).with(3).once
-            expect(subject).to receive(:sleep).with(9).once
-            expect(subject).to receive(:sleep).with(27).once
-            expect(subject).to receive(:sleep).with(60).twice
+            expect(subject).to receive(:sleep_with_wakeup_events).with(3, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(9, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(27, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(60, Numeric, Hash).twice
             check_pool_response[testcase] = 1
 
             subject.check_pool(pool_object,maxloop,loop_delay_min,loop_delay_max,loop_decay)
@@ -2209,7 +2275,7 @@ EOT
       [:checked_pending_vms, :discovered_vms, :cloned_vms].each do |testcase|
         describe "when #{testcase} is greater than zero" do
           it "delays the minimum delay time" do
-            expect(subject).to receive(:sleep).with(pool_loop_delay_min).exactly(maxloop).times
+            expect(subject).to receive(:sleep_with_wakeup_events).with(pool_loop_delay_min, Numeric, Hash).exactly(maxloop).times
             check_pool_response[testcase] = 1
 
             subject.check_pool(pool_object,maxloop,loop_delay_min,loop_delay_max,loop_decay)
@@ -2220,10 +2286,10 @@ EOT
       [:checked_running_vms, :checked_ready_vms, :destroyed_vms, :migrated_vms].each do |testcase|
         describe "when #{testcase} is greater than zero" do
           it "delays increases with a decay" do
-            expect(subject).to receive(:sleep).with(7).once
-            expect(subject).to receive(:sleep).with(17).once
-            expect(subject).to receive(:sleep).with(42).once
-            expect(subject).to receive(:sleep).with(70).twice
+            expect(subject).to receive(:sleep_with_wakeup_events).with(7, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(17, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(42, Numeric, Hash).once
+            expect(subject).to receive(:sleep_with_wakeup_events).with(70, Numeric, Hash).twice
             check_pool_response[testcase] = 1
 
             subject.check_pool(pool_object,maxloop,loop_delay_min,loop_delay_max,loop_decay)