[知识整理] Mysql Driver & 协议解析

曾经在负责Mysql时，那会觉得IO就应该是异步的，但是C# 版本Mysql Driver竟然只支持同步调用方式，简直不可接受哇。

然后就研究Mysql Driver，并改了一版异步的实现，并且做了一次小组分享： PPT，抓包分析 (2011/11/3)。

接下来为应对故障排查，也做了一版将Mysql 的Pcap抓包解析成为明文文本的工具，就是下文了。想想那会儿真是比较闲了，还能折腾点儿事，最近两年陷入业务不可自拔，荒废了。。。

Mysql的客户服服务器协议还是比较简单的，官方有详细的协议说明：http://dev.mysql.com/doc/internals/en/client-server-protocol.html

曾经想写协议解析工具，是为了mysql相关的问题排查和mysql请求统计分析，后来觉得重要性不高，没时间做、只完成request部分解析。

代码写了一部分，不过也能正常工作了，记录一下：

class MysqlQueryParser : MysqlParser
    {
        private enum ParsingState { PacketHead, Body }
        private StreamBuffer _buffer;
        private ParsingState _state;
        private MysqlQueryPacket _packet;
        public MysqlQueryParser()
        {
            SipcStack.Initialize();
            _buffer = new StreamBuffer();
            _state = ParsingState.PacketHead;
            _packet = new MysqlQueryPacket();
        }

        public override void Parse(byte[] data)
        {
            _buffer.Write(data);
            TryParse();
        }

        private void TryParse()
        {
            if (_state == ParsingState.PacketHead)
            {
                if (_buffer.Size >= 4)
                {
                    byte[] head = new byte[4];
                    _buffer.Read(head, 0, 4);
                    ReadPacketHeader(head, out _packet.PacketLength, out _packet.PacketNum);
                    _state = ParsingState.Body;
                }
                else
                    return;
            }

            if (_state == ParsingState.Body)
            {
                if (_buffer.Size >= _packet.PacketLength)
                {
                    byte[] body = new byte[_packet.PacketLength];
                    _buffer.Read(body, 0, _packet.PacketLength);
                    _packet.Body = body;
                    _packet.Command = (DBCmd)body[0];
                    HandleParseOk();
                }
            }
        }

        private void HandleParseError(string msg)
        {
            _buffer.Close();
            _state = ParsingState.PacketHead;
            _packet = new MysqlQueryPacket();

            if (OnParsedError != null)
                OnParsedError();
        }

        private void HandleParseOk()
        {
            if (OnParsedPacket != null)
            {
                _packet.Time = LastTime;
                _packet.SrcEP = LastSrcEP;
                _packet.DstEP = LastDstEP;
                OnParsedPacket(_packet);
            }
            _state = ParsingState.PacketHead;
            _packet = new MysqlQueryPacket();
        }
    }

  enum ResultSetType
    {
        OK,
        Error,
        ResultSet,
    }

    class MysqlResultSetPacket : MysqlPacket
    {
        public MysqlResultSetPacket()
        {
            PacketType = PacketType.ResultSet;
        }

        public ResultSetType ResultSetType { get; set; }
    }

    class MysqlOKResultSet : MysqlResultSetPacket
    {
        public MysqlOKResultSet()
        {
            ResultSetType = ResultSetType.OK;
        }
        public int PacketLength;
        public int PacketNum;
        public int AffectRow;
        public int ServerState;
        public string Message;
    }

    class MysqlOKResultSet : MysqlResultSetPacket
    {
        public MysqlOKResultSet()
        {
            ResultSetType = ResultSetType.Error;
        }
        public int PacketLength;
        public int PacketNum;
        public int ErrorNum;
        public int SqlState;
        public string Message;
    }

    class MysqlDataResultSet : MysqlResultSetPacket
    {
        public ResultSetHeadPacket HeadPacket;
        public List<ResultSetFieldPacket> FieldPacket;
        public List<ResultSetRowPacket> RowPacket;

        public MysqlDataResultSet()
        {
            ResultSetType = ResultSetType.ResultSet;
        }
    }

    public class ResultSetHeadPacket
    {
        public int PacketLength;
        public int PacketNum;
        public int FieldNum;
    }

    public class ResultSetFieldPacket
    {
        public int PacketLength;
        public int PacketNum;
    }

    public class ResultSetRowPacket
    {
        public int PacketLength;
        public int PacketNum;
    }

