How to read base64 encoded logs from kafka through logstash pipeline

Elastic Stack Logstash
1

Hello all,
I am working on something I have never worked on before and I really do not know where to go from here and I am hoping someone might have some direction for me to attempt trying to get this parsed to Elasticsearch properly.

The application team sending logs to kafka topics with base64 encoded. Now i want to decode that base64 encoded logs through logstash pipeline. Could you please guide me on this?

Thanks
Upendar

2

You will probably need a ruby filter. See these two threads.

3

I think you are looking for the cipher filter, it has a base64 option.

4

Hi Badger,
Thanks for replying!
I have tried below ruby filter
ruby { code => 'event.set("decoded", Base64.decode64(event.get("payload")).force_encoding("UTF-8"))' }

But i am getting below error.
[ERROR] 2023-12-21 00:30:03.194 [[pecspayload]>worker1] ruby - Ruby exception occurred: undefined method unpack1' for nil:NilClass {:class=>"NoMethodError", :backtrace=>["/usr/share/logstash/vendor/jruby/lib/ruby/stdlib/base64.rb:59:in decode64'", "(ruby filter code):2:in block in filter_method'", "/usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-ruby-3.1.8/lib/logstash/filters/ruby.rb:96:in inline_script'", "/usr/share/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-filter-ruby-3.1.8/lib/logstash/filters/ruby.rb:89:in filter'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:159:in do_filter'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:178:in block inmulti_filter'", "org/jruby/RubyArray.java:1865:in each'", "/usr/share/logstash/logstash-core/lib/logstash/filters/base.rb:175:in multi_filter'", "org/logstash/config/ir/compiler/AbstractFilterDelegatorExt.java:133:in multi_filter'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:304:in `block in start_workers'"]}

i am sharing my logstash configuration.
input
{
kafka
{
bootstrap_servers => "A,B,C"
topics => "XX_XX_XX"
group_id => "XX_XX_XX"
codec => json {}
security_protocol => "SASL_SSL"
sasl_mechanism => "SCRAM-SHA-512"
sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='XX'
password='XXXXXX';"
ssl_endpoint_identification_algorithm => ""
}
}
filter
{
ruby { code => 'event.set("decoded", Base64.decode64(event.get("payload")).force_encoding("UTF-8"))' }
}

5

I think that means event.get("payload") returned nil, i.e, the field does not exists. Perhaps should use event.get("message"), or whatever field is Base64 encoded.

6

Hi Badger,

I have used below ruby filter.

ruby { code => 'event.set("decoded", Base64.decode64(event.get("message")))' }

i am getting below result.
{"tags":["_jsonparsefailure"],"message":"\u001F�\b\u0000\u0000\u0000\

7

If you ever get a � then the encoding is wrong. If you know what the encoding should be then you can use .force_encoding to fix it.

If you do not know know what the encoding should be then it is unlikely that anyone will be able to answer if you ask about it.

8

Hi Badger,
Thanks for answering!
I have checked with application team they are doing GZIP and encoding with uTF-8.
Now how can i read logs through logstash.

9

Hi Team,
Any one please guide me on this?

Thanks

10

Hello upreddy,

The ruby code snippet works it's just your data that is not encoded in full ascii readble.

Have you tried to decode it any other way is it readable ?

Base64 encoding is made to ship data with ASCII charcaters, if you try to decode it then render the binary back to ascii you might have some surprises.....

I'm not sure what's your problem is here ?

11

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.