@@ -54,6 +54,6 @@ opts.parse! EM.run do Nanite.start_mapper(options) - puts "starting nanite-admin" + Nanite::Log.info "starting nanite-admin" Rack::Handler::Thin.run(Nanite::Admin.new(Nanite.mapper), :Port => 4000) end \ No newline at end of file bin/nanite-admin @@ -1,15 +1,14 @@ module Nanite class ActorRegistry - attr_reader :actors, :log + attr_reader :actors - def initialize(log) - @log = log + def initialize @actors = {} end def register(actor, prefix) raise ArgumentError, "#{actor.inspect} is not a Nanite::Actor subclass instance" unless Nanite::Actor === actor - log.info("Registering #{actor.inspect} with prefix #{prefix.inspect}") + Nanite::Log.info("Registering #{actor.inspect} with prefix #{prefix.inspect}") prefix ||= actor.class.default_prefix actors[prefix.to_s] = actor end lib/nanite/actor_registry.rb @@ -5,7 +5,7 @@ module Nanite include ConsoleHelper include DaemonizeHelper - attr_reader :identity, :log, :options, :serializer, :dispatcher, :registry, :amq + attr_reader :identity, :options, :serializer, :dispatcher, :registry, :amq attr_accessor :status_proc DEFAULT_OPTIONS = COMMON_DEFAULT_OPTIONS.merge({:user => 'nanite', :ping_time => 15, @@ -73,13 +73,18 @@ module Nanite def initialize(opts) set_configuration(opts) - @log = Log.new(@options, @identity) + log_path = false + if @options[:daemonize] + log_path = (@options[:log_dir] || @options[:root] || Dir.pwd) + end + Log.init(@identity, log_path) + Log.log_level = @options[:log_level] || :info @serializer = Serializer.new(@options[:format]) @status_proc = lambda { parse_uptime(`uptime`) rescue 'no status' } daemonize if @options[:daemonize] @amq = start_amqp(@options) - @registry = ActorRegistry.new(@log) - @dispatcher = Dispatcher.new(@amq, @registry, @serializer, @identity, @log, @options) + @registry = ActorRegistry.new + @dispatcher = Dispatcher.new(@amq, @registry, @serializer, @identity, @options) load_actors setup_queue advertise_services @@ -116,7 +121,7 @@ module Nanite def load_actors return unless options[:root] Dir["#{options[:root]}/actors/*.rb"].each do |actor| - log.info("loading actor: #{actor}") + Nanite::Log.info("loading actor: #{actor}") require actor end init_path = File.join(options[:root], 'init.rb') @@ -127,10 +132,10 @@ module Nanite packet = serializer.load(packet) case packet when Advertise - log.debug("handling Advertise: #{packet}") + Nanite::Log.debug("handling Advertise: #{packet}") advertise_services when Request, Push - log.debug("handling Request: #{packet}") + Nanite::Log.debug("handling Request: #{packet}") dispatcher.dispatch(packet) end end @@ -149,7 +154,7 @@ module Nanite end def advertise_services - log.debug("advertise_services: #{registry.services.inspect}") + Nanite::Log.debug("advertise_services: #{registry.services.inspect}") amq.fanout('registration', :no_declare => options[:secure]).publish(serializer.dump(Register.new(identity, registry.services, status_proc.call))) end lib/nanite/agent.rb @@ -1,15 +1,14 @@ module Nanite class Cluster - attr_reader :agent_timeout, :nanites, :reaper, :log, :serializer, :identity, :amq + attr_reader :agent_timeout, :nanites, :reaper, :serializer, :identity, :amq - def initialize(amq, agent_timeout, identity, log, serializer) + def initialize(amq, agent_timeout, identity, serializer) @amq = amq @agent_timeout = agent_timeout @identity = identity @serializer = serializer @nanites = {} @reaper = Reaper.new(agent_timeout) - @log = log setup_queues end @@ -25,7 +24,7 @@ module Nanite def register(reg) nanites[reg.identity] = { :services => reg.services, :status => reg.status } reaper.timeout(reg.identity, agent_timeout + 1) { nanites.delete(reg.identity) } - log.info("registered: #{reg.identity}, #{nanites[reg.identity]}") + Nanite::Log.info("registered: #{reg.identity}, #{nanites[reg.identity]}") end def route(request, targets) @@ -95,14 +94,14 @@ module Nanite def setup_heartbeat_queue amq.queue("heartbeat-#{identity}", :exclusive => true).bind(amq.fanout('heartbeat', :durable => true)).subscribe do |ping| - log.debug('got heartbeat') + Nanite::Log.debug('got heartbeat') handle_ping(serializer.load(ping)) end end def setup_registration_queue amq.queue("registration-#{identity}", :exclusive => true).bind(amq.fanout('registration', :durable => true)).subscribe do |msg| - log.debug('got registration') + Nanite::Log.debug('got registration') register(serializer.load(msg)) end end lib/nanite/cluster.rb @@ -4,9 +4,9 @@ module Nanite exit if fork Process.setsid exit if fork - $stdin.reopen("/dev/null") - $stdout.reopen(log.file, "a") - $stderr.reopen($stdout) + #$stdin.reopen("/dev/null") + #$stdout.reopen(log.file, "a") + #$stderr.reopen($stdout) end end end \ No newline at end of file lib/nanite/daemonize.rb @@ -1,13 +1,12 @@ module Nanite class Dispatcher - attr_reader :registry, :serializer, :identity, :log, :amq, :options + attr_reader :registry, :serializer, :identity, :amq, :options - def initialize(amq, registry, serializer, identity, log, options) + def initialize(amq, registry, serializer, identity, options) @amq = amq @registry = registry @serializer = serializer @identity = identity - @log = log @options = options end @@ -36,7 +35,7 @@ module Nanite def handle_exception(actor, meth, deliverable, e) error = describe_error(e) - log.error(error) + Nanite::Log.error(error) begin if actor.class.instance_exception_callback case actor.class.instance_exception_callback @@ -51,7 +50,7 @@ module Nanite end rescue Exception => e1 error = describe_error(e1) - log.error(error) + Nanite::Log.error(error) end error end lib/nanite/dispatcher.rb @@ -1,10 +1,9 @@ module Nanite class JobWarden - attr_reader :serializer, :jobs, :log + attr_reader :serializer, :jobs - def initialize(serializer, log) + def initialize(serializer) @serializer = serializer - @log = log @jobs = {} end @@ -16,7 +15,7 @@ module Nanite def process(msg) msg = serializer.load(msg) - log.debug("processing message: #{msg.inspect}") + Nanite::Log.debug("processing message: #{msg.inspect}") if job = jobs[msg.token] job.process(msg) if job.completed? lib/nanite/job.rb @@ -17,13 +17,13 @@ module Nanite # If this method is called with no arguments, it will log to STDOUT at the :info level. # # It also configures the Logger instance it creates to use the custom Nanite::Log::Formatter class. - def init(*opts) - if opts.length == 0 - @logger = Logger.new(STDOUT) - else - @logger = Logger.new(*opts) + def init(identity, path = false) + @file = STDOUT + if path + @file = File.join(path, "nanite.#{identity}.log") end - @logger.formatter = Nanite::Log::Formatter.new() + @logger = Logger.new(@file) + @logger.formatter = Nanite::Log::Formatter.new level(@log_level = :info) end @@ -58,7 +58,7 @@ module Nanite # this method gets hit before a call to Nanite::Logger.init has been made, it will call # Nanite::Logger.init() with no arguments. def method_missing(method_symbol, *args) - init() unless @logger + init(identity) unless @logger if args.length > 0 @logger.send(method_symbol, *args) else lib/nanite/log.rb @@ -21,7 +21,7 @@ module Nanite include ConsoleHelper include DaemonizeHelper - attr_reader :cluster, :identity, :job_warden, :options, :serializer, :log, :amq + attr_reader :cluster, :identity, :job_warden, :options, :serializer, :amq DEFAULT_OPTIONS = COMMON_DEFAULT_OPTIONS.merge({:user => 'mapper', :identity => Identity.generate, :agent_timeout => 15, :offline_redelivery_frequency => 10, :persistent => false, :offline_failsafe => false}) unless defined?(DEFAULT_OPTIONS) @@ -83,15 +83,30 @@ module Nanite end def initialize(options) - @options = DEFAULT_OPTIONS.merge(options) + @options = DEFAULT_OPTIONS.clone.merge(options) + root = options[:root] || @options[:root] + custom_config = if root + file = File.expand_path(File.join(root, 'config.yml')) + File.exists?(file) ? (YAML.load(IO.read(file)) || {}) : {} + else + {} + end + options.delete(:identity) unless options[:identity] + @options.update(custom_config.merge(options)) @identity = "mapper-#{@options[:identity]}" - @log = Log.new(@options, @identity) + @options[:file_root] ||= File.join(@options[:root], 'files') + log_path = false + if @options[:daemonize] + log_path = (@options[:log_dir] || @options[:root] || Dir.pwd) + end + Log.init(@identity, log_path) + Log.log_level = @options[:log_level] @serializer = Serializer.new(@options[:format]) daemonize if @options[:daemonize] - @amq =start_amqp(@options) - @cluster = Cluster.new(@amq, @options[:agent_timeout], @options[:identity], @log, @serializer) - @job_warden = JobWarden.new(@serializer, @log) - @log.info('starting mapper') + @amq = start_amqp(@options) + @cluster = Cluster.new(@amq, @options[:agent_timeout], @options[:identity], @serializer) + @job_warden = JobWarden.new(@serializer) + Nanite::Log.info('starting mapper') setup_queues start_console if @options[:console] && !@options[:daemonize] end lib/nanite/mapper.rb @@ -89,14 +89,14 @@ module Nanite def handle_packet(packet) case packet when Nanite::FileChunk - log.debug "written chunk to #{@dest.inspect}" + Nanite::Log.debug "written chunk to #{@dest.inspect}" @data << packet.chunk if @write @dest.write(packet.chunk) end when Nanite::FileEnd - log.debug "#{@dest.inspect} receiving is completed" + Nanite::Log.debug "#{@dest.inspect} receiving is completed" if @write @dest.close end @@ -108,7 +108,7 @@ module Nanite end def subscribe_to_files(domain='global', write=false, &blk) - log.info "subscribing to file broadcasts for #{domain}" + Nanite::Log.info "subscribing to file broadcasts for #{domain}" @files ||= {} amq.queue("files#{domain}").bind(amq.topic('file broadcast'), :key => "nanite.filepeer.#{domain}").subscribe do |packet| case msg = load_packet(packet) lib/nanite/streaming.rb c6af46d9d747550bcb21adb88c6744bc73a5ede0 Benjamin Black bb@joyeuse.local http://github.com/ezmobius/nanite/commit/122bdbe30fce7848fabb88b139bc1144c6b430de 122bdbe30fce7848fabb88b139bc1144c6b430de 2009-03-05T15:35:42-08:00 2009-03-05T15:35:42-08:00 [central_logger] updated everything for central logging. thin still chatty, though. 0fb52d3b275f20eb79e30edee6ac5427429b248e Benjamin Black bb@joyeuse.local