abstract class MysqlParser
    {
        public Action<MysqlPacket> OnParsedPacket;
        public  Action OnParsedError;

        public DateTime LastTime { get; set; }
        public IPEndPoint LastSrcEP { get; set; }
        public IPEndPoint LastDstEP { get; set; }

        public abstract void Parse(byte[] data);

        //The Packet Header
        //Bytes                 Name
        // -----                 ----
        // 3                     Packet Length
        // 1                     Packet Number

        // Packet Length: The length, in bytes, of the packet
        //                that follows the Packet Header. There
        //                may be some special values in the most
        //                significant byte. The maximum packet
        //                length is (2**24 -1),about 16MB.

        // Packet Number: A serial number which can be used to
        //                ensure that all packets are present
        //                and in order. The first packet of a
        //                client query will have Packet Number = 0
        //                Thus, when a new SQL statement starts,
        //                the packet number is re-initialised.

        protected void ReadPacketHeader(byte[] header,out int packetLength,out int packetNum)
        {
            if (header == null || header.Length != 4)
                throw new ArgumentException();

            packetLength = (int)header[0] + (((int)header[1]) << 8) + (((int)header[3]) << 16);
            packetNum = header[3];
        }
    }

class MysqlQueryPacket : MysqlPacket
    {
        public MysqlQueryPacket()
        {
            this.PacketType = PacketType.Query;
        }
        public int PacketLength;
        public int PacketNum;
        public byte[] Body;
        public DBCmd Command;

        private string _commandArgs;
        public string CommandArgs
        {
            get
            {
                if (Command == DBCmd.QUERY && Body != null && Body.Length > 0)
                {
                    if (_commandArgs == null)
                        _commandArgs = Encoding.UTF8.GetString(Body,1,Body.Length-1);
                }

                return _commandArgs;
            }
        }

        public MysqlCommandArgs ResolveCommand()
        {
            MysqlCommandArgs args = new MysqlCommandArgs();
            args.CmdType = CommandType.Text;
            if (CommandArgs != null)
            {
                if (CommandArgs.Substring(0, 4) == "CALL")
                {
                    args.CmdType = CommandType.StoredProcedure;

                    int start = CommandArgs.IndexOf("(");
                    if (start > 3)
                        args.ResolveQuery = CommandArgs.Substring(3, start - 3);
                }
                else
                {
                    //todo replace ='xx' to ={}
                }
            }

            return args;
        }
    }

    class MysqlCommandArgs
    {
        public CommandType CmdType { get; set; }
        public string ResolveQuery { get; set; }
    }

enum PacketType
   {
       Query,
       ResultSet,
   }
   class MysqlPacket
   {
       public DateTime Time { get; set; }
       public IPEndPoint SrcEP { get; set; }
       public IPEndPoint DstEP { get; set; }

       public PacketType PacketType { get; set; }
   }

enum DBCmd : byte
    {
        SLEEP = 0,
        QUIT = 1,
        INIT_DB = 2,
        QUERY = 3,
        FIELD_LIST = 4,
        CREATE_DB = 5,
        DROP_DB = 6,
        RELOAD = 7,
        SHUTDOWN = 8,
        STATISTICS = 9,
        PROCESS_INFO = 10,
        CONNECT = 11,
        PROCESS_KILL = 12,
        DEBUG = 13,
        PING = 14,
        TIME = 15,
        DELAYED_INSERT = 16,
        CHANGE_USER = 17,
        BINLOG_DUMP = 18,
        TABLE_DUMP = 19,
        CONNECT_OUT = 20,
        REGISTER_SLAVE = 21,
        PREPARE = 22,
        EXECUTE = 23,
        LONG_DATA = 24,
        CLOSE_STMT = 25,
        RESET_STMT = 26,
        SET_OPTION = 27,
        FETCH = 28
    }