- first hack at metrics. TCP input now counts how many events it reads.

- agent now every few seconds logs all current metrics.

This will get cooler once jls-grok and other libraries (which use cabin)
start using metrics.

Gemfile

@@ -9,7 +9,7 @@ gem "bunny" # for amqp support, MIT-style license
 gem "uuidtools" # for naming amqp queues, License ???
 
 gem "filewatch", "0.3.3"  # for file tailing, BSD License
-gem "jls-grok", "0.9.6" # for grok filter, BSD License
+gem "jls-grok", "0.10.2" # for grok filter, BSD License
 jruby? and gem "jruby-elasticsearch", "0.0.11" # BSD License
 gem "stomp" # for stomp protocol, Apache 2.0 License
 gem "json" # Ruby license

Gemfile.lock

@@ -1,71 +1,67 @@
 GEM
   remote: http://rubygems.org/
   specs:
-    awesome_print (0.4.0)
+    awesome_print (1.0.2)
     bouncy-castle-java (1.5.0146.1)
-    bson (1.4.0)
-    bson (1.4.0-java)
+    bson (1.5.2-java)
     bunny (0.7.8)
-    cabin (0.1.7)
+    cabin (0.2.2)
       json
     childprocess (0.3.0)
       ffi (~> 1.0.6)
     choice (0.1.4)
-    ffi (1.0.11)
     ffi (1.0.11-java)
     ffi-rzmq (0.9.0)
-    filewatch (0.3.2)
-    gelf (1.1.3)
+    filewatch (0.3.3)
+    gelf (1.3.2)
       json
     gelfd (0.2.0)
     gmetric (0.1.3)
-    haml (3.1.3)
-    jls-grok (0.9.6)
-      cabin (= 0.1.7)
+    haml (3.1.4)
+    jls-grok (0.10.2)
+      cabin (~> 0.2.2)
     jruby-elasticsearch (0.0.11)
     jruby-openssl (0.7.5)
       bouncy-castle-java (>= 1.5.0146.1)
-    json (1.6.3)
-    json (1.6.3-java)
-    minitest (2.6.1)
+    json (1.6.5-java)
+    minitest (2.11.1)
     mizuno (0.5.0)
       childprocess (>= 0.2.6)
       choice (>= 0.1.0)
       ffi (>= 1.0.0)
       rack (>= 1.0.0)
-    mongo (1.4.0)
-      bson (= 1.4.0)
-    rack (1.3.4)
-    rack-protection (1.1.4)
+    mongo (1.5.2)
+      bson (= 1.5.2)
+    rack (1.4.1)
+    rack-protection (1.2.0)
       rack
     redis (2.2.2)
-    sass (3.1.10)
-    sinatra (1.3.1)
-      rack (~> 1.3, >= 1.3.4)
-      rack-protection (~> 1.1, >= 1.1.2)
+    sass (3.1.13)
+    sinatra (1.3.2)
+      rack (~> 1.3, >= 1.3.6)
+      rack-protection (~> 1.2)
       tilt (~> 1.3, >= 1.3.3)
     statsd-ruby (0.3.0)
-    stomp (1.1.9)
+    stomp (1.2.0)
     tilt (1.3.3)
     uuidtools (2.1.2)
     xmpp4r (0.5)
 
 PLATFORMS
   java
-  ruby
 
 DEPENDENCIES
   awesome_print
   bunny
-  cabin (= 0.1.7)
+  cabin (= 0.2.2)
   ffi
   ffi-rzmq (= 0.9.0)
-  filewatch (= 0.3.2)
+  filewatch (= 0.3.3)
   gelf
   gelfd (= 0.2.0)
   gmetric (= 0.1.3)
   haml
-  jls-grok (= 0.9.6)
+  jls-grok (= 0.10.2)
   jruby-elasticsearch (= 0.0.11)
   jruby-openssl
   json

lib/logstash/agent.rb

@@ -415,6 +415,16 @@ class LogStash::Agent
     # like tests, etc.
     yield if block_given?
 
+    Thread.new do
+      while true
+        @logger.metrics.each do |identifier, metric|
+          instance, name = identifier
+          @logger.info("metric #{instance.class.name}.#{name}", :value => metric.value)
+        end
+        sleep 3
+      end
+    end
+
     # TODO(sissel): Monitor what's going on? Sleep forever? what?
     while sleep 5
     end

lib/logstash/inputs/tcp.rb

@@ -39,12 +39,18 @@ class LogStash::Inputs::Tcp < LogStash::Inputs::Base
     end # def peer
   end # module SocketPeer
 
+  def initialize(*args)
+    super(*args)
+  end # def initialize
+
   public
   def register
     if server?
       @logger.info("Starting tcp input listener", :address => "#{@host}:#{@port}")
       @server_socket = TCPServer.new(@host, @port)
     end
+    @event_meter = @logger.metrics.meter(self, "events")
+    @logger.info("tcp input", :meter => @event_meter)
   end # def register
 
   private
@@ -56,10 +62,10 @@ class LogStash::Inputs::Tcp < LogStash::Inputs::Base
         # or socket dies
         # TODO(sissel): Why do we have a timeout here? What's the point?
         if @data_timeout == -1
-          buf = socket.readline
+          buf = readline(socket)
         else
           Timeout::timeout(@data_timeout) do
-            buf = socket.readline
+            buf = readline(socket)
           end
         end
         e = self.to_event(buf, event_source)
@@ -87,6 +93,12 @@ class LogStash::Inputs::Tcp < LogStash::Inputs::Base
     @mode == "server"
   end # def server?
 
+  private
+  def readline(socket)
+    @event_meter.mark
+    line = socket.readline
+  end # def readline
+
   public
   def run(output_queue)
     if server?