Introduce configuration override

README.md

@@ -47,6 +47,45 @@ environment variables can be used to determine environment).  For example in
     production:
       "foo,bar,baz": 4
 
+### Custom Configuration
+
+Sometimes, a static YAML file doesn't cut it. At high levels of stress, you
+may want to update Resque::Pool's configuration settings on the go, maybe via
+other YAML files on the server or a database-backed solution. This is
+accomplished by creating a custom configuration object.
+
+    class QueueConfigurator
+      def call(config)
+        config.merge { 'b' => 3 }
+      end
+    end
+
+    pool = Resque::Pool.new({ 'a' => 1 }, QueueConfigurator.new)
+    pool.config
+    # => { 'a' => 1, 'b' => 3 }
+
+The configuration object can be any sort of `call`-able object, and it must
+return a Resque::Pool compatible hash configuration.
+
+If you are running Resque pool from the command line, you will need to do two
+things:
+
+1.) Create a configuration file passing your config override.
+
+    # ./config/resque_pool_config.rb
+    Resque::Pool.config_override = ->(config) { { "foo,bar,baz" => 100 } }
+
+2.) Run `resque-pool` and pass it the `--config-manager` option
+
+    $ resque-pool --config-manager ./config/resque_pool_config.rb
+
+Your pool's configuration will now pull from your override.
+
+**Warning**: *Using a custom configuration manager is risky. Only attempt to use
+this feature if you are absolutely sure that your configuration manager does
+not break your systems. You can completely override your pool configuration if
+you are not careful!*
+
 ### Rake task config
 
 Require the rake tasks (`resque/pool/tasks`) in your `Rakefile`, load your

lib/resque/pool.rb

@@ -10,7 +10,8 @@
 
 module Resque
   class Pool
-    SIG_QUEUE_MAX_SIZE = 5
+    DEFAULT_OVERRIDE_PROC   = ->(config) { config }
+    SIG_QUEUE_MAX_SIZE      = 5
     DEFAULT_WORKER_INTERVAL = 5
     QUEUE_SIGS = [ :QUIT, :INT, :TERM, :USR1, :USR2, :CONT, :HUP, :WINCH, ]
     CHUNK_SIZE = (16 * 1024)
@@ -20,12 +21,27 @@ class Pool
     attr_reader :config
     attr_reader :workers
 
-    def initialize(config)
-      init_config(config)
+    def initialize(configuration, config_proc = nil)
+      init_config(configuration)
+      @config_proc = config_proc || self.class.config_override
       @workers = Hash.new { |workers, queues| workers[queues] = {} }
       procline "(initialized)"
     end
 
+    # Config Override:
+    #
+    def self.config_override
+      @config_override || DEFAULT_OVERRIDE_PROC
+    end
+
+    def self.config_override=(override)
+      if override.respond_to? :call
+        @config_override = override
+      else
+        procline "Config override #{override.inspect} is not a callable object."
+      end
+    end
+
     # Config: after_prefork {{{
 
     # The `after_prefork` hook will be run in workers if you are using the
@@ -77,7 +93,7 @@ def self.run
       if GC.respond_to?(:copy_on_write_friendly=)
         GC.copy_on_write_friendly = true
       end
-      Resque::Pool.new(choose_config_file).start.join
+      Resque::Pool.new(choose_config_file, config_override).start.join
     end
 
     # }}}
@@ -103,8 +119,8 @@ def load_config
       else
         @config ||= {}
       end
-      environment and @config[environment] and config.merge!(@config[environment])
-      config.delete_if {|key, value| value.is_a? Hash }
+      environment and @config[environment] and @config.merge!(@config[environment])
+      @config.delete_if {|key, value| value.is_a? Hash }
     end
 
     def environment
@@ -340,6 +356,14 @@ def signal_all_workers(signal)
     # }}}
     # ???: maintain_worker_count, all_known_queues {{{
 
