Method: LogStashLogger::Buffer#buffer_flush

Defined in:
lib/logstash-logger/buffer.rb

#buffer_flush(options = {}) ⇒ Fixnum

Try to flush events.

Returns immediately if flushing is not necessary/possible at the moment:

  • :max_items have not been accumulated

  • :max_interval seconds have not elapased since the last flush

  • another flush is in progress

buffer_flush(:force => true) will cause a flush to occur even if :max_items or :max_interval have not been reached. A forced flush will still return immediately (without flushing) if another flush is currently in progress.

buffer_flush(:final => true) is identical to buffer_flush(:force => true), except that if another flush is already in progress, buffer_flush(:final => true) will block/wait for the other flush to finish before proceeding.

Parameters:

  • options (Hash) (defaults to: {}) —

    Optional. May be {:force => true} or {:final => true}.

Returns:

  • (Fixnum) —

    The number of items successfully passed to flush.



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
 
# File 'lib/logstash-logger/buffer.rb', line 202

def buffer_flush(options={})
  force = options[:force] || options[:final]
  final = options[:final]

  # final flush will wait for lock, so we are sure to flush out all buffered events
  if options[:final]
    @buffer_state[:flush_mutex].lock
  elsif ! @buffer_state[:flush_mutex].try_lock # failed to get lock, another flush already in progress
    return 0
  end

  items_flushed = 0

  begin
    time_since_last_flush = (Time.now - @buffer_state[:last_flush])

    return 0 if @buffer_state[:pending_count] == 0
    return 0 if (!force) &&
       (@buffer_state[:pending_count] < @buffer_config[:max_items]) &&
       (time_since_last_flush < @buffer_config[:max_interval])

    @buffer_state[:pending_mutex].synchronize do
      @buffer_state[:outgoing_items] = @buffer_state[:pending_items]
      @buffer_state[:outgoing_count] = @buffer_state[:pending_count]
      buffer_clear_pending
    end

    @buffer_config[:logger].debug do
      debug_output = {
        :outgoing_count => @buffer_state[:outgoing_count],
        :time_since_last_flush => time_since_last_flush,
        :outgoing_events => @buffer_state[:outgoing_items],
        :batch_timeout => @buffer_config[:max_interval],
        :force => force,
        :final => final
      }
      "Flushing output: #{debug_output}"
    end if @buffer_config[:logger]

    @buffer_state[:outgoing_items].each do |group, events|
      begin
        if group.nil?
          flush(events,final)
        else
          flush(events, group, final)
        end

        @buffer_state[:outgoing_items].delete(group)
        events_size = events.size
        @buffer_state[:outgoing_count] -= events_size
        items_flushed += events_size
        @buffer_state[:last_flush] = Time.now

      rescue => e

        @buffer_config[:logger].warn do
          warn_output = {
            :outgoing_count => @buffer_state[:outgoing_count],
            :exception => e.class.name,
            :backtrace => e.backtrace
          }
          "Failed to flush outgoing items: #{warn_output}"
        end if @buffer_config[:logger]

        if @buffer_config[:has_on_flush_error]
          on_flush_error e
        end

        if @buffer_config[:drop_messages_on_flush_error]
          reset_buffer
        else
          cancel_flush
        end

      end
    end

  ensure
    @buffer_state[:flush_mutex].unlock
  end

  return items_flushed
end