# How to use user-defined plugin

**URL:** <https://discuss.elastic.co/t/how-to-use-user-defined-plugin/328224>\
**Category:** Logstash\
**Created:** [March 22, 2023, 9:24am UTC](https://discuss.elastic.co/t/how-to-use-user-defined-plugin/328224 "2023-03-22T09:24:06Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![wendywong0020](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/wendywong0020/32/118747_2.png) [@wendywong0020](https://discuss.elastic.co/u/wendywong0020)\
**Post date:** [March 22, 2023, 9:24am UTC](https://discuss.elastic.co/t/how-to-use-user-defined-plugin/328224/1 "2023-03-22T09:24:06Z")

</div>

logstash.conf:

```auto
input {
                file {
                                type => "_doc"
                                path => "/data/mysql_*_log/slow.log"
                                codec => multiline {
                                        pattern => "^# User@Host:"
                                                negate => true
                                                what => "previous"
                                                max_bytes => "100kib"
                                                auto_flush_interval => 5
                                }
                }
}

filter {
        if [type] == "_doc" {
                  if ("multiline_codec_max_bytes_reached" in [tags]) {
                        drop {}
                  }

                  grok {
# User@Host: logstash[logstash] @ localhost [127.0.0.1]
# User@Host: logstash[logstash] @ [127.0.0.1]
#patterns_dir => ["./patterns"]
                          match => ["message", "^# User@Host: (?<user_ignore>[\s\S]*)\[(?<user>[\s\S]*)\] @ (?<dbhost>[\s\S]*)? \[%{IP:clientip}\](\s+Id:\s+%{NUMBER:thread_id:int})?" ]
#match => ["message", "^# User@Host: %{ALLWORD:user}\[%{ALLWAORD}\] @ %{ALLWORD:dbhost}? \[%{IP:clientip}\]" ]
                  }
                  grok {
# Query_time: 102.413328 Lock_time: 0.000167 Rows_sent: 0 Rows_examined: 1970
                          match => ["message", "^# Query_time: %{NUMBER:Query_time:float}%{SPACE}Lock_time: %{NUMBER:lock_wait:float}%{SPACE}Rows_sent: %{NUMBER:Rows_sent:int}%{SPACE}Rows_examined:%{SPACE}%{NUMBER:Rows_examined:int}"]
                  }

# Capture the time the query happened
                  grok {
                          match => ["message", "^SET timestamp=%{NUMBER:time};%{SPACE}(?<sql>[\s\S]*)" ]
                  }

                  if "_grokparsefailure" in [tags] { drop {} }
                  if [sql] =~ "^call p_status.*" { drop{} }
                  mutate {
                        convert => { "time" => "integer" }
                  }

                  date {
                        match => ["time", "UNIX"]
                        remove_field => ["host", "message","user_ignore", "tags"]
                add_field => {"engine_type" => "mysqlslowlog" }
                  }
        }
    dbms_business {}
}

output {
  }
  stdout { codec =>rubydebug }
}

```

dbms\_business.rb:

```auto
# encoding: utf-8
require "logstash/filters/base"
require "logstash/namespace"
require "mysql2"

# This example filter will replace the contents of the default
# message field with whatever you specify in the configuration.
#
# It is only intended to be used as an example.
class LogStash::Filters::Dbms_business < LogStash::Filters::Base

   # Setting the config_name here is required. This is how you
   # configure this filter from your Logstash config.
   #
   # filter {
   # example {
   # message => "My message..."
   # }
   # }
   #
   config_name "dbms_business"

   # Replace the message with this value.
   config :message, :validate => :string, :default => "Hello World!"

   public
   def register
      # Add instance variables
      @last_ip_time = 0
      @last_ip = ''
      @port_to_businesses=Hash.new()
      @valid_time = 86400
   end # def register

   public
   def filter(event)
      return unless filter?(event)
      ts = Time.now.to_i
      if event.get("engine_type") == "mongodbslowlog"
         business,port,groupname,productname,developer,operator,developer_dw,operator_dw,test,model = get_mongodb_business(ts, event.get("path"))

         if ( business == "" || business == nil )
            event.cancel()
         end
         event.set("business_name",business)
         event.set("ip",get_ip_address(ts))
         event.set("port",port)
         event.set("group_name",groupname)
         event.set("product_name",productname)
         event.set("developer",developer)
         event.set("operator",operator)
         event.set("developer_dw",developer_dw)
         event.set("operator_dw",operator_dw)
         event.set("test",test)
         event.set("buisness_model",model)
         event.set("time",event.get("@timestamp").to_i * 1000)
      elsif event.get("engine_type") == "mysqlslowlog"
         p,ignore = event.get("path").split(File::SEPARATOR).last(2)
         port = p.split('_')[1].to_i

         ip = get_ip_address(ts)
         business,groupname,productname,developer,operator,developer_dw,operator_dw,test,model,password = get_mysql_business(ts, port)
         event.set("business_name",business)
         event.set("ip",get_ip_address(ts))
         event.set("port",port)
         event.set("group_name",groupname)
         event.set("product_name",productname)
         event.set("developer",developer)
         event.set("operator",operator)
         event.set("developer_dw",developer_dw)
         event.set("operator_dw",operator_dw)
         event.set("test",test)
         event.set("buisness_model",model)
         event.set("time",event.get("time") * 1000)
		 thread_id=event.get("thread_id")
		 if (thread_id == "" || thread_id == nil)
		    event.set("dbname","")
		 else 
		    event.set("dbname",get_db_from_thread(port,password,thread_id))
		 end
      end
      # correct debugging log statement for reference
      # using the event.get API
      @logger.debug? && @logger.debug("Message is now: #{event.get("message")}")
      # filter_matched should go in the last line of our successful code
      filter_matched(event)
   end # def filter

   def get_conf_value_from_file(file,key)
      configfile=File.new(file, 'r')
      while ( line = configfile.gets)
         vars = line.chomp.split('=')
         if ( vars.first == key )
            vvv = vars.last
            break
         end
      end
      configfile.close
      return vvv
   rescue => err
      return nil
   end

   ......
   
   def get_db_from_thread(port,password,thread_id)
        client = Mysql2::Client.new(
			:host => '127.0.0.1', 
			:port => port,
			:username => 'db_monitor', 
			:password => password,
			:database => 'mysql',
			:encoding => 'utf8'
	    )
		 
		result = client.query("select db from information_schema.processlist where id=#{thread_id};")
		result.each do |item|
		   if item["db"]
		      return item["db"]
		   else
              return ""		   
           end
		end   
   end 
   

end # class LogStash::Filters::Dbms_business

```

recently,i add block code "get\_db\_from\_thread" function and `require "mysql2"` ,then it does not work now. but before the added code,it worked well.

the reported error:

```auto
[2023-03-22T11:02:10,137][ERROR][logstash.agent] Cannot create pipeline {:reason=>"Couldn't find any filter plugin named 'dbms_business'. Are you sure this is correct? Trying to load the dbms_business filter plugin resulted in this error: Problems loading the requested plugin named dbms_business of type filter. Error: NameError NameError"}

```

i infered that require statement is wrong，but i don't know how to fix

thanks in advance

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 19, 2023, 9:24am UTC](https://discuss.elastic.co/t/how-to-use-user-defined-plugin/328224/2 "2023-04-19T09:24:45Z")

</div>

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