introduce log types, in addition to tags

etc/logstash-standalone.yaml

@@ -5,29 +5,29 @@
 #
 inputs:
 # Give a list of inputs. Tag them for easy query/filter later.
-  linux-syslog: # this is the 'linux-syslog' tag
+  linux-syslog: # this is the 'linux-syslog' type
   - /var/log/messages # watch /var/log/messages (uses eventmachine-tail)
   - /var/log/kern.log
   - /var/log/auth.log
   - /var/log/user.log
-  apache-access: # similar, different tag.
+  apache-access: # similar, different type.
   - /var/log/apache2/access.log
   - /b/access
   apache-error:
   - /var/log/apache2/error.log
 filters:
 - grok:
-    linux-syslog: # for logs tagged 'linux-syslog'
+    linux-syslog: # for logs of type 'linux-syslog'
       patterns:
       - %{SYSLOGLINE}
-    apache-access: # for logs tagged 'apache-error'
+    apache-access: # for logs of type 'apache-error'
       patterns:
       - %{COMBINEDAPACHELOG}
 - date:
-    linux-syslog:  # for logs tagged 'linux-syslog'
+    linux-syslog:  # for logs of type 'linux-syslog'
       # Look for a field 'timestamp' with this format, parse and it for the timestamp
       # This field comes from the SYSLOGLINE pattern
-      timestamp: %b %e %H:%M:%S
+      timestamp: "%b %e %H:%M:%S"
     apache-access:
       timestamp: "%d/%b/%Y:%H:%M:%S %Z"
 outputs:

lib/logstash/agent.rb

@@ -35,21 +35,19 @@ class LogStash::Agent
     if @config.include?("inputs")
       inputs = @config["inputs"]
       inputs.each do |value|
-        # If 'url' is an array, then inputs is a hash and the key is a tag
+        # If 'url' is an array, then inputs is a hash and the key is the type
         if inputs.is_a?(Hash)
-          tag, urls = value
+          type, urls = value
         else
-          tag = nil
-          urls = value
+          raise "config error, no type for url #{urls.inspect}"
         end
 
         # url could be a string or an array.
         urls = [urls] if !urls.is_a?(Array)
 
         urls.each do |url|
-          @logger.debug("Using input #{url} with tag #{tag}")
-          input = LogStash::Inputs.from_url(url) { |event| receive(event) }
-          input.tag(tag) if tag
+          @logger.debug("Using input #{url} of type #{type}")
+          input = LogStash::Inputs.from_url(url, type) { |event| receive(event) }
           input.register
           @inputs << input
         end

lib/logstash/event.rb

@@ -7,6 +7,7 @@ module LogStash; class Event
     @cancelled = false
     @data = {
       "@source" => "unknown",
+      "@type" => nil,
       "@tags" => [],
       "@fields" => {},
     }.merge(data)
@@ -38,6 +39,8 @@ module LogStash; class Event
   def source=(val); @data["@source"] = val; end # def source=
   def message; @data["@message"]; end # def message
   def message=; @data["@message"] = val; end # def message=
+  def type; @data["@type"]; end # def type
+  def type=(val); @data["@type"] = val; end # def type=
   def tags; @data["@tags"]; end # def tags
 
   # field-related access

lib/logstash/filters/date.rb

@@ -10,48 +10,45 @@ class LogStash::Filters::Date < LogStash::Filters::Base
   #
   # filters:
   #   date:
-  #     <tagname>:
+  #     <type>:
   #       <fieldname>: <format>
-  #     <tagname2>
+  #     <type>
   #       <fieldname>: <format>
   #
   # The format is whatever is supported by Ruby's DateTime.strptime
   def initialize(config = {})
     super
 
-    @tags = Hash.new { |h,k| h[k] = [] }
+    @types = Hash.new { |h,k| h[k] = [] }
   end # def initialize
 
   def register
-    @config.each do |tag, tagconfig|
-      @tags[tag] << tagconfig
+    @config.each do |type, typeconfig|
+      @logger.debug "Setting type #{type.inspect} to the config #{typeconfig.inspect}"
+      raise "date filter type \"#{type}\" defined more than once" unless @types[type].empty?
+      @types[type] = typeconfig
     end # @config.each
   end # def register
 
   def filter(event)
