This fixes a bug where the rpc client stops processing incoming rpc responses
with the user specified block when encoutering the reply of a broken or
misconfigured node.

bin/mcollectived

@@ -8,11 +8,15 @@ require 'mcollective'
require 'getoptlong'

opts = GetoptLong.new(
  [ '--daemonize', GetoptLong::NO_ARGUMENT],
  [ '--no-daemonize', GetoptLong::NO_ARGUMENT],
  [ '--help', '-h', GetoptLong::NO_ARGUMENT ],
  [ '--config', '-c', GetoptLong::REQUIRED_ARGUMENT],
  [ '--pidfile', '-p', GetoptLong::REQUIRED_ARGUMENT]
)

daemonize = nil

if MCollective::Util.windows?
  configfile = File.join(MCollective::Util.windows_prefix, "etc", "server.cfg")
else
@@ -23,22 +27,31 @@ pid = ""
opts.each do |opt, arg|
  case opt
    when '--help'
      puts "Usage: mcollectived.rb [--config /path/to/config] [--pidfile /path/to/pid]"
      puts "Usage: mcollectived.rb [--config /path/to/config] [--pidfile /path/to/pid] [--daemonize | --no-daemonize]"
      exit
    when '--config'
      configfile = arg
    when '--pidfile'
      pid = arg
    when '--daemonize'
      daemonize = true
    when '--no-daemonize'
      daemonize = false
  end
end

config = MCollective::Config.instance

config.loadconfig(configfile) unless config.configured

# If daemonize has not been set on the cli, reach into the config file to decide
if daemonize == nil
  daemonize = config.daemonize
end

MCollective::Log.info("The Marionette Collective #{MCollective::VERSION} started logging at #{config.loglevel} level")

if config.daemonize
if daemonize
  MCollective::Log.debug("Starting in the background (#{config.daemonize})")

  if MCollective::Util.windows?

ext/Makefile

@@ -11,7 +11,6 @@ install: install-bin install-lib install-conf install-plugins install-doc
install-bin:
	install -d $(DESTDIR)/usr/sbin
	install -d $(DESTDIR)/usr/bin
	cp bin/mc-* $(DESTDIR)/usr/sbin
	cp bin/mco $(DESTDIR)/usr/bin
	cp bin/mcollectived $(DESTDIR)/usr/sbin/mcollectived

ext/debian/mcollective.init

@@ -53,7 +53,7 @@ case "$1" in
    start)
        echo "Starting daemon: " $name
        # start the program
        start-stop-daemon -S -p ${pidfile} --oknodo -q -a ${mcollectived} -- ${daemonopts}
        start-stop-daemon -S -p ${pidfile} --oknodo -q -a ${mcollectived} -- ${daemonopts} --daemonize
        [ $? = 0 ] && { exit 0 ; } || { exit 1 ; }
        log_success_msg "mcollective started"
        touch $lock

ext/redhat/mcollective.init

