@@ -11,3 +11,4 @@ config.yml spec/config.yml *.tmproj nbproject +.idea .gitignore @@ -164,28 +164,32 @@ module Nanite end def receive(packet) - packet = serializer.load(packet) - case packet - when Advertise - Nanite::Log.debug("handling Advertise: #{packet}") - advertise_services - when Request, Push - Nanite::Log.debug("handling Request: #{packet}") - if @security && !@security.authorize(packet) - if packet.kind_of?(Request) - r = Result.new(packet.token, packet.reply_to, @deny_token, identity) - amq.queue(packet.reply_to, :no_declare => options[:secure]).publish(serializer.dump(r)) + begin + packet = serializer.load(packet) + case packet + when Advertise + Nanite::Log.debug("handling Advertise: #{packet}") + advertise_services + when Request, Push + Nanite::Log.debug("handling Request: #{packet}") + if @security && !@security.authorize(packet) + if packet.kind_of?(Request) + r = Result.new(packet.token, packet.reply_to, @deny_token, identity) + amq.queue(packet.reply_to, :no_declare => options[:secure]).publish(serializer.dump(r)) + end + else + dispatcher.dispatch(packet) end - else - dispatcher.dispatch(packet) + when Result + Nanite::Log.debug("handling Result: #{packet}") + @mapper_proxy.handle_result(packet) + when IntermediateMessage + Nanite::Log.debug("handling Intermediate Result: #{packet}") + @mapper_proxy.handle_intermediate_result(packet) end - when Result - Nanite::Log.debug("handling Result: #{packet}") - @mapper_proxy.handle_result(packet) - when IntermediateMessage - Nanite::Log.debug("handling Intermediate Result: #{packet}") - @mapper_proxy.handle_intermediate_result(packet) - end + rescue Exception => e + Nanite::Log.error("Error handling packet #{packet.inspect}:\n#{e.message} at #{e.backtrace[0]}") + end end def tag(*tags) lib/nanite/agent.rb @@ -56,7 +56,7 @@ module Nanite # access to it. begin old_target = request.target - request.target = target + request.target = target unless target == 'mapper-offline' amq.queue(target).publish(serializer.dump(request), :persistent => request.persistent) ensure request.target = old_target @@ -68,32 +68,40 @@ module Nanite # updates nanite information (last ping timestamps, status) # when heartbeat message is received def handle_ping(ping) - if nanite = nanites[ping.identity] - nanite[:status] = ping.status - reaper.reset_with_autoregister_hack(ping.identity, agent_timeout + 1) { nanites.delete(ping.identity) } - else - amq.queue(ping.identity).publish(serializer.dump(Advertise.new)) + begin + if nanite = nanites[ping.identity] + nanite[:status] = ping.status + reaper.reset_with_autoregister_hack(ping.identity, agent_timeout + 1) { nanites.delete(ping.identity) } + else + amq.queue(ping.identity).publish(serializer.dump(Advertise.new)) + end + rescue Exception => e + Nanite::Log.error("Error handling ping #{ping.inspect}:\n#{e.message} at #{e.backtrace[0]}") end end # forward request coming from agent def handle_request(request) - if @security.authorize_request(request) - result = Result.new(request.token, request.from, nil, mapper.identity) - intm_handler = lambda do |res| - result.results = res - forward_response(result, request.persistent) - end - ok = mapper.send_request(request, :intermediate_handler => intm_handler) do |res| - result.results = res - forward_response(result, request.persistent) - end - if ok == false - forward_response(result, request.persistent) + begin + if @security.authorize_request(request) + result = Result.new(request.token, request.from, nil, mapper.identity) + intm_handler = lambda do |res| + result.results = res + forward_response(result, request.persistent) + end + ok = mapper.send_request(request, :intermediate_handler => intm_handler) do |res| + result.results = res + forward_response(result, request.persistent) + end + if ok == false + forward_response(result, request.persistent) + end + else + Nanite::Log.warning("request #{request.inspect} not authorized") end - else - Nanite::Log.warning("request #{request.inspect} not authorized") - end + rescue Exception => e + Nanite::Log.error("Error handling request #{request.inspect}:\n#{e.message} at #{e.backtrace[0]}") + end end # forward response back to agent that originally made the request lib/nanite/cluster.rb @@ -14,45 +14,48 @@ module Nanite end def process(msg) - msg = serializer.load(msg) - Nanite::Log.debug("processing message: #{msg.inspect}") + begin + msg = serializer.load(msg) + Nanite::Log.debug("processing message: #{msg.inspect}") - if job = jobs[msg.token] - job.process(msg) + if job = jobs[msg.token] + job.process(msg) - if job.intermediate_handler && (job.pending_keys.size > 0) + if job.intermediate_handler && (job.pending_keys.size > 0) - unless job.pending_keys.size == 1 - raise "IntermediateMessages are currently dispatched as they arrive, shouldn't have more than one key in pending_keys: #{job.pending_keys.inspect}" - end + unless job.pending_keys.size == 1 + raise "IntermediateMessages are currently dispatched as they arrive, shouldn't have more than one key in pending_keys: #{job.pending_keys.inspect}" + end - key = job.pending_keys.first - handler = job.intermediate_handler_for_key(key) - if handler - case handler.arity - when 3 - handler.call(key, msg.from, job.intermediate_state[msg.from][key].last) - when 4 - handler.call(key, msg.from, job.intermediate_state[msg.from][key].last, job) + key = job.pending_keys.first + handler = job.intermediate_handler_for_key(key) + if handler + case handler.arity + when 3 + handler.call(key, msg.from, job.intermediate_state[msg.from][key].last) + when 4 + handler.call(key, msg.from, job.intermediate_state[msg.from][key].last, job) + end end - end - job.reset_pending_intermediate_state_keys - end + job.reset_pending_intermediate_state_keys + end - if job.completed? - jobs.delete(job.token) - if job.completed - case job.completed.arity - when 1 - job.completed.call(job.results) - when 2 - job.completed.call(job.results, job) + if job.completed? + jobs.delete(job.token) + if job.completed + case job.completed.arity + when 1 + job.completed.call(job.results) + when 2 + job.completed.call(job.results, job) + end end end end - end + rescue Exception => e + Nanite::Log.error("Error handling message #{msg.inspect}:\n#{e.message} at #{e.backtrace[0]}") end end # JobWarden lib/nanite/job.rb bf99b8bc9652806136b6e916923402d415ba5569 raphael raphael@Laela.(none) http://github.com/ezmobius/nanite/commit/5ee047b854fcbf2785cab38abae271d402e4bdef 5ee047b854fcbf2785cab38abae271d402e4bdef 2009-05-25T19:36:09-07:00 2009-05-25T19:36:09-07:00 Harden agent and mapper by adding exception handlers in all AMQP messages processing functions. e8ff0538ef32778e5830c2b66589475de7bce680 raphael raphael@Laela.(none)