How to install logstash ruby filer plugin

Elastic Stack Logstash
3

Thanks Magnus for your kind help. I was able to successfully install with logstash_plugin install logstash-ruby-filter and it is getting listed too.

However , I get this exception when I try to run the logstash, Please find below the config file

[2018-03-26T12:56:29,980][ERROR][logstash.agent ] Failed to execute action {:action=>LogStash::PipelineAction::Crea te/pipeline_id:main, :exception=>"LogStash::ConfigurationError", :message=>"Expected one of #, => at line 14, column 18 (byte 392) after filter {\n mutate {\n add_field => {\n "host_ip" => "localhost" # change the IP and set something meaningful\n }\n rename => { "message" => "rawdata" } # Put the content of th e message to the PNDA Avro 'rawdata' field\n ruby ", :backtrace=>["/opt/pnda/logstash-6.2.3/logstash-core/lib/logs tash/compiler.rb:42:in compile_imperative'", "/opt/pnda/logstash-6.2.3/logstash-core/lib/logstash/compiler.rb:50:incompile _graph'", "/opt/pnda/logstash-6.2.3/logstash-core/lib/logstash/compiler.rb:12:in block in compile_sources'", "org/jruby/Ruby Array.java:2486:inmap'", "/opt/pnda/logstash-6.2.3/logstash-core/lib/logstash/compiler.rb:11:in compile_sources'", "/opt/p nda/logstash-6.2.3/logstash-core/lib/logstash/pipeline.rb:51:ininitialize'", "/opt/pnda/logstash-6.2.3/logstash-core/lib/lo gstash/pipeline.rb:169:in initialize'", "/opt/pnda/logstash-6.2.3/logstash-core/lib/logstash/pipeline_action/create.rb:40:inexecute'", "/opt/pnda/logsta

 input {
      tcp {
         port => 20518
         add_field => [ "src", "syslog" ]
        }
    }

    filter {
        mutate {
            add_field => {
                "host_ip" => "localhost" # change the IP and set something meaningful
            }
            rename => { "message" => "rawdata" } # Put the content of the message to the PNDA Avro 'rawdata' field
            ruby {
                # Convert the Logstash timestamp to a milliseconds timestamp integer
                # You can use whatever makes sense to you as long as the timestamp is a valid timestamp integer in milliseconds
                code => "event.set('timestamp', (event.get('@timestamp').to_f * 1000).to_f * 1000).to_i)"
            }
        }
    }

    output {
            kafka {
                    codec => pnda-avro { schema_uri => "/opt/pnda/pnda.avsc" } # The PNDA Avro schema to use
                    bootstrap_servers => "localhost:9092"
                    topic_id => 'test'
                    compression_type => "none" # "none", "gzip", "snappy", "lz4"
                    value_serializer => 'org.apache.kafka.common.serialization.ByteArraySerializer'
            }
    }