Merge branch 'master' of github.com:logstash/logstash

CHANGELOG

@@ -27,6 +27,7 @@
  - log4j input: now supports MDC 'event properties' which are stored as fields
    in the logstash event. (#216, #179. Patches by Charles Robertson and Jurjan
    Woltman)
+ - pipe input: fixed to actually work now.
 
  ## filters
  - new: kv: useful for parsing log formats taht use 'foo=bar baz=fizz' and

lib/logstash/agent.rb

@@ -466,6 +466,7 @@ def run_with_config(config)
       if @plugins.values.count { |p| p.alive? } == 0
         @logger.warn("no plugins running, shutting down")
         shutdown
+        break
       end
       @logger.debug("heartbeat")
     end
@@ -484,7 +485,18 @@ def shutdown
     shutdown_plugins(@plugins)
     # When we get here, all inputs have finished, all messages are done
     @logger.info("Shutdown complete")
-    exit(0)
+
+    # The 'unless $TESTING' is a hack for now to work around the test suite
+    # needing the pipeline to finish cleanly. We should just *not* exit here,
+    # but many plugins don't shutdown correctly. Fixing that shutdown problem
+    # will require a new pipeline design that has shutdown contracts built-in
+    # to the plugin<->agent protocol.
+    #
+    # For now, to make SIGINT/SIGTERM actually shutdown, exit. Unless we are
+    # testing, in which case wait properly for shutdown. Shitty solution, but
+    # whatever. We'll hopefully have a new pipeline/plugin protocol design
+    # shortly (by November 2012?) that will resolve this hack.
+    exit(0) unless $TESTING
   end # def shutdown
 
   def shutdown_plugins(plugins)

lib/logstash/config/mixin.rb

@@ -305,8 +305,21 @@ def validate_value(value, validator)
             if value.size % 2 == 1
               return false, "This field must contain an even number of items, got #{value.size}"
             end
-            # Use Hash[] (works in 1.8.7, anyway) to coerce into a hash.
-            result = Hash[*value]
+
+            # Convert the array the config parser produces into a hash.
+            result = {}
+            value.each_slice(2) do |key, value|
+              entry = result[key]
+              if entry.nil?
+                result[key] = value
+              else
+                if entry.is_a?(Array)
+                  entry << value
+                else
+                  result[key] = [entry, value]
+                end
+              end
+            end
           when :array
             result = value
           when :string

lib/logstash/filters/base.rb

@@ -99,8 +99,12 @@ def threadsafe?
   def filter_matched(event)
     (@add_field or {}).each do |field, value|
       event[field] ||= []
-      event[field] = [event[field]] if !event[field].is_a?(Array)
-      event[field] << event.sprintf(value)
+      if value.is_a?(Array)
+        event[field] += value
+      else
+        event[field] = [event[field]] if !event[field].is_a?(Array)
+        event[field] << event.sprintf(value)
+      end
       @logger.debug("filters/#{self.class.name}: adding value to field",
                     :field => field, :value => value)
     end

lib/logstash/filters/grep.rb

@@ -56,6 +56,22 @@ def filter(event)
 
     @logger.debug("Running grep filter", :event => event, :config => config)
     matches = 0
+
+    # If negate is set but no patterns are given, drop the event.
+    # This is useful in cases where you want to drop all events with
+    # a given type or set of tags
+    #
+    # filter {
+    #   grep {
+    #     negate => true
+    #     type => blah
+    #   }
+    # }
+    if @negate && @patterns.empty?
+      event.cancel
+      return
+    end
+
     @patterns.each do |field, regexes|
       # For each match object, we have to match everything in order to
       # apply any fields/tags.

lib/logstash/inputs/pipe.rb

@@ -11,7 +11,11 @@ class LogStash::Inputs::Pipe < LogStash::Inputs::Base
   config_name "pipe"
   plugin_status "experimental"
 
-  # Command line to run and read events from.
+  # Command to run and read events from, one line at a time.
+  #
+  # Example:
+  #
+  #    command => "echo hello world"
   config :command, :validate => :string, :required => true
 
   public
@@ -25,7 +29,8 @@ def run(queue)
     @pipe = IO.popen(command, mode="r")
     hostname = Socket.gethostname
 
-    @pipe.readline do |line|
+    @pipe.each do |line|
+      line = line.chomp
       source = "pipe://#{hostname}/#{command}"
       @logger.debug("Received line", :command => command, :line => line)
       e = to_event(line, source)

spec/filters/noop.rb

@@ -15,8 +15,8 @@
     }
     CONFIG
 
-    sample "" do
-      insist { subject["new_field"]} == ["new_value", "new_value_2"]
+    sample "example" do
+      insist { subject["new_field"] } == ["new_value", "new_value_2"]
     end
   end
 
@@ -31,11 +31,11 @@
     CONFIG
 
     sample({"@type" => "noop"})  do
