1. 首先安装logstash
   上传或者下载后直接执行

wget  http://dba.corp.shiqiao.com/static/download/logstash-7.7.1.rpm
rpm -ivh  logstash-7.7.1.rpm

配置文件修改

vim /etc/logstash/logstash.yml
path.data: /var/lib/logstash
path.logs: /var/log/logstash
path.config: /etc/logstash/conf.d/slowlog_y.conf

cat /etc/logstash/logstash.yml | grep path.data:
cat /etc/logstash/logstash.yml | grep path.config:
cat /etc/logstash/logstash.yml | grep path.logs:

其中主要修改path.config:

mysql慢日志配置文件
vi /etc/logstash/conf.d/slowlog.conf
slowlog.conf 内容

input {

        file {
                path => "/data/mysql3354/log/mysql-slow.log"
                type => "mysql_slowlog"
                start_position => "beginning"
                stat_interval => "2"

                add_field => {
                        "mysql_ip" => "1.2.2.2"
                        "port" => "3354"
                        "node_id" => "1480"
                }
                codec => multiline {
                  pattern => "^#\ Time:"
                  negate  => true
                  what    => "previous"
                }
        }

        file {
                path => "/data/mysql3354/log/mysql-error.log"
                type => "mysql_errorlog"
                start_position => "beginning"
                stat_interval => "2"

                add_field => {
                        "mysql_ip" => "1.2.2.2"
                        "port" => "3354"
                        "node_id" => "1480"
                }
                codec => multiline {
                  pattern => "^%{TIMESTAMP_ISO8601}"
                  negate  => true
                  what    => "previous"
                }
        }



}


filter {
        if [type] == "mysql_slowlog" {
                mutate{
                        remove_field =>["host"]
                        }
                grok {
                        match => { "message" => "SELECT SLEEP" }
                        add_tag => [ "sleep_drop" ]
                        tag_on_failure => []
                }
                if "sleep_drop" in [tags] {
                        drop {}
                }
                grok {
                        match => [ "message", "(?m)^# Time:.*\s+# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s*Id: %{NUMBER:id:int}\s+# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)$" ]
                        remove_field => [ "message" ]
                }
                date {
                        match => [ "timestamp", "UNIX" ]
                        remove_field => [ "timestamp" ]
                }
        }

        if [type] == "mysql_errorlog" {
                 mutate{
                        remove_field =>["host"]
                        }

                grok {
                        match => [ "message", "(?m)^%{TIMESTAMP_ISO8601:start_time} %{NUMBER:error_code} \[%{WORD:log_type}\] %{GREEDYDATA:info}$" ]
                        remove_field => [ "message" ]
                }
        }

}




output {
        if [type] == "mysql_slowlog" {
                elasticsearch {
                    user => "user"
                    password => "123456"
                    hosts => ["1.1.1.1:9200","1.1.1.2:9200"]
                    index => ["allslowlog-%{+YYYY.MM}"]
                }
        }


        if [type] == "mysql_errorlog" {
                elasticsearch {
                    user => "user"
                    password => "123456"
                    hosts => ["1.1.1.1:9200","1.1.1.2:9200"]
                    index => ["mysql_errorlog-%{+YYYY.MM}"]
                }
        }
}

由于服务启动方式总有问题 ，可以采用直接指定配置文件启动方式
nohup /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/slowlog_y.conf &