in_cloudwatch_logs.rb in fluent-plugin-cloudwatch-logs-0.9.1

- old
+ new

@@ -1,6 +1,7 @@
 require 'date'
+require 'time'
 require 'fluent/plugin/input'
 require 'fluent/plugin/parser'
 require 'yajl'
 
 module Fluent::Plugin
@@ -24,10 +25,13 @@
     config_param :fetch_interval, :time, default: 60
     config_param :http_proxy, :string, default: nil
     config_param :json_handler, :enum, list: [:yajl, :json], default: :yajl
     config_param :use_todays_log_stream, :bool, default: false
     config_param :use_aws_timestamp, :bool, default: false
+    config_param :start_time, :string, default: nil
+    config_param :end_time, :string, default: nil
+    config_param :time_range_format, :string, default: "%Y-%m-%d %H:%M:%S"
 
     config_section :parse do
       config_set_default :@type, 'none'
     end
 
@@ -40,10 +44,16 @@
 
     def configure(conf)
       compat_parameters_convert(conf, :parser)
       super
       configure_parser(conf)
+
+      @start_time = (Time.strptime(@start_time, @time_range_format).to_f * 1000).floor if @start_time
+      @end_time = (Time.strptime(@end_time, @time_range_format).to_f * 1000).floor if @end_time
+      if @start_time && @end_time && (@end_time < @start_time)
+        raise Fluent::ConfigError, "end_time(#{@end_time}) should be greater than start_time(#{@start_time})."
+      end
     end
 
     def start
       super
       options = {}
@@ -161,9 +171,11 @@
     def get_events(log_stream_name)
       request = {
         log_group_name: @log_group_name,
         log_stream_name: log_stream_name
       }
+      request.merge!(start_time: @start_time) if @start_time
+      request.merge!(end_time: @end_time) if @end_time
       log_next_token = next_token(log_stream_name)
       request[:next_token] = log_next_token if !log_next_token.nil? && !log_next_token.empty?
       response = @logs.get_log_events(request)
       if valid_next_token(log_next_token, response.next_forward_token)
         store_next_token(response.next_forward_token, log_stream_name)