Redis count getting filled up with unprocessed logs

[giveen]

I'm trying to figure out where this is coming from. From what I can tell its not seeing zero as a integer and so it becomes a invalid source port.

{ "convert":        { "field": "source.port",       "type": "integer",                     "ignore_missing": true  } }

[2023-02-02T14:50:08,355][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. status: 400, action: ["index", {:_id=>nil, :_index=>"so-zeek", :routing=>nil, :pipeline=>"zeek.conn"}, {"tags"=>["beats_input_codec_plain_applied"], "dataset"=>"conn", "ecs"=>{"version"=>"8.0.0"}, "message"=>"{\"ts\":1675345796.253667,\"uid\":\"CTHZ9O3FmONKSDaPZ2\",\"id.orig_h\":\"10.100.139.59\",\"id.orig_p\":0,\"id.resp_h\":\"17.253.31.205\",\"id.resp_p\":0,\"proto\":\"tcp\",\"duration\":0.00007891654968261719,\"orig_bytes\":0,\"resp_bytes\":0,\"conn_state\":\"OTH\",\"local_orig\":true,\"local_resp\":false,\"missed_bytes\":0,\"history\":\"R\",\"orig_pkts\":3,\"orig_ip_bytes\":120,\"resp_pkts\":0,\"resp_ip_bytes\":0,\"orig_mac_oui\":\"Cisco Systems, Inc\",\"community_id\":\"1:5+b0BDShOc/McV5IQWsj7TSlPF8=\"}", "agent"=>{"ephemeral_id"=>"ead6b64d-4da5-47b8-a4dd-0ab7d1cb78a8", "type"=>"filebeat", "id"=>"d10e123e-8b19-44de-b246-9224f3601e49", "version"=>"8.5.3", "name"=>"sorfh-lxp-04"}, "@timestamp"=>2023-02-02T13:50:02.084Z, "module"=>"zeek", "type"=>"redis-input", "category"=>"network", "host"=>{"name"=>"sorfh-lxp-04"}, "log"=>{"offset"=>1023436654, "file"=>{"path"=>"/nsm/zeek/logs/current/conn.log"}}, "metadata"=>{"beat"=>"filebeat", "type"=>"_doc", "version"=>"8.5.3", "ip_address"=>"10.21.132.23"}, "@version"=>"1"}], response: {"index"=>{"_index"=>"so-zeek", "_id"=>nil, "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"invalid source port [0]"}}}

[2023-02-02T14:50:09,041][WARN ][logstash.outputs.redis   ] Redis key size has hit a congestion threshold 50000000 suspending output for 1 seconds
[2023-02-02T14:50:09,829][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. status: 400, action: ["index", {:_id=>nil, :_index=>"so-zeek", :routing=>nil, :pipeline=>"zeek.conn"}, {"tags"=>["beats_input_codec_plain_applied"], "dataset"=>"conn", "ecs"=>{"version"=>"8.0.0"}, "message"=>"{\"ts\":1675345796.256351,\"uid\":\"CxeDb74auh1uCWTKJb\",\"id.orig_h\":\"10.100.139.59\",\"id.orig_p\":0,\"id.resp_h\":\"17.253.31.205\",\"id.resp_p\":0,\"proto\":\"tcp\",\"conn_state\":\"OTH\",\"local_orig\":true,\"local_resp\":false,\"missed_bytes\":0,\"history\":\"R\",\"orig_pkts\":3,\"orig_ip_bytes\":120,\"resp_pkts\":0,\"resp_ip_bytes\":0,\"orig_mac_oui\":\"Apple, Inc.\",\"community_id\":\"1:5+b0BDShOc/McV5IQWsj7TSlPF8=\"}", "agent"=>{"ephemeral_id"=>"1b2b1cf9-e590-415d-9cd1-67aa5ee11b20", "type"=>"filebeat", "id"=>"8a336e16-3dec-4c7f-800e-78c93b08699a", "version"=>"8.5.3", "name"=>"sohp-lxp-01"}, "@timestamp"=>2023-02-02T13:50:04.512Z, "module"=>"zeek", "type"=>"redis-input", "category"=>"network", "host"=>{"name"=>"sohp-lxp-01"}, "log"=>{"offset"=>864682081, "file"=>{"path"=>"/nsm/zeek/logs/current/conn.log"}}, "metadata"=>{"beat"=>"filebeat", "type"=>"_doc", "version"=>"8.5.3", "ip_address"=>"10.21.135.14"}, "@version"=>"1"}], response: {"index"=>{"_index"=>"so-zeek", "_id"=>nil, "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"invalid source port [0]"}}}
[2023-02-02T14:50:10,149][INFO ][org.logstash.beats.BeatsHandler] [local: 172.17.0.13:5044, remote: 10.25.223.9:32286] Handling exception: java.net.SocketException: Connection reset (caused by: java.net.SocketException: Connection reset)
[2023-02-02T14:50:10,149][WARN ][io.netty.channel.DefaultChannelPipeline] An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
java.net.SocketException: Connection reset
        at sun.nio.ch.SocketChannelImpl.throwConnectionReset(SocketChannelImpl.java:394) ~[?:?]
        at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:426) ~[?:?]
        at io.netty.buffer.PooledByteBuf.setBytes(PooledByteBuf.java:253) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1132) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:350) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:151) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) ~[netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) [netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [netty-all-4.1.65.Final.jar:4.1.65.Final]
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-all-4.1.65.Final.jar:4.1.65.Final]
        at java.lang.Thread.run(Thread.java:833) [?:?]

[giveen]

I ended up adding a zeek bpf of port 0 to this to stop this, still getting a full redis

[cm-ops]

Can you share your architecture, resources?

[palichx]

I have the same issue with source.port and destination.port:

invalid source port

[2023-03-17T00:18:10,323][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. status: 400, action: ["index", {:_id=>nil, :_index=>"so-zeek", :routing=>nil, :pipeline=>"zeek.conn"}, {"message"=>"{\"ts\":1679012202.939426,\"uid\":\"C2KqSR14ewmHtvuKBj\",\"id.orig_h\":\"0.0.0.0\",\"**id.orig_p\":0**,\"id.resp_h\":\"255.255.255.255\",\"id.resp_p\":0,\"proto\":\"udp\",\"duration\":23.736912965774536,\"orig_bytes\":1095,\"resp_bytes\":0,\"conn_state\":\"S0\",\"local_orig\":false,\"local_resp\":false,\"missed_bytes\":0,\"history\":\"D\",\"orig_pkts\":3,\"orig_ip_bytes\":1179,\"resp_pkts\":0,\"resp_ip_bytes\":0,\"orig_mac_oui\":\"Dell Inc.\",\"community_id\":\"1:Umv47YEUoUWvlS0s3Clq9jNmCLs=\"}", "agent"=>{"version"=>"8.6.2", "id"=>"fc9c598e-fb07-4ddd-a678-e7e6a0fa8da7", "type"=>"filebeat", "name"=>"node-10", "ephemeral_id"=>"01a247e0-0572-42cb-8f0f-ac48ad609f57"}, "type"=>"redis-input", "module"=>"zeek", "host"=>{"name"=>"node-10"}, "log"=>{"file"=>{"path"=>"/nsm/zeek/logs/current/conn.log"}, "offset"=>191594396}, "@Version"=>"1", "dataset"=>"conn", "category"=>"network", "ecs"=>{"version"=>"8.0.0"}, "@timestamp"=>2023-03-17T00:18:06.990Z, "metadata"=>{"version"=>"8.6.2", "beat"=>"filebeat", "type"=>"_doc", "ip_address"=>"10.0.115.160"}, "tags"=>["beats_input_codec_plain_applied"]}], response: {"index"=>{"_index"=>"so-zeek", "_id"=>nil, "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"invalid source port [0]"}}}

invalid destination port

[2023-03-17T00:26:27,776][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. status: 400, action: ["index", {:_id=>nil, :_index=>"so-zeek", :routing=>nil, :pipeline=>"zeek.conn"}, {"message"=>"{\"ts\":1679012782.465766,\"uid\":\"CQ8DjC4UmDBcSAvwTc\",\"id.orig_h\":\"175.31.254.92\",\"id.orig_p\":8716,\"id.resp_h\":\"xx.xx.xx.xx\",\"id.resp_p\":0,\"proto\":\"tcp\",\"conn_state\":\"S0\",\"local_orig\":false,\"local_resp\":true,\"missed_bytes\":0,\"history\":\"S\",\"orig_pkts\":1,\"orig_ip_bytes\":40,\"resp_pkts\":0,\"resp_ip_bytes\":0,\"orig_mac_oui\":\"Nokia\",\"community_id\":\"1:6PtCKFXHM1aJG3FmCtTsl73dPt0=\"}", "@timestamp"=>2023-03-17T00:26:28.242Z, "log"=>{"file"=>{"path"=>"/nsm/zeek/logs/current/conn.log"}, "offset"=>65589171}, "tags"=>["beats_input_codec_plain_applied"], "module"=>"zeek", "host"=>{"name"=>"node-11"}, "metadata"=>{"version"=>"8.6.2", "ip_address"=>"10.200.3.33", "beat"=>"filebeat", "type"=>"_doc"}, "agent"=>{"version"=>"8.6.2", "ephemeral_id"=>"46469f23-963b-4c37-8a4a-16988a4c0d51", "id"=>"0fab6ba2-cb54-4a23-adb2-0d525b776839", "type"=>"filebeat", "name"=>"node-11"}, "category"=>"network", "ecs"=>{"version"=>"8.0.0"}, "dataset"=>"conn", "@Version"=>"1", "type"=>"redis-input"}], response: {"index"=>{"_index"=>"so-zeek", "_id"=>nil, "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"invalid destination port [0]"}}}

Distributed deployment. Current version is 2.3.220

[giveen]

@palichx

Open your global.sls, this is how I resolved it.

zeek:
  bpf:
   - not port 0

[samanosuke26]

@giveen has there been any further interest in this? I'm also having this error. I know I can add a bpf but I would prefer to retain this data in case its useful to my analysts someday.

[giveen]

@giveen has there been any further interest in this? I'm also having this error. I know I can add a bpf but I would prefer to retain this data in case its useful to my analysts someday.

no update on this beyond what i had posted.

[cm-ops]

Would you check Zeek pipelines to see where it is failing? Looks like zeek.conn, but would probably also check zeek.common. You can check with so-elasticsearch-pipeline-stats zeek.conn and so-elasticsearch-pipeline-stats zeek.common.