-    # TODO(sissel): crazy deep nesting here, refactor/redesign.
-    return if event.tags.empty?
-    event.tags.each do |tag|
-      next unless @tags.include?(tag)
-      @tags[tag].each do |tagconfig|
-        tagconfig.each do |field, format|
-          # TODO(sissel): check event.message, too.
-          if (event.fields.include?(field) rescue false)
-            fieldvalue = event.fields[field]
-            fieldvalue = [fieldvalue] if fieldvalue.is_a?(String)
-            fieldvalue.each do |value|
-              #value = event["fields"][field]
-              begin
-                time = DateTime.strptime(value, format)
-                event.timestamp = LogStash::Time.to_iso8601(time)
-                @logger.debug "Parsed #{value.inspect} as #{event.timestamp}"
-              rescue => e
-                @logger.warn "Failed parsing date #{value.inspect} from field #{field} with format #{format.inspect}. Exception: #{e}"
-              end
-            end # fieldvalue.each 
-          end # if this event has a field we expect to be a timestamp
-        end # tagconfig.each
-      end # @tags[tag].each
-    end # event.tags.each
+    @logger.debug "DATE FILTER: received event of type #{event.type}"
+    return unless @types.member?(event.type)
+    @types[event.type].each do |field, format|
+      @logger.debug "DATE FILTER: type #{event.type}, looking for field #{field.inspect} with format #{format.inspect}"
+      # TODO(sissel): check event.message, too.
+      if event.fields.member?(field)
+        fieldvalue = event.fields[field]
+        fieldvalue = [fieldvalue] if fieldvalue.is_a?(String)
+        fieldvalue.each do |value|
+          begin
+            time = DateTime.strptime(value, format)
+            event.timestamp = LogStash::Time.to_iso8601(time)
+            @logger.debug "Parsed #{value.inspect} as #{event.timestamp}"
+          rescue
+            @logger.warn "Failed parsing date #{value.inspect} from field #{field} with format #{format.inspect}: #{$!}"
+          end
+        end # fieldvalue.each 
+      end # if this event has a field we expect to be a timestamp
+    end # @types[event.type].each
   end # def filter
 end # class LogStash::Filters::Date

lib/logstash/filters/grok.rb

@@ -12,15 +12,15 @@ class LogStash::Filters::Grok < LogStash::Filters::Base
 
   def register
     # TODO(sissel): Make patterns files come from the config
-    @config.each do |tag, tagconfig|
-      @logger.debug("Registering tag with grok: #{tag}")
+    @config.each do |type, typeconfig|
+      @logger.debug("Registering type with grok: #{type}")
       pile = Grok::Pile.new
       pile.add_patterns_from_file("patterns/grok-patterns")
       pile.add_patterns_from_file("patterns/linux-syslog")
-      tagconfig["patterns"].each do |pattern|
+      typeconfig["patterns"].each do |pattern|
         pile.compile(pattern)
       end
-      @grokpiles[tag] = pile
+      @grokpiles[type] = pile
     end # @config.each
   end # def register
 
@@ -29,20 +29,14 @@ class LogStash::Filters::Grok < LogStash::Filters::Base
     message = event.message
     match = false
 
-    if !event.tags.empty?
-      event.tags.each do |tag|
-        if @grokpiles.include?(tag)
-          pile = @grokpiles[tag]
-          grok, match = pile.match(message)
-          break if match
-        end # @grokpiles.include?(tag)
-      end # event.tags.each
-    else 
+    if event.type
+      if @grokpiles.include?(event.type)
+        pile = @grokpiles[event.type]
+        grok, match = pile.match(message)
+      end # @grokpiles.include?(event.type)
       # TODO(2.0): support grok pattern discovery
-      #pattern = @grok.discover(message)
-      #@grok.compile(pattern)
-      #match = @grok.match(message)
-      @logger.info("No known tag for #{event.source} (tags: #{event.tags.inspect})")
+    else
+      @logger.info("Unknown type for #{event.source} (type: #{event.type})")
       @logger.debug(event.to_hash)
     end
 

