lib/logstash/filters/aggregate.rb
def remove_expired_map_based_on_event_timestamp(task_id, event)

    @logger.debug("Aggregate remove_expired_map_based_on_event_timestamp call with task_id : '#{@task_id}'")

    # get aggregate map element
    element = @current_pipeline.aggregate_maps[@task_id][task_id]
    return nil if element.nil?

    init_pipeline_timeout_management()

    event_to_flush = nil
    event_timestamp = reference_timestamp(event)
    min_timestamp = element.timeout ? event_timestamp - element.timeout : event_timestamp - @timeout
    min_inactivity_timestamp = element.inactivity_timeout ? event_timestamp - element.inactivity_timeout : event_timestamp - @inactivity_timeout

    if element.creation_timestamp < min_timestamp || element.lastevent_timestamp < min_inactivity_timestamp
      if @push_previous_map_as_event || @push_map_as_event_on_timeout
        event_to_flush = create_timeout_event(element.map, task_id)
      end
      @current_pipeline.aggregate_maps[@task_id].delete(task_id)
      @logger.debug("Aggregate remove expired map with task_id=#{task_id}")
      metric.increment(:task_timeouts)
    end

    return event_to_flush
  end
# Remove the expired Aggregate map associated to task_id if it is older than timeout or if no new event has been received since inactivity_timeout (relative to current event timestamp).
# If @push_previous_map_as_event option is set, or @push_map_as_event_on_timeout is set, expired map is returned as new event to be flushed to Logstash pipeline.
Remove the expired Aggregate map associated to task_id if it is older than timeout or if no new event has been received since inactivity_timeout (relative to current event timestamp).
If @push_previous_map_as_event option is set, or @push_map_as_event_on_timeout is set, expired map is returned as new event to be flushed to Logstash pipeline.
Remove the expired Aggregate map associated to task_id if it is older than timeout or if no new event has been received since inactivity_timeout ( relative to current event timestamp ) . If @push_previous_map_as_event option is set or @push_map_as_event_on_timeout is set expired map is returned as new event to be flushed to Logstash pipeline .
def remove_expired_map_based_on_event_timestamp(task_id, event)
    @logger.debug("Aggregate remove_expired_map_based_on_event_timestamp call with task_id : '#{@task_id}'")
    element = @current_pipeline.aggregate_maps[@task_id][task_id]
    return nil if element.nil?
    init_pipeline_timeout_management()
    event_to_flush = nil
    event_timestamp = reference_timestamp(event)
    min_timestamp = element.timeout ? event_timestamp - element.timeout : event_timestamp - @timeout
    min_inactivity_timestamp = element.inactivity_timeout ? event_timestamp - element.inactivity_timeout : event_timestamp - @inactivity_timeout
    if element.creation_timestamp < min_timestamp || element.lastevent_timestamp < min_inactivity_timestamp
      if @push_previous_map_as_event || @push_map_as_event_on_timeout
        event_to_flush = create_timeout_event(element.map, task_id)
      end
      @current_pipeline.aggregate_maps[@task_id].delete(task_id)
      @logger.debug("Aggregate remove expired map with task_id=#{task_id}")
      metric.increment(:task_timeouts)
    end
    return event_to_flush
  end
def remove_expired_map_based_on_event_timestamp ( task_id , event ) @logger . debug ( "Aggregate remove_expired_map_based_on_event_timestamp call with task_id : '#{@task_id}'" ) element = @current_pipeline . aggregate_maps [ @task_id ] [ task_id ] return nil if element . nil? init_pipeline_timeout_management ( ) event_to_flush = nil event_timestamp = reference_timestamp ( event ) min_timestamp = element . timeout ? event_timestamp - element . timeout : event_timestamp - @timeout min_inactivity_timestamp = element . inactivity_timeout ? event_timestamp - element . inactivity_timeout : event_timestamp - @inactivity_timeout if element . creation_timestamp < min_timestamp || element . lastevent_timestamp < min_inactivity_timestamp if @push_previous_map_as_event || @push_map_as_event_on_timeout event_to_flush = create_timeout_event ( element . map , task_id ) end @current_pipeline . aggregate_maps [ @task_id ] . delete ( task_id ) @logger . debug ( "Aggregate remove expired map with task_id=#{task_id}" ) metric . increment ( :task_timeouts ) end return event_to_flush end
Remove the expired Aggregate map associated to task_id if it is older than timeout or if no new event has been received since inactivity_timeout (relative to current event timestamp).
Remove the expired Aggregate map associated to task_id if it is older than timeout or if no new event has been received since inactivity_timeout ( relative to current event timestamp ) .
# get aggregate map element
task_id
event