module MCollective module RPC # The main component of the Simple RPC client system, this wraps around MCollective::Client # and just brings in a lot of convention and standard approached. class Client attr_accessor :timeout, :verbose, :filter, :config, :progress, :ttl, :reply_to attr_reader :client, :stats, :ddl, :agent, :limit_targets, :limit_method, :output_format, :batch_size, :batch_sleep_time, :batch_mode attr_reader :discovery_options, :discovery_method, :limit_seed @@initial_options = nil # Creates a stub for a remote agent, you can pass in an options array in the flags # which will then be used else it will just create a default options array with # filtering enabled based on the standard command line use. # # rpc = RPC::Client.new("rpctest", :configfile => "client.cfg", :options => options) # # You typically would not call this directly you'd use MCollective::RPC#rpcclient instead # which is a wrapper around this that can be used as a Mixin def initialize(agent, flags = {}) if flags.include?(:options) initial_options = flags[:options] elsif @@initial_options initial_options = Marshal.load(@@initial_options) else oparser = MCollective::Optionparser.new({:verbose => false, :progress_bar => true, :mcollective_limit_targets => false, :batch_size => nil, :batch_sleep_time => 1}, "filter") initial_options = oparser.parse do |parser, opts| if block_given? yield(parser, opts) end Helpers.add_simplerpc_options(parser, opts) end @@initial_options = Marshal.dump(initial_options) end @initial_options = initial_options @config = initial_options[:config] @client = MCollective::Client.new(@config) @client.options = initial_options @stats = Stats.new @agent = agent @timeout = initial_options[:timeout] || 5 @verbose = initial_options[:verbose] @filter = initial_options[:filter] || Util.empty_filter @discovered_agents = nil @progress = initial_options[:progress_bar] @limit_targets = initial_options[:mcollective_limit_targets] @limit_method = Config.instance.rpclimitmethod @limit_seed = initial_options[:limit_seed] || nil @output_format = initial_options[:output_format] || :console @force_direct_request = false @reply_to = initial_options[:reply_to] @discovery_method = initial_options[:discovery_method] || Config.instance.default_discovery_method @discovery_options = initial_options[:discovery_options] || [] @force_display_mode = initial_options[:force_display_mode] || false @batch_size = Integer(initial_options[:batch_size] || 0) @batch_sleep_time = Float(initial_options[:batch_sleep_time] || 1) @batch_mode = @batch_size > 0 agent_filter agent @discovery_timeout = @initial_options.fetch(:disctimeout, nil) @collective = @client.collective @ttl = initial_options[:ttl] || Config.instance.ttl # if we can find a DDL for the service override # the timeout of the client so we always magically # wait appropriate amounts of time. # # We add the discovery timeout to the ddl supplied # timeout as the discovery timeout tends to be tuned # for local network conditions and fact source speed # which would other wise not be accounted for and # some results might get missed. # # We do this only if the timeout is the default 5 # seconds, so that users cli overrides will still # get applied # # DDLs are required, failure to find a DDL is fatal @ddl = DDL.new(agent) @stats.ddl = @ddl @timeout = @ddl.meta[:timeout] + discovery_timeout if @timeout == 5 # allows stderr and stdout to be overridden for testing # but also for web apps that might not want a bunch of stuff # generated to actual file handles if initial_options[:stderr] @stderr = initial_options[:stderr] else @stderr = STDERR @stderr.sync = true end if initial_options[:stdout] @stdout = initial_options[:stdout] else @stdout = STDOUT @stdout.sync = true end end # Disconnects cleanly from the middleware def disconnect @client.disconnect end # Returns help for an agent if a DDL was found def help(template) @ddl.help(template) end # Creates a suitable request hash for the SimpleRPC agent. # # You'd use this if you ever wanted to take care of sending # requests on your own - perhaps via Client#sendreq if you # didn't care for responses. # # In that case you can just do: # # msg = your_rpc.new_request("some_action", :foo => :bar) # filter = your_rpc.filter # # your_rpc.client.sendreq(msg, msg[:agent], filter) # # This will send a SimpleRPC request to the action some_action # with arguments :foo = :bar, it will return immediately and # you will have no indication at all if the request was receieved or not # # Clearly the use of this technique should be limited and done only # if your code requires such a thing def new_request(action, data) callerid = PluginManager["security_plugin"].callerid raise 'callerid received from security plugin is not valid' unless PluginManager["security_plugin"].valid_callerid?(callerid) {:agent => @agent, :action => action, :caller => callerid, :data => data} end # For the provided arguments and action the input arguments get # modified by supplying any defaults provided in the DDL for arguments # that were not supplied in the request # # We then pass the modified arguments to the DDL for validation def validate_request(action, args) raise "No DDL found for agent %s cannot validate inputs" % @agent unless @ddl @ddl.set_default_input_arguments(action, args) @ddl.validate_rpc_request(action, args) end # Magic handler to invoke remote methods # # Once the stub is created using the constructor or the RPC#rpcclient helper you can # call remote actions easily: # # ret = rpc.echo(:msg => "hello world") # # This will call the 'echo' action of the 'rpctest' agent and return the result as an array, # the array will be a simplified result set from the usual full MCollective::Client#req with # additional error codes and error text: # # { # :sender => "remote.box.com", # :statuscode => 0, # :statusmsg => "OK", # :data => "hello world" # } # # If :statuscode is 0 then everything went find, if it's 1 then you supplied the correct arguments etc # but the request could not be completed, you'll find a human parsable reason in :statusmsg then. # # Codes 2 to 5 maps directly to UnknownRPCAction, MissingRPCData, InvalidRPCData and UnknownRPCError # see below for a description of those, in each case :statusmsg would be the reason for failure. # # To get access to the full result of the MCollective::Client#req calls you can pass in a block: # # rpc.echo(:msg => "hello world") do |resp| # pp resp # end # # In this case resp will the result from MCollective::Client#req. Instead of returning simple # text and codes as above you'll also need to handle the following exceptions: # # UnknownRPCAction - There is no matching action on the agent # MissingRPCData - You did not supply all the needed parameters for the action # InvalidRPCData - The data you did supply did not pass validation # UnknownRPCError - Some other error prevented the agent from running # # During calls a progress indicator will be shown of how many results we've received against # how many nodes were discovered, you can disable this by setting progress to false: # # rpc.progress = false # # This supports a 2nd mode where it will send the SimpleRPC request and never handle the # responses. It's a bit like UDP, it sends the request with the filter attached and you # only get back the requestid, you have no indication about results. # # You can invoke this using: # # puts rpc.echo(:process_results => false) # # This will output just the request id. # # Batched processing is supported: # # printrpc rpc.ping(:batch_size => 5) # # This will do everything exactly as normal but communicate to only 5 # agents at a time def method_missing(method_name, *args, &block) # set args to an empty hash if nothings given args = args[0] args = {} if args.nil? action = method_name.to_s @stats.reset validate_request(action, args) # if a global batch size is set just use that else set it # in the case that it was passed as an argument batch_mode = args.include?(:batch_size) || @batch_mode batch_size = args.delete(:batch_size) || @batch_size batch_sleep_time = args.delete(:batch_sleep_time) || @batch_sleep_time # if we were given a batch_size argument thats 0 and batch_mode was # determined to be on via global options etc this will allow a batch_size # of 0 to disable or batch_mode for this call only batch_mode = (batch_mode && Integer(batch_size) > 0) # Handle single target requests by doing discovery and picking # a random node. Then do a custom request specifying a filter # that will only match the one node. if @limit_targets target_nodes = pick_nodes_from_discovered(@limit_targets) Log.debug("Picked #{target_nodes.join(',')} as limited target(s)") custom_request(action, args, target_nodes, {"identity" => /^(#{target_nodes.join('|')})$/}, &block) elsif batch_mode call_agent_batched(action, args, options, batch_size, batch_sleep_time, &block) else call_agent(action, args, options, :auto, &block) end end # Constructs custom requests with custom filters and discovery data # the idea is that this would be used in web applications where you # might be using a cached copy of data provided by a registration agent # to figure out on your own what nodes will be responding and what your # filter would be. # # This will help you essentially short circuit the traditional cycle of: # # mc discover / call / wait for discovered nodes # # by doing discovery however you like, contructing a filter and a list of # nodes you expect responses from. # # Other than that it will work exactly like a normal call, blocks will behave # the same way, stats will be handled the same way etcetc # # If you just wanted to contact one machine for example with a client that # already has other filter options setup you can do: # # puppet.custom_request("runonce", {}, ["your.box.com"], {:identity => "your.box.com"}) # # This will do runonce action on just 'your.box.com', no discovery will be # done and after receiving just one response it will stop waiting for responses # # If direct_addressing is enabled in the config file you can provide an empty # hash as a filter, this will force that request to be a directly addressed # request which technically does not need filters. If you try to use this # mode with direct addressing disabled an exception will be raise def custom_request(action, args, expected_agents, filter = {}, &block) validate_request(action, args) if filter == {} && !Config.instance.direct_addressing raise "Attempted to do a filterless custom_request without direct_addressing enabled, preventing unexpected call to all nodes" end @stats.reset custom_filter = Util.empty_filter custom_options = options.clone # merge the supplied filter with the standard empty one # we could just use the merge method but I want to be sure # we dont merge in stuff that isnt actually valid ["identity", "fact", "agent", "cf_class", "compound"].each do |ftype| if filter.include?(ftype) custom_filter[ftype] = [filter[ftype], custom_filter[ftype]].flatten end end # ensure that all filters at least restrict the call to the agent we're a proxy for custom_filter["agent"] << @agent unless custom_filter["agent"].include?(@agent) custom_options[:filter] = custom_filter # Fake out the stats discovery would have put there @stats.discovered_agents([expected_agents].flatten) # Handle fire and forget requests # # If a specific reply-to was set then from the client perspective this should # be a fire and forget request too since no response will ever reach us - it # will go to the reply-to destination if args[:process_results] == false || @reply_to return fire_and_forget_request(action, args, custom_filter) end # Now do a call pretty much exactly like in method_missing except with our own # options and discovery magic if block_given? call_agent(action, args, custom_options, [expected_agents].flatten) do |r| block.call(r) end else call_agent(action, args, custom_options, [expected_agents].flatten) end end def discovery_timeout return @discovery_timeout if @discovery_timeout return @client.discoverer.ddl.meta[:timeout] end def discovery_timeout=(timeout) @discovery_timeout = Float(timeout) # we calculate the overall timeout from the DDL of the agent and # the supplied discovery timeout unless someone specifically # specifies a timeout to the constructor # # But if we also then specifically set a discovery_timeout on the # agent that has to override the supplied timeout so we then # calculate a correct timeout based on DDL timeout and the # supplied discovery timeout @timeout = @ddl.meta[:timeout] + discovery_timeout end # Sets the discovery method. If we change the method there are a # number of steps to take: # # - set the new method # - if discovery options were provided, re-set those to initially # provided ones else clear them as they might now apply to a # different provider # - update the client options so it knows there is a new discovery # method in force # - reset discovery data forcing a discover on the next request # # The remaining item is the discovery timeout, we leave that as is # since that is the user supplied timeout either via initial options # or via specifically setting it on the client. def discovery_method=(method) @discovery_method = method if @initial_options[:discovery_options] @discovery_options = @initial_options[:discovery_options] else @discovery_options.clear end @client.options = options reset end def discovery_options=(options) @discovery_options = [options].flatten reset end # Sets the class filter def class_filter(klass) @filter["cf_class"] << klass @filter["cf_class"].compact! reset end # Sets the fact filter def fact_filter(fact, value=nil, operator="=") return if fact.nil? return if fact == false if value.nil? parsed = Util.parse_fact_string(fact) @filter["fact"] << parsed unless parsed == false else parsed = Util.parse_fact_string("#{fact}#{operator}#{value}") @filter["fact"] << parsed unless parsed == false end @filter["fact"].compact! reset end # Sets the agent filter def agent_filter(agent) @filter["agent"] << agent @filter["agent"].compact! reset end # Sets the identity filter def identity_filter(identity) @filter["identity"] << identity @filter["identity"].compact! reset end # Set a compound filter def compound_filter(filter) @filter["compound"] << Matcher.create_compound_callstack(filter) reset end # Resets various internal parts of the class, most importantly it clears # out the cached discovery def reset @discovered_agents = nil end # Reet the filter to an empty one def reset_filter @filter = Util.empty_filter agent_filter @agent end # Does discovery based on the filters set, if a discovery was # previously done return that else do a new discovery. # # Alternatively if identity filters are given and none of them are # regular expressions then just use the provided data as discovered # data, avoiding discovery # # Discovery can be forced if direct_addressing is enabled by passing # in an array of nodes with :nodes or JSON data like those produced # by mcollective RPC JSON output using :json # # Will show a message indicating its doing discovery if running # verbose or if the :verbose flag is passed in. # # Use reset to force a new discovery def discover(flags={}) flags.keys.each do |key| raise "Unknown option #{key} passed to discover" unless [:verbose, :hosts, :nodes, :json].include?(key) end flags.include?(:verbose) ? verbose = flags[:verbose] : verbose = @verbose verbose = false unless @output_format == :console # flags[:nodes] and flags[:hosts] are the same thing, we should never have # allowed :hosts as that was inconsistent with the established terminology flags[:nodes] = flags.delete(:hosts) if flags.include?(:hosts) reset if flags[:nodes] || flags[:json] unless @discovered_agents # if either hosts or JSON is supplied try to figure out discovery data from there # if direct_addressing is not enabled this is a critical error as the user might # not have supplied filters so raise an exception if flags[:nodes] || flags[:json] raise "Can only supply discovery data if direct_addressing is enabled" unless Config.instance.direct_addressing hosts = [] if flags[:nodes] hosts = Helpers.extract_hosts_from_array(flags[:nodes]) elsif flags[:json] hosts = Helpers.extract_hosts_from_json(flags[:json]) end raise "Could not find any hosts in discovery data provided" if hosts.empty? @discovered_agents = hosts @force_direct_request = true # if an identity filter is supplied and it is all strings no regex we can use that # as discovery data, technically the identity filter is then redundant if we are # in direct addressing mode and we could empty it out but this use case should # only really be for a few -I's on the CLI # # For safety we leave the filter in place for now, that way we can support this # enhancement also in broadcast mode. # # This is only needed for the 'mc' discovery method, other methods might change # the concept of identity to mean something else so we should pass the full # identity filter to them elsif options[:filter]["identity"].size > 0 && @discovery_method == "mc" regex_filters = options[:filter]["identity"].select{|i| i.match("^\/")}.size if regex_filters == 0 @discovered_agents = options[:filter]["identity"].clone @force_direct_request = true if Config.instance.direct_addressing end end end # All else fails we do it the hard way using a traditional broadcast unless @discovered_agents @stats.time_discovery :start @client.options = options # if compound filters are used the only real option is to use the mc # discovery plugin since its the only capable of using data queries etc # and we do not want to degrade that experience just to allow compounds # on other discovery plugins the UX would be too bad raising complex sets # of errors etc. @client.discoverer.force_discovery_method_by_filter(options[:filter]) if verbose actual_timeout = @client.discoverer.discovery_timeout(discovery_timeout, options[:filter]) if actual_timeout > 0 @stderr.print("Discovering hosts using the %s method for %d second(s) .... " % [@client.discoverer.discovery_method, actual_timeout]) else @stderr.print("Discovering hosts using the %s method .... " % [@client.discoverer.discovery_method]) end end # if the requested limit is a pure number and not a percent # and if we're configured to use the first found hosts as the # limit method then pass in the limit thus minimizing the amount # of work we do in the discover phase and speeding it up significantly if @limit_method == :first and @limit_targets.is_a?(Fixnum) @discovered_agents = @client.discover(@filter, discovery_timeout, @limit_targets) else @discovered_agents = @client.discover(@filter, discovery_timeout) end @stderr.puts(@discovered_agents.size) if verbose @force_direct_request = @client.discoverer.force_direct_mode? @stats.time_discovery :end end @stats.discovered_agents(@discovered_agents) RPC.discovered(@discovered_agents) @discovered_agents end # Provides a normal options hash like you would get from # Optionparser def options {:disctimeout => discovery_timeout, :timeout => @timeout, :verbose => @verbose, :filter => @filter, :collective => @collective, :output_format => @output_format, :ttl => @ttl, :discovery_method => @discovery_method, :discovery_options => @discovery_options, :force_display_mode => @force_display_mode, :config => @config} end # Sets the collective we are communicating with def collective=(c) raise "Unknown collective #{c}" unless Config.instance.collectives.include?(c) @collective = c @client.options = options reset end # Sets and sanity checks the limit_targets variable # used to restrict how many nodes we'll target def limit_targets=(limit) if limit.is_a?(String) raise "Invalid limit specified: #{limit} valid limits are /^\d+%*$/" unless limit =~ /^\d+%*$/ begin @limit_targets = Integer(limit) rescue @limit_targets = limit end else @limit_targets = Integer(limit) end end # Sets and sanity check the limit_method variable # used to determine how to limit targets if limit_targets is set def limit_method=(method) method = method.to_sym unless method.is_a?(Symbol) raise "Unknown limit method #{method} must be :random or :first" unless [:random, :first].include?(method) @limit_method = method end # Sets the batch size, if the size is set to 0 that will disable batch mode def batch_size=(limit) raise "Can only set batch size if direct addressing is supported" unless Config.instance.direct_addressing @batch_size = Integer(limit) @batch_mode = @batch_size > 0 end def batch_sleep_time=(time) raise "Can only set batch sleep time if direct addressing is supported" unless Config.instance.direct_addressing @batch_sleep_time = Float(time) end # Pick a number of nodes from the discovered nodes # # The count should be a string that can be either # just a number or a percentage like 10% # # It will select nodes from the discovered list based # on the rpclimitmethod configuration option which can # be either :first or anything else # # - :first would be a simple way to do a distance based # selection # - anything else will just pick one at random # - if random chosen, and batch-seed set, then set srand # for the generator, and reset afterwards def pick_nodes_from_discovered(count) if count =~ /%$/ pct = Integer((discover.size * (count.to_f / 100))) pct == 0 ? count = 1 : count = pct else count = Integer(count) end return discover if discover.size <= count result = [] if @limit_method == :first return discover[0, count] else # we delete from the discovered list because we want # to be sure there is no chance that the same node will # be randomly picked twice. So we have to clone the # discovered list else this method will only ever work # once per discovery cycle and not actually return the # right nodes. haystack = discover.clone if @limit_seed haystack.sort! srand(@limit_seed) end count.times do rnd = rand(haystack.size) result << haystack.delete_at(rnd) end # Reset random number generator to fresh seed # As our seed from options is most likely short srand if @limit_seed end [result].flatten end def load_aggregate_functions(action, ddl) return nil unless ddl return nil unless ddl.action_interface(action).keys.include?(:aggregate) return Aggregate.new(ddl.action_interface(action)) rescue => e Log.error("Failed to load aggregate functions, calculating summaries disabled: %s: %s (%s)" % [e.backtrace.first, e.to_s, e.class]) return nil end def aggregate_reply(reply, aggregate) return nil unless aggregate aggregate.call_functions(reply) return aggregate rescue Exception => e Log.error("Failed to calculate aggregate summaries for reply from %s, calculating summaries disabled: %s: %s (%s)" % [reply[:senderid], e.backtrace.first, e.to_s, e.class]) return nil end def rpc_result_from_reply(agent, action, reply) Result.new(agent, action, {:sender => reply[:senderid], :statuscode => reply[:body][:statuscode], :statusmsg => reply[:body][:statusmsg], :data => reply[:body][:data]}) end # for requests that do not care for results just # return the request id and don't do any of the # response processing. # # We send the :process_results flag with to the # nodes so they can make decisions based on that. # # Should only be called via method_missing def fire_and_forget_request(action, args, filter=nil) validate_request(action, args) req = new_request(action.to_s, args) filter = options[:filter] unless filter message = Message.new(req, nil, {:agent => @agent, :type => :request, :collective => @collective, :filter => filter, :options => options}) message.reply_to = @reply_to if @reply_to return @client.sendreq(message, nil) end # Calls an agent in a way very similar to call_agent but it supports batching # the queries to the network. # # The result sets, stats, block handling etc is all exactly like you would expect # from normal call_agent. # # This is used by method_missing and works only with direct addressing mode def call_agent_batched(action, args, opts, batch_size, sleep_time, &block) raise "Batched requests requires direct addressing" unless Config.instance.direct_addressing raise "Cannot bypass result processing for batched requests" if args[:process_results] == false batch_size = Integer(batch_size) sleep_time = Float(sleep_time) Log.debug("Calling #{agent}##{action} in batches of #{batch_size} with sleep time of #{sleep_time}") @force_direct_request = true discovered = discover results = [] respcount = 0 if discovered.size > 0 req = new_request(action.to_s, args) aggregate = load_aggregate_functions(action, @ddl) if @progress && !block_given? twirl = Progress.new @stdout.puts @stdout.print twirl.twirl(respcount, discovered.size) end @stats.requestid = nil discovered.in_groups_of(batch_size) do |hosts, last_batch| message = Message.new(req, nil, {:agent => @agent, :type => :direct_request, :collective => @collective, :filter => opts[:filter], :options => opts}) # first time round we let the Message object create a request id # we then re-use it for future requests to keep auditing sane etc @stats.requestid = message.create_reqid unless @stats.requestid message.requestid = @stats.requestid message.discovered_hosts = hosts.clone.compact @client.req(message) do |resp| respcount += 1 if block_given? aggregate = process_results_with_block(action, resp, block, aggregate) else @stdout.print twirl.twirl(respcount, discovered.size) if @progress result, aggregate = process_results_without_block(resp, action, aggregate) results << result end end @stats.noresponsefrom.concat @client.stats[:noresponsefrom] @stats.responses += @client.stats[:responses] @stats.blocktime += @client.stats[:blocktime] + sleep_time @stats.totaltime += @client.stats[:totaltime] @stats.discoverytime += @client.stats[:discoverytime] sleep sleep_time unless last_batch end @stats.aggregate_summary = aggregate.summarize if aggregate @stats.aggregate_failures = aggregate.failed if aggregate else @stderr.print("\nNo request sent, we did not discover any nodes.") end @stats.finish_request RPC.stats(@stats) @stdout.print("\n") if @progress if block_given? return stats else return [results].flatten end end # Handles traditional calls to the remote agents with full stats # blocks, non blocks and everything else supported. # # Other methods of calling the nodes can reuse this code by # for example specifying custom options and discovery data def call_agent(action, args, opts, disc=:auto, &block) # Handle fire and forget requests and make sure # the :process_results value is set appropriately # # specific reply-to requests should be treated like # fire and forget since the client will never get # the responses if args[:process_results] == false || @reply_to return fire_and_forget_request(action, args) else args[:process_results] = true end # Do discovery when no specific discovery array is given # # If an array is given set the force_direct_request hint that # will tell the message object to be a direct request one if disc == :auto discovered = discover else @force_direct_request = true if Config.instance.direct_addressing discovered = disc end req = new_request(action.to_s, args) message = Message.new(req, nil, {:agent => @agent, :type => :request, :collective => @collective, :filter => opts[:filter], :options => opts}) message.discovered_hosts = discovered.clone results = [] respcount = 0 if discovered.size > 0 message.type = :direct_request if @force_direct_request if @progress && !block_given? twirl = Progress.new @stdout.puts @stdout.print twirl.twirl(respcount, discovered.size) end aggregate = load_aggregate_functions(action, @ddl) @client.req(message) do |resp| respcount += 1 if block_given? aggregate = process_results_with_block(action, resp, block, aggregate) else @stdout.print twirl.twirl(respcount, discovered.size) if @progress result, aggregate = process_results_without_block(resp, action, aggregate) results << result end end @stats.aggregate_summary = aggregate.summarize if aggregate @stats.aggregate_failures = aggregate.failed if aggregate @stats.client_stats = @client.stats else @stderr.print("\nNo request sent, we did not discover any nodes.") end @stats.finish_request RPC.stats(@stats) @stdout.print("\n\n") if @progress if block_given? return stats else return [results].flatten end end # Handles result sets that has no block associated, sets fails and ok # in the stats object and return a hash of the response to send to the # caller def process_results_without_block(resp, action, aggregate) @stats.node_responded(resp[:senderid]) result = rpc_result_from_reply(@agent, action, resp) aggregate = aggregate_reply(result, aggregate) if aggregate if resp[:body][:statuscode] == 0 || resp[:body][:statuscode] == 1 @stats.ok if resp[:body][:statuscode] == 0 @stats.fail if resp[:body][:statuscode] == 1 else @stats.fail end [result, aggregate] end # process client requests by calling a block on each result # in this mode we do not do anything fancy with the result # objects and we raise exceptions if there are problems with # the data def process_results_with_block(action, resp, block, aggregate) @stats.node_responded(resp[:senderid]) result = rpc_result_from_reply(@agent, action, resp) aggregate = aggregate_reply(result, aggregate) if aggregate if resp[:body][:statuscode] == 0 || resp[:body][:statuscode] == 1 @stats.ok if resp[:body][:statuscode] == 0 @stats.fail if resp[:body][:statuscode] == 1 @stats.time_block_execution :start case block.arity when 1 block.call(resp) when 2 block.call(resp, result) end @stats.time_block_execution :end else @stats.fail case resp[:body][:statuscode] when 2 raise UnknownRPCAction, resp[:body][:statusmsg] when 3 raise MissingRPCData, resp[:body][:statusmsg] when 4 raise InvalidRPCData, resp[:body][:statusmsg] when 5 raise UnknownRPCError, resp[:body][:statusmsg] end end return aggregate end end end end