Class: Fluent::Plugin::NetflowipfixInput::ParserThread

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/plugin/in_netflowipfix.rb

Overview

class UdpListenerThread

Instance Method Summary collapse

Constructor Details

#initialize(udpQueue, queuesleep, eventQueue, cache_ttl, definitions, log) ⇒ ParserThread

Returns a new instance of ParserThread.



247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 247

def initialize(udpQueue, queuesleep, eventQueue, cache_ttl, definitions, log)
  @udpQueue = udpQueue
  @queuesleep = queuesleep
  @eventQueue = eventQueue
  @log = log

  @parser_v5 = NetflowipfixInput::ParserNetflowv5.new
  @parser_v9 = NetflowipfixInput::ParserNetflowv9.new
  @parser_v10 = NetflowipfixInput::ParserIPfixv10.new

  @parser_v9.configure(cache_ttl, definitions)
  @parser_v10.configure(cache_ttl, definitions)
end

Instance Method Details

#close ⇒ Object 



265
266
267
268
269
270
271
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 265

def close
  # Garbage collection
  @parser_v5 = nil
  @parser_v9 = nil
  @parser_v10 = nil
  GC.start
end

#emit(time, event, host = nil) ⇒ Object

def run



318
319
320
321
322
323
324
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 318

def emit(time, event, host = nil)
  if !host.nil?
    event["host"] = host
  end
  @eventQueue << [time, event]
    @log.trace "ParserThread::emit #{@eventQueue.length}"
end

#join ⇒ Object 



273
274
275
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 273

def join
    @thread.join
end

#run ⇒ Object 



277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 277

def run
  loop do
    if @udpQueue.length == 0
      sleep(@queuesleep)

    else
      block = method(:emit)
      ar = @udpQueue.pop
      time = ar[0]
      msg = ar[1]
      payload = msg["message"]
      host = msg["sender"]
      
      version,_ = payload[0,2].unpack('n')
      @log.trace "ParserThread::pop #{@udpQueue.length} v#{version}"

      case version
        when 5          
          packet = NetflowipfixInput::Netflow5Packet.read(payload)
          @parser_v5.handle_v5(host, packet, block)
        when 9
          packet = NetflowipfixInput::Netflow9Packet.read(payload)
          @parser_v9.handle_v9(host, packet, block)
        when 10
          packet = NetflowipfixInput::Netflow10Packet.read(payload)
          @parser_v10.handle_v10(host, packet, block)
        else
          $log.warn "Unsupported Netflow version v#{version}: #{version.class}"
      end # case

      # Free up variables for garbage collection
      ar = @udpQueue.pop
      version = nil
      time = nil
      msg = nil
      payload = nil
      host = nil

    end
  end # loop do
end

#start ⇒ Object 



260
261
262
263
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 260

def start
  @thread = Thread.new(&method(:run))
  @log.debug "ParserThread::start"
end