lib/logstash/inputs.rb

@@ -3,7 +3,7 @@ require "logstash/namespace"
 require "uri"
 
 module LogStash::Inputs
-  def self.from_url(url, &block)
+  def self.from_url(url, type, &block)
     # Assume file paths if we start with "/"
     url = "file://#{url}" if url.start_with?("/")
 
@@ -13,6 +13,6 @@ module LogStash::Inputs
     klass = uri.scheme.capitalize
     file = uri.scheme
     require "logstash/inputs/#{file}"
-    LogStash::Inputs.const_get(klass).new(uri, &block)
+    LogStash::Inputs.const_get(klass).new(uri, type, &block)
   end # def from_url
 end # module LogStash::Inputs

lib/logstash/inputs/amqp.rb

@@ -4,21 +4,21 @@ require "mq" # rubygem 'amqp'
 require "uuidtools" # rubygem 'uuidtools'
 
 class LogStash::Inputs::Amqp < LogStash::Inputs::Base
-  TYPES = [ "fanout", "queue", "topic" ]
+  MQTYPES = [ "fanout", "queue", "topic" ]
 
-  def initialize(url, config={}, &block)
+  def initialize(url, type, config={}, &block)
     super
 
     @mq = nil
 
     # Handle path /<type>/<name>
-    unused, @type, @name = @url.path.split("/", 3)
-    if @type == nil or @name == nil
-      raise "amqp urls must have a path of /<type>/name where <type> is #{TYPES.join(", ")}"
+    unused, @mqtype, @name = @url.path.split("/", 3)
+    if @mqtype == nil or @name == nil
+      raise "amqp urls must have a path of /<type>/name where <type> is #{MQTYPES.join(", ")}"
     end
 
-    if !TYPES.include?(@type)
-      raise "Invalid type '#{@type}' must be one of #{TYPES.JOIN(", ")}"
+    if !MQTYPES.include?(@mqtype)
+      raise "Invalid type '#{@mqtype}' must be one of #{MQTYPES.JOIN(", ")}"
     end
   end
 
@@ -28,7 +28,7 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Base
     @target = nil
 
     @target = @mq.queue(UUIDTools::UUID.timestamp_create)
-    case @type
+    case @mqtype
       when "fanout"
         #@target.bind(MQ.fanout(@url.path, :durable => true))
         @target.bind(MQ.fanout(@url.path))
@@ -36,7 +36,7 @@ class LogStash::Inputs::Amqp < LogStash::Inputs::Base
         @target.bind(MQ.direct(@url.path))
       when "topic"
         @target.bind(MQ.topic(@url.path))
-    end # case @type
+    end # case @mqtype
 
     @target.subscribe(:ack => true) do |header, message|
       event = LogStash::Event.from_json(message)

lib/logstash/inputs/base.rb

@@ -4,12 +4,13 @@ require "logstash/logging"
 require "uri"
 
 class LogStash::Inputs::Base
-  def initialize(url, config={}, &block)
+  def initialize(url, type, config={}, &block)
     @logger = LogStash::Logger.new(STDERR)
     @url = url
     @url = URI.parse(url) if url.is_a? String
     @config = config
     @callback = block
+    @type = type
     @tags = []
   end
 
@@ -22,6 +23,7 @@ class LogStash::Inputs::Base
   end
 
   def receive(event)
+    event.type = @type
     event.tags |= @tags # set union
     @callback.call(event)
   end

lib/logstash/inputs/file.rb

@@ -3,7 +3,7 @@ require "eventmachine-tail"
 require "socket" # for Socket.gethostname
 
 class LogStash::Inputs::File < LogStash::Inputs::Base
-  def initialize(url, config={}, &block)
+  def initialize(url, type, config={}, &block)
     super
 
     # Hack the hostname into the url.
@@ -20,6 +20,7 @@ class LogStash::Inputs::File < LogStash::Inputs::Base
     event = LogStash::Event.new({
       "@source" => @url.to_s,
       "@message" => event,
+      "@type" => @type,
       "@tags" => @tags.clone,
     })
     @callback.call(event)