mysql slow query.

mydbslower.py

# Licensed under the Apache License, Version 2.0
#
# 15-Feb-2017   Sasha Goldshtein   Created this.
# 20-Apr-2021   Meng Zhou Edited this.
from bcc import BPF

# for influxdb
from init_db import influx_client
from db_modules import write2db

from datetime import datetime

# just for mysql 5.7
# 可以传入的参数。
# threshold 是 ms
threshold = 0
threshold_ns = threshold * 1000000

regex = "\\w+dispatch_command\\w+"
# get_user_functions_and_addresses 这个函数就很离谱。
# 这再让我想一辈子也想不出来啊。
symbols = BPF.get_user_functions_and_addresses("/usr/sbin/mysqld", regex)
if len(symbols) == 0:
    print("Can't find function 'dispatch_command' in %s" % ("/usr/sbin/mysqld"))
    exit(1)

# 这个获取函数的方式也很离谱hhh
# 相当离谱。。
(mysql_func_name, addr) = symbols[0]
print(symbols)
program = """
#include <uapi/linux/ptrace.h>

DEFINE_THRESHOLD

struct temp_t {
    u64 timestamp;
    char query[256];
};

struct data_t {
    u64 pid;
    u64 timestamp;
    u64 duration;
    char query[256];
};

BPF_HASH(temp, u64, struct temp_t);
BPF_PERF_OUTPUT(events);

int query_start(struct pt_regs *ctx) {
    u64 command  = (u64) PT_REGS_PARM3(ctx);
    if (command != 3) return 0; //Trace only packets with enum_server_command == COM_QUERY
    struct temp_t tmp = {};
    tmp.timestamp = bpf_ktime_get_ns();
    void* st = (void*) PT_REGS_PARM2(ctx);
    char* query;
    // 这边相当于走了两遍。我没明白。
    bpf_probe_read_user(&query, sizeof(query), st);
    bpf_probe_read_user(&tmp.query, sizeof(tmp.query), query);
    u64 pid = bpf_get_current_pid_tgid();
    temp.update(&pid, &tmp);
    return 0;
}

int query_end(struct pt_regs *ctx) {
    struct temp_t *tempp;
    u64 pid = bpf_get_current_pid_tgid();
    tempp = temp.lookup(&pid);
    if (!tempp)
        return 0;

    u64 delta = bpf_ktime_get_ns() - tempp->timestamp;

#ifdef THRESHOLD
    if (delta >= THRESHOLD) {
#endif //THRESHOLD
        struct data_t data = {};
        data.pid = pid >> 32;   // only process id
        data.timestamp = tempp->timestamp;
        data.duration = delta;

        bpf_probe_read_kernel(&data.query, sizeof(data.query), tempp->query);

        events.perf_submit(ctx, &data, sizeof(data));
#ifdef THRESHOLD
    }
#endif //THRESHOLD
    temp.delete(&pid);
    return 0;
}
""".replace("DEFINE_THRESHOLD",
            "#define THRESHOLD %d" % threshold_ns if threshold_ns > 0 else "")
print(program)
bpf = BPF(text=program)
bpf.attach_uprobe(name="/usr/sbin/mysqld", sym=mysql_func_name,
                  fn_name="query_start")
bpf.attach_uretprobe(name="/usr/sbin/mysqld", sym=mysql_func_name,
                     fn_name="query_end")

start = BPF.monotonic_time()

class lmp_data(object):
    def __init__(self, a,b,c,d,e):
        self.time = a
        self.glob = b
        self.pid = c
        self.duration = d
        self.query = e


data_struct = {"measurement": 'MySQLSlowQueryTable',
               "time": [],
               "tags": ['glob','pid'],
               "fields": ['duration', 'query']}


def print_event(cpu, data, size):
    event = bpf["events"].event(data)
    
    print("%-14.6f %-6d %8.3f %s" % (
        float(event.timestamp - start) / 1000000000,
        event.pid, float(event.duration) / 1000000, event.query))
    
    test_data = lmp_data(datetime.now().isoformat(), 'glob', event.pid,float(event.duration) / 1000000,event.query.decode('utf-8', 'replace'))
    
    write2db(data_struct, test_data, influx_client)  

# header
print("%-14s %-6s %8s %s" % ("TIME(s)", "PID", "MS", "QUERY"))

print("Tracing database queries for application %s slower than %d ms..." % ("mysql", 1))
bpf["events"].open_perf_buffer(print_event, page_cnt=64)
while 1:
    bpf.perf_buffer_poll()