[MYSQL] 定制mysql中间件实现binlog过滤

create table db1.t20250307(id int);
insert into db1.t20250307 values(1);
insert into db1.t20250307 values(2);
insert into db1.t20250307 values(3);
insert into db1.t20250307 values(4);
commit;
flush logs; -- 我们从一个新的binlog开始, 主要是方便截图.
show master status;

# 启动中间件,监控3306端口
python3 t20250307.py  # 懒得想名字了, 就日期来命名吧...

# 使用mysqlbinlog模拟抽数工具
mysqlbinlog -vvv -h127.0.0.1 -P3306  -p123456 --stop-never --read-from-remote-server m3314.000089

begin;
insert into db1.t20250307 values(5);
delete from db1.t20250307 where id=4;
insert into db1.t20250307 values(6);
commit;

import struct
from threading import Thread
from multiprocessing import Process
import socket
import time
import sys
import ssl

def btoint(bdata,t='little'):
        return int.from_bytes(bdata,t)

def read_pack(rf):
	pack_header = rf.read(4)
	if len(pack_header) < 4:
		print(pack_header,' bye!')
		sys.exit(2)
	btrl, btrh, packet_seq = struct.unpack("<HBB", pack_header)
	pack_size = btrl + (btrh << 16)
	bdata = rf.read(pack_size)
	return pack_header+bdata


def send_repack(bdata,conn,seq,log_pos):
	seq = (seq+1)%255
	log_pos += len(bdata)
	bdata = bdata[:3] + struct.pack('<B',seq) + bdata[4:18] + struct.pack('<L',log_pos) + bdata[22:]
	conn.sendall(bdata)
	return seq,log_pos

class mmonitor(object): 
	def __init__(self):
		self.host = '0.0.0.0'
		self.port = 3306
		self.server = ('192.168.101.21',3314,)
		self.cert = '/data/mysql_3314/mysqldata/server-cert.pem'
		self.key = '/data/mysql_3314/mysqldata/server-key.pem'

	def handler_msg_fileter_delete(self,rf,sock,f):
		SEQ = 0
		LOG_POS = 0
		server_uuid = 866003314
		server_uuid = struct.pack('<L',server_uuid)
		table_map_packt = b''
		while True: # 一些客户端需要查询的信息
			bdata = read_pack(rf)
			sock.sendall(bdata)
			if bdata[4] == 0 and len(bdata) >= 25 and server_uuid == bdata[10:14]:
				bdata = read_pack(rf)
				SEQ = bdata[3] # struct.unpack('<B',bdata[3:4])
				LOG_POS = struct.unpack('<L',bdata[14:18])[0]
				sock.sendall(bdata) # GTID
				break
		
		while True:
			bdata = read_pack(rf)
			seq,_,timestamp,event_type,server_id,event_size,log_pos,flags = struct.unpack('<BBLBLLLh',bdata[3:24])
			#print(seq,_,timestamp,event_type,server_id,event_size,log_pos,flags)
			if event_type == 4: # 轮转了
				SEQ = seq
				LOG_POS = log_pos
				sock.sendall(bdata)
			elif event_type == 19: # table_map_event
				table_map_packt = bdata
			elif event_type != 32:
					if table_map_packt != b'':
						SEQ,LOG_POS = send_repack(table_map_packt,sock,SEQ,LOG_POS)
						table_map_packt = b''
					SEQ,LOG_POS = send_repack(bdata,sock,SEQ,LOG_POS)
		
	def handler_msg(self,rf,sock,f):
		#print(f'{f} start')
		while True:
			bdata = read_pack(rf)
			sock.sendall(bdata)
			print(f'{f}',btoint(bdata[3:4]),bdata)

	def handler(self,conn,addr):
		sock = socket.create_connection((self.server[0], self.server[1]))
		server_rf = sock.makefile('rb')
		bdata = read_pack(server_rf)
		conn.sendall(bdata)
		print('S->C: ',btoint(bdata[3:4]),bdata)

		client_rf = conn.makefile('rb')
		bdata = read_pack(client_rf)
		print('C->S: ',btoint(bdata[3:4]),bdata)
		sock.sendall(bdata)

		if len(bdata) < 38: #封装为SSL (32+4)
			#print('SSL')
			#封装客户端的SSL (因为相对于client, 这是server角色)
			context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
			context.load_cert_chain(certfile=self.cert, keyfile=self.key)
			conn = context.wrap_socket(conn, server_side=True)
			client_rf = conn.makefile('rb')

			#封装到server的SSL
			sock = ssl.wrap_socket(sock)
			server_rf = sock.makefile('rb')
		
		t1 = Process(target=self.handler_msg,args=(client_rf,sock,'C->S: ')) #监控客户端数据, 然后发往server端
		#t2 = Process(target=self.handler_msg,args=(server_rf,conn,'S->C: '))
		t2 = Process(target=self.handler_msg_fileter_delete,args=(server_rf,conn,'S->C: '))
		t1.start()
		t2.start()
		t1.join()
		t2.join()

	def init(self):
		socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		socket_server.bind((self.host, self.port))
		socket_server.listen(12345) #设置连接数
		self.socket_server = socket_server

		accept_client_thread = Thread(target=self.accept_client,)
		accept_client_thread.start()
		accept_client_thread.join()
		
	def accept_client(self,):
		while True:
			conn, addr = self.socket_server.accept()
			p = Process(target=self.handler,args=(conn,addr),)
			p.start()
	
aa = mmonitor()
aa.init()