Class: Fluent::Plugin::NetflowipfixInput

Inherits:
Input
  • Object
show all
Includes:
DetachMultiProcessMixin
Defined in:
lib/fluent/plugin/in_netflowipfix.rb,
lib/fluent/plugin/parser_netflow_v5.rb,
lib/fluent/plugin/parser_netflow_v9.rb,
lib/fluent/plugin/netflowipfix_records.rb

Defined Under Namespace

Classes: Header, IP4Addr, IP6Addr, MacAddr, MplsLabel, Netflow10Packet, Netflow5Packet, Netflow9Packet, OctetArray1, OctetArray2, Option10, Option9, ParserIPfixv10, ParserNetflowBase, ParserNetflowIpfix, ParserNetflowv5, ParserNetflowv9, ParserThread, PortConnection, Template10, Template9, UdpListenerThread

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 84

def configure(conf)
  super
  $log.debug "NetflowipfixInput::configure: #{@bind}:#{@port}"
  @@connections ||=  {}
  if @@connections.nil?
  end
  @@connections[@port] = PortConnection.new(@bind, @port, @tag, @cache_ttl, @definitions, @queuesleep, log)
  log.debug "NetflowipfixInput::configure NB=#{@@connections.length}"  
  @total = 0
end

#shutdown ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 110

def shutdown
  super
  $log.debug "NetflowipfixInput::shutdown NB=#{@@connections.length}"  
  if @@connections.nil?
  else
    @@connections.each do | port, conn |
      $log.debug "shutdown listening UDP on #{conn.bind}:#{conn.port}"
      conn.stop        
    end
    @@connections = nil
  end

end

#start ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 95

def start
  super
  
  $log.debug "NetflowipfixInput::start NB=#{@@connections.length}" 
  if @@connections.nil?
  else
    @@connections.each do | port, conn |
      $log.debug "start listening UDP on #{conn.bind}:#{conn.port}"
      conn.start       
    end
  end      
  
  waitForEvents
end

#waitForEvents ⇒ Object 



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/fluent/plugin/in_netflowipfix.rb', line 125

def waitForEvents
  loop do
      @@connections.each do | port, conn |
        if (conn.event_queue_length > 0) 
          $log.trace "waitForEvents: #{conn.bind}:#{conn.port} queue has #{conn.event_queue_length} elements"
          nbq = conn.event_queue_length 
          loop do
            ar = conn.event_pop     
            time = ar[0]
            record = ar[1]
            router.emit(conn.tag, EventTime.new(time.to_i), record)
            nbq = nbq - 1
            break if nbq == 0
          end 
        end
      end
      $log.trace "waitForEvents: sleep #{@queuesleep}"
      sleep(@queuesleep)

  end

end