watcher: add peak times for statistics

It could be useful to know when the first and last peak time of
a maximum was.

2 files changed, 70 insertions, 14 deletions

diff --git a/lib/raindrops/watcher.rb b/lib/raindrops/watcher.rb
index f343e43..19756a7 100644
--- a/lib/raindrops/watcher.rb
+++ b/lib/raindrops/watcher.rb
@@ -104,6 +104,7 @@ class Raindrops::Watcher
   include Rack::Utils
   include Raindrops::Linux
   DOC_URL = "http://raindrops.bogomips.org/Raindrops/Watcher.html"
+  Peak = Struct.new(:first, :last)
 
   def initialize(opts = {})
     @tcp_listeners = @unix_listeners = nil
@@ -121,6 +122,8 @@ class Raindrops::Watcher
     @active = Hash.new { |h,k| h[k] = agg_class.new }
     @queued = Hash.new { |h,k| h[k] = agg_class.new }
     @resets = Hash.new { |h,k| h[k] = start }
+    @peak_active = Hash.new { |h,k| h[k] = Peak.new(start, start) }
+    @peak_queued = Hash.new { |h,k| h[k] = Peak.new(start, start) }
     @snapshot = [ start, {} ]
     @delay = opts[:delay] || 1
     @lock = Mutex.new
@@ -146,6 +149,16 @@ class Raindrops::Watcher
     end
   end
 
+  def aggregate!(agg_hash, peak_hash, addr, number, now)
+    agg = agg_hash[addr]
+    if (max = agg.max) && number > 0 && number >= max
+      peak = peak_hash[addr]
+      peak.first = now if number > max
+      peak.last = now
+    end
+    agg << number
+  end
+
   def aggregator_thread(logger) # :nodoc:
     @socket = sock = Raindrops::InetDiagSocket.new
     thr = Thread.new do
@@ -153,11 +166,12 @@ class Raindrops::Watcher
         combined = tcp_listener_stats(@tcp_listeners, sock)
         combined.merge!(unix_listener_stats(@unix_listeners))
         @lock.synchronize do
+          now = Time.now.utc
           combined.each do |addr,stats|
-            @active[addr] << stats.active
-            @queued[addr] << stats.queued
+            aggregate!(@active, @peak_active, addr, stats.active, now)
+            aggregate!(@queued, @peak_queued, addr, stats.queued, now)
           end
-          @snapshot = [ Time.now.utc, combined ]
+          @snapshot = [ now, combined ]
           @cond.broadcast
         end
       rescue => e
@@ -171,17 +185,17 @@ class Raindrops::Watcher
 
   def active_stats(addr) # :nodoc:
     @lock.synchronize do
-      tmp = @active[addr] or return
+      tmp, peak = @active[addr], @peak_active[addr]
       time, combined = @snapshot
-      [ time, @resets[addr], tmp.dup, combined[addr].active ]
+      [ time, @resets[addr], tmp.dup, combined[addr].active, peak ]
     end
   end
 
   def queued_stats(addr) # :nodoc:
     @lock.synchronize do
-      tmp = @queued[addr] or return
+      tmp, peak = @queued[addr], @peak_queued[addr]
       time, combined = @snapshot
-      [ time, @resets[addr], tmp.dup, combined[addr].queued ]
+      [ time, @resets[addr], tmp.dup, combined[addr].queued, peak ]
     end
   end
 
@@ -192,7 +206,7 @@ class Raindrops::Watcher
     end
   end
 
-  def agg_to_hash(reset_at, agg)
+  def agg_to_hash(reset_at, agg, current, peak)
     {
       "X-Count" => agg.count.to_s,
       "X-Min" => agg.min.to_s,
@@ -202,14 +216,16 @@ class Raindrops::Watcher
       "X-Outliers-Low" => agg.outliers_low.to_s,
       "X-Outliers-High" => agg.outliers_high.to_s,
       "X-Last-Reset" => reset_at.httpdate,
+      "X-Current" => current.to_s,
+      "X-First-Peak-At" => peak.first.httpdate,
+      "X-Last-Peak-At" => peak.last.httpdate,
     }
   end
 
   def histogram_txt(agg)
