Module: LogStashLogger::Buffer
- Included in:
- Device::Connectable
- Defined in:
- lib/logstash-logger/buffer.rb
Overview
Implements a generic framework for accepting events which are later flushed in batches. Flushing occurs whenever :max_items
or :max_interval
(seconds) has been reached.
Including class must implement flush
, which will be called with all accumulated items either when the output buffer fills (:max_items
) or when a fixed amount of time (:max_interval
) passes.
batch_receive and flush
General receive/flush can be implemented in one of two ways.
batch_receive(event) / flush(events)
flush
will receive an array of events which were passed to buffer_receive
.
batch_receive('one')
batch_receive('two')
will cause a flush invocation like
flush(['one', 'two'])
batch_receive(event, group) / flush(events, group)
flush() will receive an array of events, plus a grouping key.
batch_receive('one', :server => 'a')
batch_receive('two', :server => 'b')
batch_receive('three', :server => 'a')
batch_receive('four', :server => 'b')
will result in the following flush calls
flush(['one', 'three'], {:server => 'a'})
flush(['two', 'four'], {:server => 'b'})
Grouping keys can be anything which are valid Hash keys. (They don’t have to be hashes themselves.) Strings or Fixnums work fine. Use anything which you’d like to receive in your flush
method to help enable different handling for various groups of events.
on_flush_error
Including class may implement on_flush_error
, which will be called with an Exception instance whenever buffer_flush encounters an error.
-
buffer_flush
will automatically re-try failed flushes, soon_flush_error
should not try to implement retry behavior. -
Exceptions occurring within
on_flush_error
are not handled bybuffer_flush
.
on_full_buffer_receive
Including class may implement on_full_buffer_receive
, which will be called whenever buffer_receive
is called while the buffer is full.
on_full_buffer_receive
will receive a Hash like {:pending => 30, :outgoing => 20}
which describes the internal state of the module at the moment.
final flush
Including class should call buffer_flush(:final => true)
during a teardown/shutdown routine (after the last call to buffer_receive) to ensure that all accumulated messages are flushed.
Instance Method Summary collapse
-
#buffer_flush(options = {}) ⇒ Fixnum
Try to flush events.
-
#buffer_full? ⇒ bool
Determine if
:max_items
has been reached. -
#buffer_initialize(options = {}) ⇒ Object
Initialize the buffer.
-
#buffer_receive(event, group = nil) ⇒ Object
Save an event for later delivery.
- #reset_buffer ⇒ Object
Instance Method Details
#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.
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(={}) force = [:force] || [:final] final = [:final] # final flush will wait for lock, so we are sure to flush out all buffered events if [: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 |
#buffer_full? ⇒ bool
Determine if :max_items
has been reached.
buffer_receive calls will block while buffer_full? == true
.
141 142 143 |
# File 'lib/logstash-logger/buffer.rb', line 141 def buffer_full? @buffer_state[:pending_count] + @buffer_state[:outgoing_count] >= @buffer_config[:max_items] end |
#buffer_initialize(options = {}) ⇒ Object
Initialize the buffer.
Call directly from your constructor if you wish to set some non-default options. Otherwise buffer_initialize will be called automatically during the first buffer_receive call.
Options:
-
:max_items, Max number of items to buffer before flushing. Default 50.
-
:max_interval, Max number of seconds to wait between flushes. Default 5.
-
:logger, A logger to write log messages to. No default. Optional.
-
:autoflush, Whether to immediately flush all inbound messages. Default true.
-
:drop_messages_on_flush_error, Whether to drop messages when there is a flush error. Default false.
-
:drop_messages_on_full_buffer, Whether to drop messages when the buffer is full. Default false.
85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 |
# File 'lib/logstash-logger/buffer.rb', line 85 def buffer_initialize(={}) if ! self.class.method_defined?(:flush) raise ArgumentError, "Any class including Stud::Buffer must define a flush() method." end @buffer_config = { :max_items => [:max_items] || 50, :max_interval => [:max_interval] || 5, :logger => [:logger] || nil, :autoflush => .fetch(:autoflush, true), :has_on_flush_error => self.class.method_defined?(:on_flush_error), :has_on_full_buffer_receive => self.class.method_defined?(:on_full_buffer_receive), :drop_messages_on_flush_error => .fetch(:drop_messages_on_flush_error, false), :drop_messages_on_full_buffer => .fetch(:drop_messages_on_full_buffer, false), :flush_at_exit => .fetch(:flush_at_exit, false) } if @buffer_config[:flush_at_exit] at_exit { buffer_flush(final: true) } end reset_buffer end |
#buffer_receive(event, group = nil) ⇒ Object
Save an event for later delivery
Events are grouped by the (optional) group parameter you provide. Groups of events, plus the group name, are later passed to flush
.
This call will block if :max_items
has been reached.
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 |
# File 'lib/logstash-logger/buffer.rb', line 157 def buffer_receive(event, group=nil) buffer_initialize if ! @buffer_state # block if we've accumulated too many events while buffer_full? do on_full_buffer_receive( :pending => @buffer_state[:pending_count], :outgoing => @buffer_state[:outgoing_count] ) if @buffer_config[:has_on_full_buffer_receive] if @buffer_config[:drop_messages_on_full_buffer] reset_buffer else sleep 0.1 end end @buffer_state[:pending_mutex].synchronize do @buffer_state[:pending_items][group] << event @buffer_state[:pending_count] += 1 end if @buffer_config[:autoflush] buffer_flush(force: true) end end |
#reset_buffer ⇒ Object
109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 |
# File 'lib/logstash-logger/buffer.rb', line 109 def reset_buffer reset_flush_timer_thread @buffer_state = { # items accepted from including class :pending_items => {}, :pending_count => 0, # guard access to pending_items & pending_count :pending_mutex => pending_mutex, # items which are currently being flushed :outgoing_items => {}, :outgoing_count => 0, # ensure only 1 flush is operating at once :flush_mutex => flush_mutex, # data for timed flushes :last_flush => Time.now, :timer => flush_timer_thread } # events we've accumulated buffer_clear_pending end |