Class: TaskTempest::Bookkeeper
Constant Summary collapse
- KB =
1024
- MB =
KB**2
Instance Attribute Summary collapse
-
#logger ⇒ Object
readonly
Returns the value of attribute logger.
Instance Method Summary collapse
- #format_memory(memory) ⇒ Object
- #get_files(which) ⇒ Object
- #get_memory(which) ⇒ Object
-
#initialize(options) ⇒ Bookkeeper
constructor
A new instance of Bookkeeper.
- #make_book ⇒ Object
- #report(executions) ⇒ Object
- #task_counts ⇒ Object
- #task_durations ⇒ Object
- #task_throughput ⇒ Object
- #tasks_per_thread ⇒ Object
Constructor Details
#initialize(options) ⇒ Bookkeeper
Returns a new instance of Bookkeeper.
5 6 7 |
# File 'lib/task_tempest/bookkeeper.rb', line 5 def initialize() .each{ |k, v| instance_variable_set("@#{k}", v) } end |
Instance Attribute Details
#logger ⇒ Object (readonly)
Returns the value of attribute logger.
3 4 5 |
# File 'lib/task_tempest/bookkeeper.rb', line 3 def logger @logger end |
Instance Method Details
#format_memory(memory) ⇒ Object
128 129 130 131 132 133 134 |
# File 'lib/task_tempest/bookkeeper.rb', line 128 def format_memory(memory) if memory > MB (memory / MB).to_s + "M" else (memory / KB).to_s + "K" end end |
#get_files(which) ⇒ Object
113 114 115 116 117 118 119 120 121 122 123 124 |
# File 'lib/task_tempest/bookkeeper.rb', line 113 def get_files(which) @files ||= begin output = `lsof -p #{Process.pid}` rescue "" output.split("\n") end case which when :total @files.length when :tcp @files.inject(0){ |memo, line| memo += 1 if line.downcase =~ /tcp/; memo } end end |
#get_memory(which) ⇒ Object
103 104 105 106 107 108 109 110 111 |
# File 'lib/task_tempest/bookkeeper.rb', line 103 def get_memory(which) @memory ||= `ps -o rss= -o vsz= -p #{Process.pid}`.split.collect{ |s| s.strip } rescue [nil, nil] case which when :resident @memory[0].to_i when :virtual @memory[1].to_i end end |
#make_book ⇒ Object
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
# File 'lib/task_tempest/bookkeeper.rb', line 21 def make_book # Do some setup. ObjectSpace.garbage_collect queue = @queue_factory.call book = {} # Reset memoized objects. @memory = nil @files = nil # Task success/error counts. book[:tasks] = {} book[:tasks][:counts] = task_counts book[:tasks][:per_thread] = tasks_per_thread book[:tasks][:durations] = task_durations book[:tasks][:throughput] = task_throughput # Thread (worker) info. book[:threads] = {} book[:threads][:busy] = @storm.busy_workers.length book[:threads][:idle] = @storm.size - book[:threads][:busy] book[:threads][:saturation] = (book[:threads][:busy] / @storm.size.to_f * 100).round(2) # Memory, Object, GC info. book[:memory] = {} book[:memory][:live_objects] = ObjectSpace.live_objects rescue nil book[:memory][:resident] = format_memory(get_memory(:resident)) book[:memory][:virtual] = format_memory(get_memory(:virtual)) # Open file counts. book[:files] = {} book[:files][:total_count] = get_files(:total) book[:files][:tcp_count] = get_files(:tcp) # Queue info. book[:queue] = {} book[:queue][:size] = queue.size if queue.respond_to?(:size) book[:queue][:backlog] = @storm.executions.inject(0){ |memo, e| memo += 1 unless e.started?; memo } book end |
#report(executions) ⇒ Object
9 10 11 12 13 14 15 16 17 18 19 |
# File 'lib/task_tempest/bookkeeper.rb', line 9 def report(executions) @timer ||= Time.now @executions ||= [] @executions += executions if Time.now - @timer > @interval logger.info "[STATS] " + make_book.inspect @executions.clear @timer = Time.now end end |
#task_counts ⇒ Object
64 65 66 67 68 69 70 71 72 73 74 75 |
# File 'lib/task_tempest/bookkeeper.rb', line 64 def task_counts tot = @executions.length err = @executions.sum{ |e| e.exception ? 1 : 0 } pct = begin if tot > 0 (err.to_f / tot) else 0.0 end end { :tot => tot, :err => err, :pct => pct.round(3) } end |
#task_durations ⇒ Object
94 95 96 97 98 99 100 101 |
# File 'lib/task_tempest/bookkeeper.rb', line 94 def task_durations durations = @executions.collect{ |execution| execution.duration } if durations.length > 0 { :min => durations.min.round(3), :max => durations.max.round(3), :avg => durations.avg.round(3) } else "n/a" end end |
#task_throughput ⇒ Object
77 78 79 80 81 82 |
# File 'lib/task_tempest/bookkeeper.rb', line 77 def task_throughput duration = Time.now - @timer per_sec = @executions.length.to_f / duration per_min = (per_sec * 60).round(2) "#{per_min}/m" end |
#tasks_per_thread ⇒ Object
84 85 86 87 88 89 90 91 92 |
# File 'lib/task_tempest/bookkeeper.rb', line 84 def tasks_per_thread counts_by_thread = @storm.threads.inject({}) do |memo, thread| memo[thread] = 0 memo end @executions.each{ |e| counts_by_thread[e.thread] += 1 } counts = counts_by_thread.values { :min => counts.min, :max => counts.max, :avg => counts.avg.round(2) } end |