Input format option (plain, json_event)

README.md

@@ -25,6 +25,7 @@ woodchuck --files /var/log/syslog,/var/log/apache/**/*,/var/log/nginx/*.log --ou
 * -l, --log-level - Log verbosity. [ debug, warn, info, error, fatal ]
 * -p, --paths     - A comma-separated list of files to watch for changes (file globbing is accepted).
 * -o, --output    - The output to send to [ stdout, redis ]
+* -f, --format    - Input line format
 ```
 ## Coming soon
 

lib/woodchuck/agent.rb

@@ -3,26 +3,36 @@
 require 'thread'
 
 require 'woodchuck/output'
+require 'woodchuck/input'
+require 'woodchuck/input/plain'
+require 'woodchuck/input/json_event'
 
 class Woodchuck::Agent
 
-  attr_accessor :logger, :watcher, :watcher_thread, :paths, :output
+  attr_accessor :logger, :watcher, :watcher_thread, :paths, :output, :input_format
 
   def initialize(options={})
     @paths = options[:paths]
+		options[:log_level] ||= :info
     @logger = Woodchuck::Logger.new(::STDOUT)
+		@logger.level = options[:log_level]
     @mutex = Mutex.new
     @output = case options[:output]
-                 when :stdout 
-                   Woodchuck::Output::STDOUT.new
                  when :zeromq
-                   Woodchuck::Output::ZeroMQ.new
+                   Woodchuck::Output::ZeroMQ.new(options[:log_level])
                  when :redis
-                   Woodchuck::Output::Redis.new
+                   Woodchuck::Output::Redis.new(options[:log_level])
                  else
-                   Woodchuck::Output::STDOUT.new
+                   Woodchuck::Output::STDOUT.new(options[:log_level])
                  end
-    @watcher = Woodchuck::Watcher.new(self, @paths)
+		@input_format = case options[:input_format]
+			when :json_event
+				Woodchuck::Input::JsonEvent.new
+			else
+				Woodchuck::Input::Plain.new
+			end
+
+    @watcher = Woodchuck::Watcher.new(self, options[:log_level], @input_format, @paths)
   end
 
   def start(blocking=false)

lib/woodchuck/event.rb

@@ -2,17 +2,17 @@
 
 class Woodchuck::Event
 
-  attr_accessor :path, :line, :host, :timestamp, :source, :message, :fields, :tags
+  attr_accessor :source_path, :source_host, :timestamp, :source, :message, :fields, :tags, :type
 
-  def initialize(path, line)
-    @path = path
-    @line = line
-    @host = Socket.gethostname
-    @timestamp = Time.now.utc.iso8601(6)
-    @source = Addressable::URI.new(:scheme => 'file', :host => host, :path => path)
-    @message = line.strip
-    @fields = {}
-    @tags = []
+  def initialize(init_hsh)
+    @source_path = init_hsh["@source_path"] || init_hsh[:source_path]
+    @source_host = init_hsh["@source_host"] || init_hsh[:source_host]
+    @timestamp = init_hsh["@timestamp"] || init_hsh[:timestamp]
+    @source = init_hsh["@source"] || init_hsh[:source]
+    @message = init_hsh["@message"] || init_hsh[:message]
+    @fields = init_hsh["@fields"] || init_hsh[:fields]
+    @tags = init_hsh["@tags"] || init_hsh[:tags]
+		@type = init_hsh["@type"] || init_hsh[:type]
   end
 
   def method_missing(symbol, *args, &block)
@@ -26,12 +26,12 @@ def method_missing(symbol, *args, &block)
   def to_hash
     {
       '@source' => source.to_s,
-      '@type' => source.scheme,
+      '@type' => type,
       '@tags' => tags,
       '@fields' => fields,
       '@timestamp' => timestamp,
-      '@source_host' => host,
-      '@source_path' => path,
+      '@source_host' => source_host,
+      '@source_path' => source_path,
       '@message' => message
     }
   end

lib/woodchuck/input.rb

@@ -0,0 +1,25 @@
+require 'socket'
+
+class Woodchuck::Input
+
+  def create(path, line)
+		Woodchuck::Event.new(prepare_hash(path, line))
+	end
+
+	protected
+
+	def prepare_hash(path, line)
+		host = Socket.gethostname
+
+		{
+			:source_path => path,
+			:line => line,
+			:source_host => host,
+			:timestamp => Time.now.utc.iso8601(6),
+			:source => Addressable::URI.new(:scheme => 'file', :host => host, :path => path),
+			:type => 'file',
+			:fields => {},
+			:tags => []
+		}
+	end
+end

lib/woodchuck/input/json.rb

@@ -0,0 +1,19 @@
+require 'json'
+
+class Woodchuck::Input::Plain < Woodchuck::Input
+	protected
+
+	def prepare_hash(path, line)
+		super(path, line).merge(
+			parse_json(line)
+		)
+	end
+
+	def parse_json(line)
+		begin
+			JSON.parse(line)
+		rescue
+			{}
+		end
+	end
+end

lib/woodchuck/input/json_event.rb

@@ -0,0 +1,19 @@
+require 'json'
+
+class Woodchuck::Input::JsonEvent < Woodchuck::Input
+	protected
+
+	def prepare_hash(path, line)
+		super(path, line).merge(
+			parse_json(line)
+		)
+	end
+
+	def parse_json(line)
+		begin
+			JSON.parse(line)
+		rescue
+			{}
+		end
+	end
+end

lib/woodchuck/input/plain.rb

@@ -0,0 +1,11 @@
+class Woodchuck::Input::Plain < Woodchuck::Input
+	protected
+
+	def prepare_hash(path, line)
+		super(path, line).merge(
+			{
+				:message => line.strip
+			}
+		)
+	end
+end

lib/woodchuck/output.rb

@@ -3,12 +3,13 @@
 class Woodchuck::Output
   attr_accessor :type, :logger, :host
 
-  def initialize
+  def initialize(log_level)
     @host = Socket.gethostname
     @logger = Woodchuck::Logger.new(::STDOUT)
+		@logger.level = log_level
   end
 
   def handle(event)
     true
   end
-end
+end

lib/woodchuck/output/redis.rb

@@ -4,8 +4,8 @@
 class Woodchuck::Output::Redis < Woodchuck::Output
   attr_accessor :url, :host, :port, :db, :namespace
 
-  def initialize
-    super
+  def initialize(log_level)
+    super(log_level)
     @type = :redis
     @url = Addressable::URI.parse(ENV['REDIS_URL'] || 'redis://localhost:6379/9')
     @namespace = ENV['REDIS_NAMESPACE'] || 'logstash:woodchuck'
@@ -20,4 +20,4 @@ def handle(event)
     redis.lpush("events", event.to_json)
     @logger.info "Logging event to Redis", event.to_hash
   end
-end
+end

lib/woodchuck/output/stdout.rb

@@ -1,12 +1,12 @@
 require 'woodchuck/output'
 
 class Woodchuck::Output::STDOUT < Woodchuck::Output
-  def initialize
-    super
+  def initialize(log_level)
+    super(log_level)
     @type = :stdout
   end
 
   def handle(event)
     @logger.info "Logging event to STDOUT", event.to_hash
   end
-end
+end

lib/woodchuck/output/zeromq.rb

@@ -1,12 +1,12 @@
 require 'woodchuck/output'
 
 class Woodchuck::Output::ZeroMQ < Woodchuck::Output
-  def initialize
-    super
+  def initialize(log_level)
+    super(log_level)
     @type = :zeromq
   end
 
   def handle(event)
     @logger.info event.message, event.to_hash
   end
-end
+end

lib/woodchuck/runner.rb

@@ -26,13 +26,16 @@ def read(args=ARGV)
       opts.on('-p', '--paths [PATHS]', Array, 'A list of file paths to watch') do |paths|
         options[:paths] = paths
       end
+      opts.on('-f', '--format [FORMAT]', [:plain, :json_event], 'Input line format') do |input_format|
+        options[:input_format] = input_format
+      end
     end
     optparse.parse!(args)
     options
   end
 
   def run(options={})
-    puts options.inspect
+    #puts options.inspect
     agent = Woodchuck::Agent.new(options)
     agent.start(true)
     Signal.trap('INT') do

lib/woodchuck/tail_reader.rb

@@ -0,0 +1,26 @@
+require "eventmachine-tail"
+
+class Woodchuck::TailReader < EventMachine::FileTail
+
+	attr_accessor :input_format, :output
+
+	def self.static_init(input_format, output)
+		@@input_format = input_format
+		@@output = output
+	end
+
+	def initialize(path, startpos = -1)
+		super(path, startpos)
+		@input_format = @@input_format
+		@output = @@output
+
+		@buffer = BufferedTokenizer.new
+	end
+
+	def receive_data(data)
+		@buffer.extract(data.force_encoding('utf-8')).each do |line|
+			@output.handle(@input_format.create(path, line))
+		end
+	end
+end
+

lib/woodchuck/watcher.rb

@@ -1,32 +1,35 @@
-require 'filewatch/tail'
+require "eventmachine"
+require "eventmachine-tail"
 require 'addressable/uri'
 require 'yajl'
 require 'socket'
 require 'time'
 require 'thread'
 require 'forwardable'
+require 'woodchuck/tail_reader'
 
 class Woodchuck::Watcher
   extend Forwardable
 
-  attr_accessor :tailer, :paths, :logger, :events, :agent
+  attr_accessor :paths, :logger, :events, :agent, :input_format
   def_delegator :@agent, :output, :output
 
-  def initialize(agent, *paths)
+  def initialize(agent, log_level, input_format, *paths)
     @agent = agent
     @logger = Woodchuck::Logger.new(::STDOUT)
-    @tailer = FileWatch::Tail.new
-    @paths = paths.flatten
+		@logger.level = log_level unless log_level.nil?
+    @paths = paths.flatten.compact
+		@input_format = input_format
   end
 
   def start
-    paths.each do |path|
-      tailer.tail(path)
-    end
-    tailer.subscribe do |path, line|
-      event = Woodchuck::Event.new(path, line)
-      output.handle(event)
-    end
+		Woodchuck::TailReader.static_init(input_format, output)
+
+		EventMachine.run do
+			paths.each do |path|
+				EventMachine::FileGlobWatchTail.new(path, Woodchuck::TailReader)
+			end
+		end
   end
 
   def inspect

woodchuck-json-event.gemspec

@@ -0,0 +1,34 @@
+# -*- encoding: utf-8 -*-
+lib = File.expand_path('../lib', __FILE__)
+$LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
+require 'woodchuck/version'
+
+Gem::Specification.new do |gem|
+  gem.name          = "woodchuck-json-event"
+  gem.version       = Woodchuck::VERSION
+  gem.platform    = Gem::Platform::RUBY
+  gem.authors       = ["Dan Ryan", "David Pech"]
+  gem.email         = ["dan@appliedawesome.com", "davidpechcz@gmail.com"]
+  # gem.description   = %q{TODO: Write a gem description}
+  gem.summary       = %q{lightweight log shipper for logstash with enhanced input options}
+  gem.homepage      = "https://github.com/pechdavid/woodchuck"
+
+  gem.files         = `git ls-files`.split($/)
+  gem.executables   = gem.files.grep(%r{^bin/}).map{ |f| File.basename(f) }
+  gem.test_files    = gem.files.grep(%r{^(test|spec|features)/})
+  gem.require_paths = ["lib"]
+
+  # gem.add_dependency 'listen', '~> 0.5.0'
+  gem.add_dependency 'eventmachine-tail', '~> 0.6.3'
+  gem.add_dependency 'yajl-ruby', '~> 1.1.0'
+  gem.add_dependency 'cabin', '~> 0.4.4'
+  gem.add_dependency 'addressable', '~> 2.3.2'
+  gem.add_dependency 'redis', '~> 3.0.1'
+  gem.add_dependency 'redis-namespace', '~> 1.2.1'
+  gem.add_dependency 'fallen', '~> 0.0.1'
+
+  gem.add_development_dependency 'bundler'
+  gem.add_development_dependency 'rspec', '~> 2.11.0'
+  gem.add_development_dependency 'spork', '~> 1.0.0.rc3'
+  gem.add_development_dependency 'timecop', '~> 0.4.6'
+end

woodchuck.gemspec

@@ -19,7 +19,7 @@ Gem::Specification.new do |gem|
   gem.require_paths = ["lib"]
 
   # gem.add_dependency 'listen', '~> 0.5.0'
-  gem.add_dependency 'filewatch', '~> 0.3.4'
+  gem.add_dependency 'eventmachine-tail', '~> 0.6.3'
   gem.add_dependency 'yajl-ruby', '~> 1.1.0'
   gem.add_dependency 'cabin', '~> 0.4.4'
   gem.add_dependency 'addressable', '~> 2.3.2'