@@ -50,7 +50,7 @@ start() {
    echo -n "Starting mcollective: "
    # Only try to start if not already started
    if ! rh_status_q; then
      daemon ${daemonopts} ${mcollectived} --pid=${pidfile} --config="/etc/mcollective/server.cfg"
      daemon ${daemonopts} ${mcollectived} --pid=${pidfile} --config="/etc/mcollective/server.cfg" --daemonize
    fi
    # This will be 0 if mcollective is already running
    RETVAL=$?

ext/redhat/mcollective.service

@@ -6,7 +6,7 @@ After=network.target
Type=forking
StandardOutput=syslog
StandardError=syslog
ExecStart=/usr/sbin/mcollectived --config=/etc/mcollective/server.cfg --pidfile=/var/run/mcollective.pid
ExecStart=/usr/sbin/mcollectived --config=/etc/mcollective/server.cfg --pidfile=/var/run/mcollective.pid --daemonize
ExecReload=/bin/kill -USR1 $MAINPID
PIDFile=/var/run/mcollective.pid

ext/solaris/README

@@ -17,7 +17,7 @@ Requirements, get them from opencsw:

Run requirements
- rubystomp library
  http://stomp.codehaus.org/Ruby+Client
  https://rubygems.org/gems/stomp
  Up and till version 1.0.4 it is a single file. Put in /opt/csw/lib/ruby/site_ruby/1.8/

Configuration

ext/solaris/mcollective.init

@@ -40,7 +40,7 @@ case "$1" in

	rm -f ${pidfile}

	${mcollectived} --pid=${pidfile} --config="/etc/mcollective/server.cfg"
	${mcollectived} --pid=${pidfile} --config="/etc/mcollective/server.cfg" --daemonize
	if [ $? = 0 ]; then
	    touch $lock
	    exit 0

ext/windows/README.md

@@ -4,7 +4,7 @@ Here are a few instructions for people who wish to do early adopter testing. At

Assuming you are installing MCollective into `C:\marionette-collective`:

 * Install Ruby 1.8.7 from <http://rubyinstaller.org/>
 * Install Ruby 1.9.3 from <http://rubyinstaller.org/>
      * check the boxes for "Add Ruby executables to your PATH" and "Associate .rb and .rbw files with the Ruby installation"
 * Run the following commands to install the required gems:
      * `gem install --no-rdoc --no-ri stomp win32-service sys-admin windows-api`
@@ -30,4 +30,4 @@ If it does not run:

 * Look in the log files. Edit `server.cfg` to set `loglevel` to `debug`. 
 * If the log files are empty, look at the command the service wrapper runs and run it by hand. This will show you any early exceptions preventing it from running. It wont succesfully start, but you should see why it does not get far enough to start writing logs.
 
 

lib/mcollective/application.rb

@@ -314,12 +314,27 @@ def halt_code(stats)
                       :okcount => 0,
                       :failcount => 0}.merge(stats.to_hash)

      return 4 if request_stats[:discoverytime] == 0 && request_stats[:responses] == 0
      return 3 if request_stats[:discovered] > 0 && request_stats[:responses] == 0
      return 2 if request_stats[:discovered] > 0 && request_stats[:failcount] > 0
      return 1 if request_stats[:discovered] == 0
      return 0 if request_stats[:discoverytime] == 0 && request_stats[:discovered] == request_stats[:okcount]
      return 0 if request_stats[:discovered] == request_stats[:okcount]
      if (request_stats[:discoverytime] == 0 && request_stats[:responses] == 0)
        return 4
      end

      if (request_stats[:discovered] > 0)
        if (request_stats[:responses] == 0)
          return 3
        elsif (request_stats[:failcount] > 0)
          return 2
        end
      end

      if (request_stats[:discovered] == 0)
        if (request_stats[:responses] && request_stats[:responses] > 0)
          return 0
        else
          return 1
        end
      end

      return 0
    end

    # A helper that creates a consistent exit code for applications by looking at an

lib/mcollective/client.rb

@@ -18,6 +18,11 @@ def initialize(configfile)
      @connection.connect
    end

    @@request_sequence = 0
    def self.request_sequence
      @@request_sequence
    end

    # Returns the configured main collective if no
    # specific collective is specified as options
    def collective
@@ -55,6 +60,8 @@ def createreq(msg, agent, filter ={})
        request.reply_to = @options[:reply_to] if @options[:reply_to]
      end

      @@request_sequence += 1

      request.encode!
      subscribe(agent, :reply) unless request.reply_to
      request

lib/mcollective/config.rb

@@ -15,7 +15,8 @@ class Config
    attr_reader :main_collective, :ssl_cipher, :registration_collective
    attr_reader :direct_addressing, :direct_addressing_threshold, :ttl
    attr_reader :default_discovery_method, :default_discovery_options
    attr_reader :publish_timeout, :threaded, :soft_shutdown
    attr_reader :publish_timeout, :threaded, :soft_shutdown, :activate_agents
    attr_reader :registration_splay, :discovery_timeout, :soft_shutdown_timeout

    def initialize
      @configured = false
@@ -43,6 +44,8 @@ def loadconfig(configfile)
                  @registration_collective = val
                when "registerinterval"
                  @registerinterval = Integer(val)
                when "registration_splay"
                  @registration_splay = Util.str_to_bool(val)
                when "collectives"
                  @collectives = val.split(",").map {|c| c.strip}
                when "main_collective"
@@ -87,6 +90,8 @@ def loadconfig(configfile)
                  @classesfile = val
                when /^plugin.(.+)$/
                  @pluginconf[$1] = val
                when "discovery_timeout"
                  @discovery_timeout = Integer(val)
                when "publish_timeout"
                  @publish_timeout = Integer(val)
                when "rpcaudit"