+    def refresh_config
+      cloned = @config.dup
+      @config = @config_proc.call(@config)
+    rescue => e
+      log "There was an issue updating the configuration: #{e.message} #{e.backtrace.join("\n")}"
+      cloned
+    end
+
     def maintain_worker_count
       all_known_queues.each do |queues|
         delta = worker_delta_for(queues)
@@ -349,6 +373,7 @@ def maintain_worker_count
     end
 
     def all_known_queues
+      refresh_config
       config.keys | workers.keys
     end
 

lib/resque/pool/cli.rb

@@ -14,6 +14,7 @@ def run
         redirect opts
         setup_environment opts
         set_pool_options opts
+        set_pool_configuration_override opts
         start_pool
       end
 
@@ -38,6 +39,7 @@ def parse_options
           opt :nosync, "Don't sync logfiles on every write"
           opt :pidfile, "PID file location",         :type => String,    :short => "-p"
           opt :environment, "Set RAILS_ENV/RACK_ENV/RESQUE_ENV", :type => String, :short => "-E"
+          opt :config_manager,     "Path to configuration override definition", :type => String, :short => "-C"
           opt :term_graceful_wait, "On TERM signal, wait for workers to shut down gracefully"
           opt :term_graceful,      "On TERM signal, shut down workers gracefully"
           opt :term_immediate,     "On TERM signal, shut down workers immediately (default)"
@@ -115,6 +117,19 @@ def set_pool_options(opts)
         end
       end
 
+      def set_pool_configuration_override(opts)
+        require 'pathname'
+        if opts[:config_manager]
+          file = Pathname.new(opts[:config_manager]).expand_path
+          if file.exist?
+            require file
+            Resque::Pool.log "Resque Pool configuration is overriden in #{opts[:config_manager]}"
+          else
+            Resque::Pool.log "--config-manager #{file} error: File does not exist"
+          end
+        end
+      end
+
       def setup_environment(opts)
         Resque::Pool.app_name = opts[:appname]    if opts[:appname]
         ENV["RACK_ENV"] = ENV["RAILS_ENV"] = ENV["RESQUE_ENV"] = opts[:environment] if opts[:environment]

spec/resque_pool_override.rb

@@ -0,0 +1,3 @@
+Resque::Pool.config_override = ->(config) {
+  { "foo" => 1, "bar" => 2, "baz" => 3, "foo,bar,baz" => 4 }
+}

spec/resque_pool_spec.rb

@@ -9,6 +9,56 @@
   }
 end
 
+describe Resque::Pool, "when using a custom configuration manager" do
+  let(:config) do
+    { 'foo' => 1, 'bar' => 2, 'foo,bar' => 3, 'bar,foo' => 4, }
+  end
+  subject { Resque::Pool.new(config, manager) }
+  before { subject.all_known_queues }
+
+  context "when no errors are raised" do
+    let(:manager) do
+      ->(config) { config.merge( "fooey" => 10 ) }
+    end
+    it "should merge the other values into the pool's config" do
+      subject.config["fooey"].should == 10
+      subject.config["foo"].should == 1
+      subject.config["bar"].should == 2
+      subject.config["foo,bar"].should == 3
+      subject.config["bar,foo"].should == 4
+    end
+  end
+
+  context "when an error is raised" do
+    let(:manager) do
+      ->(config) { raise "config error was raised" }
+    end
+
+    it "should replace the config of the original on an error" do
+      subject.config["foo"].should == 1
+      subject.config["bar"].should == 2
+      subject.config["foo,bar"].should == 3
+      subject.config["bar,foo"].should == 4
+    end
+  end
+
+  context "when a config override is globally set" do
+    around do |e|
+      Resque::Pool.config_override = ->(config) {
+        { "foo,bar,baz" => 100 }
+      }
+      e.run
+      Resque::Pool.config_override = nil
+    end
+    let(:manager) { nil }
+
+    it "should use the global configuration manager" do
+      subject.config.should == { "foo,bar,baz" => 100 }
+    end
+  end
+
+end
+
 describe Resque::Pool, "when loading a simple pool configuration" do
   let(:config) do
     { 'foo' => 1, 'bar' => 2, 'foo,bar' => 3, 'bar,foo' => 4, }