lib/raindrops/watcher.rb

   1 # -*- encoding: binary -*-
   2 require "thread"
   3 require "time"
   4 require "socket"
   5 require "rack"
   6 require "aggregate"
   7
   8 # Raindrops::Watcher is a stand-alone Rack application for watching
   9 # any number of TCP and UNIX listeners (all of them by default).
  10 #
  11 # It depends on the {Aggregate RubyGem}[http://rubygems.org/gems/aggregate]
  12 #
  13 # In your Rack config.ru:
  14 #
  15 #    run Raindrops::Watcher(options = {})
  16 #
  17 # It takes the following options hash:
  18 #
  19 # - :listeners - an array of listener names, (e.g. %w(0.0.0.0:80 /tmp/sock))
  20 # - :delay - interval between stats updates in seconds (default: 1)
  21 #
  22 # Raindrops::Watcher is compatible any thread-safe/thread-aware Rack
  23 # middleware.  It does not work well with multi-process web servers
  24 # but can be used to monitor them.  It consumes minimal resources
  25 # with the default :delay.
  26 #
  27 # == HTTP endpoints
  28 #
  29 # === GET /
  30 #
  31 # Returns an HTML summary listing of all listen interfaces watched on
  32 #
  33 # === GET /active/$LISTENER.txt
  34 #
  35 # Returns a plain text summary + histogram with X-* HTTP headers for
  36 # active connections.
  37 #
  38 # e.g.: curl http://raindrops-demo.bogomips.org/active/0.0.0.0%3A80.txt
  39 #
  40 # === GET /active/$LISTENER.html
  41 #
  42 # Returns an HTML summary + histogram with X-* HTTP headers for
  43 # active connections.
  44 #
  45 # e.g.: curl http://raindrops-demo.bogomips.org/active/0.0.0.0%3A80.html
  46 #
  47 # === GET /queued/$LISTENER.txt
  48 #
  49 # Returns a plain text summary + histogram with X-* HTTP headers for
  50 # queued connections.
  51 #
  52 # e.g.: curl http://raindrops-demo.bogomips.org/queued/0.0.0.0%3A80.txt
  53 #
  54 # === GET /queued/$LISTENER.html
  55 #
  56 # Returns an HTML summary + histogram with X-* HTTP headers for
  57 # queued connections.
  58 #
  59 # e.g.: curl http://raindrops-demo.bogomips.org/queued/0.0.0.0%3A80.html
  60 #
  61 # === GET /tail/$LISTENER.txt?active_min=1&queued_min=1
  62 #
  63 # Streams chunked a response to the client.
  64 # Interval is the preconfigured +:delay+ of the application (default 1 second)
  65 #
  66 # The response is plain text in the following format:
  67 #
  68 #   ISO8601_TIMESTAMP LISTENER_NAME ACTIVE_COUNT QUEUED_COUNT LINEFEED
  69 #
  70 # Query parameters:
  71 #
  72 # - active_min - do not stream a line until this active count is reached
  73 # - queued_min - do not stream a line until this queued count is reached
  74 #
  75 # == Response headers (mostly the same names as Raindrops::LastDataRecv)
  76 #
  77 # - X-Count   - number of samples polled
  78 # - X-Last-Reset - date since the last reset
  79 #
  80 # The following headers are only present if X-Count is greater than one.
  81 #
  82 # - X-Min     - lowest number of connections recorded
  83 # - X-Max     - highest number of connections recorded
  84 # - X-Mean    - mean number of connections recorded
  85 # - X-Std-Dev - standard deviation of connection count
  86 # - X-Outliers-Low - number of low outliers (hopefully many for queued)
  87 # - X-Outliers-High - number of high outliers (hopefully zero for queued)
  88 #
  89 # = Demo Server
  90 #
  91 # There is a server running this app at http://raindrops-demo.bogomips.org/
  92 # The Raindrops::Middleware demo is also accessible at
  93 # http://raindrops-demo.bogomips.org/_raindrops
  94 #
  95 # The demo server is only limited to 30 users, so be sure not to abuse it
  96 # by using the /tail/ endpoint too much.
  97 class Raindrops::Watcher
  98   # :stopdoc:
  99   attr_reader :snapshot
 100   include Rack::Utils
 101   include Raindrops::Linux
 102   DOC_URL = "http://raindrops.bogomips.org/Raindrops/Watcher.html"
 103
 104   def initialize(opts = {})
 105     @tcp_listeners = @unix_listeners = nil
 106     if l = opts[:listeners]
 107       tcp, unix = [], []
 108       Array(l).each { |addr| (addr =~ %r{\A/} ? unix : tcp) << addr }
 109       unless tcp.empty? && unix.empty?
 110         @tcp_listeners = tcp
 111         @unix_listeners = unix
 112       end
 113     end
 114
 115     agg_class = opts[:agg_class] || Aggregate
 116     start = Time.now.utc
 117     @active = Hash.new { |h,k| h[k] = agg_class.new }
 118     @queued = Hash.new { |h,k| h[k] = agg_class.new }
 119     @resets = Hash.new { |h,k| h[k] = start }
 120     @snapshot = [ start, {} ]
 121     @delay = opts[:delay] || 1
 122     @lock = Mutex.new
 123     @start = Mutex.new
 124     @cond = ConditionVariable.new
 125     @thr = nil
 126   end
 127
 128   def hostname
 129     Socket.gethostname
 130   end
 131
 132   # rack endpoint
 133   def call(env)
 134     @start.synchronize { @thr ||= aggregator_thread(env["rack.logger"]) }
 135     case env["REQUEST_METHOD"]
 136     when "HEAD", "GET"
 137       get env
 138     when "POST"
 139       post env
 140     else
 141       Rack::Response.new(["Method Not Allowed"], 405).finish
 142     end
 143   end
 144
 145   def aggregator_thread(logger) # :nodoc:
 146     @socket = sock = Raindrops::InetDiagSocket.new
 147     thr = Thread.new do
 148       begin
 149         combined = tcp_listener_stats(@tcp_listeners, sock)
 150         combined.merge!(unix_listener_stats(@unix_listeners))
 151         @lock.synchronize do
 152           combined.each do |addr,stats|
 153             @active[addr] << stats.active
 154             @queued[addr] << stats.queued
 155           end
 156           @snapshot = [ Time.now.utc, combined ]
 157           @cond.broadcast
 158         end
 159       rescue => e
 160         logger.error "#{e.class} #{e.inspect}"
 161       end while sleep(@delay) && @socket
 162       sock.close
 163     end
 164     wait_snapshot
 165     thr
 166   end
 167
 168   def active_stats(addr) # :nodoc:
 169     @lock.synchronize do
 170       tmp = @active[addr] or return
 171       [ @snapshot[0], @resets[addr], tmp.dup ]
 172     end
 173   end
 174
 175   def queued_stats(addr) # :nodoc:
 176     @lock.synchronize do
 177       tmp = @queued[addr] or return
 178       [ @snapshot[0], @resets[addr], tmp.dup ]
 179     end
 180   end
 181
 182   def wait_snapshot
 183     @lock.synchronize do
 184       @cond.wait @lock
 185       @snapshot
 186     end
 187   end
 188
 189   def agg_to_hash(reset_at, agg)
 190     {
 191       "X-Count" => agg.count.to_s,
 192       "X-Min" => agg.min.to_s,
 193       "X-Max" => agg.max.to_s,
 194       "X-Mean" => agg.mean.to_s,
 195       "X-Std-Dev" => agg.std_dev.to_s,
 196       "X-Outliers-Low" => agg.outliers_low.to_s,
 197       "X-Outliers-High" => agg.outliers_high.to_s,
 198       "X-Last-Reset" => reset_at.httpdate,
 199     }
 200   end
 201
 202   def histogram_txt(agg)
 203     updated_at, reset_at, agg = *agg
 204     headers = agg_to_hash(reset_at, agg)
 205     body = agg.to_s
 206     headers["Content-Type"] = "text/plain"
 207     headers["Expires"] = (updated_at + @delay).httpdate
 208     headers["Content-Length"] = bytesize(body).to_s
 209     [ 200, headers, [ body ] ]
 210   end
 211
 212   def histogram_html(agg, addr)
 213     updated_at, reset_at, agg = *agg
 214     headers = agg_to_hash(reset_at, agg)
 215     body = "<html>" \
 216       "<head><title>#{hostname} - #{escape_html addr}</title></head>" \
 217       "<body><table>" <<
 218       headers.map { |k,v|
 219         "<tr><td>#{k.gsub(/^X-/, '')}</td><td>#{v}</td></tr>"
 220       }.join << "</table><pre>#{escape_html agg}</pre>" \
 221       "<form action='/reset/#{escape addr}' method='post'>" \
 222       "<input type='submit' name='x' value='reset' /></form>" \
 223       "</body>"
 224     headers["Content-Type"] = "text/html"
 225     headers["Expires"] = (updated_at + @delay).httpdate
 226     headers["Content-Length"] = bytesize(body).to_s
 227     [ 200, headers, [ body ] ]
 228   end
 229
 230   def get(env)
 231     retried = false
 232     case env["PATH_INFO"]
 233     when "/"
 234       index
 235     when %r{\A/active/(.+)\.txt\z}
 236       histogram_txt(active_stats(unescape($1)))
 237     when %r{\A/active/(.+)\.html\z}
 238       addr = unescape $1
 239       histogram_html(active_stats(addr), addr)
 240     when %r{\A/queued/(.+)\.txt\z}
 241       histogram_txt(queued_stats(unescape($1)))
 242     when %r{\A/queued/(.+)\.html\z}
 243       addr = unescape $1
 244       histogram_html(queued_stats(addr), addr)
 245     when %r{\A/tail/(.+)\.txt\z}
 246       tail(unescape($1), env)
 247     else
 248       not_found
 249     end
 250     rescue Errno::EDOM
 251       raise if retried
 252       retried = true
 253       wait_snapshot
 254       retry
 255   end
 256
 257   def not_found
 258     Rack::Response.new(["Not Found"], 404).finish
 259   end
 260
 261   def post(env)
 262     case env["PATH_INFO"]
 263     when %r{\A/reset/(.+)\z}
 264       reset!(env, unescape($1))
 265     else
 266       not_found
 267     end
 268   end
 269
 270   def reset!(env, addr)
 271     @lock.synchronize do
 272       @active.include?(addr) or return not_found
 273       @active.delete addr
 274       @queued.delete addr
 275       @resets[addr] = Time.now.utc
 276       @cond.wait @lock
 277     end
 278     req = Rack::Request.new(env)
 279     res = Rack::Response.new
 280     url = req.referer || "#{req.host_with_port}/"
 281     res.redirect(url)
 282     res.content_type.replace "text/plain"
 283     res.write "Redirecting to #{url}"
 284     res.finish
 285   end
 286
 287   def index
 288     updated_at, all = snapshot
 289     headers = {
 290       "Content-Type" => "text/html",
 291       "Last-Modified" => updated_at.httpdate,
 292       "Expires" => (updated_at + @delay).httpdate,
 293     }
 294     body = "<html><head>" \
 295       "<title>#{hostname} - all interfaces</title>" \
 296       "</head><body><h3>Updated at #{updated_at.iso8601}</h3>" \
 297       "<table><tr>" \
 298         "<th>address</th><th>active</th><th>queued</th><th>reset</th>" \
 299       "</tr>" <<
 300       all.map do |addr,stats|
 301         e_addr = escape addr
 302         "<tr>" \
 303           "<td><a href='/tail/#{e_addr}.txt' " \
 304             "title='&quot;tail&quot; output in real time'" \
 305             ">#{escape_html addr}</a></td>" \
 306           "<td><a href='/active/#{e_addr}.html' " \
 307             "title='show active connection stats'>#{stats.active}</a></td>" \
 308           "<td><a href='/queued/#{e_addr}.html' " \
 309             "title='show queued connection stats'>#{stats.queued}</a></td>" \
 310           "<td><form action='/reset/#{e_addr}' method='post'>" \
 311             "<input title='reset statistics' " \
 312               "type='submit' name='x' value='x' /></form></td>" \
 313         "</tr>" \
 314       end.join << "</table>" \
 315       "<p>" \
 316         "This is running the #{self.class}</a> service, see " \
 317         "<a href='#{DOC_URL}'>#{DOC_URL}</a> " \
 318         "for more information and options." \
 319       "</p>" \
 320       "</body></html>"
 321     headers["Content-Length"] = bytesize(body).to_s
 322     [ 200, headers, [ body ] ]
 323   end
 324
 325   def tail(addr, env)
 326     Tailer.new(self, addr, env).finish
 327   end
 328
 329   # This is the response body returned for "/tail/$ADDRESS.txt".  This
 330   # must use a multi-threaded Rack server with streaming response support.
 331   # It is an internal class and not expected to be used directly
 332   class Tailer
 333     def initialize(rdmon, addr, env) # :nodoc:
 334       @rdmon = rdmon
 335       @addr = addr
 336       q = Rack::Utils.parse_query env["QUERY_STRING"]
 337       @active_min = q["active_min"].to_i
 338       @queued_min = q["queued_min"].to_i
 339       len = Rack::Utils.bytesize(addr)
 340       len = 35 if len > 35
 341       @fmt = "%20s % #{len}s % 10u % 10u\n"
 342       case env["HTTP_VERSION"]
 343       when "HTTP/1.0", nil
 344         @chunk = false
 345       else
 346         @chunk = true
 347       end
 348     end
 349
 350     def finish
 351       headers = {
 352         "Content-Type" => "text/plain",
 353         "Cache-Control" => "no-transform",
 354         "Expires" => Time.at(0).httpdate,
 355       }
 356       headers["Transfer-Encoding"] = "chunked" if @chunk
 357       [ 200, headers, self ]
 358     end
 359
 360     # called by the Rack server
 361     def each # :nodoc:
 362       begin
 363         time, all = @rdmon.wait_snapshot
 364         stats = all[@addr] or next
 365         stats.queued >= @queued_min or next
 366         stats.active >= @active_min or next
 367         body = sprintf(@fmt, time.iso8601, @addr, stats.active, stats.queued)
 368         body = "#{body.size.to_s(16)}\r\n#{body}\r\n" if @chunk
 369         yield body
 370       end while true
 371       yield "0\r\n\r\n" if @chunk
 372     end
 373   end
 374
 375   # shuts down the background thread, only for tests
 376   def shutdown
 377     @socket = nil
 378     @thr.join if @thr
 379     @thr = nil
 380   end
 381   # :startdoc:
 382 end