Elasticsearch + logstash 获取存储 mysql 的 general_log 日志

亦凉 2022-11-15 14:12 143阅读 0赞

### logstash mysql general\_log 配置脚本 ###

*   *   *  实现功能
         *  general\_log
         *  实现步骤
         *  代码

### 实现功能 ###

解析 `general_log.log`

1.  过滤特定用户
2.  为每一条数据添加用户

### general\_log ###

1.  `general_log` 只记录了 `Time, Id, Command, Argument`，没有对应的用户
2.  `general_log` 会记录所有的操作，可能会出现一部分冗余的数据

/usr/sbin/mysqld, Version: 8.0.99 (MySQL Community Server - GPL). started with:
    Tcp port: 3306  Unix socket: /var/lib/mysql/mysql.sock
    Time                 Id Command    Argument
    2021-03-30T07:36:52.165750Z    10 Query begin
    2021-03-30T07:36:52.166089Z    10 Query commit
    2021-03-30T07:36:52.166737Z    10 Query commit
    2021-03-30T07:36:52.238762Z 41345 Connect   root@localhost on root using Socket

### 实现步骤 ###

1.  解析原日志，利用`grok`：`"%{TIMESTAMP_ISO8601:insert_time}\s+%{INT:thread_id}\s%{WORD:method}\s%{GREEDYDATA:sql}"`
2.  `[method] == "Connect"`，链接时存储用户信息：`mutate` 和 `ruby`；以`thread_id`为键，`user` 为值。
3.  过滤不需要统计的用户，这里是以`exclude1`或`exclude2`为用户名前缀的用户
4.  将用户信息写入要输出的对象中
5.  `[method] == "Quit"`，关闭连接时，删除对应的链接信息。**防止内存泄露**
6.  输出，如果有对应的用户的时候，才会输出（这里是输出到屏幕）

### 代码 ###

1.  日志

# 设置时区
    SET GLOBAL log_timestamps=SYSTEM;
    # 一般这个设置是默认的
    set global general_log_file='/var/lib/mysql/localhost.log';
    set global general_log=on;

添加权限：`chmod 777 /var/lib/mysql/localhost.log`  
2. 切割 `vim /etc/logrotate.d/mysql` 添加配置

/var/lib/mysql/localhost.log {
        daily
        rotate 5
        compress
        missingok
        notifempty
        copytruncate
        }

运行切割 `logrotate /etc/logrotate.d/mysql`  
查看状态 `vim /var/lib/logrotate/logrotate.status`  
3. logstash 配置 `vim /etc/logstash/conf.d/mysql.conf`

input{
        file {
            path => "/var/lib/mysql/localhost.log"
            type => general_log
            exclude => "*.gz"       #不想监听的文件规则，基于glob匹配语法
            start_position => "beginning"   #第一次丛头开始读取文件 beginning or end
            stat_interval => "3"    #定时检查文件是否更新，默认1s
        }
    }
    filter {
        grok {
            match => { "message" => "%{TIMESTAMP_ISO8601:insert_time}\s+%{INT:thread_id}\s%{WORD:method}\s%{GREEDYDATA:sql}"}
        }
        # 过滤不需要统计的用户，这里是以exclude1或exclude2为用户名前缀的用户
        if [method] == "Connect" and [sql] !~ /^exclude1\w*/ and [sql] !~ /^exclude2\w*/{
            mutate {
                 split => { "sql" => " " }
                 add_field => { "user" => "%{[sql][0]}" }
            }
            # 获取mysql连接和user的对应关系
            ruby {
               init => '
                   @@metadata = {}
               '
               code => '
                   thread_id = event.get("thread_id")
                   user = event.get("user")
                   @@metadata[thread_id] = user
               '
            }
            # 将sql重新组合起来
            mutate {
                join => {"sql" => " "}
            }
        }
        # 获取mysql连接和user的对应关系
        ruby {
            code => '
                thread_id = event.get("thread_id")
                user = @@metadata[thread_id]
                event.set("user", user)
                '
         }
         if [method] == "Quit" {
            ruby {
                code => '
                    thread_id = event.get("thread_id")
                    @@metadata.delete(thread_id){""}
                    '
            }
        }
        # 如果用户信息为空，则直接丢掉信息
        if ![user] {
            drop {}
        }
    }
    
    output {
        elasticsearch {
            hosts => ["http://127.0.0.1:9200"]
            index => "mysql_back_log-%{+YYYY.MM.dd}"
       }
    }