Gauge stats

.gitignore

@@ -1,2 +1,11 @@
 rstatsd.yaml
 rstatsd-cfg.rb
+*.gem
+rstatsd-1.0.0.tar.gz
+rstatsd-1.0.0/
+ruby-rstatsd-1.0.0/
+ruby-rstatsd_1.0.0-1.debian.tar.gz
+ruby-rstatsd_1.0.0-1.dsc
+ruby-rstatsd_1.0.0-1_all.deb
+ruby-rstatsd_1.0.0-1_amd64.changes
+ruby-rstatsd_1.0.0.orig.tar.gz

rstatsd.rb → bin/rstatsd

@@ -10,7 +10,7 @@
 require 'optparse'
 require 'rubygems'
 require 'statsd'
-require File.dirname(__FILE__)+'/lib/rstatsd'
+require 'rstatsd'
 
 
 def setup_logger ( logfile, loglevel = "INFO"  )
@@ -27,6 +27,7 @@ def setup_logger ( logfile, loglevel = "INFO"  )
 options = { :cfg_file  => File.dirname(__FILE__)+'/rstatsd.yaml',
             :pidfile   => '/tmp/rstatsd.pid',
             :ctl_cmd   => "start",
+            :cfg_dir   => File.dirname(__FILE__)+'/conf.d',
             :daemonize => false }
 
 
@@ -38,6 +39,9 @@ def setup_logger ( logfile, loglevel = "INFO"  )
   opts.on("-c", "--config FILE", String, "Configuration file (default #{options[:cfg_file]})") do |v|
     options[:cfg_file] = v
   end
+  opts.on("-g", "--glob-dir FILE", String, "Configuration include directory (default #{options[:cfg_dir]})") do |v|
+    options[:cfg_dir] = v
+  end
   opts.on("-p", "--pidfile FILE", String, "PID file (default #{options[:pidfile]})") do |v|
     options[:pidfile] = v
   end
@@ -50,7 +54,15 @@ def setup_logger ( logfile, loglevel = "INFO"  )
   end
 end.parse!
 
+puts "Loading config from " + options[:cfg_file]
 cfg = YAML.load_file(options[:cfg_file])
+cfg[:cmds] ||= []
+
+Dir[options[:cfg_dir]+'/*.yaml'].each do |file|
+  include_cfg = YAML.load_file(file)
+  puts "Loading " + file
+  cfg[:cmds] = cfg[:cmds].push(include_cfg[0])
+end
 
 Statsd.host = cfg[:statsd][:host]
 Statsd.port = cfg[:statsd][:port]
@@ -61,6 +73,7 @@ def setup_logger ( logfile, loglevel = "INFO"  )
 
 cfg[:cmds].each do |cmd|
   params = { :every  => cmd[:every], 
+             :interval => cmd[:interval],
              :logger => logger, 
              :prefix => cfg[:metric_prefix] }
 

config/rstatsd.upstart.conf

@@ -0,0 +1,13 @@
+description "Capture metrics from logs and send to statsd server"
+start on runlevel [2345]
+stop on runlevel [!2345]
+
+respawn
+expect fork
+kill timeout 2
+
+pre-start exec /usr/bin/test -e /etc/rstatsd.yaml
+
+exec /usr/bin/rstatsd -c /etc/rstatsd.yaml -d -k start
+
+pre-stop exec /usr/bin/rstatsd -c /etc/rstatsd.yaml -d -k stop

config/test_regex.pl

@@ -0,0 +1,21 @@
+#!/usr/bin/perl
+
+# example:
+# perl test_regex.pl "Charlie has 9 girls in 12 cities" "Charlie\s+has\s+(?<girls>\d+)\s+girls\s+in\s+(?<cities>\d+)\s+cities"
+#
+# Should return:
+# girls: 9
+# cities: 12
+
+# Parse command line arguments
+$input_string = $ARGV[0];
+$input_regexp = $ARGV[1];
+
+# Perform regexp matching
+$input_string =~ m/($input_regexp)/;
+
+# Print the resulting groups and values
+while (($key, $value) = each(%+)){
+     print $key.": ".$value."\n";
+}
+

lib/rstatsd/processctl.rb

@@ -100,7 +100,7 @@ def get_running_pids
   end
 
   def get_allpids
-    @allpids = `ps -ef |sed 1d`.to_a.map { |x| a = x.strip.split(/\s+/); [a[1].to_i,a[2].to_i] }
+    @allpids = `ps -ef |sed 1d`.lines.to_a.map { |x| a = x.strip.split(/\s+/); [a[1].to_i,a[2].to_i] }
   end
 
   # thar be recursion ahead, matey

lib/rstatsd/rstatsd.rb

@@ -1,4 +1,6 @@
 module RStatsd
+  ENV['TZ'] = ":/etc/localtime"
+
   module Helpers
     def prefix_metric_name ( pieces )
       pieces.join(".")
@@ -11,13 +13,15 @@ def logit ( msg, level = Logger::INFO )
 
   class Command
     include Helpers
-    attr_accessor :command, :every
+    attr_accessor :command
     def initialize ( h = {}, &block )
-      @command = nil
-      @logger  = h[:logger]
-      @every   = h[:every] || 1
-      @prefix  = h[:prefix]
-      @regexes = []   
+      @command  = nil
+      @logger   = h[:logger]
+      @every    = h[:every] || 1
+      @interval = h[:interval] || nil
+      @prefix   = h[:prefix]
+      @regexes  = []
+
       yield self
     end
 
