Class: LogStash::Codecs::Sflow
- Inherits:
-
Base
- Object
- Base
- LogStash::Codecs::Sflow
- Defined in:
- lib/logstash/codecs/sflow.rb
Overview
The “sflow” codec is for decoding sflow v5 flows.
Instance Method Summary collapse
-
#assign_key_value(event, bindata_kv) ⇒ Object
def initialize.
- #common_sflow(event, decoded, sample) ⇒ Object
- #decode(payload) ⇒ Object
-
#initialize(params = {}) ⇒ Sflow
constructor
A new instance of Sflow.
- #register ⇒ Object
- #snmp_call(event) ⇒ Object
Constructor Details
#initialize(params = {}) ⇒ Sflow
Returns a new instance of Sflow.
31 32 33 34 |
# File 'lib/logstash/codecs/sflow.rb', line 31 def initialize(params = {}) super(params) @threadsafe = false end |
Instance Method Details
#assign_key_value(event, bindata_kv) ⇒ Object
def initialize
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 |
# File 'lib/logstash/codecs/sflow.rb', line 38 def assign_key_value(event, bindata_kv) inspection_queue = [bindata_kv] while !inspection_queue.empty? kv = inspection_queue.shift() if kv.nil? || kv.to_s.eql?('') next end kv.each_pair do |k, v| if v.is_a?(BinData::Choice) inspection_queue.push(v) elsif !@removed_field.include?(k.to_s) && !v.is_a?(BinData::Array) event.set("#{k.to_s}", v.to_s) end end end end |
#common_sflow(event, decoded, sample) ⇒ Object
61 62 63 64 65 |
# File 'lib/logstash/codecs/sflow.rb', line 61 def common_sflow(event, decoded, sample) event.set('agent_ip', decoded['agent_ip'].to_s) assign_key_value(event, decoded) assign_key_value(event, sample) end |
#decode(payload) ⇒ Object
103 104 105 106 107 108 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 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 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 183 184 185 186 187 188 189 190 191 |
# File 'lib/logstash/codecs/sflow.rb', line 103 def decode(payload) header = SFlowHeader.read(payload) unless @versions.include?(header.sflow_version) @logger.warn("Ignoring Sflow version v#{header.sflow_version}") return end decoded = SFlow.read(payload) events = [] decoded['samples'].each do |sample| @logger.debug("sample: #{sample}") #Treat case with no flow decoded (Unknown flow) if sample['sample_data'].to_s.eql? '' @logger.warn("Unknown sample entreprise #{sample['sample_entreprise'].to_s} - format #{sample['sample_format'].to_s}") next end #treat sample flow and expanded sample flow if sample['sample_entreprise'] == 0 && (sample['sample_format'] == 1 || sample['sample_format'] == 3) # Create the logstash event event = LogStash::Event.new({}) common_sflow(event, decoded, sample) sample['sample_data']['records'].each do |record| # Ensure that some data exist for the record if record['record_data'].to_s.eql? '' @logger.warn("Unknown sample_flow record: entreprise #{record['record_entreprise'].to_s}, format #{record['record_format'].to_s}") next end assign_key_value(event, record) end #compute frame_length_times_sampling_rate if event.include?('frame_length') and event.include?('sampling_rate') event.set('frame_length_times_sampling_rate', event.get('frame_length').to_i * event.get('sampling_rate').to_i) end if sample['sample_format'] == 1 event.set('sflow_type', 'flow_sample') else event.set('sflow_type', 'expanded_flow_sample') end #Get interface dfescr if snmp_interface true snmp_call(event) events.push(event) #treat counter flow and expanded counter flow elsif sample['sample_entreprise'] == 0 && (sample['sample_format'] == 2 || sample['sample_format'] == 4) sample['sample_data']['records'].each do |record| # Ensure that some data exist for the record if record['record_data'].to_s.eql? '' @logger.warn("Unknown counter_flow record: entreprise #{record['record_entreprise'].to_s}, format #{record['record_format'].to_s}") next end # Create the logstash event event = LogStash::Event.new({}) common_sflow(event, decoded, sample) assign_key_value(event, record) if sample['sample_format'] == 2 event.set('sflow_type', 'counter_sample') else event.set('sflow_type', 'expanded_counter_sample') end #Get interface dfescr if snmp_interface true snmp_call(event) events.push(event) end end end events.each do |event| yield event end rescue BinData::ValidityError, EOFError, IOError, RangeError => e @logger.warn("Invalid sflow packet received (#{e})") # Sflow.instance_variables.each { |ivar| puts "#{ivar}: #{Sflow.instance_variable_get(ivar)}" } end |
#register ⇒ Object
87 88 89 90 91 92 93 94 95 96 97 98 |
# File 'lib/logstash/codecs/sflow.rb', line 87 def register require 'logstash/codecs/sflow/datagram' require 'logstash/codecs/snmp/interface_resolver' # noinspection RubyResolve @removed_field = %w(record_length record_count record_entreprise record_format sample_entreprise sample_format sample_length sample_count sample_header data storage) | @optional_removed_field if @snmp_interface @snmp = SNMPInterfaceResolver.new(@snmp_community, @interface_cache_size, @interface_cache_ttl, @logger) end end |
#snmp_call(event) ⇒ Object
67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
# File 'lib/logstash/codecs/sflow.rb', line 67 def snmp_call(event) if @snmp_interface if event.include?('source_id_type') and event.get('source_id_type').to_s == '0' if event.include?('source_id_index') event.set('source_id_index_descr', @snmp.get_interface(event.get('agent_ip'), event.get('source_id_index'))) end if event.include?('input_interface') event.set('input_interface_descr', @snmp.get_interface(event.get('agent_ip'), event.get('input_interface'))) end if event.include?('output_interface') event.set('output_interface_descr', @snmp.get_interface(event.get('agent_ip'), event.get('output_interface'))) end if event.include?('interface_index') event.set('interface_index_descr', @snmp.get_interface(event.get('agent_ip'), event.get('interface_index'))) end end end end |