initial commit

Author: ecwws

fluent-plugin-elasticsearch-timestamp-check.gemspec

@@ -0,0 +1,19 @@
+Gem::Specification.new do |spec|
+  spec.name          = "fluent-plugin-elasticsearch-timestamp-check"
+  spec.version       = "0.1.0"
+  spec.authors       = ["Richard Li"]
+  spec.email         = ["[email protected]"]
+  spec.description   = %q{fluent filter plugin to ensure @timestamp is in proper format}
+  spec.summary       = %q{fluent timestamp checker filter}
+  spec.homepage      = "https://github.com/ecwws/fluent-plugin-elasticsearch-timestamp-check"
+  spec.license       = "MIT"
+
+  spec.files         = `git ls-files`.split($/)
+  spec.executables   = spec.files.grep(%r{^bin/}) { |f| File.basename(f) }
+  spec.test_files    = spec.files.grep(%r{^(test|spec|features)/})
+  spec.require_paths = ["lib"]
+
+  spec.add_development_dependency "bundler", "~> 1.3"
+  spec.add_development_dependency "rake"
+  spec.add_development_dependency "rspec", ">= 3.0.0"
+end

lib/fluent/plugin/filter_elasticsearch_timestamp_check.rb

@@ -0,0 +1,36 @@
+module Fluent
+  class ElasticsearchTimestampCheckFilter < Filter
+    Fluent::Plugin.register_filter('elasticsearch_timestamp_check', self)
+
+    def configure(conf)
+      super
+      require 'date'
+    end
+
+    def start
+      super
+    end
+
+    def shutdown
+      super
+    end
+
+    def filter(tag, time, record)
+      existing = record['timestamp'] || record['@timestamp']
+      if existing
+        record['@timestamp'] =
+          DateTime.parse(existing).strftime('%Y-%m-%dT%H:%M:%S%z')
+        record['fluent_converted_timestamp'] =
+          DateTime.parse(existing).strftime('%Y-%m-%dT%H:%M:%S%z')
+        $log.debug("Timestamp parsed: #{record['@timestamp']}")
+      else
+        record['@timestamp'] = Time.now.strftime('%Y-%m-%dT%H:%M:%S%z')
+        record['fluent_added_timestamp'] =
+          Time.now.strftime('%Y-%m-%dT%H:%M:%S%z')
+        $log.debug("Timestamp added: #{record['@timestamp']}")
+      end
+      record
+    end
+  end
+end
+