Merge remote-tracking branch 'upstream/master'

2025-04-23 22:27:21 -04:00 · 2013-08-20 16:14:10 -07:00 · 2013-08-20 16:14:10 -07:00 · 095a19da4e
commit 095a19da4e
parent cacafc99f2 501ff5b799
66 changed files with 1736 additions and 331 deletions
--- a/.gitignore
+++ b/.gitignore
@ -1,5 +1,7 @@
 .*.swp
 *.gem
+pkg/*.deb
+pkg/*.rpm
 *.class
 .rbx
 Gemfile.lock
@ -15,3 +17,5 @@ vendor
 data
 .buildpath
 .project
+.DS_Store
+*.pyc
--- a/11
+++ b/11
@ -5,21 +5,20 @@
  - The old logstash web ui has been replaced by Kibana 3. Kibana is a far
    superior search and analytics interface.
  - TODO(sissel): document new conditionals feature (LOGSTASH-661)
-  - TODO(sissel): document new field selection syntax (LOGSTASH-1153)
-  - ElasticSearch version 0.90.0 is included.
+  - TODO(sissel): document new field reference syntax (LOGSTASH-1153)
+  - ElasticSearch version 0.90.2 is included.
  - Many deprecated features have been removed.
    TODO(sissel): Document what these were.
  - 'type' is no longer a required setting on inputs.
-  - feature: codecs. Used to implement encoding/decoding of events
-    for input and output plugins.
+  - feature: codecs. Used to implement decoding of events for inputs and
+    encoding of events for outputs.
      TODO(nickethier): Document how to use and how to hack.
-  - The multiline filter is replaced by the multiline codec.

  ## inputs
  - bugfix: gelf: work around gelf parser errors (#476, patch by Chris McCoy)
  - broken: the twitter input is disabled because the twitter stream v1 api is
    no longer supported and I couldn't find a replacement library that works
-    under jruby.
+    under JRuby.

  ## filters
  - feature: grok: 'singles' now defaults to true.
--- a/7
+++ b/7
@ -1,2 +1,9 @@
 source :rubygems
 gemspec :name => "logstash"
+
+group :development do
+  gem "insist"
+  gem "guard"
+  gem "guard-rspec"
+  gem "parallel_tests"
+end
--- a/4
+++ b/4
@ -3,7 +3,7 @@
 #   wget or curl
 #
 JRUBY_VERSION=1.7.4
-ELASTICSEARCH_VERSION=0.90.0
+ELASTICSEARCH_VERSION=0.90.2
 #VERSION=$(shell ruby -r./lib/logstash/version -e 'puts LOGSTASH_VERSION')
 VERSION=$(shell awk -F\" '/LOGSTASH_VERSION/ {print $$2}' lib/logstash/version.rb)

@ -66,6 +66,7 @@ clean:
 	-$(QUIET)rm -rf .bundle
 	-$(QUIET)rm -rf build
 	-$(QUIET)rm -rf vendor
+	-$(QUIET)rm -f pkg/*.deb

 .PHONY: compile
 compile: compile-grammar compile-runner | build/ruby
@ -364,3 +365,4 @@ package:
 vendor/kibana: | build
 	$(QUIET)mkdir vendor/kibana || true
 	$(DOWNLOAD_COMMAND) - $(KIBANA_URL) | tar -C $@ -zx --strip-components=1
+	$(QUIET)mv vendor/kibana/dashboards/logstash.json vendor/kibana/dashboards/default.json
--- a/docs/configuration.md
+++ b/docs/configuration.md
@ -27,7 +27,8 @@ Example:

 ## Filters and Ordering

-For a given event, are applied in the order of appearance in the config file.
+For a given event, are applied in the order of appearance in the
+configuration file.

 ## Comments

@ -35,14 +36,46 @@ Comments are as in ruby, perl, and python. Starts with a '#' character. Example:

    # this is a comment

+    input { # comments can appear at the end of a line, too
+      # ...
+    }
+
+## Plugins
+
+The input, filter, and output sections all let you configure plugins. Plugins
+configuration consists of the plugin name followed by a block of settings for
+that plugin. For example, how about two file inputs:
+
+    input {
+      file {
+        path => "/var/log/messages"
+        type => "syslog"
+      }
+
+      file {
+        path => "/var/log/apache/access.log"
+        type => "apache"
+      }
+    }
+
+The above configures a two file separate inputs. Both set two
+configuration settings each: path and type. Each plugin has different
+settings for configuring it, seek the documentation for your plugin to
+learn what settings are available and what they mean. For example, the
+[file input][fileinput] documentation will explain the meanings of the
+path and type settings.
+
+[fileinput]: inputs/file
+
 ## Value Types

-The documentation for a plugin may say that a config field has a certain type.
-Examples include boolean, string, array, number, hash, etc.
+The documentation for a plugin may say that a configuration field has a
+certain type.  Examples include boolean, string, array, number, hash,
+etc.

 ### <a name="boolean"></a>Boolean

-A boolean must be either true or false.
+A boolean must be either `true` or `false`.

 Examples:

@ -68,7 +101,7 @@ Example:

 ### <a name="array"></a>Array

-An 'array' can be a single string value or multiple. If you specify the same
+An array can be a single string value or multiple. If you specify the same
 field multiple times, it appends to the array.

 Examples:
@ -80,11 +113,82 @@ The above makes 'path' a 3-element array including all 3 strings.

 ### <a name="hash"></a>Hash

-A 'hash' is basically the same syntax as Ruby hashes. 
-The 'key' and 'value' are simply pairs, such as:
+A hash is basically the same syntax as Ruby hashes. 
+The key and value are simply pairs, such as:

    match => { "field1" => "value1", "field2" => "value2", ... }

-## Further reading
+## Conditionals
+
+Sometimes you only want a filter or output to process an even under
+certain conditions. For that, you'll want to use a conditional!
+
+Conditionals in logstash look and act the same way they do in programming
+languages. You have `if`, `else if` and `else` statements. Conditionals may be
+nested if you need that.
+
+The syntax is follows:
+
+    if EXPRESSION {
+      ...
+    } else if EXPRESSION {
+      ...
+    } else {
+      ...
+    }
+
+What's an expression? Comparison tests, boolean logic, etc!
+
+The following comparison operators  are supported:
+
+* equality, etc: ==  !=  <  >  <=  >= 
+* regexp: =~  !~ 
+* inclusion: in
+
+The following boolean operators are supported:
+
+* and, or, nand, xor
+
+The following unary operators are supported:
+
+* !
+
+Expressions may contain expressions. Expressions may be negated with `!`.
+Expressions may be grouped with parentheses `(...)`.
+
+For example, if we want to remove the field `secret` if the field
+`action` has a value of `login`:
+
+    filter {
+      if [action] == "login" {
+        mutate { remove => "secret" }
+      }
+    }
+
+The above uses the field reference syntax to get the value of the
+`action` field. It is compared against the text `login` and, when equal,
+allows the mutate filter to do delete the field named `secret`
+
+How about a more complex example?
+
+* alert nagios of any apache events with status 5xx
+* record any 4xx status to elasticsearch
+* record all status code hits via statsd
+
+How about telling nagios of any http event that has a status code of 5xx?
+
+    output {
+      if [type] == "apache" {
+        if [status] =~ /^5\d\d/ {
+          nagios { ...  }
+        } else if [status] =~ /^4\d\d/ {
+          elasticsearch { ... }
+        }
+
+        statsd { increment => "apache.%{status}" }
+      }
+    }
+
+## Further Reading

 For more information, see [the plugin docs index](index)
--- a/docs/index.html.erb
+++ b/docs/index.html.erb
@ -4,13 +4,19 @@ layout: content_right
 ---
 <div id="doc_index_container">

-  <h3> general </h3>
+  <h3> for users </h3>
  <ul>
    <li> <a href="https://logstash.objects.dreamhost.com/release/logstash-%VERSION%-flatjar.jar"> download logstash %VERSION% </a> </li>
-    <li> <a href="flags"> command-line flags </a> </li>
    <li> <a href="configuration"> configuration file overview </a> </li>
-    <li> <a href="extending"> writing your own plugins </a> </li>
    <li> <a href="life-of-an-event"> the life of an event in logstash </a> </li>
+    <li> <a href="flags"> command-line flags </a> </li>
+    <li> <a href="conditionals">conditionals</a> </li>
+    <li> <a href="field-references">event field reference</a> </li>
+    <li> <a href="format-strings">text formatting</a> </li>
+  </ul>
+
+  <h3> for developers </h3>
+    <li> <a href="extending"> writing your own plugins </a> </li>
  </ul>

  <h3> use cases and tutorials </h3>
--- a/docs/plugin-doc.html.erb
+++ b/docs/plugin-doc.html.erb
@ -51,7 +51,7 @@ This is what it might look like in your config file:
     end
     annotation += ", default: #{config[:default].inspect}" if config.include?(:default)
 -%>
-    <a href="#setting_<%= name %>"><%= name %></a> => ... # <%= annotation %>
+    <a href="#<%= name %>"><%= name %></a> => ... # <%= annotation %>
 <% end -%>
  }
 }
@ -69,7 +69,7 @@ This is what it might look like in your config file:
     end
 -%>
 <h4> 
-  <a name="setting_<%= name %>">
+  <a name="<%= name %>">
    <%= name %><%= " (required setting)" if config[:required] %>
    <%= " <strong>DEPRECATED</strong>" if config[:deprecated] %>
 </a>
--- a/lib/logstash/JRUBY-6970.rb
+++ b/lib/logstash/JRUBY-6970.rb
@ -41,16 +41,7 @@ class File
          return expand_path_JRUBY_6970(path, dir)
        else
          resource = expand_path_JRUBY_6970(resource, dir)
-          # TODO(sissel): use LogStash::Util::UNAME
-          if RbConfig::CONFIG["host_os"] == "mswin32"
-            # 'expand_path' on "/" will return "C:/" on windows.
-            # So like.. we don't want that because technically this
-            # is the root of the jar, not of a disk.
-            puts :expand_path => [path, "#{jar}!#{resource.gsub(/^[A-Za-z]:/, "")}"]
-            return "#{jar}!#{resource.gsub(/^[A-Za-z]:/, "")}"
-          else
-            return "#{jar}!#{resource}"
-          end
+          return fix_jar_path(jar, resource)
        end
      elsif dir =~ /(jar:)?file:\/.*\.jar!/
        jar, dir = dir.split("!", 2)
@ -58,11 +49,26 @@ class File
          # sometimes the original dir is just 'file:/foo.jar!'
          return File.join("#{jar}!", path) 
        end
-        return "#{jar}!#{expand_path_JRUBY_6970(path, dir)}"
+        dir = expand_path_JRUBY_6970(path, dir)
+        return fix_jar_path(jar, dir)
      else
        return expand_path_JRUBY_6970(path, dir)
      end
    end
  end
-end

+  protected
+
+  def self.fix_jar_path(jar, resource)
+    # TODO(sissel): use LogStash::Util::UNAME
+    if RbConfig::CONFIG["host_os"] == "mswin32"
+      # 'expand_path' on "/" will return "C:/" on windows.
+      # So like.. we don't want that because technically this
+      # is the root of the jar, not of a disk.
+      puts :fix_jar_path => ["#{jar}!#{resource.gsub(/^[A-Za-z]:/, "")}"]
+      return "#{jar}!#{resource.gsub(/^[A-Za-z]:/, "")}"
+    else
+      return "#{jar}!#{resource}"
+    end
+  end
+end
--- a/lib/logstash/config/config_ast.rb
+++ b/lib/logstash/config/config_ast.rb
@ -53,11 +53,12 @@ module LogStash; module Config; module AST
      end

      # start inputs
-      code << "class << self"
+      #code << "class << self"
      definitions = []
        
      ["filter", "output"].each do |type|
-        definitions << "def #{type}(event)"
+        #definitions << "def #{type}(event)"
+        definitions << "@#{type}_func = lambda do |event, &block|"
        if type == "filter"
          definitions << "  extra_events = []"
        end
@ -68,13 +69,13 @@ module LogStash; module Config; module AST
        end

        if type == "filter"
-          definitions << "  extra_events.each { |e| yield e }"
+          definitions << "  extra_events.each(&block)"
        end
        definitions << "end"
      end

      code += definitions.join("\n").split("\n", -1).collect { |l| "  #{l}" }
-      code << "end"
+      #code << "end"
      return code.join("\n")
    end
  end
@ -169,7 +170,7 @@ module LogStash; module Config; module AST
            "  extra_events << newevent",
            "end",
            "if event.cancelled?",
-            "  extra_events.each { |e| yield e }",
+            "  extra_events.each(&block)",
            "  return",
            "end",
          ].map { |l| "#{l}\n" }.join("")
@ -193,7 +194,8 @@ module LogStash; module Config; module AST
      return %Q(#{name.compile} => #{value.compile})
    end
  end
-  class Value < Node; end
+  class RValue < Node; end
+  class Value < RValue; end
  class Bareword < Value
    def compile
      return text_value.inspect
@ -262,11 +264,19 @@ module LogStash; module Config; module AST
  end
  module Expression
    def compile
+      # Hack for compiling 'in' support.
+      # This really belongs elsewhere, I think.
+      cmp = recursive_select(LogStash::Config::AST::ComparisonOperator)
+      if cmp.count == 1 && cmp.first.text_value == "in"
+        # item 'in' list
+        # technically anything that responds to #include? is accepted.
+        item, list = recursive_select(LogStash::Config::AST::RValue)
+        return "(x = #{list.compile}; x.respond_to?(:include?) && x.include?(#{item.compile}))"
+        #return "#{list.compile}.include?(#{item.compile})"
+      end
      return "(#{super})"
    end
  end
-  class Rvalue < Node
-  end
  class MethodCall < Node
    def compile
      arguments = recursive_inject { |e| [String, Number, Selector, Array, MethodCall].any? { |c| e.is_a?(c) } }
@ -283,7 +293,7 @@ module LogStash; module Config; module AST
      return " #{text_value} "
    end
  end
-  class Selector < Node
+  class Selector < RValue
    def compile
      return "event[#{text_value.inspect}]"
    end
--- a/lib/logstash/config/grammar.rb
+++ b/lib/logstash/config/grammar.rb
@ -1652,7 +1652,7 @@ module LogStashConfig
      elements[0]
    end

-    def elsif
+    def else_if
      elements[1]
    end
  end
@ -1695,7 +1695,7 @@ module LogStashConfig
        r4 = _nt__
        s3 << r4
        if r4
-          r5 = _nt_elsif
+          r5 = _nt_else_if
          s3 << r5
        end
        if s3.last
@ -1873,7 +1873,7 @@ module LogStashConfig
    r0
  end

-  module Elsif0
+  module ElseIf0
    def branch_or_plugin
      elements[0]
    end
@ -1883,29 +1883,33 @@ module LogStashConfig
    end
  end

-  module Elsif1
+  module ElseIf1
    def _1
      elements[1]
    end

-    def condition
-      elements[2]
-    end
-
    def _2
      elements[3]
    end

+    def condition
+      elements[4]
+    end
+
    def _3
      elements[5]
    end

+    def _4
+      elements[7]
+    end
+
  end

-  def _nt_elsif
+  def _nt_else_if
    start_index = index
-    if node_cache[:elsif].has_key?(index)
-      cached = node_cache[:elsif][index]
+    if node_cache[:else_if].has_key?(index)
+      cached = node_cache[:else_if][index]
      if cached
        cached = SyntaxNode.new(input, index...(index + 1)) if cached == true
        @index = cached.interval.end
@ -1914,11 +1918,11 @@ module LogStashConfig
    end

    i0, s0 = index, []
-    if has_terminal?("elsif", false, index)
-      r1 = instantiate_node(SyntaxNode,input, index...(index + 5))
-      @index += 5
+    if has_terminal?("else", false, index)
+      r1 = instantiate_node(SyntaxNode,input, index...(index + 4))
+      @index += 4
    else
-      terminal_parse_failure("elsif")
+      terminal_parse_failure("else")
      r1 = nil
    end
    s0 << r1
@ -1926,57 +1930,71 @@ module LogStashConfig
      r2 = _nt__
      s0 << r2
      if r2
-        r3 = _nt_condition
+        if has_terminal?("if", false, index)
+          r3 = instantiate_node(SyntaxNode,input, index...(index + 2))
+          @index += 2
+        else
+          terminal_parse_failure("if")
+          r3 = nil
+        end
        s0 << r3
        if r3
          r4 = _nt__
          s0 << r4
          if r4
-            if has_terminal?("{", false, index)
-              r5 = instantiate_node(SyntaxNode,input, index...(index + 1))
-              @index += 1
-            else
-              terminal_parse_failure("{")
-              r5 = nil
-            end
+            r5 = _nt_condition
            s0 << r5
            if r5
              r6 = _nt__
              s0 << r6
              if r6
-                s7, i7 = [], index
-                loop do
-                  i8, s8 = index, []
-                  r9 = _nt_branch_or_plugin
-                  s8 << r9
-                  if r9
-                    r10 = _nt__
-                    s8 << r10
-                  end
-                  if s8.last
-                    r8 = instantiate_node(SyntaxNode,input, i8...index, s8)
-                    r8.extend(Elsif0)
-                  else
-                    @index = i8
-                    r8 = nil
-                  end
-                  if r8
-                    s7 << r8
-                  else
-                    break
-                  end
+                if has_terminal?("{", false, index)
+                  r7 = instantiate_node(SyntaxNode,input, index...(index + 1))
+                  @index += 1
+                else
+                  terminal_parse_failure("{")
+                  r7 = nil
                end
-                r7 = instantiate_node(SyntaxNode,input, i7...index, s7)
                s0 << r7
                if r7
-                  if has_terminal?("}", false, index)
-                    r11 = instantiate_node(SyntaxNode,input, index...(index + 1))
-                    @index += 1
-                  else
-                    terminal_parse_failure("}")
-                    r11 = nil
+                  r8 = _nt__
+                  s0 << r8
+                  if r8
+                    s9, i9 = [], index
+                    loop do
+                      i10, s10 = index, []
+                      r11 = _nt_branch_or_plugin
+                      s10 << r11
+                      if r11
+                        r12 = _nt__
+                        s10 << r12
+                      end
+                      if s10.last
+                        r10 = instantiate_node(SyntaxNode,input, i10...index, s10)
+                        r10.extend(ElseIf0)
+                      else
+                        @index = i10
+                        r10 = nil
+                      end
+                      if r10
+                        s9 << r10
+                      else
+                        break
+                      end
+                    end
+                    r9 = instantiate_node(SyntaxNode,input, i9...index, s9)
+                    s0 << r9
+                    if r9
+                      if has_terminal?("}", false, index)
+                        r13 = instantiate_node(SyntaxNode,input, index...(index + 1))
+                        @index += 1
+                      else
+                        terminal_parse_failure("}")
+                        r13 = nil
+                      end
+                      s0 << r13
+                    end
                  end
-                  s0 << r11
                end
              end
            end
@ -1986,13 +2004,13 @@ module LogStashConfig
    end
    if s0.last
      r0 = instantiate_node(LogStash::Config::AST::Elsif,input, i0...index, s0)
-      r0.extend(Elsif1)
+      r0.extend(ElseIf1)
    else
      @index = i0
      r0 = nil
    end

-    node_cache[:elsif][start_index] = r0
+    node_cache[:else_if][start_index] = r0

    r0
  end
--- a/lib/logstash/config/grammar.treetop
+++ b/lib/logstash/config/grammar.treetop
@ -115,7 +115,7 @@ grammar LogStashConfig

  # Conditions
  rule branch
-    if (_ elsif)* (_ else)?
+    if (_ else_if)* (_ else)?
    <LogStash::Config::AST::Branch>
  end

@ -124,8 +124,8 @@ grammar LogStashConfig
    <LogStash::Config::AST::If>
  end

-  rule elsif
-    "elsif" _ condition _ "{" _ ( branch_or_plugin _)* "}"
+  rule else_if
+    "else" _ "if" _ condition _ "{" _ ( branch_or_plugin _)* "}"
    <LogStash::Config::AST::Elsif>
  end

--- a/lib/logstash/config/mixin.rb
+++ b/lib/logstash/config/mixin.rb
@ -62,7 +62,14 @@ module LogStash::Config::Mixin
    self.class.get_config.each do |name, opts|
      next if params.include?(name.to_s)
      if opts.include?(:default) and (name.is_a?(Symbol) or name.is_a?(String))
-        params[name.to_s] = opts[:default] unless params.include?(name.to_s)
+        # default values should be cloned if possible
+        # cloning prevents 
+        case opts[:default]
+          when FalseClass, TrueClass, NilClass, Fixnum
+            params[name.to_s] = opts[:default]
+          else
+            params[name.to_s] = opts[:default].clone
+        end
      end
    end

--- a/lib/logstash/event.rb
+++ b/lib/logstash/event.rb
@ -2,6 +2,7 @@ require "json"
 require "time"
 require "date"
 require "logstash/namespace"
+require "logstash/util/fieldreference"

 # Use a custom serialization for jsonifying Time objects.
 # TODO(sissel): Put this in a separate file.
@ -110,40 +111,40 @@ class LogStash::Event
  
  # field-related access
  public
-  def [](key)
-    if key[0] == '['
-      val = @data
-      key.gsub(/(?<=\[).+?(?=\])/).each do |tok|
-        if val.is_a? Array
-          val = val[tok.to_i]
-        else
-          val = val[tok]
-        end
-      end
-      return val
+  def [](str)
+    if str[0,1] == "+"
    else
-      return @data[key]
+      return LogStash::Util::FieldReference.exec(str, @data)
    end
  end # def []
  
  public
-  def []=(key, value)
-    if key[0] == '['
-      val = @data
-      keys = key.scan(/(?<=\[).+?(?=\])/)
-      last = keys.pop
+  def []=(str, value)
+    r = LogStash::Util::FieldReference.exec(str, @data) do |obj, key|
+      obj[key] = value
+    end

-      keys.each do |tok|
-        if val.is_a? Array
-          val = val[tok.to_i]
+    # The assignment can fail if the given field reference (str) does not exist
+    # In this case, we'll want to set the value manually.
+    if r.nil?
+      # TODO(sissel): Implement this in LogStash::Util::FieldReference
+      if str[0,1] != "["
+        return @data[str] = value
+      end
+
+      # No existing element was found, so let's set one.
+      *parents, key = str.scan(/(?<=\[)[^\]]+(?=\])/)
+      obj = @data
+      parents.each do |p|
+        if obj.include?(p)
+          obj = obj[p]
        else
-          val = val[tok]
+          obj[p] = {}
        end
      end
-      val[last] = value
-    else
-      @data[key] = value
+      obj[key] = value
    end
+    return value
  end # def []=

  public
@ -177,10 +178,13 @@ class LogStash::Event
    LogStash::Util.hash_merge(@data, event.to_hash)
  end # append

-  # Remove a field. Returns the value of that field when deleted
+  # Remove a field or field reference. Returns the value of that field when
+  # deleted
  public
-  def remove(field)
-    return @data.delete(field)
+  def remove(str)
+    return LogStash::Util::FieldReference.exec(str, @data) do |obj, key|
+      next obj.delete(key)
+    end
  end # def remove

  # sprintf. This could use a better method name.
@ -201,6 +205,7 @@ class LogStash::Event
  # is an array (or hash?) should be. Join by comma? Something else?
  public
  def sprintf(format)
+    format = format.to_s
    if format.index("%").nil?
      return format
    end
--- a/lib/logstash/filters/json_encode.rb
+++ b/lib/logstash/filters/json_encode.rb
@ -0,0 +1,59 @@
+require "logstash/filters/base"
+require "logstash/namespace"
+
+# JSON encode filter. Takes a field and serializes it into JSON
+class LogStash::Filters::JsonEncode < LogStash::Filters::Base
+
+  config_name "json_encode"
+  plugin_status "beta"
+
+  # Config for json_encode is:
+  # 
+  # * source => dest
+  #
+  # For example, if you have a field named 'foo', and you want to store the
+  # JSON encoded string in 'bar', do this:
+  #
+  #     filter {
+  #       json_encode {
+  #         foo => bar
+  #       }
+  #     }
+  #
+  # Note: if the "dest" field already exists, it will be overridden.
+  config /[A-Za-z0-9_@-]+/, :validate => :string
+
+  public
+  def register
+    @json = {}
+
+    @config.each do |field, dest|
+      next if RESERVED.member?(field)
+
+      @json[field] = dest
+    end
+  end # def register
+
+  public
+  def filter(event)
+    return unless filter?(event)
+
+    @logger.debug("Running JSON encoder", :event => event)
+
+    @json.each do |key, dest|
+      next unless event[key]
+
+      begin
+        event[dest] = JSON.pretty_generate(event[key])
+        filter_matched(event)
+      rescue => e
+        event.tags << "_jsongeneratefailure"
+        @logger.warn("Trouble encoding JSON", :key => key, :raw => event[key],
+                      :exception => e)
+        next
+      end
+    end
+
+    @logger.debug("Event after JSON encoder", :event => event)
+  end # def filter
+end # class LogStash::Filters::JsonEncode
--- a/lib/logstash/filters/kv.rb
+++ b/lib/logstash/filters/kv.rb
@ -42,6 +42,21 @@ class LogStash::Filters::KV < LogStash::Filters::Base
  #     }
  config :trim, :validate => :string

+  # A string of characters to trim from the key. This is useful if your
+  # key are wrapped in brackets or starts with space
+  #
+  # These characters form a regex character class and thus you must escape special regex
+  # characters like [ or ] using \.
+  #
+  # Example, to strip '<' '>' '[' ']' and ',' characters from keys:
+  #
+  #     filter {
+  #       kv {
+  #         trimkey => "<>\[\],"
+  #       }
+  #     }
+  config :trimkey, :validate => :string
+
  # A string of characters to use as delimiters for parsing out key-value pairs.
  #
  # These characters form a regex character class and thus you must escape special regex
@ -143,6 +158,7 @@ class LogStash::Filters::KV < LogStash::Filters::Base

  def register
    @trim_re = Regexp.new("[#{@trim}]") if !@trim.nil?
+    @trimkey_re = Regexp.new("[#{@trimkey}]") if !@trimkey.nil?
    @scan_re = Regexp.new("((?:\\\\ |[^"+@field_split+@value_split+"])+)["+@value_split+"](?:\"([^\"]+)\"|'([^']+)'|((?:\\\\ |[^"+@field_split+"])+))")
  end # def register

@ -186,6 +202,7 @@ class LogStash::Filters::KV < LogStash::Filters::Base
    end
    text.scan(@scan_re) do |key, v1, v2, v3|
      value = v1 || v2 || v3
+      key = @trimkey.nil? ? key : key.gsub(@trimkey_re, "")      
      key = @prefix + key
      next if not @include_keys.empty? and not @include_keys.include?(key)
      next if @exclude_keys.include?(key)
--- a/lib/logstash/filters/railsparallelrequest.rb
+++ b/lib/logstash/filters/railsparallelrequest.rb
@ -26,9 +26,9 @@ class LogStash::Filters::Railsparallelrequest < LogStash::Filters::Base

  def filter(event)
    return unless filter?(event)
-    return if event.tags.include? CONFIG_NAME
+    return if event.tags.include? self.class.config_name

-    event.tags << CONFIG_NAME
+    event.tags << self.class.config_name

    line = event["message"]

--- a/lib/logstash/filters/xml.rb
+++ b/lib/logstash/filters/xml.rb
@ -71,36 +71,30 @@ class LogStash::Filters::Xml < LogStash::Filters::Base
  public
  def filter(event)
    return unless filter?(event)
-
    matched = false

    @logger.debug("Running xml filter", :event => event)

-    key = @source
-    dest = @target
+    return unless event.include?(@source)

-    return unless event.include?(key)
-    if event[key].is_a?(String)
-      event[key] = [event[key]]
-    end
+    value = event[@source]

-    if event[key].length > 1
+    if value.is_a?(Array) && value.length > 1
      @logger.warn("XML filter only works on fields of length 1",
-                   :key => key, :value => event[key])
+                   :source => @source, :value => value)
      return
    end

-    raw = event[key].first
-
-    # for some reason, an empty string is considered valid XML
-    return if raw.strip.length == 0
+    # Do nothing with an empty string.
+    return if value.strip.length == 0

    if @xpath
      begin
-        doc = Nokogiri::XML(raw)
+        doc = Nokogiri::XML(value)
      rescue => e
+        p :failed => value
        event.tag("_xmlparsefailure")
-        @logger.warn("Trouble parsing xml", :key => key, :raw => raw,
+        @logger.warn("Trouble parsing xml", :source => @source, :value => value,
                     :exception => e, :backtrace => e.backtrace)
        return
      end
@ -129,12 +123,12 @@ class LogStash::Filters::Xml < LogStash::Filters::Base

    if @store_xml
      begin
-        event[dest] = XmlSimple.xml_in(raw)
+        event[@target] = XmlSimple.xml_in(value)
        matched = true
      rescue => e
        event.tag("_xmlparsefailure")
-        @logger.warn("Trouble parsing xml with XmlSimple", :key => key,
-                     :raw => raw, :exception => e, :backtrace => e.backtrace)
+        @logger.warn("Trouble parsing xml with XmlSimple", :source => @source,
+                     :value => value, :exception => e, :backtrace => e.backtrace)
        return
      end
    end # if @store_xml
--- a/lib/logstash/inputs/imap.rb
+++ b/lib/logstash/inputs/imap.rb
@ -1,5 +1,6 @@
 require "logstash/inputs/base"
 require "logstash/namespace"
+require "stud/interval"
 require "socket" # for Socket.gethostname

 # Read mail from IMAP servers
@ -53,7 +54,7 @@ class LogStash::Inputs::IMAP < LogStash::Inputs::Base
    # EOFError, OpenSSL::SSL::SSLError
    imap = connect
    imap.select("INBOX")
-    ids = imap.search("ALL")
+    ids = imap.search("NOT SEEN")

    ids.each_slice(@fetch_count) do |id_set|
      items = imap.fetch(id_set, "RFC822")
@ -61,6 +62,7 @@ class LogStash::Inputs::IMAP < LogStash::Inputs::Base
        mail = Mail.read_from_string(item.attr["RFC822"])
        queue << mail_to_event(mail)
      end
+      imap.store(id_set, '+FLAGS', :Seen)
    end

    imap.close
@ -79,7 +81,7 @@ class LogStash::Inputs::IMAP < LogStash::Inputs::Base
    end

    event = to_event(message, "imap://#{@user}@#{@host}/#{m.from.first rescue ""}")
-   
+
    # Use the 'Date' field as the timestamp
    t = mail.date.to_time.gmtime
    event["@timestamp"] = sprintf(ISO8601_STRFTIME, t.year, t.month,
--- a/lib/logstash/inputs/lumberjack.rb
+++ b/lib/logstash/inputs/lumberjack.rb
@ -40,22 +40,14 @@ class LogStash::Inputs::Lumberjack < LogStash::Inputs::Base
  public
  def run(output_queue)
    @lumberjack.run do |l|
-      file = l.delete("file")
-      if file[0,1] == "/"
-        source = "lumberjack://#{l.delete("host")}#{file}"
-      else
-        source = "lumberjack://#{l.delete("host")}/#{file}"
-      end
-      event = to_event(l.delete("line"), source)
-
-      # TODO(sissel): We shoudln't use 'fields' here explicitly, but the new
-      # 'event[key]' code seems... slow, so work around it for now.
-      # TODO(sissel): Once Event_v1 is live, we can just merge 'l' directly into it.
-      #l.each do |key, value|
-        #event[key] = value
+      line = l.delete("line")
+      #if file[0,1] == "/"
+        #source = "lumberjack://#{l.delete("host")}#{file}"
+      #else
+        #source = "lumberjack://#{l.delete("host")}/#{file}"
      #end
-      event.fields.merge!(l)
-
+      event = LogStash::Event.new(l)
+      event["message"] = line
      output_queue << event
    end
  end # def run
--- a/lib/logstash/inputs/s3.rb
+++ b/lib/logstash/inputs/s3.rb
@ -0,0 +1,265 @@
+require "logstash/inputs/base"
+require "logstash/namespace"
+
+require "aws-sdk"
+require "time"
+require "tmpdir"
+
+# Stream events from files from a S3 bucket.
+#
+# Each line from each file generates an event.
+# Files ending in '.gz' are handled as gzip'ed files.
+class LogStash::Inputs::S3 < LogStash::Inputs::Base
+  config_name "s3"
+  milestone 1
+
+  # The credentials of the AWS account used to access the bucket.
+  # Credentials can be specified:
+  # - As an ["id","secret"] array
+  # - As a path to a file containing AWS_ACCESS_KEY_ID=... and AWS_SECRET_ACCESS_KEY=...
+  # - In the environment (variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY)
+  config :credentials, :validate => :array, :default => nil
+
+  # The name of the S3 bucket.
+  config :bucket, :validate => :string, :required => true
+
+  # The AWS region for your bucket.
+  config :region, :validate => :string, :default => 'us-east-1'
+
+  # If specified, the prefix the filenames in the bucket must match (not a regexp)
+  config :prefix, :validate => :string, :default => nil
+
+  # Where to write the since database (keeps track of the date
+  # the last handled file was added to S3). The default will write
+  # sincedb files to some path matching "$HOME/.sincedb*"
+  config :sincedb_path, :validate => :string, :default => nil
+
+  # Name of a S3 bucket to backup processed files to.
+  config :backup_to_bucket, :validate => :string, :default => nil
+
+  # Path of a local directory to backup processed files to.
+  config :backup_to_dir, :validate => :string, :default => nil
+
+  # Whether to delete processed files from the original bucket.
+  config :delete, :validate => :boolean, :default => false
+
+  # Interval to wait between to check the file list again after a run is finished.
+  # Value is in seconds.
+  config :interval, :validate => :number, :default => 60
+
+  public
+  def register
+    LogStash::Util::set_thread_name("input|s3|#{bucket}");
+    @logger.info("Registering s3 input", :bucket => @bucket)
+
+    if @credentials.nil?
+      @access_key_id = ENV['AWS_ACCESS_KEY_ID']
+      @secret_access_key = ENV['AWS_SECRET_ACCESS_KEY']
+    elsif @credentials.is_a? Array
+      if @credentials.length ==1
+        File.open(@credentials[0]) { |f| f.each do |line|
+          unless (/^\#/.match(line))
+            if(/\s*=\s*/.match(line))
+              param, value = line.split('=', 2)
+              param = param.chomp().strip()
+              value = value.chomp().strip()
+              if param.eql?('AWS_ACCESS_KEY_ID')
+                @access_key_id = value
+              elsif param.eql?('AWS_SECRET_ACCESS_KEY')
+                @secret_access_key = value
+              end
+            end
+          end
+        end
+        }
+      elsif @credentials.length == 2
+        @access_key_id = @credentials[0]
+        @secret_access_key = @credentials[1]
+      else
+        raise ArgumentError.new('Credentials must be of the form "/path/to/file" or ["id", "secret"]')
+      end
+    end
+    if @access_key_id.nil? or @secret_access_key.nil?
+      raise ArgumentError.new('Missing AWS credentials')
+    end
+
+    if @bucket.nil?
+      raise ArgumentError.new('Missing AWS bucket')
+    end
+
+    if @sincedb_path.nil?
+      if ENV['HOME'].nil?
+        raise ArgumentError.new('No HOME or sincedb_path set')
+      end
+      @sincedb_path = File.join(ENV["HOME"], ".sincedb_" + Digest::MD5.hexdigest("#{@bucket}+#{@prefix}"))
+    end
+
+    s3 = AWS::S3.new(
+      :access_key_id => @access_key_id,
+      :secret_access_key => @secret_access_key,
+      :region => @region
+    )
+    @s3bucket = s3.buckets[@bucket]
+
+    unless @backup_to_bucket.nil?
+      @backup_bucket = s3.buckets[@backup_to_bucket]
+      unless @backup_bucket.exists?
+        s3.buckets.create(@backup_to_bucket)
+      end
+    end
+
+    unless @backup_to_dir.nil?
+      Dir.mkdir(@backup_to_dir, 0700) unless File.exists?(@backup_to_dir)
+    end
+
+  end # def register
+
+  public
+  def run(queue)
+    loop do
+      process_new(queue)
+      sleep(@interval)
+    end
+    finished
+  end # def run
+
+  private
+  def process_new(queue, since=nil)
+
+    if since.nil?
+        since = sincedb_read()
+    end
+
+    objects = list_new(since)
+    objects.each do |k|
+      @logger.debug("S3 input processing", :bucket => @bucket, :key => k)
+      lastmod = @s3bucket.objects[k].last_modified
+      process_log(queue, k)
+      sincedb_write(lastmod)
+    end
+
+  end # def process_new
+
+  private
+  def list_new(since=nil)
+
+    if since.nil?
+      since = Time.new(0)
+    end
+
+    objects = {}
+    @s3bucket.objects.with_prefix(@prefix).each do |log|
+      if log.last_modified > since
+        objects[log.key] = log.last_modified
+      end
+    end
+
+    return sorted_objects = objects.keys.sort {|a,b| objects[a] <=> objects[b]}
+
+  end # def list_new
+
+  private
+  def process_log(queue, key)
+
+    source = "s3://#{bucket}/#{key}"
+    object = @s3bucket.objects[key]
+    tmp = Dir.mktmpdir("logstash-")
+    begin
+      filename = File.join(tmp, File.basename(key))
+      File.open(filename, 'wb') do |s3file|
+        object.read do |chunk|
+          s3file.write(chunk)
+        end
+      end
+      process_local_log(queue, filename, source)
+      unless @backup_to_bucket.nil?
+        backup_object = @backup_bucket.objects[key]
+        backup_object.write(Pathname.new(filename))
+      end
+      unless @backup_to_dir.nil?
+        FileUtils.cp(filename, @backup_to_dir)
+      end
+      if @delete
+        object.delete()
+      end
+    end
+    FileUtils.remove_entry_secure(tmp, force=true)
+
+  end # def process_log
+
+  private
+  def process_local_log(queue, filename, source)
+
+    metadata = {
+      :version => nil,
+      :format => nil,
+      :source => source
+    }
+    File.open(filename) do |file|
+      if filename.end_with?('.gz')
+        gz = Zlib::GzipReader.new(file)
+        gz.each_line do |line|
+          metadata = process_line(queue, metadata, line)
+        end
+      else
+        file.each do |line|
+          metadata = process_line(queue, metadata, line)
+        end
+      end
+    end
+
+  end # def process_local_log
+
+  private
+  def process_line(queue, metadata, line)
+
+    if /#Version: .+/.match(line)
+      junk, version = line.strip().split(/#Version: (.+)/)
+      unless version.nil?
+        metadata[:version] = version
+      end
+    elsif /#Fields: .+/.match(line)
+      junk, format = line.strip().split(/#Fields: (.+)/)
+      unless format.nil?
+        metadata[:format] = format
+      end
+    else
+      @codec.decode(line) do |event|
+        unless metadata[:version].nil?
+          event["cloudfront_version"] = metadata[:version]
+        end
+        unless metadata[:format].nil?
+          event["cloudfront_fields"] = metadata[:format]
+        end
+        event["source"] = metadata[:source]
+        queue << event
+      end
+    end
+    return metadata
+
+  end # def process_line
+
+  private
+  def sincedb_read()
+
+    if File.exists?(@sincedb_path)
+      since = Time.parse(File.read(@sincedb_path).chomp.strip)
+    else
+      since = Time.new(0)
+    end
+    return since
+
+  end # def sincedb_read
+
+  private
+  def sincedb_write(since=nil)
+
+    if since.nil?
+      since = Time.now()
+    end
+    File.open(@sincedb_path, 'w') { |file| file.write(since.to_s) }
+
+  end # def sincedb_write
+
+end # class LogStash::Inputs::S3
+
--- a/lib/logstash/outputs/datadog_metrics.rb
+++ b/lib/logstash/outputs/datadog_metrics.rb
@ -1,5 +1,6 @@
 require "logstash/outputs/base"
 require "logstash/namespace"
+require "stud/buffer"

 # This output lets you send metrics to
 # DataDogHQ based on Logstash events.
@ -9,6 +10,8 @@ require "logstash/namespace"

 class LogStash::Outputs::DatadogMetrics < LogStash::Outputs::Base

+  include Stud::Buffer
+
  config_name "datadog_metrics"
  milestone 1

@ -38,13 +41,11 @@ class LogStash::Outputs::DatadogMetrics < LogStash::Outputs::Base
  # prior to schedule set in @timeframe
  config :queue_size, :validate => :number, :default => 10

-  # How often to flush queued events to Datadog
-  config :timeframe, :validate => :string, :default => "10s"
+  # How often (in seconds) to flush queued events to Datadog
+  config :timeframe, :validate => :number, :default => 10

  public
  def register
-    require "thread"
-    require "rufus/scheduler"
    require 'time'
    require "net/https"
    require "uri"
@ -55,13 +56,11 @@ class LogStash::Outputs::DatadogMetrics < LogStash::Outputs::Base
    @client.use_ssl = true
    @client.verify_mode = OpenSSL::SSL::VERIFY_NONE
    @logger.debug("Client", :client => @client.inspect)
-
-    @event_queue = SizedQueue.new(@queue_size)
-    @scheduler = Rufus::Scheduler.start_new
-    @job = @scheduler.every @timeframe do
-      @logger.info("Scheduler Activated")
-      flush_metrics
-    end
+    buffer_initialize(
+      :max_items => @queue_size,
+      :max_interval => @timeframe,
+      :logger => @logger
+    )
  end # def register

  public
@ -84,35 +83,24 @@ class LogStash::Outputs::DatadogMetrics < LogStash::Outputs::Base
    end
    dd_metrics['tags'] = tagz if tagz

-    if (@event_queue.length >= @event_queue.max)
-      @job.trigger
-      @logger.warn("Event queue full! Flushing before schedule. Consider increasing queue_size.")
-    end
-
    @logger.info("Queueing event", :event => dd_metrics)
-    @event_queue << dd_metrics
+    buffer_receive(dd_metrics)
  end # def receive

-  private
-  def flush_metrics
+  public
+  def flush(events, final=false)
    dd_series = Hash.new
    dd_series['series'] = []

-    while !@event_queue.empty? do
+    events.each do |event|
      begin
-        event = @event_queue.pop(true)
        dd_series['series'] << event
-      rescue Exception => e
-        @logger.warn("Exception!  Breaking count loop", :exception => e)
-        break
+      rescue
+        @logger.warn("Error adding event to series!", :exception => e)
+        next
      end
    end

-    if dd_series['series'].empty?
-        @logger.info("Datadog metrics queue empty. Skipping.")
-        return
-    end
-
    request = Net::HTTP::Post.new("#{@uri.path}?api_key=#{@api_key}")

    begin
@ -124,7 +112,7 @@ class LogStash::Outputs::DatadogMetrics < LogStash::Outputs::Base
    rescue Exception => e
      @logger.warn("Unhandled exception", :request => request.inspect, :response => response.inspect, :exception => e.inspect)
    end
-  end # def flush_metrics
+  end # def flush

  private
  def to_epoch(t)
--- a/lib/logstash/outputs/elasticsearch.rb
+++ b/lib/logstash/outputs/elasticsearch.rb
@ -32,9 +32,6 @@ class LogStash::Outputs::ElasticSearch < LogStash::Outputs::Base
  config_name "elasticsearch"
  milestone 3

-  # ElasticSearch server name. This is optional if your server is discoverable.
-  config :host, :validate => :string
-
  # The index to write events to. This can be dynamic using the %{foo} syntax.
  # The default value will partition your indices by day so you can more easily
  # delete old data or only search specific date ranges.
--- a/lib/logstash/outputs/rabbitmq/bunny.rb
+++ b/lib/logstash/outputs/rabbitmq/bunny.rb
@ -45,6 +45,8 @@ class LogStash::Outputs::RabbitMQ
        return if terminating?

        sleep n
+        connect
+        declare_exchange
        retry
      end
    end
--- a/lib/logstash/outputs/rabbitmq/hot_bunnies.rb
+++ b/lib/logstash/outputs/rabbitmq/hot_bunnies.rb
@ -56,6 +56,7 @@ class LogStash::Outputs::RabbitMQ

        connect
        declare_exchange
+        retry
      end
    end

--- a/lib/logstash/outputs/tcp.rb
+++ b/lib/logstash/outputs/tcp.rb
@ -21,6 +21,9 @@ class LogStash::Outputs::Tcp < LogStash::Outputs::Base
  # When mode is `server`, the port to listen on.
  # When mode is `client`, the port to connect to.
  config :port, :validate => :number, :required => true
+  
+  # When connect failed,retry interval in sec.
+  config :reconnect_interval, :validate => :number, :default => 10

  # Mode to operate in. `server` listens for client connections,
  # `client` connects to a server.
@ -116,7 +119,9 @@ class LogStash::Outputs::Tcp < LogStash::Outputs::Base
      rescue => e
        @logger.warn("tcp output exception", :host => @host, :port => @port,
                     :exception => e, :backtrace => e.backtrace)
+        sleep @reconnect_interval
        @client_socket = nil
+        retry
      end
    end
  end # def receive
--- a/lib/logstash/outputs/websocket.rb
+++ b/lib/logstash/outputs/websocket.rb
@ -4,7 +4,7 @@ require "logstash/outputs/base"
 # This output runs a websocket server and publishes any 
 # messages to all connected websocket clients.
 #
-# You can connect to it with ws://<host>:<port>/
+# You can connect to it with ws://<host\>:<port\>/
 #
 # If no clients are connected, any messages received are ignored.
 class LogStash::Outputs::WebSocket < LogStash::Outputs::Base
--- a/lib/logstash/pipeline.rb
+++ b/lib/logstash/pipeline.rb
@ -199,7 +199,7 @@ class LogStash::Pipeline
      output(event)
    end # while true
    @outputs.each(&:teardown)
-  end # def filterworker
+  end # def outputworker

  # Shutdown this pipeline.
  #
@ -225,4 +225,12 @@ class LogStash::Pipeline
    klass = LogStash::Plugin.lookup(plugin_type, name)
    return klass.new(*args)
  end
+
+  def filter(event, &block)
+    @filter_func.call(event, &block)
+  end
+
+  def output(event)
+    @output_func.call(event)
+  end
 end # class Pipeline
--- a/lib/logstash/runner.rb
+++ b/lib/logstash/runner.rb
@ -156,7 +156,8 @@ class LogStash::Runner
      end,
      "irb" => lambda do
        require "irb"
-        return IRB.start(__FILE__)
+        IRB.start(__FILE__)
+        return []
      end,
      "pry" => lambda do
        require "pry"
--- a/lib/logstash/util/fieldreference.rb
+++ b/lib/logstash/util/fieldreference.rb
@ -0,0 +1,48 @@
+require "logstash/namespace"
+require "logstash/util"
+
+module LogStash::Util::FieldReference
+  def compile(str)
+    if str[0,1] != '['
+      return <<-"CODE"
+        lambda do |e, &block|
+          return block.call(e, #{str.inspect}) unless block.nil?
+          return e[#{str.inspect}]
+        end
+      CODE
+    end
+
+    code = "lambda do |e, &block|\n"
+    selectors = str.scan(/(?<=\[).+?(?=\])/)
+    selectors.each_with_index do |tok, i|
+      last = (i == selectors.count() - 1)
+      code << "   # [#{tok}]#{ last ? " (last selector)" : "" }\n"
+     
+      if last
+        code << <<-"CODE"
+          return block.call(e, #{tok.inspect}) unless block.nil?
+        CODE
+      end
+
+      code << <<-"CODE"
+        if e.is_a?(Array)
+          e = e[#{tok.to_i}]
+        else
+          e = e[#{tok.inspect}]
+        end
+      CODE
+      
+    end
+    code << "return e\nend"
+    #puts code
+    return code
+  end # def compile
+
+  def exec(str, obj, &block)
+    @__fieldeval_cache ||= {}
+    @__fieldeval_cache[str] ||= eval(compile(str))
+    return @__fieldeval_cache[str].call(obj, &block)
+  end
+
+  extend self
+end # module LogStash::Util::FieldReference
--- a/locales/en.yml
+++ b/locales/en.yml
@ -37,7 +37,7 @@ en:
      milestone:
        "0": >-
          Using milestone 0 %{type} plugin '%{name}'. This plugin isn't well
-          supported by the commnity and likely has no maintainer. For more
+          supported by the community and likely has no maintainer. For more
          information on plugin milestones, see
          http://logstash.net/docs/%{LOGSTASH_VERSION}/plugin-milestones
        "1": >-
--- a/logstash-event.gemspec
+++ b/logstash-event.gemspec
@ -25,5 +25,7 @@ Gem::Specification.new do |gem|
  gem.version       = LOGSTASH_VERSION
  
  gem.add_development_dependency "rspec"
+  gem.add_development_dependency "guard"
+  gem.add_development_dependency "guard-rspec"
  gem.add_development_dependency "insist", "0.0.8"
 end
--- a/logstash.gemspec
+++ b/logstash.gemspec
@ -89,7 +89,7 @@ Gem::Specification.new do |gem|
  end

  if RUBY_PLATFORM != 'java'
-    gem.add_runtime_dependency "bunny",       ["~> 0.9.2"]  #(MIT license)
+    gem.add_runtime_dependency "bunny",       ["~> 0.9.8"]  #(MIT license)
  else
    gem.add_runtime_dependency "hot_bunnies", ["~> 2.0.0.pre9"] #(MIT license)
  end
--- a/misc/patterns/mysql
+++ b/misc/patterns/mysql
@ -0,0 +1,11 @@
+MYSQL_ERROR_TIMESTAMP %{NUMBER:date} %{TIME}
+MYSQL_ERORR_LOG_CONTENT_P1 \[%{WORD:mysql_error_log_level}\] %{GREEDYDATA:mysql_error_log_content}
+MYSQL_ERORR_LOG_CONTENT_P2 %{GREEDYDATA:mysql_error_log_content}
+MYSQL_ERROR_LOG %{MYSQL_ERROR_TIMESTAMP} (%{MYSQL_ERORR_LOG_CONTENT_P1}|%{MYSQL_ERORR_LOG_CONTENT_P2})
+
+MYSQL_SLOW_FROM ^# User@Host: %{USER:user}\[[^\]]+\] @ %{HOST:host} \[%{IP:ip_addr}?]
+MYSQL_SLOW_STAT ^# Query_time: %{NUMBER:duration:float} \s*Lock_time: %{NUMBER:lock_wait:float} \s*Rows_sent: %{NUMBER:results:int} \s*Rows_examined: %{NUMBER:scanned:int}
+MYSQL_SLOW_TIMESTAMP ^SET timestamp=%{NUMBER:timestamp};
+MYSQL_SLOW_DB ^use %{WORD:db_name};
+MYSQL_SLOW_QUERY ^%{WORD:action}%{SPACE}%{GREEDYDATA};
+
--- a/misc/patterns/php5
+++ b/misc/patterns/php5
@ -0,0 +1,6 @@
+PHP_LOG_CONTENT (.+)
+PHP_DATE_TIME %{MONTHDAY}-%{MONTH}-%{YEAR}\s+%{TIME}
+PHP_TZ_NAME [A-Z]{3}
+PHP_ERROR_LOG \s*+\[%{PHP_DATE_TIME:timestamp} %{PHP_TZ_NAME}\] PHP %{LOGLEVEL:php_log_level} error: %{PHP_LOG_CONTENT:php_log_content}
+PHP_FPM_ERROR_LOG \[%{PHP_DATE_TIME:timestamp}\] %{LOGLEVEL:php_log_level}: (\[%{GREEDYDATA:php_fpm_pool}\] child %{POSINT}, %{GREEDYDATA:php_log_content}|%{GREEDYDATA:php_log_content})
+PHP_FPM_SLOW_LOG \[%{GREEDYDATA:stack_addr}\] %{GREEDYDATA:func_name} %{UNIXPATH:script_path}
--- a/misc/patterns/redis
+++ b/misc/patterns/redis
@ -0,0 +1,2 @@
+REDISLOG_WITH_LEVEL \[%{POSINT:pid}\] %{REDISTIMESTAMP:timestamp} # %{LOGLEVEL:redis_log_level} %{GREEDYDATA}
+REDISLOG_FIXED (%{REDISLOG}|%{REDISLOG_WITH_LEVEL})
--- a/misc/pl.rb
+++ b/misc/pl.rb
--- a/misc/pl2.rb
+++ b/misc/pl2.rb
--- a/misc/screencast/000.intro
+++ b/misc/screencast/000.intro
--- a/misc/screencast/001.config
+++ b/misc/screencast/001.config
--- a/misc/screencast/002.webdemo
+++ b/misc/screencast/002.webdemo
--- a/misc/screencast/README
+++ b/misc/screencast/README
--- a/misc/screencast/run.rb
+++ b/misc/screencast/run.rb
--- a/misc/shipper_config_generator.py
+++ b/misc/shipper_config_generator.py
@ -0,0 +1,416 @@
+#!/usr/bin/env python
+"""
+generate logstash shipper configuration file
+"""
+import logging
+import os
+import re
+import subprocess
+import sys
+import httplib
+import socket
+
+from subprocess import Popen, CalledProcessError
+from subprocess import STDOUT, PIPE
+
+logging.getLogger("shipper_config_generator").setLevel(logging.DEBUG)
+
+PWD = os.path.dirname(os.path.realpath(__file__))
+
+
+def _get_first_ip_addr_by_sock():
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.connect(('baidu.com', 80))
+    ip_addr = sock.getsockname()[0]
+    return ip_addr
+   
+def _get_first_ip_addr_by_http():
+    conn = httplib.HTTPConnection(host='ifconfig.me', port=80, timeout=3.0)
+    conn.request('GET', '/ip')
+    resp = conn.getresponse()
+    ip_addr = resp.read().strip()
+    return ip_addr
+  
+def get_first_ip_addr():
+    try:
+        return _get_first_ip_addr_by_http()
+    except Exception:
+        return _get_first_ip_addr_by_sock()
+
+
+# this function copy from Python 2.7 subprocess.py::check_output
+def func_check_output(*popenargs, **kwargs):
+    r"""Run command with arguments and return its output as a byte string.
+
+    If the exit code was non-zero it raises a CalledProcessError.  The
+    CalledProcessError object will have the return code in the returncode
+    attribute and output in the output attribute.
+
+    The arguments are the same as for the Popen constructor.  Example:
+
+    >>> check_output(["ls", "-l", "/dev/null"])
+    'crw-rw-rw- 1 root root 1, 3 Oct 18  2007 /dev/null\n'
+
+    The stdout argument is not allowed as it is used internally.
+    To capture standard error in the result, use stderr=STDOUT.
+
+    >>> check_output(["/bin/sh", "-c",
+    ...               "ls -l non_existent_file ; exit 0"],
+    ...              stderr=STDOUT)
+    'ls: non_existent_file: No such file or directory\n'
+    """
+    if 'stdout' in kwargs:
+        raise ValueError('stdout argument not allowed, it will be overridden.')
+    process = Popen(stdout=PIPE, *popenargs, **kwargs)
+    output, unused_err = process.communicate()
+    retcode = process.poll()
+    if retcode:
+        cmd = kwargs.get("args")
+        if cmd is None:
+            cmd = popenargs[0]
+        raise CalledProcessError(retcode, cmd, output=output)
+    return output
+
+
+def check_output(*popenargs, **kwargs):
+    if sys.version_info[0] == 2 and sys.version_info[1] < 7:
+        _check_output = func_check_output
+    else:
+        _check_output = subprocess.check_output
+
+    return _check_output(*popenargs, **kwargs)
+
+def check_output_wrapper(s):
+    return check_output(s, shell=True).strip()
+
+"""
+Usage: 
+s = "ps aux | grep redis-server | grep -v 'grep' | awk '{print $NF}'"
+print check_output(s, shell=True)
+print check_output_wrapper(s)
+"""
+
+
+def get_redis_log_full_path_list():
+    log_full_path_list = set()
+    ls_rds_inst = "ps aux | grep redis-server | grep -v 'grep' | awk '{print $NF}'"
+    
+    for config_path in check_output_wrapper(ls_rds_inst).split():
+        if not os.path.exists(config_path):
+           sys.stderr.write('[redis] %s not exists or not a absolute path \n' % config_path)
+           continue
+
+        with open(config_path) as f:
+            for line in f.readlines():
+                if line.startswith('logfile'):
+                    splits = line.split()
+                    if len(splits) == 2:
+                        key, val = splits[0], splits[1].strip()
+                        if os.path.exists(val):
+                            log_full_path_list.add(val)
+    return log_full_path_list
+
+def get_php_fpm_log_full_path_list():
+    error_log_full_path_list = set()
+    slow_log_full_path_list = set()
+
+    get_config_abs_path = "ps aux |grep php-fpm |grep master | awk '{print $NF}' | tr -d '()'"
+
+    for config_path in check_output_wrapper(get_config_abs_path).split():
+        config_path = config_path.strip().strip('"').strip("'")
+        if not config_path:
+            continue
+        if not os.path.exists(config_path):
+            sys.stderr.write('[php-fpm] %s not exits or not a absolute path \n' % config_path)
+            continue
+
+        s = file(config_path).read()
+        pool_name_list = [i for i in re.findall('^\[(?P<pool_name>\w+)\]', s, re.MULTILINE)
+                    if i != 'global']
+        if len(pool_name_list) != 1:
+            sys.stderr.write("[php-fpm] %s php-fpm log detector doesn't supports multiple pool \n" % config_path)
+            continue
+        pool_name = pool_name_list[0]
+
+        with open(config_path) as f:
+            for line in f.readlines():
+                if line.startswith('error_log'):
+                    splits = line.split('=')
+                    error_log = splits[-1].strip().strip(';').replace('$pool',  pool_name)
+                    if not os.path.exists(error_log):
+                        sys.stderr.write('[php-fpm] %s not exits or not a absolute path \n' % error_log)
+                        continue
+                    error_log_full_path_list.add(error_log)
+
+                if line.startswith('slowlog'):
+                    splits = line.split('=')
+                    slow_log = splits[-1].strip().strip(';').replace('$pool', pool_name)
+                    if not os.path.exists(slow_log):
+                        sys.stderr.write('[php-fpm] %s not exits or not a absolute path \n' % slow_log)
+                        continue
+                    slow_log_full_path_list.add(slow_log)
+
+    return error_log_full_path_list, slow_log_full_path_list
+
+
+def get_mysql_log_full_path_list():
+    error_list = set()
+    slow_list = set()
+
+    for pid in check_output_wrapper('pidof mysqld').split():
+        pid = pid.strip()
+        meta = {
+            'config-file': None,
+            'error-log': None,
+            'slow-log': None,
+            }
+        for line in check_output_wrapper('ps -p %s -f | grep %s' % (pid, pid)).split():
+            line = line.strip().replace('_', '-')
+            if line.startswith("--defaults-file"):
+                meta['config-file'] = line.replace('--defaults-file=', '')
+            elif line.startswith('--log-error'):
+                meta['error-log'] = line.replace('--log-error=', '')
+            elif line.startswith('--slow-query-log-file'):
+                meta['slow-log'] = line.replace('--slow-query-log-file=', '')
+
+        if meta['config-file']:
+            with open(meta['config-file']) as f:
+                for line in f.readlines():
+                    line = line.replace('_', '-')
+                    if line.startswith('slow-query-log-file'):
+                        meta['slow-log'] = line.replace('slow-query-log-file', '').replace('=', '').strip()
+                    elif line.startswith('log-error'):
+                        meta['error-log'] = line.replace('error-log', '').replace('=', '').strip()                    
+
+        if meta['slow-log']:
+            slow_list.add(meta['slow-log'])
+        if meta['error-log']:
+            error_list.add(meta['error-log'])
+            
+    return list(error_list), list(slow_list)
+
+
+TEMPLATE_INPUT_FILE = """  file {{
+    charset => 'UTF-8'
+    type => '{logstash_type}'
+    path => '{file_path}'
+    format => 'plain'
+  }}
+"""
+
+def generte_input_file_block(file_path, logstash_type):
+     return TEMPLATE_INPUT_FILE.format(
+          logstash_type=logstash_type,
+          file_path=file_path, 
+          )
+
+CONFIG_TEMPLATE_FILTER_PHP = """  multiline {{
+    type => 'php-error'
+    pattern => '^(\s|#|Stack)'
+    what => 'previous'
+  }}
+
+  multiline {{
+    type => 'php-fpm-slow'
+    pattern => '^$'
+    what => 'previous'
+    negate => true
+  }}
+
+  grok {{
+    type => 'php-error'
+    patterns_dir => '{patterns_dir}'
+    pattern => '%{{PHP_ERROR_LOG}}'
+    singles => true
+  }}
+
+  grok {{
+    type => 'php-fpm-error'
+    patterns_dir => '{patterns_dir}'
+    pattern => '%{{PHP_FPM_ERROR_LOG}}'
+    singles => true
+  }}
+
+  grok {{
+    type => 'php-fpm-slow'
+    patterns_dir => '{patterns_dir}'
+    pattern => '%{{PHP_FPM_SLOW_LOG}}'
+    singles => true
+  }}
+"""
+
+CONFIG_TEMPLATE_INPUTS = """input {{
+  stdin {{
+    type => 'stdin-type'
+  }}
+
+  tcp {{
+    type => 'test-pattern'
+    host => '127.0.0.1'
+    port => 9100
+    mode => server
+    debug => true
+    format => plain
+  }}
+
+{input_blocks}
+
+}}
+
+"""
+
+CONFIG_TEMPLATE_FILTERS_PREFIX = """filter {{
+"""
+CONFIG_TEMPLATE_FILTERS_SUFFIX = """  date {{
+    match => ['timestamp', 'dd-MMM-YYYY HH:mm:ss z', 'dd-MMM-YYYY HH:mm:ss']
+  }}
+
+  mutate {{
+    replace => ["@source_host", "DEFAULT_SOURCE_HOST"]
+  }}
+
+}}
+
+"""
+
+CONFIG_TEMPLATE_FILTER_MYSQL = """  multiline {{
+    type => 'mysql-slow'
+    pattern => "^# User@Host: "
+    negate => true
+    what => previous
+  }}
+
+  multiline {{
+    type => 'mysql-error'
+    what => previous
+    pattern => '^\s'
+  }}
+
+  grok {{
+    type => 'mysql-error'
+    patterns_dir => '{patterns_dir}'
+    pattern => '%{{MYSQL_ERROR_LOG}}'
+  }}
+
+  grep {{
+    type => 'mysql-slow'
+    match => [ "@message", "^# Time: " ]
+    negate => true
+  }}
+
+  grok {{
+    type => 'mysql-slow'
+    singles => true
+    patterns_dir => '{patterns_dir}'
+    pattern => [
+      "%{{MYSQL_SLOW_FROM}}",
+      "%{{MYSQL_SLOW_STAT}}",
+      "%{{MYSQL_SLOW_TIMESTAMP}}",
+      "%{{MYSQL_SLOW_DB}}",
+      "%{{MYSQL_SLOW_QUERY}}"
+     ]
+   }}
+
+  date {{
+    type => 'mysql-slow'
+    match => ['timestamp', 'YYddMM HH:mm:ss']
+  }}
+
+  mutate {{
+    type => 'mysql-slow'
+    remove => "timestamp"
+  }}
+"""
+
+CONFIG_TEMPLATE_FILTER_REDIS = """  grok {{
+    type => 'redis'
+    patterns_dir => '{patterns_dir}'
+    pattern => '%{{REDISLOG_FIXED}}'
+  }}
+"""
+
+
+CONFIG_TEMPLATE_OUTPUTS = """output {{
+#  stdout {{
+#    debug => true
+#    debug_format => "json"
+#  }}
+
+  redis {{
+    host => "{output_redis_host}"
+    port => {output_redis_port}
+    data_type => "list"
+    key => "logstash"
+  }}
+
+{output_blocks}
+}}
+"""
+
+def main():
+    output_redis_host = '10.20.60.85'
+    output_redis_port = 6380
+    patterns_dir = '/usr/local/logstash/patterns'
+
+
+    chunks = []
+    for path in get_redis_log_full_path_list():
+        sys.stdout.write("%s %s found \n" % ("redis", path))
+        chunks.append(generte_input_file_block(path, "redis"))
+
+    error_list, slow_list = get_php_fpm_log_full_path_list()
+    for path in error_list:
+        sys.stdout.write("%s %s found \n" % ("php-fpm-error", path))
+        chunks.append(generte_input_file_block(path, "php-fpm-error"))
+    for path in slow_list:
+        sys.stdout.write("%s %s found \n" % ("php-fpm-slow", path))
+        chunks.append(generte_input_file_block(path, "php-fpm-slow"))
+
+    error_list, slow_list = get_mysql_log_full_path_list()
+    for path in error_list:
+        sys.stdout.write("%s %s found \n" % ("mysql-error", path))
+        chunks.append(generte_input_file_block(path, "mysql-error"))
+    for path in slow_list:
+        sys.stdout.write("%s %s found \n" % ("mysql-slow", path))
+        chunks.append(generte_input_file_block(path, "mysql-slow"))    
+    input_blocks = '\n'.join(chunks)
+
+    t = CONFIG_TEMPLATE_INPUTS + \
+        CONFIG_TEMPLATE_FILTERS_PREFIX + \
+        CONFIG_TEMPLATE_FILTER_REDIS + \
+        '\n' + \
+        CONFIG_TEMPLATE_FILTER_MYSQL + \
+        '\n' + \
+        CONFIG_TEMPLATE_FILTER_PHP + \
+        CONFIG_TEMPLATE_FILTERS_SUFFIX + \
+        CONFIG_TEMPLATE_OUTPUTS
+
+    output_blocks = ""
+    content = t.format(
+         input_blocks=input_blocks, 
+        output_blocks=output_blocks,
+        patterns_dir=patterns_dir,
+         output_redis_host=output_redis_host,
+         output_redis_port=output_redis_port,
+         )
+
+    ip_addr = None
+    try:
+        ip_addr = get_first_ip_addr()
+    except Exception:
+        pass
+    if ip_addr:
+        content = content.replace("DEFAULT_SOURCE_HOST", ip_addr)
+
+    FOLDER_PARENT = os.path.dirname(PWD)
+    save_to_prefix = os.path.join(FOLDER_PARENT, "conf")
+    save_to = os.path.join(save_to_prefix, "shipper-dev.conf")
+    
+    # save_to = os.path.join(PWD, 'shipper.conf')
+    
+    with open(save_to, 'w') as f:
+         f.write(content)
+    sys.stdout.write("save to %s \n" % save_to)
+
+if __name__ == '__main__':
+    main()
--- a/patterns/grok-patterns
+++ b/patterns/grok-patterns
@ -92,4 +92,4 @@ SYSLOGBASE %{SYSLOGTIMESTAMP:timestamp} (?:%{SYSLOGFACILITY} )?%{SYSLOGHOST:logs
 COMBINEDAPACHELOG %{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:referrer} %{QS:agent}

 # Log Levels
-LOGLEVEL ([T|t]race|TRACE|[D|d]ebug|DEBUG|[N|n]otice|NOTICE|[I|i]nfo|INFO|[W|w]arn?(?:ing)?|WARN?(?:ING)?|[E|e]rr?(?:or)?|ERR?(?:OR)?|[C|c]rit?(?:ical)?|CRIT?(?:ICAL)?|[F|f]atal|FATAL|[S|s]evere|SEVERE)
+LOGLEVEL ([T|t]race|TRACE|[D|d]ebug|DEBUG|[N|n]otice|NOTICE|[I|i]nfo|INFO|[W|w]arn?(?:ing)?|WARN?(?:ING)?|[E|e]rr?(?:or)?|ERR?(?:OR)?|[C|c]rit?(?:ical)?|CRIT?(?:ICAL)?|[F|f]atal|FATAL|[S|s]evere|SEVERE|EMERG(?:ENCY)?|[Ee]merg(?:ency)?)
--- a/pkg/build.sh
+++ b/pkg/build.sh
@ -15,7 +15,7 @@ release=$2

 echo "Building package for $os $release"

-destdir=build/
+destdir=build/$(echo "$os" | tr ' ' '_')
 prefix=/opt/logstash

 if [ "$destdir/$prefix" != "/" -a -d "$destdir/$prefix" ] ; then
@ -49,21 +49,31 @@ case $os@$release in
    install -m755 logstash.sysv.redhat $destdir/etc/init.d/logstash
    ;;
  ubuntu@*)
+    mkdir -p $destdir/etc/logstash/conf.d
    mkdir -p $destdir/etc/logrotate.d
    mkdir -p $destdir/etc/init
    mkdir -p $destdir/var/log/logstash
-    touch $destdir/etc/sysconfig/logstash
-    install -m644 logrotate.conf $destdir/etc/logrotate.d/
+    mkdir -p $destdir/etc/default
+    touch $destdir/etc/default/logstash
+    install -m644 logrotate.conf $destdir/etc/logrotate.d/logstash
+    install -m644 logstash.default $destdir/etc/default/logstash
+    install -m644 logstash-web.default $destdir/etc/default/logstash-web
    install -m755 logstash.upstart.ubuntu $destdir/etc/init/logstash.conf
+    install -m755 logstash-web.upstart.ubuntu $destdir/etc/init/logstash-web.conf
    ;;
  debian@*)
+    mkdir -p $destdir/etc/logstash/conf.d
    mkdir -p $destdir/etc/logrotate.d
    mkdir -p $destdir/etc/init.d
    mkdir -p $destdir/var/lib/logstash
-    mkdir -p $destdir/var/run/logstash
    mkdir -p $destdir/var/log/logstash
-    install -m644 logrotate.conf $destdir/etc/logrotate.d/
+    mkdir -p $destdir/etc/default
+    touch $destdir/etc/default/logstash
+    install -m644 logrotate.conf $destdir/etc/logrotate.d/logstash
+    install -m644 logstash.default $destdir/etc/default/logstash
+    install -m644 logstash-web.default $destdir/etc/default/logstash-web
    install -m755 logstash.sysv.debian $destdir/etc/init.d/logstash
+    install -m755 logstash-web.sysv.debian $destdir/etc/init.d/logstash-web
    ;;
  *) 
    echo "Unknown OS: $os $release"
@ -86,9 +96,10 @@ case $os in
    fpm -s dir -t deb -n logstash -v "$VERSION" \
      -a all --iteration 1-$os \
      -d "java6-runtime" \
-      --before-install ubuntu/before-install.sh \
-      --before-remove ubuntu/before-remove.sh \
-      --after-install ubuntu/after-install.sh \
+      --deb-user root --deb-group root \
+      --before-install $os/before-install.sh \
+      --before-remove $os/before-remove.sh \
+      --after-install $os/after-install.sh \
      -f -C $destdir .
    ;;
 esac
--- a/pkg/debian/after-install.sh
+++ b/pkg/debian/after-install.sh
@ -1,4 +1,5 @@
 #!/bin/sh

-mkdir -p /home/logstash
-chown logstash:logstash /home/logstash
+chown -R logstash:logstash /opt/logstash
+chown logstash /var/log/logstash
+chown logstash:logstash /var/lib/logstash
--- a/pkg/debian/before-install.sh
+++ b/pkg/debian/before-install.sh
@ -7,6 +7,7 @@ fi

 # create logstash user
 if ! getent passwd logstash >/dev/null; then
-  useradd -r -g logstash -d /home/logstash \
-    -s /sbin/nologin -c "logstash" logstash
+  useradd -M -r -g logstash -d /var/lib/logstash \
+    -s /sbin/nologin -c "LogStash Service User" logstash
 fi
+
--- a/pkg/debian/before-remove.sh
+++ b/pkg/debian/before-remove.sh
@ -1,17 +1,13 @@
 #!/bin/sh

 if [ $1 == "remove" ]; then
-  /etc/init.d/logstash > /dev/null 2>&1 || true
+  /etc/init.d/logstash stop >/dev/null 2>&1 || true

  if getent passwd logstash >/dev/null ; then
    userdel logstash
  fi

-  if getent group logstash > /dev/null ; then
+  if getent group logstash >/dev/null ; then
    groupdel logstash
  fi
-
-  if [ -d "/home/logstash" ] ; then
-    rm -rf /home/logstash
-  fi
 fi
--- a/pkg/logstash-web.default
+++ b/pkg/logstash-web.default
@ -0,0 +1,41 @@
+# defaults for logstash
+
+# Start logstash on boot?
+START=no
+
+# pulled in from the init script; makes things easier.
+NAME=logstash-web
+
+# location of java
+JAVA=/usr/bin/java
+
+# arguments to pass to java
+LS_JAVA_OPTS="-Xmx256m -Djava.io.tmpdir=/var/lib/logstash/"
+
+PIDFILE=/var/run/logstash-web.pid
+
+# user id to be invoked as
+LS_USER=logstash
+
+# location of the logstas jar file
+LS_JAR=/opt/logstash/logstash.jar
+
+# logstash home location
+LS_HOME=/var/lib/logstash
+
+# logstash log directory
+LOG_DIR=/var/log/logstash
+
+# logstash log file
+LOG_FILE=$LOG_DIR/$NAME.log
+
+# logstash configuration directory
+CONF_DIR=/etc/logstash/conf.d
+
+# Open file limit
+OPEN_FILES=2048
+
+# Nice level
+NICE=19
+
+HOME=/var/lib/logstash
--- a/pkg/logstash-web.sysv.debian
+++ b/pkg/logstash-web.sysv.debian
@ -0,0 +1,201 @@
+#!/bin/bash
+#
+# /etc/init.d/logstash -- startup script for LogStash.
+#
+### BEGIN INIT INFO
+# Provides:          logstash-web
+# Required-Start:    $all
+# Required-Stop:     $all
+# Default-Start:     2 3 4 5
+# Default-Stop:      0 1 6
+# Short-Description: Starts the LogStash web server
+# Description:       Starts logstash-web using start-stop-daemon
+### END INIT INFO
+
+set -e
+
+PATH=/bin:/usr/bin:/sbin:/usr/sbin
+NAME=logstash-web
+DESC="Logstash Web Server"
+DEFAULT=/etc/default/$NAME
+
+if [ `id -u` -ne 0 ]; then
+   echo "You need root privileges to run this script"
+   exit 1
+fi
+
+. /lib/lsb/init-functions
+
+if [ -r /etc/default/rcS ]; then
+   . /etc/default/rcS
+fi
+
+# The following variables can be overwritten in $DEFAULT
+
+# Run logstash as this user ID and group ID
+LS_USER=logstash
+LS_GROUP=logstash
+
+JAVA=/usr/bin/java
+
+# Directory where the logstash all in one jar lives
+LS_HOME=/var/lib/logstash
+
+# Additional Java OPTS
+LS_JAVA_OPTS=" -Djava.io.tmpdir=/var/logstash/"
+
+# logstash log directory
+LOG_DIR=/var/log/logstash
+
+# logstash configuration directory
+CONF_DIR=/etc/logstash/conf.d
+
+# logstash log file
+LOG_FILE=$LOG_DIR/$NAME.log
+
+# Open File limit
+OPEN_FILES=2048
+
+
+
+# Nice level
+NICE=19
+
+# End of variables that can be overwritten in $DEFAULT
+
+# overwrite settings from default file
+if [ -f "$DEFAULT" ]; then
+   . "$DEFAULT"
+fi
+
+# Define other required variables
+PID_FILE=/var/run/$NAME.pid
+DAEMON=$LS_JAR
+DAEMON_OPTS="web"
+
+is_true() {
+    if [ "x$1" = "xtrue" -o "x$1" = "xyes" -o "x$1" = "x1" ] ; then
+        return 0
+    else
+        return 1
+    fi
+}
+
+# Check DAEMON exists
+if ! test -e $DAEMON; then
+   log_failure_msg "Daemon $DAEMON doesn't exist"
+   exit 1
+fi
+
+case "$1" in
+   start)
+      if ! is_true "$START" ; then
+         echo "logstash not configured to start, please edit /etc/default/logstash to enable"
+         exit 0
+      fi
+
+      if [ -z "$JAVA" ]; then
+         log_failure_msg "no JDK found - $JAVA"
+         exit 1
+      fi
+
+      # Check if a config file exists
+      if ! test -e $CONF_DIR/*.conf; then
+         log_failure_msg "There aren't any configuration files in $CONF_DIR"
+         exit 1
+      fi
+
+      log_daemon_msg "Starting $DESC"
+
+      if start-stop-daemon --test --start --pidfile "$PID_FILE" \
+         --user "$LS_USER" --exec "$JAVA" \
+      >/dev/null; then
+         # Prepare environment
+         ulimit -n $OPEN_FILES
+
+         # Start Daemon
+         start-stop-daemon --start -b --user "$LS_USER" -c "$LS_USER":"$LS_GROUP" \
+           -d "$LS_HOME" --pidfile "$PID_FILE" --make-pidfile \
+           --exec "$JAVA" -- $LS_JAVA_OPTS -jar $DAEMON $DAEMON_OPTS
+
+         sleep 1
+
+         if start-stop-daemon --test --start --pidfile "$PID_FILE" \
+            --user "$LS_USER" --exec "$JAVA" \
+         >/dev/null; then
+
+            if [ -f "$PID_FILE" ]; then
+               rm -f "$PID_FILE"
+            fi
+
+            log_end_msg 1
+         else
+            log_end_msg 0
+         fi
+      else
+         log_progress_msg "(already running)"
+         log_end_msg 0
+      fi
+   ;;
+   stop)
+      log_daemon_msg "Stopping $DESC"
+
+      set +e
+
+      if [ -f "$PID_FILE" ]; then
+         start-stop-daemon --stop --pidfile "$PID_FILE" \
+            --user "$LS_USER" \
+            --retry=TERM/20/KILL/5 >/dev/null
+
+         if [ $? -eq 1 ]; then
+            log_progress_msg "$DESC is not running but pid file exists, cleaning up"
+         elif [ $? -eq 3 ]; then
+            PID="`cat $PID_FILE`"
+            log_failure_msg "Failed to stop $DESC (pid $PID)"
+            exit 1
+         fi
+
+         rm -f "$PID_FILE"
+      else
+         log_progress_msg "(not running)"
+      fi
+
+      log_end_msg 0
+      set -e
+   ;;
+   status)
+      set +e
+
+      start-stop-daemon --test --start --pidfile "$PID_FILE" \
+         --user "$LS_USER" --exec "$JAVA" \
+      >/dev/null 2>&1
+
+      if [ "$?" = "0" ]; then
+         if [ -f "$PID_FILE" ]; then
+            log_success_msg "$DESC is not running, but pid file exists."
+            exit 1
+         else
+            log_success_msg "$DESC is not running."
+            exit 3
+         fi
+      else
+         log_success_msg "$DESC is running with pid `cat $PID_FILE`"
+      fi
+
+      set -e
+   ;;
+   restart|force-reload)
+      if [ -f "$PID_FILE" ]; then
+         $0 stop
+         sleep 1
+      fi
+
+      $0 start
+   ;;
+   *)
+      log_success_msg "Usage: $0 {start|stop|restart|force-reload|status}"
+      exit 1
+   ;;
+esac
+
+exit 0
--- a/pkg/logstash-web.upstart.ubuntu
+++ b/pkg/logstash-web.upstart.ubuntu
@ -0,0 +1,18 @@
+# logstash-web - web server
+#
+
+description "logstash-web web server"
+
+start on virtual-filesystems
+stop on runlevel [06]
+
+# Respawn it if the process exits
+respawn
+
+setuid logstash
+setgid logstash
+
+# Change into a writable directory
+chdir /var/lib/logstash
+console log
+exec /usr/bin/java -jar /opt/logstash/logstash.jar web
--- a/pkg/logstash.default
+++ b/pkg/logstash.default
@ -0,0 +1,45 @@
+# defaults for logstash
+
+# Start logstash on boot?
+START=no
+
+# pulled in from the init script; makes things easier.
+NAME=logstash
+
+# location of java
+JAVA=/usr/bin/java
+
+# arguments to pass to java
+LS_JAVA_OPTS="-Xmx256m -Djava.io.tmpdir=/var/lib/logstash/"
+
+PIDFILE=/var/run/logstash.pid
+
+# user id to be invoked as
+LS_USER=logstash
+
+# location of the logstas jar file
+LS_JAR=/opt/logstash/logstash.jar
+
+# logstash home location
+LS_HOME=/var/lib/logstash
+
+# logstash log directory
+LOG_DIR=/var/log/logstash
+
+# logstash log file
+LOG_FILE=$LOG_DIR/$NAME.log
+
+# logstash configuration directory
+CONF_DIR=/etc/logstash/conf.d
+
+# Open file limit
+OPEN_FILES=2048
+
+# Nice level
+NICE=19
+
+# Set LogStash options
+LS_OPTS="--log ${LOG_FILE}"
+
+# Set a home directory
+HOME=/var/lib/logstash
--- a/pkg/logstash.sysv.debian
+++ b/pkg/logstash.sysv.debian
@ -56,7 +56,8 @@ LOG_FILE=$LOG_DIR/$NAME.log
 # Open File limit
 OPEN_FILES=2048

-
+# LogStash options
+LS_OPTS="--log ${LOG_DIR}/${NAME}.log"

 # Nice level
 NICE=19
@ -71,7 +72,7 @@ fi
 # Define other required variables
 PID_FILE=/var/run/$NAME.pid
 DAEMON=$LS_JAR
-DAEMON_OPTS="agent -f ${CONF_DIR} --log ${LOG_FILE}"
+DAEMON_OPTS="agent -f ${CONF_DIR} ${LS_OPTS}"

 is_true() {
    if [ "x$1" = "xtrue" -o "x$1" = "xyes" -o "x$1" = "x1" ] ; then
--- a/pkg/logstash.upstart.ubuntu
+++ b/pkg/logstash.upstart.ubuntu
@ -15,6 +15,6 @@ setgid logstash

 # You need to chdir somewhere writable because logstash needs to unpack a few
 # temporary files on startup.
-chdir /home/logstash
+chdir /var/lib/logstash
 console log
-exec /usr/bin/java -jar logstash.jar agent -f /etc/logstash/agent.conf
+exec /usr/bin/java -jar /opt/logstash/logstash.jar agent -f /etc/logstash/conf.d
--- a/pkg/ubuntu/after-install.sh
+++ b/pkg/ubuntu/after-install.sh
@ -1,4 +1,5 @@
 #!/bin/sh

-mkdir -p /home/logstash
-chown logstash:logstash /home/logstash
+chown -R logstash:logstash /opt/logstash
+chown logstash /var/log/logstash
+chown logstash:logstash /var/lib/logstash
--- a/pkg/ubuntu/before-install.sh
+++ b/pkg/ubuntu/before-install.sh
@ -7,6 +7,6 @@ fi

 # create logstash user
 if ! getent passwd logstash >/dev/null; then
-  useradd -r -g logstash -d /home/logstash \
-    -s /sbin/nologin -c "logstash" logstash
+  useradd -M -r -g logstash -d /var/lib/logstash \
+    -s /sbin/nologin -c "LogStash Service User" logstash
 fi
--- a/pkg/ubuntu/before-remove.sh
+++ b/pkg/ubuntu/before-remove.sh
@ -1,17 +1,13 @@
 #!/bin/sh

 if [ $1 == "remove" ]; then
-  stop logstash > /dev/null 2>&1 || true
+  stop logstash >/dev/null 2>&1 || true

  if getent passwd logstash >/dev/null ; then
    userdel logstash
  fi

-  if getent group logstash > /dev/null ; then
+  if getent group logstash >/dev/null ; then
    groupdel logstash
  fi
-
-  if [ -d "/home/logstash" ] ; then
-    rm -rf /home/logstash
-  fi
 fi
--- a/spec/conditionals/test.rb
+++ b/spec/conditionals/test.rb
@ -9,7 +9,7 @@ describe "conditionals" do
        mutate { add_field => { "always" => "awesome" } }
        if [foo] == "bar" {
          mutate { add_field => { "hello" => "world" } }
-        } elsif [bar] == "baz" {
+        } else if [bar] == "baz" {
          mutate { add_field => { "fancy" => "pants" } }
        } else {
          mutate { add_field => { "free" => "hugs" } }
@ -46,7 +46,7 @@ describe "conditionals" do
          mutate { add_field => { "always" => "awesome" } }
          if [foo] == "bar" {
            mutate { add_field => { "hello" => "world" } }
-          } elsif [bar] == "baz" {
+          } else if [bar] == "baz" {
            mutate { add_field => { "fancy" => "pants" } }
          } else {
            mutate { add_field => { "free" => "hugs" } }
@ -97,4 +97,34 @@ describe "conditionals" do
      insist { subject["tags"] }.include?("woot")
    end
  end
+
+  describe "the 'in' operator" do
+    config <<-CONFIG
+      filter {
+        if [foo] in [foobar] {
+          mutate { add_tag => "field in field" }
+        }
+        if [foo] in "foo" {
+          mutate { add_tag => "field in string" }
+        }
+        if "hello" in [greeting] {
+          mutate { add_tag => "string in field" }
+        }
+        if [foo] in ["hello", "world", "foo"] {
+          mutate { add_tag => "field in list" }
+        }
+        if [missing] in [alsomissing] {
+          mutate { add_tag => "shouldnotexist" }
+        }
+      }
+    CONFIG
+
+    sample("foo" => "foo", "foobar" => "foobar", "greeting" => "hello world") do
+      insist { subject["tags"] }.include?("field in field")
+      insist { subject["tags"] }.include?("field in string")
+      insist { subject["tags"] }.include?("string in field")
+      insist { subject["tags"] }.include?("field in list")
+      reject { subject["tags"] }.include?("shouldnotexist")
+    end
+  end
 end
--- a/spec/event.rb
+++ b/spec/event.rb
@ -45,6 +45,10 @@ describe LogStash::Event do
      insist { subject.sprintf("%{[j][k1]}") } == "v"
      insist { subject.sprintf("%{[j][k2][0]}") } == "w"
    end
+
+    it "should be able to take a non-string for the format" do
+      insist { subject.sprintf(2) } == "2"
+    end
  end
  
  context "#[]" do
@ -63,6 +67,14 @@ describe LogStash::Event do
      insist { subject['[j][5]'] } == 7

    end
+
+    it "should be fast?" do
+      2.times do
+        start = Time.now
+        100000.times { subject["[j][k1]"] }
+        puts "Duration: #{Time.now - start}"
+      end
+    end
  end

  context "#append" do
--- a/spec/examples/parse-haproxy-logs.rb
+++ b/spec/examples/parse-haproxy-logs.rb
@ -25,7 +25,7 @@ describe "haproxy httplog format" do
    # See http://rubydoc.info/gems/insist for more info.

    # Require that grok does not fail to parse this event.
-    reject { subject["@tags"] }.include?("_grokparsefailure")
+    insist { subject["tags"] }.nil?


    # Ensure that grok captures certain expected fields.
@ -71,45 +71,45 @@ describe "haproxy httplog format" do

 #    # Ensure that those fields match expected values from the event.

-    insist{ subject["syslog_timestamp"] } == ["Feb  6 12:14:14"]
-    insist{ subject["syslog_server"] } == ["localhost"]
-    insist{ subject["program"] } == ["haproxy"]
-    insist{ subject["pid"] } == ["14389"]
-    insist{ subject["client_ip"] } == ["10.0.1.2"]
-    insist{ subject["client_port"] } == ["33317"]
-    insist{ subject["accept_date"] } == ["06/Feb/2009:12:14:14.655"]
-    insist{ subject["haproxy_monthday"] } == ["06"]
-    insist{ subject["haproxy_month"] } == ["Feb"]
-    insist{ subject["haproxy_year"] } == ["2009"]
-    insist{ subject["haproxy_time"] } == ["12:14:14"]
-    insist{ subject["haproxy_hour"] } == ["12"]
-    insist{ subject["haproxy_minute"] } == ["14"]
-    insist{ subject["haproxy_second"] } == ["14"]
-    insist{ subject["haproxy_milliseconds"] } == ["655"]
-    insist{ subject["frontend_name"] } == ["http-in"]
-    insist{ subject["backend_name"] } == ["static"]
-    insist{ subject["server_name"] } == ["srv1"]
-    insist{ subject["time_request"] } == ["10"]
-    insist{ subject["time_queue"] } == ["0"]
-    insist{ subject["time_backend_connect"] } == ["30"]
-    insist{ subject["time_backend_response"] } == ["69"]
-    insist{ subject["time_duration"] } == ["109"]
-    insist{ subject["http_status_code"] } == ["200"]
-    insist{ subject["bytes_read"] } == ["2750"]
-    insist{ subject["captured_request_cookie"] } == ["-"]
-    insist{ subject["captured_response_cookie"] } == ["-"]
-    insist{ subject["termination_state"] } == ["----"]
-    insist{ subject["actconn"] } == ["1"]
-    insist{ subject["feconn"] } == ["1"]
-    insist{ subject["beconn"] } == ["1"]
-    insist{ subject["srvconn"] } == ["1"]
-    insist{ subject["retries"] } == ["0"]
-    insist{ subject["srv_queue"] } == ["0"]
-    insist{ subject["backend_queue"] } == ["0"]
-    insist{ subject["captured_request_headers"] } == ["1wt.eu"]
-    insist{ subject["http_verb"] } == ["GET"]
-    insist{ subject["http_request"] } == ["/index.html"]
-    insist{ subject["http_version"] } == ["1.1"]
+    insist{ subject["syslog_timestamp"] } == "Feb  6 12:14:14"
+    insist{ subject["syslog_server"] } == "localhost"
+    insist{ subject["program"] } == "haproxy"
+    insist{ subject["pid"] } == "14389"
+    insist{ subject["client_ip"] } == "10.0.1.2"
+    insist{ subject["client_port"] } == "33317"
+    insist{ subject["accept_date"] } == "06/Feb/2009:12:14:14.655"
+    insist{ subject["haproxy_monthday"] } == "06"
+    insist{ subject["haproxy_month"] } == "Feb"
+    insist{ subject["haproxy_year"] } == "2009"
+    insist{ subject["haproxy_time"] } == "12:14:14"
+    insist{ subject["haproxy_hour"] } == "12"
+    insist{ subject["haproxy_minute"] } == "14"
+    insist{ subject["haproxy_second"] } == "14"
+    insist{ subject["haproxy_milliseconds"] } == "655"
+    insist{ subject["frontend_name"] } == "http-in"
+    insist{ subject["backend_name"] } == "static"
+    insist{ subject["server_name"] } == "srv1"
+    insist{ subject["time_request"] } == "10"
+    insist{ subject["time_queue"] } == "0"
+    insist{ subject["time_backend_connect"] } == "30"
+    insist{ subject["time_backend_response"] } == "69"
+    insist{ subject["time_duration"] } == "109"
+    insist{ subject["http_status_code"] } == "200"
+    insist{ subject["bytes_read"] } == "2750"
+    insist{ subject["captured_request_cookie"] } == "-"
+    insist{ subject["captured_response_cookie"] } == "-"
+    insist{ subject["termination_state"] } == "----"
+    insist{ subject["actconn"] } == "1"
+    insist{ subject["feconn"] } == "1"
+    insist{ subject["beconn"] } == "1"
+    insist{ subject["srvconn"] } == "1"
+    insist{ subject["retries"] } == "0"
+    insist{ subject["srv_queue"] } == "0"
+    insist{ subject["backend_queue"] } == "0"
+    insist{ subject["captured_request_headers"] } == "1wt.eu"
+    insist{ subject["http_verb"] } == "GET"
+    insist{ subject["http_request"] } == "/index.html"
+    insist{ subject["http_version"] } == "1.1"
  end

 end
--- a/spec/examples/syslog.rb
+++ b/spec/examples/syslog.rb
@ -44,6 +44,5 @@ describe "parse syslog" do
    insist { subject.tags }.nil?
    insist { subject["syslog_pri"] } == "164"
    #insist { subject.timestamp } == "2012-10-26T15:19:25.000Z"
-    puts subject.to_hash
  end
 end
--- a/spec/filters/grok.rb
+++ b/spec/filters/grok.rb
@ -263,6 +263,34 @@ describe LogStash::Filters::Grok do
    end
  end

+  describe "grok on %{LOGLEVEL}" do
+    config <<-'CONFIG'
+      filter {
+        grok {
+          pattern => "%{LOGLEVEL:level}: error!"
+        }
+      }
+    CONFIG
+
+    log_level_names = %w(
+      trace Trace TRACE
+      debug Debug DEBUG
+      notice Notice Notice
+      info Info INFO
+      warn warning Warn Warning WARN WARNING
+      err error Err Error ERR ERROR
+      crit critical Crit Critical CRIT CRITICAL
+      fatal Fatal FATAL
+      severe Severe SEVERE
+      emerg emergency Emerg Emergency EMERG EMERGENCY
+    )
+    log_level_names.each do |level_name|
+      sample "#{level_name}: error!" do
+        insist { subject['level'] } == level_name
+      end
+    end
+  end
+
  describe "tagging on failure" do
    config <<-CONFIG
      filter {
@ -311,4 +339,35 @@ describe LogStash::Filters::Grok do
      insist { subject["foo-bar"] } == "hello"
    end
  end
+
+  describe "performance test" do
+    event_count = 100000
+    min_rate = 4000
+
+    max_duration = event_count / min_rate
+    input = "Nov 24 01:29:01 -0800"
+    config <<-CONFIG
+      input {
+        generator {
+          count => #{event_count}
+          message => "Mar 16 00:01:25 evita postfix/smtpd[1713]: connect from camomile.cloud9.net[168.100.1.3]"
+        }
+      }
+      filter {
+        grok {
+          match => [ "message", "%{SYSLOGLINE}" ]
+          singles => true
+          overwrite => [ "message" ]
+        }
+      }
+      output { null { } }
+    CONFIG
+
+    2.times do
+      agent do
+        puts "grok parse rate: #{event_count / @duration}"
+        insist { @duration } < max_duration
+      end
+    end
+  end
 end
--- a/spec/support/akamai-grok.rb
+++ b/spec/support/akamai-grok.rb
@ -20,7 +20,7 @@ describe "..." do
  line = '192.168.1.1 - - [25/Mar/2013:20:33:56 +0000] "GET /www.somewebsite.co.uk/dwr/interface/AjaxNewsletter.js HTTP/1.1" 200 794 "http://www.somewebsite.co.uk/money/index.html" "Mozilla/5.0 (Linux; U; Android 2.3.6; en-gb; GT-I8160 Build/GINGERBREAD) AppleWebKit/533.1 (KHTML, like Gecko) Version/4.0 Mobile Safari/533.1" "NREUM=s=1364243891214&r=589267&p=101913; __utma=259942479.284548354.1358973919.1363109625.1364243485.15; __utmb=259942479.4.10.1364243485; __utmc=259942479; __utmz=259942479.1359409342.3.3.utmcsr=investing.somewebsite.co.uk|utmccn=(referral)|utmcmd=referral|utmcct=/performance/overview/; asi_segs=D05509_10903|D05509_11337|D05509_11335|D05509_11341|D05509_11125|D05509_11301|D05509_11355|D05509_11508|D05509_11624|D05509_10784|D05509_11003|D05509_10699|D05509_11024|D05509_11096|D05509_11466|D05509_11514|D05509_11598|D05509_11599|D05509_11628|D05509_11681; rsi_segs=D05509_10903|D05509_11337|D05509_11335|D05509_11341|D05509_11125|D05509_11301|D05509_11355|D05509_11508|D05509_11624|D05509_10784|D05509_11003|D05509_10699|D05509_11024|D05509_11096|D05509_11466|D05509_11514|D05509_11598|D05509_11599|D05509_11628|D05509_11681|D05509_11701|D05509_11818|D05509_11850|D05509_11892|D05509_11893|D05509_12074|D05509_12091|D05509_12093|D05509_12095|D05509_12136|D05509_12137|D05509_12156|D05509_0; s_pers=%20s_nr%3D1361998955946%7C1364590955946%3B%20s_pn2%3D/money/home%7C1364245284228%3B%20s_c39%3D/money/home%7C1364245522830%3B%20s_visit%3D1%7C1364245693534%3B; s_sess=%20s_pn%3D/money/home%3B%20s_cc%3Dtrue%3B%20s_sq%3D%3B"'

  sample line do
-    puts subject["@timestamp"]
-    puts subject["timestamp"]
+    #puts subject["@timestamp"]
+    #puts subject["timestamp"]
  end
 end
--- a/spec/test_utils.rb
+++ b/spec/test_utils.rb
@ -25,33 +25,33 @@ end
 module LogStash
  module RSpec
    def config(configstr)
-      @config_str = configstr
+      let(:config) { configstr }
    end # def config

    def type(default_type)
-      @default_type = default_type
+      let(:default_type) { default_type }
    end
    
    def tags(*tags)
-      @default_tags = tags
+      let(:default_tags) { tags }
      puts "Setting default tags: #{@default_tags}"
    end

-    def sample(event, &block)
-      pipeline = LogStash::Pipeline.new(@config_str)
-
-      name = event.is_a?(String) ? event : event.to_json
+    def sample(sample_event, &block)
+      name = sample_event.is_a?(String) ? sample_event : sample_event.to_json
      name = name[0..50] + "..." if name.length > 50

      describe "\"#{name}\"" do
-        before :each do
-          # Coerce to an array of LogStash::Event
-          event = [event] unless event.is_a?(Array)
-          event = event.collect do |e| 
+        let(:pipeline) { LogStash::Pipeline.new(config) }
+        let(:event) do
+          sample_event = [sample_event] unless sample_event.is_a?(Array)
+          next sample_event.collect do |e|
            e = { "message" => e } if e.is_a?(String)
            next LogStash::Event.new(e)
          end
+        end

+        let(:results) do
          results = []
          count = 0
          pipeline.instance_eval { @filters.each(&:register) }
@ -66,19 +66,19 @@ module LogStash

          # TODO(sissel): pipeline flush needs to be implemented.
          #results += pipeline.flush
-          @results = results
-        end # before :all
+          next results
+        end
+
+        subject { results.length > 1 ? results: results.first }

-        subject { @results.length > 1 ? @results: @results.first }
        it("when processed", &block)
      end
    end # def sample

    def input(&block)
-      config_str = @config_str
      it "inputs" do
        queue = Queue.new
-        pipeline = LogStash::Pipeline.new(config_str)
+        pipeline = LogStash::Pipeline.new(config)
        #(class << pipeline; self; end).send(:define_method, :output) do |event|
          #p :event => event
          #queue << event
@ -94,11 +94,10 @@ module LogStash
      require "logstash/pipeline"

      # scoping is hard, let's go shopping!
-      config_str = @config_str
      describe "agent(#{@agent_count}) #{caller[1]}" do
        before :each do
          start = ::Time.now
-          pipeline = LogStash::Pipeline.new(config_str)
+          pipeline = LogStash::Pipeline.new(config)
          pipeline.run
          @duration = ::Time.now - start
        end
--- a/spec/util/fieldeval_spec.rb
+++ b/spec/util/fieldeval_spec.rb
@ -0,0 +1,44 @@
+require "test_utils"
+require "logstash/util/fieldreference"
+
+describe LogStash::Util::FieldReference, :if => true do
+  it "should permit simple key names" do
+    str = "hello"
+    m = eval(subject.compile(str))
+    data = { "hello" => "world" }
+    insist { m.call(data) } == data[str]
+  end
+
+  it "should permit [key][access]" do
+    str = "[hello][world]"
+    m = eval(subject.compile(str))
+    data = { "hello" => { "world" => "foo", "bar" => "baz" } }
+    insist { m.call(data) } == data["hello"]["world"]
+  end
+  it "should permit [key][access]" do
+    str = "[hello][world]"
+    m = eval(subject.compile(str))
+    data = { "hello" => { "world" => "foo", "bar" => "baz" } }
+    insist { m.call(data) } == data["hello"]["world"]
+  end
+  
+  it "should permit blocks" do
+    str = "[hello][world]"
+    code = subject.compile(str)
+    m = eval(subject.compile(str))
+    data = { "hello" => { "world" => "foo", "bar" => "baz" } }
+    m.call(data) { |obj, key| obj.delete(key) }
+
+    # Make sure the "world" key is removed.
+    insist { data["hello"] } == { "bar" => "baz" }
+  end
+
+  it "should permit blocks #2" do
+    str = "simple"
+    code = subject.compile(str)
+    m = eval(subject.compile(str))
+    data = { "simple" => "things" }
+    m.call(data) { |obj, key| obj.delete(key) }
+    insist { data }.empty?
+  end
+end
--- a/src/net/logstash/logstash.java
+++ b/src/net/logstash/logstash.java
@ -1,34 +0,0 @@
-/* This is the runner for logstash when it is packed up in a jar file. 
- * It exists to work around http://jira.codehaus.org/browse/JRUBY-6015
- */
-package net.logstash;
-import org.jruby.embed.ScriptingContainer;
-import org.jruby.embed.PathType;
-import org.jruby.CompatVersion;
-import java.io.InputStream;
-
-public class logstash {
-  private ScriptingContainer container;
-
-  public static void main(String[] args) {
-    // Malkovich malkovich? Malkovich!
-    logstash logstash = new logstash();
-    logstash.run(args);
-  } /* void main */
-
-
-  public logstash() {
-    this.container = new ScriptingContainer();
-    this.container.setCompatVersion(CompatVersion.RUBY1_9);
-  }
-
-  public void run(String[] args) {
-    final String script_path = "logstash/runner.rb";
-    ClassLoader loader = this.getClass().getClassLoader();
-    InputStream script = loader.getResourceAsStream(script_path);
-    //container.runScriptlet(PathType.RELATIVE, "logstash/runner.rb");
-    this.container.setArgv(args);
-    this.container.runScriptlet(script, script_path);
-  }
-}
-