@@ -30,26 +34,41 @@ def execute!
       trap(:INT)  { logit("Caught SIGINT.  Exiting");  Process.kill(:INT,  pipe.pid) }
       trap(:TERM) { logit("Caught SIGTERM.  Exiting"); Process.kill(:TERM, pipe.pid) }
 
-      timers, counters = {}, {}
+      timers, counters, gauges = {}, {}, {}
 
-      logit("Starting thread for command #{@command}")
+      logit("Starting thread for command #{@command}; interval #{@interval}, every #{@every}")
       pipe = IO.popen(@command)
       begin
+        if @interval
+          next_update = Time.now.to_f + @interval
+        end
+
         pipe.each_with_index do |l,i|
           @regexes.each do |r| 
             if r.statsd_type == RegExData::Timer
               timers = r.get_increments(l, timers)
+            elsif r.statsd_type == RegExData::Gauge
+              gauges = r.get_increments(l, gauges)
             else
               counters = r.get_increments(l, counters)
             end
           end
 
-          # only send to statsd every x lines - this is to avoid UDP floods
-          if (i % @every) == 0
-            logit("#{i} lines, sending to statsd", Logger::DEBUG) 
+          if @interval
+              do_update = Time.now.to_f >= next_update
+          else
+              do_update = (i % @every) == 0
+          end
+
+          if do_update
+            logit("#{@command}: #{i} lines, sending to statsd", Logger::DEBUG)
             statsd_send(counters) 
+            statsd_send(gauges, 'gauge') 
             statsd_send(divide_hash(timers, @every), 'timer')
-            timers, counters = {}, {}
+            timers, counters, gauges = {}, {}, {}
+            if @interval
+              next_update = Time.now.to_f + @interval
+            end
           end
           # this is for debugging
 #sleep(rand/100)
@@ -73,6 +92,9 @@ def statsd_send ( h, statsd_type = nil )
         if statsd_type == 'timer'
           Statsd.timing(k,v)
           logit("Set timer value #{k} to #{v}",Logger::DEBUG) 
+        elsif statsd_type == 'gauge'
+          Statsd.gauge(k,v)
+          logit("Set gauge value #{k} to #{v}",Logger::DEBUG) 
         else
           Statsd.update_counter(k,v)
           logit("Incremented #{k} by #{v}",Logger::DEBUG) 
@@ -87,13 +109,15 @@ class RegExData
     attr_writer :logger
     attr_reader :regex, :statsd_type
 
-    Counter, Timer = 1, 2
+    Counter, Timer, Gauge = 1, 2, 3
 
     def initialize ( h )
       @regex       = which_regex h[:regex]
       @metrics     = h[:metrics] || []
+      @unmetrics   = h[:unmetrics] || []
       @statsd_type = case h[:statsd_type]
                      when 'timer' then Timer 
+                     when 'gauge' then Gauge
                      else Counter 
                      end
       @statsd      = h[:statsd] || true
@@ -110,13 +134,23 @@ def initialize ( h )
     def get_increments ( line, h )
       h ||= {}
       @matches = @regex.match(line)
-      return h unless @matches
-      if has_captures?
-        @matches.names.each do |name|
-          h = build_and_increment(h, name )
+      if @matches
+        if has_captures?
+          @matches.names.each do |name|
+            # If we use nested captures, we can get empty MatchData members,
+            # we should skip those.
+            if !@matches[name.to_sym].nil?
+              h = build_and_increment(h, name )
+            end
+          end
+        else
+          h = build_and_increment(h)
+        end
+      elsif @unmetrics
+        @unmetrics.each do |unmetric|
+          h[unmetric] ||= 0
+          h[unmetric] += 1
         end
-      else
-        h = build_and_increment(h)
       end
       h
     end
@@ -132,11 +166,13 @@ def build_and_increment ( h, name = nil )
           h[metric_name] ||= 0
           if @statsd_type == Timer
             h[metric_name] += @matches[name.to_sym].to_f
+          elsif @statsd_type == Gauge
+            h[metric_name] = @matches[name.to_sym].to_f
           else
             h[metric_name] += @matches[name.to_sym].to_i
           end
         else
-          # the value of the named capture will be used as a leaf node in the mtric name
+          # the value of the named capture will be used as a leaf node in the metric name
           metric_name = prefix_metric_name( [ metric_name, name, @matches[name.to_sym] ] ) if name
           h[metric_name] ||= 0
           h[metric_name] += 1

rstatsd.gemspec

@@ -0,0 +1,19 @@
+$:.push File.expand_path("../lib", __FILE__)
+
+Gem::Specification.new do |s|
+  s.name        = 'rstatsd'
+  s.version     = '1.0.3'
+  s.date        = '2014-11-09'
+  s.summary     = "Tool to turn logs into statsd metrics."
+  s.description = "rstatsd is a daemon that takes the output of multiple commands and sends
+  the output to statsd based on a regular expressions with optional named capture
+  groups."
+  s.authors     = ["Aaron Brown", "Ulrik Holmen", "Stefan Bergstrom"]
+  s.email       = '9minutesnooze@github.com'
+  s.files       = `git ls-files -- bin lib config`.split("\n")
+  s.homepage    = 'https://github.com/ideeli/rstatsd'
+  s.executables = `git ls-files -- bin/*`.split("\n").map{ |f| File.basename(f) }
+  s.require_paths = ["lib"]
+  s.default_executable = 'rstatsd'
+  s.add_dependency 'statsd-client'
+end