@@ -115,6 +120,10 @@ def loadconfig(configfile)
                  @default_discovery_method = val
                when "soft_shutdown"
                  @soft_shutdown = Util.str_to_bool(val)
                when "soft_shutdown_timeout"
                  @soft_shutdown_timeout = Integer(val)
                when "activate_agents"
                  @activate_agents = Util.str_to_bool(val)
                when "topicprefix", "topicsep", "queueprefix", "rpchelptemplate", "helptemplatedir"
                  Log.warn("Use of deprecated '#{key}' option.  This option is ignored and should be removed from '#{configfile}'")
                else
@@ -165,6 +174,7 @@ def set_config_defaults(configfile)
      @registration = "Agentlist"
      @registerinterval = 0
      @registration_collective = nil
      @registration_splay = false
      @classesfile = "/var/lib/puppet/state/classes.txt"
      @rpcaudit = false
      @rpcauditprovider = ""
@@ -193,6 +203,8 @@ def set_config_defaults(configfile)
      @publish_timeout = 2
      @threaded = false
      @soft_shutdown = false
      @soft_shutdown_timeout = nil
      @activate_agents = true
    end

    def read_plugin_config_dir(dir)

lib/mcollective/connector/base.rb

@@ -17,6 +17,8 @@ module MCollective
  module Connector
    class Base
      def self.inherited(klass)
        plugin_name = klass.to_s.split("::").last.downcase
        ddl = DDL.new(plugin_name, :connector)
        PluginManager << {:type => "connector_plugin", :class => klass.to_s}
      end
    end

lib/mcollective/facts/base.rb

@@ -34,11 +34,7 @@ def get_fact(fact=nil)
              # Force reset to last known good state on empty facts
              raise "Got empty facts" if tfacts.empty?

              @facts.clear

              tfacts.each_pair do |key,value|
                @facts[key.to_s] = value.to_s
              end
              @facts = normalize_facts(tfacts)

              @last_good_facts = @facts.clone
              @last_facts_load = Time.now.to_i
@@ -81,6 +77,23 @@ def has_fact?(fact)
      def force_reload?
        false
      end

      private

      def normalize_facts(value)
        case value
        when Array
          return value.map { |v| normalize_facts(v) }
        when Hash
          new_hash = {}
          value.each do |k,v|
            new_hash[k.to_s] = normalize_facts(v)
          end
          return new_hash
        else
          return value.to_s
        end
      end
    end
  end
end

lib/mcollective/log.rb

@@ -44,6 +44,13 @@ def cycle_level
        @logger.cycle_level if @configured
      end

      # reopen log files
      def reopen
        if @configured
          @logger.reopen
        end
      end

      # logs a message at a certain level
      def log(level, msg)
        configure unless @configured

lib/mcollective/logger/base.rb

@@ -34,6 +34,18 @@ def set_level(level)
        @active_level = level.to_sym
      end

      def start
        raise "The logging class did not supply a start method"
      end

      def log(level, from, msg)
        raise "The logging class did not supply a log method"
      end

      def reopen
        # reopen may not make sense to all Loggers, but we expect it of the API
      end

      private
      def map_level(level)
        raise "Logger class do not know how to handle #{level} messages" unless valid_levels.include?(level.to_sym)
@@ -60,14 +72,6 @@ def get_next_level
      def valid_levels
        raise "The logging class did not supply a valid_levels method"
      end

      def log(level, from, msg)
        raise "The logging class did not supply a log method"
      end

      def start
        raise "The logging class did not supply a start method"
      end
    end
  end
end

lib/mcollective/logger/file_logger.rb

@@ -41,6 +41,13 @@ def log(level, from, msg)
        # STDERR it as last resort
        STDERR.puts("#{level}: #{msg}")
      end

      def reopen
        level = @logger.level
        @logger.close
        start
        @logger.level = level
      end
    end
  end
end

lib/mcollective/message.rb

@@ -200,7 +200,7 @@ def validate
        if msg_age > ttl
          PluginManager["global_stats"].ttlexpired

          raise(MsgTTLExpired, "message #{requestid} from #{cid} created at #{msgtime} is #{msg_age} seconds old, TTL is #{ttl}")
          raise(MsgTTLExpired, "message #{requestid} from #{cid} created at #{msgtime} is #{msg_age} seconds old, TTL is #{ttl}.  Rejecting message.")
        end
      end

lib/mcollective/optionparser.rb

@@ -177,6 +177,10 @@ def add_common_options
      @parser.on("--threaded", "Start publishing requests and receiving responses in threaded mode.") do |v|
        @options[:threaded] = true
      end

      @parser.on("--sort", "Sort the output of an RPC call before processing.") do |v|
        @options[:sort] = true
      end
    end

    private