-    updated_at, reset_at, agg, current = *agg
-    headers = agg_to_hash(reset_at, agg)
+    updated_at, reset_at, agg, current, peak = *agg
+    headers = agg_to_hash(reset_at, agg, current, peak)
     body = agg.to_s
-    headers["X-Current"] = current.to_s
     headers["Content-Type"] = "text/plain"
     headers["Expires"] = (updated_at + @delay).httpdate
     headers["Content-Length"] = bytesize(body).to_s
@@ -217,8 +233,8 @@ class Raindrops::Watcher
   end
 
   def histogram_html(agg, addr)
-    updated_at, reset_at, agg, current = *agg
-    headers = agg_to_hash(reset_at, agg)
+    updated_at, reset_at, agg, current, peak = *agg
+    headers = agg_to_hash(reset_at, agg, current, peak)
     body = "<html>" \
       "<head><title>#{hostname} - #{escape_html addr}</title></head>" \
       "<body><table>" <<
@@ -228,7 +244,6 @@ class Raindrops::Watcher
       "<form action='/reset/#{escape addr}' method='post'>" \
       "<input type='submit' name='x' value='reset' /></form>" \
       "</body>"
-    headers["X-Current"] = current.to_s
     headers["Content-Type"] = "text/html"
     headers["Expires"] = (updated_at + @delay).httpdate
     headers["Content-Length"] = bytesize(body).to_s
diff --git a/test/test_watcher.rb b/test/test_watcher.rb
index 11a2d86..f22a954 100644
--- a/test/test_watcher.rb
+++ b/test/test_watcher.rb
@@ -117,4 +117,45 @@ class TestWatcher < Test::Unit::TestCase
     status, headers, body = @app.call(env)
     assert_equal "1", headers["X-Current"], headers.inspect
   end
+
+  def test_peaks
+    env = @req.class.env_for "/active/#@addr.txt"
+    status, headers, body = @app.call(env.dup)
+    start = headers["X-First-Peak-At"]
+    assert headers["X-First-Peak-At"], headers.inspect
+    assert headers["X-Last-Peak-At"], headers.inspect
+    assert_nothing_raised { Time.parse(headers["X-First-Peak-At"]) }
+    assert_nothing_raised { Time.parse(headers["X-Last-Peak-At"]) }
+    before = headers["X-Last-Peak-At"]
+
+    env = @req.class.env_for "/queued/#@addr.txt"
+    status, headers, body = @app.call(env)
+    assert_nothing_raised { Time.parse(headers["X-First-Peak-At"]) }
+    assert_nothing_raised { Time.parse(headers["X-Last-Peak-At"]) }
+    assert_equal before, headers["X-Last-Peak-At"], "should not change"
+
+    sleep 2
+    env = @req.class.env_for "/active/#@addr.txt"
+    status, headers, body = @app.call(env.dup)
+    assert_equal before, headers["X-Last-Peak-At"], headers.inspect
+
+    @ios << @srv.accept
+    assert_raises(Errno::EAGAIN) { @srv.accept_nonblock }
+    sleep 0.1
+    status, headers, body = @app.call(env.dup)
+    assert headers["X-Last-Peak-At"], headers.inspect
+    assert_nothing_raised { Time.parse(headers["X-Last-Peak-At"]) }
+    assert before != headers["X-Last-Peak-At"]
+
+    queued_before = headers["X-Last-Peak-At"]
+
+    sleep 2
+
+    env = @req.class.env_for "/queued/#@addr.txt"
+    status, headers, body = @app.call(env)
+    assert_equal "0", headers["X-Current"]
+    assert_nothing_raised { Time.parse(headers["X-Last-Peak-At"]) }
+    assert_equal queued_before, headers["X-Last-Peak-At"], "should not change"
+    assert_equal start, headers["X-First-Peak-At"]
+  end
 end if RUBY_PLATFORM =~ /linux/