Fluentd+mongodb实现log文件的过滤

 

Fluentd+mongodb 的安装不多说了，官网上很详细各种操作系统版本的都有。由于，我的项目中需要将多个log文件进行初步的过滤，所以，官网上提供的in_tail 插件不能够满足需求，需要基于in_tail插件写一个适合自己需求的插件。 

 

直接上代码：

 

module Fluent
    require 'fluent/plugin/in_tail'
	class LtAbcInput < Fluent::TailInput
	  Fluent::Plugin.register_input('lt_abc', self)

	  # Override the 'configure_parser(conf)' method.
	  # You can get config parameters in this method.
	  def configure_parser(conf)
		@time_format = conf['time_format'] || '%Y%m%d%H%M%S'
	  end

	  # Override the 'parse_line(line)' method that returns the time and record.
	  # This example method assumes the following log format:
	  #   %Y-%m-%d %H:%M:%S\tkey1\tvalue1\tkey2\tvalue2...
	  #   %Y-%m-%d %H:%M:%S\tkey1\tvalue1\tkey2\tvalue2...
	  #   ...
	  def parse_line(line)
		elements = line.split("|") //获取输入的每一行的信息

		time = elements.shift 
		time = Time.strptime(time, @time_format).to_i

		# [k1, v1, k2, v2, ...] -> {k1=>v1, k2=>v2, ...}
		record = {}
		
		mark = elements.shift
		record["mark"] = mark //这里是我的log中每行数据的类型
		
		while (p = elements.shift)
			pair = p.split(":")
			if (k = pair.shift) && (v = pair.shift)
				record[k] = v
			end
			
			if (p == "E")
				record["E"] = "E";
			end
			
			if (p == "N")
				record["N"] = "N";
			end
		end

		return time, record
	  end
	end
end

 

 

下面是log文件的格式

 

20131228182935|HEART|id:40354|obu:102198|type:0x1
20131228182935|GPS|id:40444|obu:104751|type:0|lat:363936|lon:1170118|speed:0.0000|direction:0|time:182935|E|N|mile:27647448
20131228182935|TIME SYNC|id:46576|obu:105450|termtime:1388255374:20131229022934
20131228182935|GPS|id:40886|obu:103611|type:0|lat:363819|lon:1165236|speed:0.0000|direction:0|time:182933|E|N|mile:3140237
20131228182935|GPS|id:50446|obu:104756|type:0|lat:363936|lon:1170135|speed:31.687|direction:90|time:182934|E|N|mile:12300444
20131228182935|GPS|id:41464|obu:103569|type:0|lat:364212|lon:1165313|speed:0.0000|direction:0|time:182934|E|N|mile:4211590
20131228182935|GPS|id:47602|obu:103170|type:0|lat:364309|lon:1170235|speed:13.575|direction:310|time:192249|E|N|mile:17175434
20131228182935|HEART|id:43991|obu:102578|type:0x1
20131228182935|GPS|id:40221|obu:104703|type:0|lat:364051|lon:1165917|speed:36.632|direction:181|time:182934|E|N|mile:938709
20131228182935|HEART|id:40303|obu:104036|type:0x1
20131228182935|INOUT|id:42517|obu:102431|io:0|auto^hand:1|route:00027|service:00w|station:00271048|time:182934