lib/logstash/filters/aggregate.rb
# This method is invoked by LogStash every 5 seconds.
This method is invoked by LogStash every 5 seconds.
This method is invoked by LogStash every 5 seconds .
def flush(options = {})
    @logger.trace("Aggregate flush call with #{options}")
    init_pipeline_timeout_management()
    if @current_pipeline.flush_instance_map[@task_id] == self && @current_pipeline.aggregate_maps[@task_id] && (!@current_pipeline.last_flush_timestamp_map.has_key?(@task_id) || Time.now > @current_pipeline.last_flush_timestamp_map[@task_id] + @inactivity_timeout / 2 || options[:final] || @check_expired_maps_on_every_flush)
      events_to_flush = remove_expired_maps()
      @current_pipeline.mutex.synchronize do
        if options[:final] && @push_previous_map_as_event && !@current_pipeline.aggregate_maps[@task_id].empty?
          events_to_flush << extract_previous_map_as_event()
        end
      end
      update_aggregate_maps_metric()
      if options[:final]
        events_to_flush.each { |event_to_flush| event_to_flush.tag("_aggregatefinalflush") }
      end
      @current_pipeline.last_flush_timestamp_map[@task_id] = Time.now
      return events_to_flush
    else
      return []
    end
  end
def flush ( options = { } ) @logger . trace ( "Aggregate flush call with #{options}" ) init_pipeline_timeout_management ( ) if @current_pipeline . flush_instance_map [ @task_id ] == self && @current_pipeline . aggregate_maps [ @task_id ] && ( ! @current_pipeline . last_flush_timestamp_map . has_key? ( @task_id ) || Time . now > @current_pipeline . last_flush_timestamp_map [ @task_id ] + @inactivity_timeout / 2 || options [ :final ] || @check_expired_maps_on_every_flush ) events_to_flush = remove_expired_maps ( ) @current_pipeline . mutex . synchronize do if options [ :final ] && @push_previous_map_as_event && ! @current_pipeline . aggregate_maps [ @task_id ] . empty? events_to_flush << extract_previous_map_as_event ( ) end end update_aggregate_maps_metric ( ) if options [ :final ] events_to_flush . each { | event_to_flush | event_to_flush . tag ( "_aggregatefinalflush" ) } end @current_pipeline . last_flush_timestamp_map [ @task_id ] = Time . now return events_to_flush else return [ ] end end
This method is invoked by LogStash every 5 seconds.
This method is invoked by LogStash every 5 seconds .
# init flush/timeout properties for current pipeline # launch timeout management only every interval of (@inactivity_timeout / 2) seconds or at Logstash shutdown # at Logstash shutdown, if push_previous_map_as_event is enabled, it's important to force flush (particularly for jdbc input plugin) # tag flushed events, indicating "final flush" special event # update last flush timestamp # return events to flush into Logstash pipeline
options