Class: MetriksdReporter::Reporter
- Inherits:
-
Object
- Object
- MetriksdReporter::Reporter
- Defined in:
- lib/metriksd_reporter/reporter.rb
Instance Method Summary collapse
- #append_to_packet(data) ⇒ Object
- #extract_from_metric(metric, *keys) ⇒ Object
- #flush ⇒ Object
- #flush_packet ⇒ Object
- #flush_packet_if_full ⇒ Object
-
#initialize(options = {}) ⇒ Reporter
constructor
A new instance of Reporter.
- #join ⇒ Object
- #max_packet_size_with_compression_ratio ⇒ Object
- #restart ⇒ Object
- #sleep_for_up_to(duration) ⇒ Object
- #sleep_until_deadline ⇒ Object
- #start ⇒ Object
- #stop ⇒ Object
- #write ⇒ Object
- #write_metric(name, type, metric, keys, snapshot_keys = []) ⇒ Object
Constructor Details
#initialize(options = {}) ⇒ Reporter
Returns a new instance of Reporter.
6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# File 'lib/metriksd_reporter/reporter.rb', line 6 def initialize( = {}) missing_keys = %w(port host) - .keys.map(&:to_s) unless missing_keys.empty? raise ArgumentError, "Missing required options: #{missing_keys * ', '}" end @port = [:port] @host = [:host] @client_id = [:client_id] || "#{Socket.gethostname}:#{$$}" @extras = [:extras] || {} @registry = [:registry] || Metriks::Registry.default @max_packet_size = [:max_packet_size] || 1000 @interval = [:interval] || 60 @interval_offset = [:interval_offset] || 0 @flush_delay = [:flush_delay] || 0.6 @on_error = [:on_error] || proc { |ex| } end |
Instance Method Details
#append_to_packet(data) ⇒ Object
114 115 116 117 118 |
# File 'lib/metriksd_reporter/reporter.rb', line 114 def append_to_packet(data) @packet ||= '' @packet << data.to_msgpack flush_packet_if_full end |
#extract_from_metric(metric, *keys) ⇒ Object
148 149 150 151 152 153 154 155 156 157 |
# File 'lib/metriksd_reporter/reporter.rb', line 148 def extract_from_metric(metric, *keys) h = {} keys.flatten.collect do |key| name = key.to_s.gsub(/^get_/, '') h[name] = metric.send(key) end h end |
#flush ⇒ Object
68 69 70 |
# File 'lib/metriksd_reporter/reporter.rb', line 68 def flush write end |
#flush_packet ⇒ Object
127 128 129 130 131 132 133 134 135 136 137 138 |
# File 'lib/metriksd_reporter/reporter.rb', line 127 def flush_packet if @packet && @packet.length > 0 compressed = Snappy.deflate(@packet) # Calculate the compression ratio @compression_ratio = @packet.length / compressed.length # Send the packet @socket.send(compressed, 0, @host, @port) @packet = '' end end |
#flush_packet_if_full ⇒ Object
120 121 122 123 124 125 |
# File 'lib/metriksd_reporter/reporter.rb', line 120 def flush_packet_if_full if @packet && @packet.length > 0 && @packet.length > max_packet_size_with_compression_ratio flush_packet sleep_for_up_to(0.6) end end |
#join ⇒ Object
57 58 59 60 61 |
# File 'lib/metriksd_reporter/reporter.rb', line 57 def join if @thread @thread.join end end |
#max_packet_size_with_compression_ratio ⇒ Object
140 141 142 143 144 145 146 |
# File 'lib/metriksd_reporter/reporter.rb', line 140 def max_packet_size_with_compression_ratio if @compression_ratio @max_packet_size * @compression_ratio * 0.9 else @max_packet_size end end |
#restart ⇒ Object
63 64 65 66 |
# File 'lib/metriksd_reporter/reporter.rb', line 63 def restart stop start end |
#sleep_for_up_to(duration) ⇒ Object
190 191 192 193 194 195 |
# File 'lib/metriksd_reporter/reporter.rb', line 190 def sleep_for_up_to(duration) duration *= rand if duration > 0 sleep(duration) end end |
#sleep_until_deadline ⇒ Object
177 178 179 180 181 182 183 184 185 186 187 188 |
# File 'lib/metriksd_reporter/reporter.rb', line 177 def sleep_until_deadline # Ensure we round up when we should now = Time.now.to_f + @interval_offset rounded = now - (now % @interval) next_rounded = rounded + @interval - @interval_offset sleep_time = next_rounded - Time.now.to_f if sleep_time > 0 sleep(sleep_time) end end |
#start ⇒ Object
27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
# File 'lib/metriksd_reporter/reporter.rb', line 27 def start @socket ||= UDPSocket.new @thread ||= Thread.new do loop do sleep_until_deadline begin write rescue Exception => ex @on_error[ex] rescue nil end end end end |
#stop ⇒ Object
43 44 45 46 47 48 49 50 51 52 53 54 55 |
# File 'lib/metriksd_reporter/reporter.rb', line 43 def stop if @thread @thread.exit @thread = nil end flush if @socket @socket.close @socket = nil end end |
#write ⇒ Object
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
# File 'lib/metriksd_reporter/reporter.rb', line 72 def write @registry.each do |name, metric| case metric when Metriks::Meter write_metric name, 'meter', metric, [ :count, :one_minute_rate, :five_minute_rate, :fifteen_minute_rate, :mean_rate ] when Metriks::Counter write_metric name, 'counter', metric, [ :count ] when Metriks::UtilizationTimer write_metric name, 'utilization_timer', metric, [ :count, :one_minute_rate, :five_minute_rate, :fifteen_minute_rate, :mean_rate, :min, :max, :mean, :stddev, :one_minute_utilization, :five_minute_utilization, :fifteen_minute_utilization, :mean_utilization, ], [ :median, :get_95th_percentile ] when Metriks::Timer write_metric name, 'timer', metric, [ :count, :one_minute_rate, :five_minute_rate, :fifteen_minute_rate, :mean_rate, :min, :max, :mean, :stddev ], [ :median, :get_95th_percentile ] when Metriks::Histogram write_metric name, 'histogram', metric, [ :count, :min, :max, :mean, :stddev ], [ :median, :get_95th_percentile ] end end flush_packet end |
#write_metric(name, type, metric, keys, snapshot_keys = []) ⇒ Object
159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
# File 'lib/metriksd_reporter/reporter.rb', line 159 def write_metric(name, type, metric, keys, snapshot_keys = []) = @extras.merge( :client_id => @client_id, :time => Time.now.to_i, :name => name, :type => type ) .merge!(extract_from_metric(metric, keys)) unless snapshot_keys.empty? snapshot = metric.snapshot .merge!(extract_from_metric(snapshot, snapshot_keys)) end append_to_packet() end |