-      insist { subject.tags} == ["test"]
+      insist { subject.tags } == ["test"]
     end
 
     sample({"@type" => "not_noop"})  do
-      insist { subject.tags} == []
+      insist { subject.tags } == []
     end
   end
 

spec/inputs/redis.rb

@@ -1,8 +1,8 @@
 require "test_utils"
+require "redis"
 
 describe "inputs/redis" do
   extend LogStash::RSpec
-  require "redis"
 
   describe "read events from a list" do
     key = 10.times.collect { rand(10).to_s }.join("")
@@ -20,12 +20,14 @@
     CONFIG
 
     # populate the redis list
-    before :all do
+    before :each do
       require "logstash/event"
       redis = Redis.new(:host => "localhost")
       event_count.times do |value|
         event = LogStash::Event.new("@fields" => { "sequence" => value })
-        redis.rpush(key, event.to_json)
+        Stud::try(10.times) do
+          redis.rpush(key, event.to_json)
+        end
       end
     end
 

spec/outputs/elasticsearch.rb

@@ -0,0 +1,54 @@
+require "test_utils"
+require "ftw"
+
+describe "outputs/elasticsearch" do
+  extend LogStash::RSpec
+
+  describe "ship lots of events" do
+    # Generate a random index name
+    index = 10.times.collect { rand(10).to_s }.join("")
+
+    # Write about 10000 events. Add jitter to increase likeliness of finding
+    # boundary-related bugs.
+    event_count = 10000 + rand(500)
+
+    embedded_http_port = rand(20000) + 10000
+
+    config <<-CONFIG
+      input {
+        generator {
+          message => "hello world"
+          count => #{event_count}
+          type => "generator"
+        }
+      }
+      output {
+        elasticsearch {
+          embedded => true
+          embedded_http_port => #{embedded_http_port}
+          cluster => "#{index}"
+          index => "#{index}"
+          index_type => "testing"
+        }
+      }
+    CONFIG
+
+    agent do
+      # Try a few times to check if we have the correct number of events stored
+      # in ES.
+      #
+      # We try multiple times to allow final agent flushes as well as allowing
+      # elasticsearch to finish processing everything.
+      Stud::try(10.times) do
+        ftw = FTW::Agent.new
+        data = ""
+        response = ftw.get!("http://127.0.0.1:#{embedded_http_port}/#{index}/_count?q=*")
+        response.read_body { |chunk| data << chunk }
+        count = JSON.parse(data)["count"]
+        insist { count } == event_count
+      end
+
+      puts "Rate: #{event_count / @duration}/sec"
+    end
+  end
+end

spec/outputs/redis.rb

@@ -1,6 +1,8 @@
 require "test_utils"
+require "logstash/outputs/redis"
+require "redis"
 
-describe "outputs/redis" do
+describe LogStash::Outputs::Redis do
   extend LogStash::RSpec
 
   describe "ship lots of events to a list" do

spec/test_utils.rb

@@ -3,6 +3,7 @@
 require "insist"
 require "stud/try"
 
+$TESTING = true
 if RUBY_VERSION < "1.9.2"
   $stderr.puts "Ruby 1.9.2 or later is required. (You are running: " + RUBY_VERSION + ")"
   $stderr.puts "Options for fixing this: "
@@ -13,7 +14,6 @@
 
 module LogStash
   module RSpec
-
     def config(configstr)
       @config_str = configstr
     end # def config
@@ -86,7 +86,7 @@ def agent(&block)
       # scoping is hard, let's go shopping!
       config_str = @config_str
       describe "agent(#{@agent_count}) #{caller[1]}" do
-        before :all do
+        before :each do
           start = ::Time.now
           @agent = LogStash::Agent.new
           @agent.run(["-e", config_str])
@@ -97,6 +97,7 @@ def agent(&block)
       end
       @agent_count += 1
     end # def agent
+
   end # module RSpec
 end # module LogStash
 
@@ -109,4 +110,3 @@ def <<(event)
     @block.call(event)
   end
 end # class Shiftback
-

test/services/redis/Makefile

@@ -0,0 +1,29 @@
+VERSION?=2.4.17
+TARBALL=redis-$(VERSION).tar.gz
+DIR=redis-$(VERSION)
+BASEURL=http://redis.googlecode.com/files
+
+default: build
+
+run: | build
+	$(DIR)/src/redis-server
+
+fetch: $(TARBALL)
+extract: $(DIR) | fetch
+
+build: $(DIR)/src/redis-server
+
+$(DIR)/src/redis-server: | extract
+	cd $(DIR); make
+
+clean:
+	-rm -rf $(DIR)
+
+distclean:
+	-rm -f $(TARBALL)
+
+$(TARBALL):
+	wget $(BASEURL)/$@
+
+$(DIR): | $(TARBALL)
+	tar -zxf $(TARBALL)