examples/linux-listener-stats.rb

   1 #!/usr/bin/ruby
   2 # -*- encoding: binary -*-
   3 # frozen_string_literal: false
   4 $stdout.sync = $stderr.sync = true
   5 # this is used to show or watch the number of active and queued
   6 # connections on any listener socket from the command line
   7
   8 require 'raindrops'
   9 require 'optparse'
  10 require 'ipaddr'
  11 require 'time'
  12 begin
  13   require 'sleepy_penguin'
  14 rescue LoadError
  15 end
  16 usage = "Usage: #$0 [-d DELAY] [-t QUEUED_THRESHOLD] ADDR..."
  17 ARGV.size > 0 or abort usage
  18 delay = false
  19 queued_thresh = -1
  20 # "normal" exits when driven on the command-line
  21 trap(:INT) { exit 130 }
  22 trap(:PIPE) { exit 0 }
  23
  24 OptionParser.new('', 24, '  ') do |opts|
  25   opts.banner = usage
  26   opts.on('-d', '--delay=DELAY', Float) { |n| delay = n }
  27   opts.on('-t', '--queued-threshold=INT', Integer) { |n| queued_thresh = n }
  28   opts.on('-a', '--all') { } # noop
  29   opts.parse! ARGV
  30 end
  31
  32 begin
  33   require 'aggregate'
  34 rescue LoadError
  35   $stderr.puts "Aggregate missing, USR1 and USR2 handlers unavailable"
  36 end if delay
  37
  38 if delay && defined?(SleepyPenguin::TimerFD)
  39   @tfd = SleepyPenguin::TimerFD.new
  40   @tfd.settime nil, delay, delay
  41   def delay_for(seconds)
  42     @tfd.expirations
  43   end
  44 else
  45   alias delay_for sleep
  46 end
  47
  48 agg_active = agg_queued = nil
  49 if delay && defined?(Aggregate)
  50   agg_active = Aggregate.new
  51   agg_queued = Aggregate.new
  52
  53   def dump_aggregate(label, agg)
  54     $stderr.write "--- #{label} ---\n"
  55     %w(count min max outliers_low outliers_high mean std_dev).each do |f|
  56       $stderr.write "#{f}=#{agg.__send__ f}\n"
  57     end
  58     $stderr.write "#{agg}\n\n"
  59   end
  60
  61   trap(:USR1) do
  62     dump_aggregate "active", agg_active
  63     dump_aggregate "queued", agg_queued
  64   end
  65   trap(:USR2) do
  66     agg_active = Aggregate.new
  67     agg_queued = Aggregate.new
  68   end
  69   $stderr.puts "USR1(dump_aggregate) and USR2(reset) handlers ready for PID=#$$"
  70 end
  71
  72 ARGV.each do |addr|
  73   addr =~ %r{\A(127\..+):(\d+)\z} or next
  74   host, port = $1, $2
  75   hex_port = '%X' % port.to_i
  76   ip_addr = IPAddr.new(host)
  77   hex_host = ip_addr.hton.each_byte.inject('') { |s,o| s << '%02X' % o }
  78   socks = File.readlines('/proc/net/tcp')
  79   hex_addr = "#{hex_host}:#{hex_port}"
  80   if socks.grep(/^\s+\d+:\s+#{hex_addr}\s+/).empty? &&
  81      ! socks.grep(/^\s+\d+:\s+00000000:#{hex_port}\s+/).empty?
  82     warn "W: #{host}:#{port} (#{hex_addr}) not found in /proc/net/tcp"
  83     warn "W: Did you mean 0.0.0.0:#{port}?"
  84   end
  85 end
  86
  87 len = "address".size
  88 now = nil
  89 tcp, unix = [], []
  90 ARGV.each do |addr|
  91   bs = addr.respond_to?(:bytesize) ? addr.bytesize : addr.size
  92   len = bs if bs > len
  93   (addr =~ %r{\A/} ? unix : tcp) << addr
  94 end
  95 combined = {}
  96 tcp_args = unix_args = nil
  97 unless tcp.empty? && unix.empty?
  98   tcp_args = tcp
  99   unix_args = unix
 100 end
 101 sock = Raindrops::InetDiagSocket.new if tcp
 102
 103 len = 35 if len > 35
 104 fmt = "%20s % #{len}s % 10u % 10u\n"
 105 $stderr.printf fmt.tr('u','s'), *%w(timestamp address active queued)
 106
 107 begin
 108   if now
 109     combined.clear
 110     now = nil
 111   end
 112   combined.merge! Raindrops::Linux.tcp_listener_stats(tcp_args, sock)
 113   combined.merge! Raindrops::Linux.unix_listener_stats(unix_args)
 114   combined.each do |addr,stats|
 115     active, queued = stats.active, stats.queued
 116     if agg_active
 117       agg_active << active
 118       agg_queued << queued
 119     end
 120     next if queued < queued_thresh
 121     printf fmt, now ||= Time.now.utc.iso8601, addr, active, queued
 122   end
 123 end while delay && delay_for(delay)