Issues when using logstash-filter-rest and logstash-netflow-codec together

[WolfangAukang]

I have a config file that uses the UDP input and the Netflow codec plugins. We have enabled a config file with the rest plugin today, but we received the following error:

 [2017-09-06T16:42:26,284][ERROR][logstash.inputs.udp      ] Exception in inputworker {"exception"=>#<RuntimeError: can't modify frozen array>, "backtrace"=>["org/jruby/RubyArray.java:2640:in `reject!'", "org/jruby/RubyArray.java:2653:in `delete_if'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-filter-rest-0.5.4/lib/logstash/filters/rest.rb:22:in `compact'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/struct.rb:168:in `each_pair'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:356:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:342:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:200:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:196:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:118:in `inputworker'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:89:in `udp_listener'"]}

The rest config works as expected, but the Netflow and UDP config crashed.

[gandalfb]

Please provide debug log and config.

And please use proper markdown code block to make reading more comfortable.

Thanks.

[WolfangAukang]

Thanks for answering @gandalfb and excuse me for the improper formatting.

On the logs, the error on the first comment appears two times. If you want more logs (debug ones, for example), the problem is that this is from a production server, where we integrated the plugin (we can't integrate both Netflow and Rest plugin together on the test server because the Netflow feed is huge), and there are a lot of logs to send (and some won't be useful for debugging this). What do you recommend in this case?

Here is the config file: Netflow input part

input {
  udp {
    port => 4430
    codec => netflow {
      versions => [5,9,10]
    }
    type => "cd_events"
  }
}

Rest filter part

filter {
#Source
  if [type] == "ab_events" and "" in [source_ip] {
    rest {
      request => {
        url => "https://api.det.xyz/2/api/%{source_ip}"
        auth => {
          user => "user"
          password => "pass"
        }
      }
      json => true                         # boolean (optional, default = true)
      target => "ABsource"                   # string (mandatory, no default)
    }
    if [ABsource][result][records][infoX] != [] {
      split {
        field => "[ABsource][result][records][infoX]"
        target => "infoXSource"
        add_tag => "infoXSourceSplit"
      }
    }
    mutate {
      remove_field => ["[ABsource][result][records][infoX]"]
    }
    if "" in [infoXSource][A] {
      mutate {
        add_field => { "source_A_address" => "%{[infoXSource][A]}" }
      }
    }
    if "" in [infoXSource][B] {
      mutate {
        add_field => { "source_hostB" => "%{[infoXSource][B]}" }
      }
    }
    if "" in [infoXSource][C] {
      mutate {
        add_field => { "source_C" => "%{[infoXSource][C]}" }
      }
    }
    mutate {
      remove_field => [ "[infoXSource]", "[ABsource]" ]
    }
  }
#Destination
  if [type] == "ab_events" and "" in [destination_ip] {
    rest {
      request => {
        url => "https://api.det.xyz/2/api/%{destination_ip}"
        auth => {
          user => "user"
          password => "pass"
        }
      }
      json => true                         # boolean (optional, default = true)
      target => "ABdestination"                   # string (mandatory, no default)
    }
    if [ABdestination][result][records][infoX] != [] {
      split {
        field => "[ABdestination][result][records][infoX]"
        target => "infoXDestination"
        add_tag => "infoXDestinationSplit"
      }
    }
    mutate {
      remove_field => ["[ABdestination][result][records][infoX]"]
    }
    if "" in [infoXDestination][A] {
      mutate {
        add_field => { "destination_A_address" => "%{[infoXDestination][A]}" }
      }
    }
    if "" in [infoXDestination][B] {
      mutate {
        add_field => { "destination_hostB" => "%{[infoXDestination][B]}" }
      }
    }
    if "" in [infoXDestination][C] {
      mutate {
        add_field => { "destination_C" => "%{[infoXDestination][C]}" }
      }
    }
    mutate {
      remove_field => [ "[infoXDestination]", "[ABdestination]" ]
    }
  }
}

The Rest filter never touches the Netflow input (this is what is intended to do).

[gandalfb]

Thank you. This really helps.

Wild guess would be the "monkey patch" of 'compact'.
this could interfere with other plugins, maybe.

is this reproducible in test with same config? You can reproduce events on the same machine via 'nc' or something. Without any live data.

As you mentioned the thing should be triggered anyway, even if the 'rest' section is not "processed"?

[WolfangAukang]

I couldn't reproduce the same on test, I didn't get any errors. I was also looking at the debug logs on production, and these are the only errors I get:

[2017-09-07T10:49:05,167][ERROR][logstash.codecs.json     ] JSON parse error, original data now in message field {:error=>#<LogStash::Json::ParserError: Unexpected character ('<' (code 60)): expected a valid value (number, String, array, object, 'true', 'false' or 'null')
[2017-09-07T10:49:05,749][ERROR][logstash.inputs.udp      ] Exception in inputworker {"exception"=>#<RuntimeError: can't modify frozen array>, "backtrace"=>["org/jruby/RubyArray.java:2640:in `reject!'", "org/jruby/RubyArray.java:2653:in `delete_if'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-filter-rest-0.5.4/lib/logstash/filters/rest.rb:22:in `compact'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/struct.rb:168:in `each_pair'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:356:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:342:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:200:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:196:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:118:in `inputworker'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:89:in `udp_listener'"]}
[2017-09-07T10:49:05,846][ERROR][logstash.inputs.udp      ] Exception in inputworker {"exception"=>#<RuntimeError: can't modify frozen array>, "backtrace"=>["org/jruby/RubyArray.java:2640:in `reject!'", "org/jruby/RubyArray.java:2653:in `delete_if'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-filter-rest-0.5.4/lib/logstash/filters/rest.rb:22:in `compact'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/struct.rb:168:in `each_pair'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:356:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:342:in `decode_netflow9'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:200:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/bindata-2.4.0/lib/bindata/array.rb:208:in `each'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-netflow-3.4.1/lib/logstash/codecs/netflow.rb:196:in `decode'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:118:in `inputworker'", "/opt/apps/elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-udp-3.1.1/lib/logstash/inputs/udp.rb:89:in `udp_listener'"]}

~~I attached the JSON error because the rest filter is the only one processing JSONs. Also, something that makes sense is that the JSON received contains arrays (on which I have to split later, as indicated on the config file), even though there is no error related to the split filter plugin.~~ There is another plugin that does it, which is the http_poller input plugin. It works fine with the Netflow plugin, though.