lib/logstash/filters/aggregate.rb
def remove_expired_maps()
    events_to_flush = []
    default_min_timestamp = Time.now - @timeout
    default_min_inactivity_timestamp = Time.now - @inactivity_timeout

    @current_pipeline.mutex.synchronize do

      @logger.debug("Aggregate remove_expired_maps call with '#{@task_id}' pattern and #{@current_pipeline.aggregate_maps[@task_id].length} maps")

      @current_pipeline.aggregate_maps[@task_id].delete_if do |key, element|
        min_timestamp = element.timeout ? Time.now - element.timeout : default_min_timestamp
        min_inactivity_timestamp = element.inactivity_timeout ? Time.now - element.inactivity_timeout : default_min_inactivity_timestamp
        if element.creation_timestamp + element.difference_from_creation_to_now < min_timestamp || element.lastevent_timestamp + element.difference_from_lastevent_to_now < min_inactivity_timestamp
          if @push_previous_map_as_event || @push_map_as_event_on_timeout
            events_to_flush << create_timeout_event(element.map, key)
          end
          @logger.debug("Aggregate remove expired map with task_id=#{key}")
          metric.increment(:task_timeouts)
          next true
        end
        next false
      end
    end
    
    # disable check_expired_maps_on_every_flush if there is not anymore maps
    if @current_pipeline.aggregate_maps[@task_id].length == 0 && @check_expired_maps_on_every_flush
      @check_expired_maps_on_every_flush = nil
    end

    return events_to_flush
  end
# Remove the expired Aggregate maps from @current_pipeline.aggregate_maps if they are older than timeout or if no new event has been received since inactivity_timeout.
# If @push_previous_map_as_event option is set, or @push_map_as_event_on_timeout is set, expired maps are returned as new events to be flushed to Logstash pipeline.
Remove the expired Aggregate maps from @current_pipeline.aggregate_maps if they are older than timeout or if no new event has been received since inactivity_timeout.
If @push_previous_map_as_event option is set, or @push_map_as_event_on_timeout is set, expired maps are returned as new events to be flushed to Logstash pipeline.
Remove the expired Aggregate maps from @current_pipeline . aggregate_maps if they are older than timeout or if no new event has been received since inactivity_timeout . If @push_previous_map_as_event option is set or @push_map_as_event_on_timeout is set expired maps are returned as new events to be flushed to Logstash pipeline .
def remove_expired_maps()
    events_to_flush = []
    default_min_timestamp = Time.now - @timeout
    default_min_inactivity_timestamp = Time.now - @inactivity_timeout
    @current_pipeline.mutex.synchronize do
      @logger.debug("Aggregate remove_expired_maps call with '#{@task_id}' pattern and #{@current_pipeline.aggregate_maps[@task_id].length} maps")
      @current_pipeline.aggregate_maps[@task_id].delete_if do |key, element|
        min_timestamp = element.timeout ? Time.now - element.timeout : default_min_timestamp
        min_inactivity_timestamp = element.inactivity_timeout ? Time.now - element.inactivity_timeout : default_min_inactivity_timestamp
        if element.creation_timestamp + element.difference_from_creation_to_now < min_timestamp || element.lastevent_timestamp + element.difference_from_lastevent_to_now < min_inactivity_timestamp
          if @push_previous_map_as_event || @push_map_as_event_on_timeout
            events_to_flush << create_timeout_event(element.map, key)
          end
          @logger.debug("Aggregate remove expired map with task_id=#{key}")
          metric.increment(:task_timeouts)
          next true
        end
        next false
      end
    end
    if @current_pipeline.aggregate_maps[@task_id].length == 0 && @check_expired_maps_on_every_flush
      @check_expired_maps_on_every_flush = nil
    end
    return events_to_flush
  end
def remove_expired_maps ( ) events_to_flush = [ ] default_min_timestamp = Time . now - @timeout default_min_inactivity_timestamp = Time . now - @inactivity_timeout @current_pipeline . mutex . synchronize do @logger . debug ( "Aggregate remove_expired_maps call with '#{@task_id}' pattern and #{@current_pipeline.aggregate_maps[@task_id].length} maps" ) @current_pipeline . aggregate_maps [ @task_id ] . delete_if do | key , element | min_timestamp = element . timeout ? Time . now - element . timeout : default_min_timestamp min_inactivity_timestamp = element . inactivity_timeout ? Time . now - element . inactivity_timeout : default_min_inactivity_timestamp if element . creation_timestamp + element . difference_from_creation_to_now < min_timestamp || element . lastevent_timestamp + element . difference_from_lastevent_to_now < min_inactivity_timestamp if @push_previous_map_as_event || @push_map_as_event_on_timeout events_to_flush << create_timeout_event ( element . map , key ) end @logger . debug ( "Aggregate remove expired map with task_id=#{key}" ) metric . increment ( :task_timeouts ) next true end next false end end if @current_pipeline . aggregate_maps [ @task_id ] . length == 0 && @check_expired_maps_on_every_flush @check_expired_maps_on_every_flush = nil end return events_to_flush end
Remove the expired Aggregate maps from @current_pipeline.aggregate_maps if they are older than timeout or if no new event has been received since inactivity_timeout.
Remove the expired Aggregate maps from @current_pipeline . aggregate_maps if they are older than timeout or if no new event has been received since inactivity_timeout .
# disable check_expired_maps_on_every_flush if there is not anymore maps