Kafka input convert to json

Elastic Stack Logstash
1

I have use input as kafka and my message is like

2020-01-20 10:47:48: partition=0 offset=187965 machine=host1 queue=qhost1 comman=1298316293 action=REMOVE

I want this to be separated out. how do I use filter

input {
   kafka {
      bootstrap_servers => "kaf001:9092"
      topics => "events.commands"
   }
}

filter {}

output   {
   stdout { codec => rubydubug }
}

I want output like

{
   mydate: 2020-01-20 10:47:48
   partition: 0
   offset:187965
   machine: host1
   queue: qhost1 
   command: 1298316293 
   action: REMOVE
}

but output has some kind of special character. and Looks like this at this time.

"message" => "\u0001\u000E\u0010qhost1����\t\fREMOVE\u0002�����[\u0000"

2

That looks to me as though you have the wrong encoding. You can specify this using the charset option on the codec for the input. That is, add something like

codec => plain { charset => "UTF-8" }

I do not think that text is UTF-8, I am just giving you an example.

Once your messages look good I would use dissect to take the prefix off the line and then a kv filter, similar to this.

3

ohh found out that they are using avro to put message in kafka and got avro schema file from them but now getting different error.

Error: negative length -1 given
  Exception: ArgumentError
  Stack: org/jruby/ext/stringio/StringIO.java:852:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:106:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:93:in `read_bytes'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:99:in `read_string'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:276:in `read_data'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:363:in `block in read_record'
org/jruby/RubyArray.java:1800:in `each'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:361:in `read_record'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:287:in `read_data'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/avro-1.9.1/lib/avro/io.rb:252:in `read'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-codec-avro-3.2.3-java/lib/logstash/codecs/avro.rb:77:in `decode'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.0.0-java/lib/logstash/inputs/kafka.rb:258:in `bk in thread_runner'
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.0.0-java/lib/logstash/inputs/kafka.rb:257:in `bk in thread_runner'
[FATAL] 2020-01-21 09:28:35.646 [LogStash::Runner] runner - An unexpected error occurred! {:error=>#<ArgumentError: negative len -1 given>
4

found more that they are encoding data. how do I decode it?

5

Team who is putting this message in log telling me to remove first character.
it does not make sense to me but i have to test and prove them

how do I remove very first character from message that I get ?

6

I use this, ignore last request
mutate { gsub => [ "message", "^.", ""] }

7

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