Merge pull request #34 from fankux/fix_auth

fix obmysql auth

CMakeLists.txt

@@ -4,10 +4,11 @@ project(oblogproxy CXX)
 macro(ob_define VAR DEFAULT)
     if (NOT DEFINED ${VAR})
         set(${VAR} ${DEFAULT})
-    endif()
+    endif ()
 endmacro()
 
 ob_define(OBLOGPROXY_RELEASEID 1)
+
 option(WITH_DEBUG "With debug symbols" ON)
 option(WITH_ASAN "Compile with AddressSanitizer" OFF)
 option(WITH_TEST "With Tests" OFF)
@@ -15,6 +16,7 @@ option(WITH_DEMO "With Demos" OFF)
 option(WITH_JNI_LIB "With oblogreader jni lib" OFF)
 option(WITH_GLOG "With google log" ON)
 option(WITH_DEPS "With precompiled deps" ON)
+option(WITH_LOGMSG "compiled logmsg from source files" OFF)
 option(USE_OBCDC_NS "With libobcdc" ON)
 option(USE_LIBOBLOG "With precompiled liboblog" OFF)
 option(USE_CXX11_ABI "Build with C++11 ABI" OFF)
@@ -31,9 +33,10 @@ SET(FIND_LIBOBLOG ON)
 if (USE_OBCDC_NS)
     SET(OBCDC_NAME "libobcdc")
     SET(OBCDC_NAME_VAR "-DUSE_OBCDC_NS")
-else()
+else ()
     SET(OBCDC_NAME "liboblog")
 endif ()
+
 if (WITH_DEPS)
     execute_process(
             COMMAND bash deps/dep_create.sh ${OMS_PROJECT_BUILD_PATH}/deps
@@ -172,14 +175,21 @@ add_library(PROTO_OBJS OBJECT ${PROTO_SRCS} ${PROTO_HDRS})
 message("protoc: ${PROTOBUF_PROTOC_EXECUTABLE}, proto srcs : ${PROTO_SRCS}")
 
 # oblogmsg
-SET(OBLOGMSG_MAPPING "-DLOGMSG_BY_LIBOBLOG=1 -DLogMsgLocalInit=\"if((_t_s_lmb=new(std::nothrow)LogMsgBuf())==nullptr){OMS_ERROR<<\\\"Failed to alloc LogMsgBuf\\\";stop();return;}\" -DLogMsgLocalDestroy=\"delete _t_s_lmb\"")
-SET(OBLOGMSG_INCLUDE_DIR ${LIBOBLOG_INCLUDE_PATH} ${LIBOBLOG_INCLUDE_PATH}/oblogmsg)
+set(LOGMSG_BY_LIBOBLOG_DEFINE "")
+if (LOGMSG_BY_LIBOBLOG)
+    set(LOGMSG_BY_LIBOBLOG_DEFINE "-DLOGMSG_BY_LIBOBLOG=1")
+    SET(OBLOGMSG_MAPPING "${LOGMSG_BY_LIBOBLOG_DEFINE} -DLogMsgLocalInit=\"if((_t_s_lmb=new(std::nothrow)LogMsgBuf())==nullptr){OMS_ERROR<<\\\"Failed to alloc LogMsgBuf\\\";stop();return;}\" -DLogMsgLocalDestroy=\"delete _t_s_lmb\"")
+    SET(OBLOGMSG_INCLUDE_DIR ${LIBOBLOG_INCLUDE_PATH} ${LIBOBLOG_INCLUDE_PATH}/oblogmsg)
 
-SET(OBLOGMSG_LIBRARIES ${LIBOBLOG_LIBRARIES})
-GET_FILENAME_COMPONENT(OBLOGMSG_LIB_DIR ${OBLOGMSG_LIBRARIES} DIRECTORY)
+    SET(OBLOGMSG_LIBRARIES ${LIBOBLOG_LIBRARIES})
+    GET_FILENAME_COMPONENT(OBLOGMSG_LIB_DIR ${OBLOGMSG_LIBRARIES} DIRECTORY)
 
-ADD_LIBRARY(oblogmsg STATIC IMPORTED GLOBAL)
-SET_PROPERTY(TARGET oblogmsg PROPERTY IMPORTED_LOCATION ${OBLOGMSG_LIBRARIES})
+    ADD_LIBRARY(oblogmsg STATIC IMPORTED GLOBAL)
+    SET_PROPERTY(TARGET oblogmsg PROPERTY IMPORTED_LOCATION ${OBLOGMSG_LIBRARIES})
+
+else ()
+    include(oblogmsg)
+endif ()
 
 # oblog
 if (FIND_LIBOBLOG)
@@ -310,7 +320,7 @@ target_include_directories(common PUBLIC ${COMMON_INC})
 
 if (USE_OBCDC_NS)
     set(DEP_OBCDC_LIB obcdc)
-else()
+else ()
     set(DEP_OBCDC_LIB oblog)
 endif ()
 # oblogreader
@@ -353,7 +363,7 @@ add_dependencies(logproxy logproxy_static)
 target_include_directories(logproxy PUBLIC ${DEP_INC} ${LOGPROXY_INC})
 target_link_directories(logproxy PUBLIC ${DEP_LIB_PATH})
 target_link_libraries(logproxy ${BASE_LIBS} ${DEP_OBCDC_LIB} ${DEP_LIBS})
-target_link_options(logproxy PUBLIC "${ASAN_LINK_OPTION}")
+target_link_options(logproxy PUBLIC -static-libstdc++ ${ASAN_LINK_OPTION})
 
 if (WITH_DEMO)
     # demo client
@@ -363,18 +373,17 @@ if (WITH_DEMO)
     target_include_directories(demo_client PUBLIC ${COMMON_INC})
     target_link_directories(demo_client PUBLIC ${DEP_LIB_PATH})
     target_link_libraries(demo_client libcommon.a ${DEP_LIBS})
-    target_link_libraries(demo_client libcommon.a ${DEP_LIBS})
-    target_link_options(demo_client PUBLIC ${ASAN_LINK_OPTION})
+    target_link_options(demo_client PUBLIC -static-libstdc++ ${ASAN_LINK_OPTION})
 endif ()
 
 if (WITH_TEST)
     # test_base
     file(GLOB TEST_BASE_SRC ./src/test/test_entry.cpp)
     add_executable(test_base ${TEST_BASE_SRC})
-    add_dependencies(test_base logproxy_static gtest)
+    add_dependencies(test_base common oblogmsg gtest)
     target_include_directories(test_base PUBLIC ${LOGPROXY_INC})
     target_link_directories(test_base PUBLIC ${DEP_LIB_PATH})
-    target_link_libraries(test_base ${BASE_LIBS} ${DEP_LIBS})
+    target_link_libraries(test_base libcommon.a ${DEP_LIBS})
     target_link_options(test_base PUBLIC ${ASAN_LINK_OPTION})
 
     # test_oblogreader

cmake/gtest.cmake

@@ -16,7 +16,7 @@ ExternalProject_Add(
         ${EXTERNAL_PROJECT_LOG_ARGS}
         DEPENDS gflags
         GIT_REPOSITORY "https://github.com/google/googletest.git"
-        GIT_TAG "release-1.6.0"
+        GIT_TAG "release-1.11.0"
         PREFIX ${GTEST_SOURCES_DIR}
         UPDATE_COMMAND ""
         CMAKE_ARGS -DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER}
@@ -31,7 +31,7 @@ ExternalProject_Add(
         -DCMAKE_INSTALL_LIBDIR=${GTEST_INSTALL_DIR}/lib
         -DCMAKE_POSITION_INDEPENDENT_CODE=ON
         -DBUILD_GMOCK=OFF
-        -Dbuild_gtest_samples=OFF
+        -Dgtest_build_samples=OFF
         -Dgtest_build_tests=OFF
         -DCMAKE_BUILD_TYPE=${THIRD_PARTY_BUILD_TYPE}
         -DCMAKE_PREFIX_PATH=${prefix_path}

cmake/oblogmsg.cmake

@@ -7,7 +7,7 @@ endif ()
 SET(OBLOGMSG_SOURCES_DIR ${THIRD_PARTY_PATH}/oblogmsg)
 SET(OBLOGMSG_DOWNLOAD_DIR "${OBLOGMSG_SOURCES_DIR}/src/extern_oblogmsg")
 SET(OBLOGMSG_INSTALL_DIR ${THIRD_PARTY_PATH}/install/oblogmsg)
-SET(OBLOGMSG_INCLUDE_DIR "${OBLOGMSG_INSTALL_DIR}/include" CACHE PATH "oblogmsg include directory." FORCE)
+SET(OBLOGMSG_INCLUDE_DIR "${OBLOGMSG_INSTALL_DIR}/include;${THIRD_PARTY_PATH}/install/oblogmsg/drcmessage" CACHE PATH "oblogmsg include directory." FORCE)
 
 SET(OBLOGMSG_LIB_DIR "${OBLOGMSG_INSTALL_DIR}/lib/" CACHE FILEPATH "oblogmsg library directory." FORCE)
 SET(OBLOGMSG_LIBRARIES "liboblogmsg.a" CACHE FILEPATH "oblogmsg library." FORCE)

script/run.sh

@@ -74,6 +74,34 @@ stop() {
   kill_proc_9
 }
 
+do_config_sys() {
+  username=$1
+  password=$2
+  if [[ -z "${username}" ]] || [[ -z "${password}" ]]; then
+    echo "No input sys username or password"
+    exit -1
+  fi
+
+  username_x=`./bin/${BIN} -x ${username}`
+  password_x=`./bin/${BIN} -x ${password}`
+
+  cp ./conf/conf.json ./conf/conf.json.new
+  sed -r -i 's/"ob_sys_username"[ ]*:[ ]*"[0-9a-zA-Z]+/"ob_sys_username": "'${username_x}'/' ./conf/conf.json.new
+  sed -r -i 's/"ob_sys_password"[ ]*:[ ]*"[0-9a-zA-Z]+/"ob_sys_password": "'${password_x}'/' ./conf/conf.json.new
+  diff ./conf/conf.json ./conf/conf.json.new
+  echo ""
+
+  read -r -p "!!DANGER!! About to update logproxy conf/conf.json, Please confirm? [Y/n] " response
+  if [ "${response}" != "y" ] && [ "${response}" != "Y" ]; then
+      echo "Cancel!"
+      rm -rf ./conf/conf.json.new
+      exit 0
+  fi
+
+  cp ./conf/conf.json ./conf/conf.json.bak
+  mv ./conf/conf.json.new ./conf/conf.json
+}
+
 case C"$1" in
 Cstop)
   stop
@@ -93,6 +121,9 @@ Cstatus)
   echo "status : ${status}"
   exit ${status}
   ;;
+Cconfig_sys)
+  do_config_sys $2 $3
+  ;;
 C*)
   echo "Usage: $0 {start|stop|status}"
   ;;

src/codec/codec_endian.h

@@ -17,75 +17,44 @@
 namespace oceanbase {
 namespace logproxy {
 
-enum class Endian {
-  BIG,
-  LITTLE,
-};
+#if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
 
-inline bool is_little_endian()
-{
-  int32_t i = 0x01020304;
-  char* buf = (char*)&i;
-  return buf[0] == 0x04;
-}
+#define le_to_cpu(integer) (integer)
+#define cpu_to_le(integer) (integer)
 
-template <class Integer>
-Integer bswap(Integer integer)
+inline uint16_t be_to_cpu(uint16_t integer)
 {
-  char* src_buf = (char*)&integer;
-  Integer ret = 0;
-  char* dst_buf = (char*)&ret;
-  const int bytes = sizeof(integer);
-  for (int i = 0; i < bytes; i++) {
-    dst_buf[i] = src_buf[bytes - i - 1];
-  }
-  return ret;
+  return __builtin_bswap16(integer);
 }
 
-template <class Integer>
-Integer le_to_cpu(Integer integer)
+inline uint32_t be_to_cpu(uint32_t integer)
 {
-  if (is_little_endian()) {
-    return integer;
-  }
-  return bswap<Integer>(integer);
+  return __builtin_bswap32(integer);
 }
 
-template <class Integer>
-Integer cpu_to_le(Integer integer)
+inline uint64_t be_to_cpu(uint64_t integer)
 {
-  return le_to_cpu<Integer>(integer);
+  return __builtin_bswap64(integer);
 }
 
-template <class Integer>
-Integer be_to_cpu(Integer integer)
+inline uint16_t cpu_to_be(uint16_t integer)
 {
-  if (!is_little_endian()) {
-    return integer;
-  }
-  return bswap<Integer>(integer);
+  return __builtin_bswap16(integer);
 }
 
-template <class Integer>
-Integer cpu_to_be(Integer integer)
+inline uint32_t cpu_to_be(uint32_t integer)
 {
-  return be_to_cpu<Integer>(integer);
+  return __builtin_bswap32(integer);
 }
 
-template <class Integer, Endian endian>
-Integer transform_endian(Integer integer)
+inline uint64_t cpu_to_be(uint64_t integer)
 {
-  if (endian == Endian::BIG) {
-    if (!is_little_endian()) {
-      return integer;
-    }
-    return bswap(integer);
-  } else {
-    if (is_little_endian()) {
-      return integer;
-    }
-    return bswap(integer);
-  }
+  return __builtin_bswap64(integer);
 }
+
+#else
+
+#endif
+
 }  // namespace logproxy
 }  // namespace oceanbase

src/codec/legacy_decoder.cpp

@@ -30,7 +30,7 @@ static PacketError decode_v1(Channel* ch, Message*& message)
     OMS_ERROR << "Failed to read message payload size, ch:" << ch->peer().id() << ", error:" << strerror(errno);
     return PacketError::NETWORK_ERROR;
   }
-  payload_size = be_to_cpu<uint32_t>(payload_size);
+  payload_size = be_to_cpu(payload_size);
   // FIXME.. use an mem pool
   char* payload_buf = (char*)malloc(payload_size);
   if (nullptr == payload_buf) {
@@ -86,12 +86,12 @@ PacketError LegacyDecoder::decode(Channel* ch, MessageVersion version, Message*&
   }
 
   // type
-  int32_t type = -1;
+  uint32_t type = -1;
   if (ch->readn((char*)&type, 4) != OMS_OK) {
     OMS_ERROR << "Failed to read message header, ch:" << ch->peer().id() << ", error:" << strerror(errno);
     return PacketError::NETWORK_ERROR;
   }
-  type = be_to_cpu<int32_t>(type);
+  type = be_to_cpu(type);
   if (!is_type_available(type)) {
     OMS_ERROR << "Invalid packet type:" << type << ", ch:" << ch->peer().id();
     return PacketError::PROTOCOL_ERROR;
@@ -117,7 +117,7 @@ static int read_varstr(Channel* ch, std::string& val)
   if (ch->readn((char*)&len, 4) != OMS_OK) {
     return OMS_FAILED;
   }
-  len = be_to_cpu<uint32_t>(len);
+  len = be_to_cpu(len);
 
   char* buf = (char*)malloc(len);
   FreeGuard<char*> ff(buf);

src/codec/legacy_encoder.cpp

@@ -64,8 +64,8 @@ static int compress_data(const RecordDataMessage& msg, MsgBuf& buffer)
   size_t offset = 0;
   for (auto& ptr : ptrs) {
     size_t block_size = ptr.second;
-    uint32_t seq_be = cpu_to_be<uint32_t>(idx++);
-    uint32_t size_be = cpu_to_be<uint32_t>(block_size);
+    uint32_t seq_be = cpu_to_be(idx++);
+    uint32_t size_be = cpu_to_be((uint32_t)block_size);
     memcpy(raw + offset, &seq_be, 4);
     memcpy(raw + offset + 4, &size_be, 4);
     memcpy(raw + offset + 8, ptr.first, block_size);
@@ -81,9 +81,9 @@ static int compress_data(const RecordDataMessage& msg, MsgBuf& buffer)
     OMS_DEBUG << "compress packet raw from size:" << total_size << " to compressed size:" << compressed_size;
   }
 
-  uint32_t packet_len_be = cpu_to_be<uint32_t>(compressed_size + 9);
-  uint32_t orginal_size_be = cpu_to_be<uint32_t>(total_size);
-  uint32_t compressed_size_be = cpu_to_be<uint32_t>(compressed_size);
+  uint32_t packet_len_be = cpu_to_be((uint32_t)compressed_size + 9);
+  uint32_t orginal_size_be = cpu_to_be(total_size);
+  uint32_t compressed_size_be = cpu_to_be((uint32_t)compressed_size);
   char* buf = (char*)malloc(13);
   memcpy(buf, &packet_len_be, 4);
   memset(buf + 4, (uint8_t)CompressType::LZ4, 1);
@@ -170,8 +170,8 @@ LegacyEncoder::LegacyEncoder()
                   << ", size: " << header->m_size;
       }
 
-      uint32_t seq_be = cpu_to_be<uint32_t>(idx++);
-      uint32_t size_be = cpu_to_be<uint32_t>(size);
+      uint32_t seq_be = cpu_to_be(idx++);
+      uint32_t size_be = cpu_to_be((uint32_t)size);
       buffer.push_back_copy((char*)&seq_be, 4);
       buffer.push_back_copy((char*)&size_be, 4);
       buffer.push_back((char*)logmsg_buf, size, false);
@@ -179,8 +179,8 @@ LegacyEncoder::LegacyEncoder()
       total_size += (size + 8);
     }
 
-    uint32_t packet_len_be = cpu_to_be<uint32_t>(total_size + 9);
-    total_size = cpu_to_be<uint32_t>(total_size);
+    uint32_t packet_len_be = cpu_to_be(total_size + 9);
+    total_size = cpu_to_be(total_size);
 
     char* buf = (char*)malloc(4 + 1 + 4 + 4);
     memcpy(buf, &packet_len_be, 4);
@@ -208,9 +208,9 @@ LegacyEncoder::LegacyEncoder()
     }
 
     // Error message
-    uint32_t code_be = cpu_to_be<uint32_t>(msg.code);
+    uint32_t code_be = cpu_to_be((uint32_t)msg.code);
     memcpy(buf, &code_be, 4);
-    uint32_t varlen_be = cpu_to_be<uint32_t>(msg.message.size());
+    uint32_t varlen_be = cpu_to_be((uint32_t)msg.message.size());
     memcpy(buf + 4, &varlen_be, 4);
     if (msg.message.size() != 0) {
       memcpy(buf + 8, msg.message.c_str(), msg.message.size());
@@ -237,7 +237,7 @@ int LegacyEncoder::encode(const Message& msg, MsgBuf& buffer)
   memset(buf, 0, 2);
 
   // response type code
-  uint32_t msg_type_be = cpu_to_be<uint32_t>((uint32_t)msg.type());
+  uint32_t msg_type_be = cpu_to_be((uint32_t)msg.type());
   memcpy(buf + 2, &msg_type_be, 4);
   buffer.push_front(buf, len);
   return ret;

src/codec/msg_buf.h

@@ -63,11 +63,6 @@ public:
     bool _owned = true;
   };
 
-public:
-  using ChunksType = std::deque<Chunk>;
-  using ChunkIterator = ChunksType::iterator;
-  using ChunkConstIterator = ChunksType::const_iterator;
-
 public:
   MsgBuf() = default;
 
@@ -107,12 +102,12 @@ public:
     return _chunks.size();
   }
 
-  ChunkConstIterator begin() const
+  std::deque<Chunk>::const_iterator begin() const
   {
     return _chunks.begin();
   }
 
-  ChunkConstIterator end() const
+  std::deque<Chunk>::const_iterator end() const
   {
     return _chunks.end();
   }
@@ -120,7 +115,7 @@ public:
   size_t byte_size() const;
 
 private:
-  ChunksType _chunks;
+  std::deque<Chunk> _chunks;
 };
 
 // TODO messageBufferWriter
@@ -150,14 +145,14 @@ public:
 
   int read_uint64(uint64_t& i);
 
-  template <class Integer, Endian endian = Endian::LITTLE>
+  template <class Integer>
   int read_int(Integer& i)
   {
     int ret = read((char*)&i, sizeof(i));
     if (ret != 0) {
       return ret;
     }
-    i = transform_endian<Integer, endian>(i);
+    i = le_to_cpu(i);
     return 0;
   }
 
@@ -182,7 +177,7 @@ public:
 
 private:
   const MsgBuf& _buffer;
-  MsgBuf::ChunkConstIterator _iter;
+  std::deque<MsgBuf::Chunk>::const_iterator _iter;
   size_t _pos = 0;
   size_t _byte_size = 0;
   size_t _read_size = 0;

src/codec/protobuf_decoder.cpp

@@ -42,7 +42,6 @@ PacketError ProtobufDecoder::decode(Channel* ch, MessageVersion version, Message
   // type
   int8_t type = -1;
   memcpy(&type, header_buf, 1);
-  type = be_to_cpu<int8_t>(type);
   if (!is_type_available(type)) {
     OMS_ERROR << "Invalid packet type:" << type << ", ch:" << ch->peer().id();
     return PacketError::PROTOCOL_ERROR;
@@ -51,7 +50,7 @@ PacketError ProtobufDecoder::decode(Channel* ch, MessageVersion version, Message
   // payload size
   uint32_t payload_size = 0;
   memcpy(&payload_size, header_buf + 1, 4);
-  payload_size = be_to_cpu<uint32_t>(payload_size);
+  payload_size = be_to_cpu(payload_size);
 
   // TODO... suppose that no large message
   if (payload_size > Config::instance().max_packet_bytes.val()) {

src/codec/protobuf_encoder.cpp

@@ -67,7 +67,7 @@ static char* encode_message_header(MessageType type, int packet_size, bool magic
     offset = sizeof(PACKET_MAGIC);
   }
 
-  int16_t version = cpu_to_be<int16_t>((int16_t)MessageVersion::V2);
+  uint16_t version = cpu_to_be((uint16_t)MessageVersion::V2);
   memcpy(buffer + offset, &version, sizeof(version));
   offset += sizeof(version);
 
@@ -75,7 +75,7 @@ static char* encode_message_header(MessageType type, int packet_size, bool magic
   memcpy(buffer + offset, &message_type, sizeof(message_type));
   offset += sizeof(message_type);
 
-  int32_t pb_packet_size = cpu_to_be<int32_t>(packet_size);
+  uint32_t pb_packet_size = cpu_to_be((uint32_t)packet_size);
   memcpy(buffer + offset, &pb_packet_size, sizeof(pb_packet_size));
   return buffer;
 }

src/communication/communicator.cpp

@@ -392,7 +392,7 @@ PacketError Communicator::receive_message(Channel* ch, Message*& msg)
     }
   }
 
-  version = be_to_cpu<uint16_t>(version);
+  version = be_to_cpu(version);
   if (!is_version_available(version)) {
     OMS_ERROR << "Invalid packet version:" << version << ", ch:" << ch->peer().id();
     return PacketError::PROTOCOL_ERROR;

src/communication/http.cpp

@@ -75,7 +75,7 @@ static void http_request_error_cb(enum evhttp_request_error error, void* arg)
 
 void http_conn_close_cb(struct evhttp_connection*, void* arg)
 {
-  OMS_WARN << "HTTP request conn closed";
+  OMS_DEBUG << "HTTP request conn closed";
   event_base_loopexit(((HttpContext*)arg)->base, nullptr);
 }
 

src/obaccess/mysql_protocol.cpp

@@ -46,9 +46,84 @@ int MysqlProtocol::connect_to_server()
 {
   int ret = connect(_hostname.c_str(), _port, false, _detect_timeout, _sockfd);
   if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to connect to server. server=" << _hostname << ':' << _port;
+    OMS_ERROR << "Failed to connect to server: " << _hostname << ':' << _port << ", user: " << _username;
   } else {
-    OMS_INFO << "Connect to server success. server=" << _hostname << ':' << _port;
+    OMS_INFO << "Connect to server success: " << _hostname << ':' << _port << ", user: " << _username;
+  }
+  return ret;
+}
+
+int MysqlProtocol::login(const std::string& host, int port, const std::string& username, const std::string& passwd_sha1,
+    const std::string& database)
+{
+  _hostname = host;
+  _port = port;
+  _username = username;
+  _passwd_sha1 = passwd_sha1;
+
+  // https://dev.mysql.com/doc/internals/en/secure-password-authentication.html
+  // 1 connect to the server
+  int ret = connect_to_server();
+  if (ret < 0) {
+    OMS_ERROR << "Failed to connect to server " << _hostname << ':' << _port;
+    return OMS_CONNECT_FAILED;
+  }
+
+  // 2 receive initial handshake
+  MsgBuf msgbuf;
+  uint8_t sequence = 0;
+  uint32_t packet_length = 0;
+  ret = recv_mysql_packet(_sockfd, _detect_timeout, packet_length, sequence, msgbuf);
+  if (ret != OMS_OK) {
+    OMS_ERROR << "Failed to receive handshake packet from: " << _hostname << ':' << _port
+              << ", error: " << strerror(errno);
+    return ret;
+  }
+  OMS_DEBUG << "Receive handshake packet from server: " << _hostname << ':' << _port << ", user: " << _username;
+
+  MySQLInitialHandShakePacket handshake_packet;
+  ret = handshake_packet.decode(msgbuf);
+  if (ret != OMS_OK || !handshake_packet.scramble_valid()) {
+    OMS_ERROR << "Failed to decode_payload initial handshake packet or does not has a valid scramble:"
+              << handshake_packet.scramble_valid() << ". length=" << packet_length << ", server: " << _hostname << ':'
+              << _port;
+    return ret;
+  }
+
+  const std::vector<char>& scramble = handshake_packet.scramble();
+
+  // 3 calculate the password combined with scramble buffer -> auth information
+  std::vector<char> auth;
+  ret = calc_mysql_auth_info(scramble, auth);
+  if (ret != OMS_OK) {
+    OMS_ERROR << "Failed to calc login info from: " << _hostname << ':' << _port << ", user: " << _username;
+    return ret;
+  }
+
+  // 4 send the handshake response with auth information
+  ret = send_auth(auth, database);
+  if (ret != OMS_OK) {
+    OMS_ERROR << "Failed to send handshake response message to " << _hostname << ':' << _port << ", user=" << _username;
+    return OMS_FAILED;
+  }
+
+  // 5 receive response from server
+  msgbuf.reset();
+  ret = recv_mysql_packet(_sockfd, _detect_timeout, packet_length, sequence, msgbuf);
+  if (ret != OMS_OK) {
+    OMS_ERROR << "Failed to recv handshake auth response from: " << _hostname << ':' << _port
+              << ", user: " << _username;
+    return OMS_FAILED;
+  }
+
+  MySQLOkPacket ok_packet;
+  ret = ok_packet.decode(msgbuf);
+  if (ret == OMS_OK) {
+    OMS_INFO << "Auth user success of server: " << _hostname << ':' << _port << ", user: " << _username;
+  } else {
+    MySQLErrorPacket error_packet;
+    error_packet.decode(msgbuf);
+    OMS_ERROR << "Auth user failed of server: " << _hostname << ':' << _port << ", user: " << _username;
   }
   return ret;
 }
@@ -61,7 +136,7 @@ static inline void my_xor(const unsigned char* s1, const unsigned char* s2, uint
   }
 }
 
-int MysqlProtocol::calc_mysql_auth_info(const std::vector<char>& scramble_buffer, std::vector<char>& auth_info)
+int MysqlProtocol::calc_mysql_auth_info(const std::vector<char>& scramble, std::vector<char>& auth)
 {
   // SHA1( password ) XOR SHA1( "20-bytes random data from server" <concat> SHA1( SHA1( password ) ) )
   // SHA1(password) -> stage1
@@ -82,8 +157,8 @@ int MysqlProtocol::calc_mysql_auth_info(const std::vector<char>& scramble_buffer
   }
 
   std::vector<char> scramble_combined;
-  scramble_combined.reserve(scramble_buffer.size() + passwd_stage2.size());
-  scramble_combined.assign(scramble_buffer.begin(), scramble_buffer.end());
+  scramble_combined.reserve(scramble.size() + passwd_stage2.size());
+  scramble_combined.assign(scramble.begin(), scramble.end());
   scramble_combined.insert(scramble_combined.end(), passwd_stage2.begin(), passwd_stage2.end());
 
   sha1.reset();
@@ -107,127 +182,45 @@ int MysqlProtocol::calc_mysql_auth_info(const std::vector<char>& scramble_buffer
     return OMS_FAILED;
   }
 
-  auth_info.resize(sha_combined.size());
+  auth.resize(sha_combined.size());
   my_xor((const unsigned char*)_passwd_sha1.data(),
       (const unsigned char*)sha_combined.data(),
-      auth_info.size(),
-      (unsigned char*)auth_info.data());
+      auth.size(),
+      (unsigned char*)auth.data());
   return OMS_OK;
 }
 
-int MysqlProtocol::send_auth_info(const std::vector<char>& auth_info, uint8_t sequence)
+int MysqlProtocol::send_auth(const std::vector<char>& auth_info, const std::string& database)
 {
-  MysqlHandShakeResponsePacket hand_shake_response_packet(_username, "", auth_info, sequence);
+  MySQLHandShakeResponsePacket handshake_response_packet(_username, database, auth_info);
   MsgBuf msgbuf;
-  int ret = hand_shake_response_packet.encode(msgbuf);
+  int ret = handshake_response_packet.encode(msgbuf);
   if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to encode hand shake response packet";
+    OMS_ERROR << "Failed to encode handshake response packet";
     return -1;
   }
 
-  for (const auto& iter : msgbuf) {
-    ret = writen(_sockfd, iter.buffer(), iter.size());
-    if (ret != OMS_OK) {
-      OMS_ERROR << "Failed to send hand shake response message. error=" << strerror(errno) << ". peer=" << _hostname
-                << ":" << _port;
-      return ret;
-    }
-  }
-  return OMS_OK;
-}
-
-int MysqlProtocol::is_mysql_response_ok()
-{
-  MsgBuf msgbuf;
-  int ret = recv_mysql_packet(_sockfd, _detect_timeout, msgbuf);
+  ret = send_mysql_packet(_sockfd, msgbuf, 0);
   if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to receive ok packet from server " << _hostname << ':' << _port
-              << ", error=" << strerror(errno);
-    return OMS_FAILED;
-  }
-  MysqlOkPacket ok_packet;
-  return ok_packet.decode(msgbuf);
-}
-
-int MysqlProtocol::login(const std::string& host, int port, const std::string& username, const std::string& passwd_sha1,
-    const std::string& database)
-{
-  _hostname = host;
-  _port = port;
-  _username = username;
-  _passwd_sha1 = passwd_sha1;
-
-  // https://dev.mysql.com/doc/internals/en/secure-password-authentication.html
-  // 1 connect to the server
-  int ret = connect_to_server();
-  if (ret < 0) {
-    OMS_ERROR << "Failed to connect to server " << _hostname << ':' << _port;
-    return OMS_CONNECT_FAILED;
-  }
-
-  // 2 receive initial hand shake
-  uint32_t packet_length = 0;
-  uint8_t sequence = 0;
-  MsgBuf msgbuf;
-  ret = recv_mysql_packet(_sockfd, _detect_timeout, packet_length, sequence, msgbuf);
-  if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to receive hand shake packet. error=" << strerror(errno) << ". server=" << _hostname << ':'
-              << _port;
-    return ret;
-  }
-  OMS_DEBUG << "receive hand shake packet from server=" << _hostname << ':' << _port;
-
-  MysqlInitialHandShakePacket handshake_packet;
-  ret = handshake_packet.decode(msgbuf);
-  if (ret != OMS_OK || !handshake_packet.scramble_buffer_valid()) {
-    OMS_ERROR << "Failed to decode_payload initial hand shake packet or does not has a valid scramble:"
-              << handshake_packet.scramble_buffer_valid() << ". length=" << packet_length << ", server=" << _hostname
-              << ':' << _port;
-    return ret;
-  }
-
-  const std::vector<char>& scramble_buffer = handshake_packet.scramble_buffer();
-
-  // 3 calculate the password combined with scramble buffer -> auth information
-  std::vector<char> auth_info;
-  ret = calc_mysql_auth_info(scramble_buffer, auth_info);
-  if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to calc login info. server=" << _hostname << ':' << _port << ", user=" << _username;
+    OMS_ERROR << "Failed to send handshake response to server: " << _hostname << ":" << _port;
     return ret;
   }
-
-  // 4 send the hand shake response with auth information
-  ret = send_auth_info(auth_info, sequence + 1);
-  if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to send hand shake response message to " << _hostname << ':' << _port
-              << ", user=" << _username;
-    return OMS_FAILED;
-  }
-
-  // 5 receive response from server
-  ret = is_mysql_response_ok();
-
-  if (ret == OMS_OK) {
-    OMS_INFO << "Auth user success. server=" << _hostname << ':' << _port << ", user=" << _username;
-  } else {
-    OMS_DEBUG << "Auth user failed. server=" << _hostname << ':' << _port << ", user=" << _username;
-  }
-  return ret;
+  return OMS_OK;
 }
 
-int MysqlProtocol::query(const std::string& sql, MysqlResultSet& rs)
+int MysqlProtocol::query(const std::string& sql, MySQLResultSet& rs)
 {
 
-  OMS_INFO << "query obmysql:" << sql;
+  OMS_INFO << "Query obmysql SQL:" << sql;
 
-  MysqlQueryPacket packet(sql);
   MsgBuf msgbuf;
+  MySQLQueryPacket packet(sql);
   int ret = packet.encode_inplace(msgbuf);
   if (ret != OMS_OK) {
-    OMS_ERROR << "Failed to encode mysql sql packet ,ret:" << ret;
+    OMS_ERROR << "Failed to encode observer sql packet, ret:" << ret;
     return ret;
   }
-  ret = send_mysql_packet(_sockfd, msgbuf);
+  ret = send_mysql_packet(_sockfd, msgbuf, 0);
   if (ret != OMS_OK) {
     OMS_ERROR << "Failed to send query packet to server:" << _hostname << ':' << _port;
     return OMS_FAILED;
@@ -239,11 +232,11 @@ int MysqlProtocol::query(const std::string& sql, MysqlResultSet& rs)
     return OMS_FAILED;
   }
 
-  MysqlQueryResponsePacket query_resp;
+  MySQLQueryResponsePacket query_resp;
   ret = query_resp.decode(msgbuf);
   if (ret != OMS_OK || query_resp.col_count() == 0) {
-    OMS_ERROR << "Failed to query mysql:" << query_resp._err._message
-              << (query_resp.col_count() == 0 ? ", unexpected column count 0" : "");
+    OMS_ERROR << "Failed to query observer:" << query_resp._err._message
+              << (query_resp.col_count() == 0 ? ", unexpected column count: 0" : "");
     return OMS_FAILED;
   }
 
@@ -257,7 +250,7 @@ int MysqlProtocol::query(const std::string& sql, MysqlResultSet& rs)
       return OMS_FAILED;
     }
 
-    MysqlCol column;
+    MySQLCol column;
     column.decode(msgbuf);
     rs.cols[i] = column;
   }
@@ -268,7 +261,7 @@ int MysqlProtocol::query(const std::string& sql, MysqlResultSet& rs)
     OMS_ERROR << "Failed to recv eof packet from server:" << _hostname << ':' << _port;
     return OMS_FAILED;
   }
-  MysqlEofPacket eof;
+  MySQLEofPacket eof;
   ret = eof.decode(msgbuf);
   if (ret != OMS_OK) {
     OMS_ERROR << "Failed to decode eof packet from server:" << _hostname << ':' << _port;
@@ -287,7 +280,7 @@ int MysqlProtocol::query(const std::string& sql, MysqlResultSet& rs)
       break;
     }
 
-    MysqlRow row(query_resp.col_count());
+    MySQLRow row(query_resp.col_count());
     row.decode(msgbuf);
     rs.rows.emplace_back(row);
   }

src/obaccess/mysql_protocol.h

@@ -37,7 +37,7 @@ public:
   int login(const std::string& host, int port, const std::string& username, const std::string& passwd_sha1,
       const std::string& database = "");
 
-  int query(const std::string& sql, MysqlResultSet& rs);
+  int query(const std::string& sql, MySQLResultSet& rs);
 
   /**
    * 设置接收网络消息包时，首次检测是否有消息到达的超时时间。单位毫秒
@@ -59,11 +59,9 @@ public:
 private:
   int connect_to_server();
 
-  int calc_mysql_auth_info(const std::vector<char>& scramble_buffer, std::vector<char>& auth_info);
+  int calc_mysql_auth_info(const std::vector<char>& scramble, std::vector<char>& auth);
 
-  int send_auth_info(const std::vector<char>& auth_info, uint8_t sequence);
-
-  int is_mysql_response_ok();
+  int send_auth(const std::vector<char>& auth_info, const std::string& database);
 
 private:
   std::string _username;

src/obaccess/ob_access.cpp

@@ -96,21 +96,26 @@ static int parse_cluster_url(const std::string& cluster_url, std::vector<ObAcces
   return OMS_OK;
 }
 
-int ObAccess::init(const OblogConfig& config)
+int ObAccess::init(const OblogConfig& hs_config)
 {
-  if (!config.cluster_url.empty()) {
-    if (parse_cluster_url(config.cluster_url.val(), _servers) != OMS_OK) {
+  _user = hs_config.user.val();
+  _user_to_conn = _user;
+  std::vector<std::string> sections;
+
+  if (!hs_config.cluster_url.empty()) {
+    if (parse_cluster_url(hs_config.cluster_url.val(), _servers) != OMS_OK) {
       return OMS_FAILED;
     }
+    _user_to_conn = ObUsername(_user).name_without_cluster();
+
   } else {
 
-    const std::string& root_servers = config.root_servers.val();
+    const std::string& root_servers = hs_config.root_servers.val();
     if (root_servers.empty()) {
       OMS_ERROR << "Failed to init ObAccess caused by empty root_servers";
       return OMS_FAILED;
     }
 
-    std::vector<std::string> sections;
     int ret = split(root_servers, ';', sections);
     if (ret == 0) {
       OMS_ERROR << "Failed to init ObAccess caused by invalid root_servers";
@@ -138,26 +143,27 @@ int ObAccess::init(const OblogConfig& config)
     }
   }
 
-  _user = config.user.val();
-  hex2bin(config.password.val().c_str(), config.password.val().size(), _password_sha1);
+  hex2bin(hs_config.password.val().c_str(), hs_config.password.val().size(), _password_sha1);
 
-  if (_user.empty() || _password_sha1.empty()) {
+  if (_user_to_conn.empty() || _password_sha1.empty()) {
     OMS_ERROR << "Failed to init ObAccess caused by empty user or password";
     return OMS_FAILED;
   }
 
-  _sys_user = config.sys_user.empty() ? Config::instance().ob_sys_username.val() : config.sys_user.val();
-  if (config.sys_password.empty()) {
-    MysqlProtocol::do_sha_password(Config::instance().ob_sys_password.val(), _sys_password_sha1);
+  _sys_user = !hs_config.sys_user.empty() ? hs_config.sys_user.val() : Config::instance().ob_sys_username.val();
+
+  // preconfiged sys password was encrypted, otherwise was plain text as handshake param
+  if (!hs_config.sys_password.empty()) {
+    MysqlProtocol::do_sha_password(hs_config.sys_password.val(), _sys_password_sha1);
   } else {
-    MysqlProtocol::do_sha_password(config.sys_password.val(), _sys_password_sha1);
+    MysqlProtocol::do_sha_password(Config::instance().ob_sys_password.val(), _sys_password_sha1);
   }
   if (_sys_user.empty() || _sys_password_sha1.empty()) {
     OMS_ERROR << "Failed to init ObAccess caused by empty sys_user or sys_password";
     return OMS_FAILED;
   }
 
-  int ret = _table_whites.from(config.table_whites.val());
+  int ret = _table_whites.from(hs_config.table_whites.val());
   if (ret != OMS_OK) {
     return ret;
   }
@@ -168,7 +174,8 @@ int ObAccess::init(const OblogConfig& config)
 int ObAccess::auth()
 {
   for (auto& server : _servers) {
-    int ret = _table_whites.all_tenant ? auth_sys(server) : auth_tenant(server);
+    bool auth_by_sys = _table_whites.all_tenant || _table_whites.with_sys;
+    int ret = auth_by_sys ? auth_sys(server) : auth_tenant(server);
     if (ret != OMS_OK) {
       return OMS_FAILED;
     }
@@ -180,22 +187,22 @@ int ObAccess::auth_sys(const ServerInfo& server)
 {
   MysqlProtocol auther;
   // all tenant must be sys tenant;
-  int ret = auther.login(server.host, server.port, _user, _password_sha1);
+  int ret = auther.login(server.host, server.port, _user_to_conn, _password_sha1);
   if (ret != OMS_OK) {
     return ret;
   }
 
-  MysqlResultSet rs;
+  MySQLResultSet rs;
   ret = auther.query("show tenant", rs);
   if (ret != OMS_OK || rs.rows.empty()) {
     OMS_ERROR << "Failed to auth, show tenant for sys all match mode, ret:" << ret;
     return OMS_FAILED;
   }
 
-  const MysqlRow& row = rs.rows.front();
+  const MySQLRow& row = rs.rows.front();
   const std::string& tenant = row.fields().front();
   if (tenant.size() < 3 || strncasecmp("sys", tenant.c_str(), 3) != 0) {
-    OMS_ERROR << "Failed to auth, all tenant mode must be sys tenant, current: " << tenant;
+    OMS_ERROR << "Failed to auth, all tenant mode or sys tenant must be connected as sys tenant, current: " << tenant;
     return OMS_FAILED;
   }
   return OMS_OK;
@@ -213,9 +220,9 @@ int ObAccess::auth_tenant(const ServerInfo& server)
   ObUsername ob_user(_user);
 
   // 2. for each of tenant servers, login it.
-  MysqlResultSet rs;
+  MySQLResultSet rs;
   for (auto& tenant_entry : _table_whites.tenants) {
-    OMS_INFO << "About to auth tenant:" << tenant_entry.first << " of user:" << _user;
+    OMS_INFO << "About to auth tenant:" << tenant_entry.first << " of user:" << _user_to_conn;
 
     rs.reset();
     ret = sys_auther.query(
@@ -235,15 +242,12 @@ int ObAccess::auth_tenant(const ServerInfo& server)
                 << ", col count:" << rs.col_count << ", ret:" << ret;
       return OMS_FAILED;
     }
-    const MysqlRow& row = rs.rows.front();
+    const MySQLRow& row = rs.rows.front();
     const std::string& host = row.fields()[0];
     const uint16_t sql_port = atoi(row.fields()[1].c_str());
 
     MysqlProtocol user_auther;
-    std::string conn_user = ob_user.username;
-    conn_user.append("@");
-    conn_user.append(ob_user.tenant.empty() ? tenant_entry.first : ob_user.tenant);
-    ret = user_auther.login(host, sql_port, conn_user, _password_sha1);
+    ret = user_auther.login(host, sql_port, ob_user.name_without_cluster(tenant_entry.first), _password_sha1);
     if (ret != OMS_OK) {
       OMS_ERROR << "Failed to auth from tenant server: " << host << ":" << sql_port << ", ret:" << ret;
       return ret;
@@ -288,5 +292,18 @@ ObUsername::ObUsername(const std::string& full_name)
   }
 }
 
+std::string ObUsername::name_without_cluster(const std::string& in_tenant)
+{
+  std::string conn_user = username;
+  if (!tenant.empty()) {
+    conn_user.append("@");
+    conn_user.append(tenant);
+  } else if (!in_tenant.empty()) {
+    conn_user.append("@");
+    conn_user.append(in_tenant);
+  }
+  return conn_user;
+}
+
 }  // namespace logproxy
 }  // namespace oceanbase

src/obaccess/ob_access.h

@@ -46,6 +46,7 @@ private:
 private:
   std::vector<ServerInfo> _servers;
   std::string _user;
+  std::string _user_to_conn;
   std::string _password_sha1;
   std::string _sys_user;
   std::string _sys_password_sha1;
@@ -60,6 +61,8 @@ struct ObUsername {
   std::string username;
 
   explicit ObUsername(const std::string& full_name);
+
+  std::string name_without_cluster(const std::string& in_tenant = "");
 };
 
 }  // namespace logproxy

src/obaccess/ob_mysql_packet.h

@@ -21,9 +21,8 @@ namespace logproxy {
 class MsgBuf;
 
 /**
- * 接收一个mysql消息包
- * 参考: https://dev.mysql.com/doc/internals/en/mysql-packet.html
- * @param fd 接收消息的描述符
+ * rece a mysql protocol packet
+ * @param fd
  * @param timout 等待有消息的超时时间(一旦判断有消息到达，就不再关注timeout). 单位 毫秒
  * @param[out] packet_length 消息包长度
  * @param[out] sequence 消息sequence，参考mysql说明
@@ -34,9 +33,9 @@ int recv_mysql_packet(int fd, int timeout, uint32_t& packet_length, uint8_t& seq
 
 int recv_mysql_packet(int fd, int timeout, MsgBuf& msgbuf);
 
-int send_mysql_packet(int fd, MsgBuf& msgbuf);
+int send_mysql_packet(int fd, MsgBuf& msgbuf, uint8_t sequence);
 
-class MysqlResponse {
+class MySQLResponse {
 public:
   friend class MysqlProtocol;
 
@@ -44,12 +43,12 @@ protected:
   virtual int decode(const MsgBuf& msgbuf) = 0;
 };
 
-class MysqlOkPacket : public MysqlResponse {
+class MySQLOkPacket : public MySQLResponse {
 public:
   int decode(const MsgBuf& msgbuf) override;
 };
 
-class MysqlEofPacket : public MysqlResponse {
+class MySQLEofPacket : public MySQLResponse {
 public:
   int decode(const MsgBuf& msgbuf) override;
 
@@ -60,7 +59,7 @@ private:
   uint16_t _status_flags;
 };
 
-class MysqlErrorPacket : public MysqlResponse {
+class MySQLErrorPacket : public MySQLResponse {
 public:
   friend class MysqlProtocol;
 
@@ -74,17 +73,13 @@ private:
   std::string _message;
 };
 
-class MysqlInitialHandShakePacket {
+class MySQLInitialHandShakePacket {
 public:
-  /**
-   * 连接上mysql之后，mysql就向客户端发送一个握手数据包。握手数据包中包含了一个随机生成的字符串(20字节）。
-   * 这个随机字符串称为scramble，与密码一起做运算后，发送给mysql server做鉴权
-   */
   int decode(const MsgBuf& msgbuf);
 
-  bool scramble_buffer_valid() const;
+  bool scramble_valid() const;
 
-  const std::vector<char>& scramble_buffer() const;
+  const std::vector<char>& scramble() const;
 
   uint8_t sequence() const;
 
@@ -92,34 +87,28 @@ private:
   uint8_t _sequence = 0;
   uint8_t _protocol_version = 0;
   uint32_t _capabilities_flag = 0;
-  std::vector<char> _scramble_buffer;
-  bool _scramble_buffer_valid = false;
+  std::string _auth_plugin_name;
+
+  bool _scramble_valid = false;
+  std::vector<char> _scramble;
 };
 
-class MysqlHandShakeResponsePacket {
+class MySQLHandShakeResponsePacket {
 public:
-  MysqlHandShakeResponsePacket(const std::string& username, const std::string& database,
-      const std::vector<char>& auth_response, int8_t sequence);
+  MySQLHandShakeResponsePacket(
+      const std::string& username, const std::string& database, const std::vector<char>& auth_response);
 
-  /**
-   * 客户端创建socket连接成功mysql server后，MySQL会发一个握手包，之后客户端向MySQL server回复一个消息。
-   * 这里就负责这条消息的编码。
-   */
   int encode(MsgBuf& msgbuf);
 
-private:
-  uint32_t calc_capabilities_flag();
-
 private:
   std::string _username;
   std::string _database;
   std::vector<char> _auth_response;
-  int8_t _sequence = 0;
 };
 
-class MysqlQueryPacket {
+class MySQLQueryPacket {
 public:
-  explicit MysqlQueryPacket(const std::string& sql);
+  explicit MySQLQueryPacket(const std::string& sql);
 
   // use memory in-stack, none any heap memory would be alloc
   int encode_inplace(MsgBuf& msgbuf);
@@ -129,7 +118,7 @@ private:
   const std::string& _sql;
 };
 
-class MysqlCol : public MysqlResponse {
+class MySQLCol : public MySQLResponse {
 public:
   int decode(const MsgBuf& msgbuf) override;
 
@@ -149,11 +138,11 @@ private:
   uint16_t _filler;
 };
 
-class MysqlRow : public MysqlResponse {
+class MySQLRow : public MySQLResponse {
 public:
   friend class MysqlProtocol;
 
-  explicit MysqlRow(uint64_t col_count);
+  explicit MySQLRow(uint64_t col_count);
 
   int decode(const MsgBuf& msgbuf) override;
 
@@ -172,7 +161,7 @@ private:
   std::vector<std::string> _fields;
 };
 
-class MysqlQueryResponsePacket : public MysqlResponse {
+class MySQLQueryResponsePacket : public MySQLResponse {
 public:
   friend class MysqlProtocol;
 
@@ -186,15 +175,15 @@ public:
 private:
   uint64_t _col_count = 0;
 
-  MysqlErrorPacket _err;
+  MySQLErrorPacket _err;
 };
 
-struct MysqlResultSet {
+struct MySQLResultSet {
   void reset();
 
   uint64_t col_count = 0;
-  std::vector<MysqlCol> cols;
-  std::vector<MysqlRow> rows;
+  std::vector<MySQLCol> cols;
+  std::vector<MySQLRow> rows;
 };
 
 }  // namespace logproxy

src/obaccess/oblog_config.cpp

@@ -113,7 +113,11 @@ int TenantDbTable::from(const std::string& table_whites)
     }
 
     const std::string& tenant = items[0];
+    if (tenant == "sys") {
+      with_sys = true;
+    }
     if (tenant == "*") {
+      with_sys = true;
       all_tenant = true;
       tenants.clear();
       return OMS_OK;

src/obaccess/oblog_config.h

@@ -60,7 +60,8 @@ struct DbTable {
 };
 
 struct TenantDbTable {
-  bool all_tenant;
+  bool with_sys = false;
+  bool all_tenant = false;
   std::map<std::string, DbTable> tenants;
 
   int from(const std::string& table_whites);

src/test/test_common.cpp

@@ -29,13 +29,6 @@ TEST(COMMON, hex2bin)
   std::string binstr;
   hex2bin(hexstr.data(), hexstr.size(), binstr);
   ASSERT_STREQ(binstr.c_str(), text);
-
-  binstr.clear();
-  hexstr.insert(0, 1, ' ');
-  hexstr.insert(hexstr.size() / 2, 1, ' ');
-  hexstr.insert(hexstr.size(), 1, ' ');
-  hex2bin(hexstr.data(), hexstr.size(), binstr);
-  ASSERT_STREQ(binstr.c_str(), text);
 }
 
 TEST(COMMON, json)

src/test/test_ob_mysql.cpp

@@ -51,7 +51,7 @@ int do_sha_password(const std::string& pswd, std::string& sha_password)
 }
 
 void login(const std::string& host, int port, const std::string& user, const std::string& passwd,
-    const std::string& sql, MysqlResultSet& rs)
+    const std::string& sql, MySQLResultSet& rs)
 {
   std::string sha_password;
   int ret = do_sha_password(passwd, sha_password);
@@ -65,7 +65,7 @@ void login(const std::string& host, int port, const std::string& user, const std
 
   LogStream ls(0, "", 0, nullptr);
   ls << "| column counts:" << rs.cols.size() << " |\n";
-  for (MysqlRow& row : rs.rows) {
+  for (MySQLRow& row : rs.rows) {
     ls << "| ";
     for (const std::string& field : row.fields()) {
       ls << field << " | ";
@@ -89,32 +89,35 @@ static std::string sys_password_sha1;
 
 static std::string cluster_url = "";
 
-static std::vector<std::string> sqls = {"show tenant",
-    "select version()",
+// static std::vector<std::string> sqls = {"select 1"};
+static std::vector<std::string> sqls = {"show tenant", "select version()"};
+static std::vector<std::string> sqls_for_sys = {
     "SELECT server.svr_ip, server.svr_port, server.zone, tenant.tenant_id, tenant.tenant_name from "
     "oceanbase.__all_resource_pool AS pool, oceanbase.__all_unit AS unit, oceanbase.__all_server AS "
     "server, oceanbase.__all_tenant AS tenant WHERE tenant.tenant_id = pool.tenant_id AND "
     "unit.resource_pool_id = pool.resource_pool_id AND unit.svr_ip = server.svr_ip AND "
     "unit.svr_port = server.svr_port AND tenant.tenant_name='" +
-        tenant + "'"};
+    tenant + "'"};
 
 TEST(MYSQL_AUTH, query)
 {
-  MysqlResultSet rs;
+  MySQLResultSet rs;
   for (auto& sql : sqls) {
     login(host, port, user, password, sql, rs);
+  }
+  for (auto& sql : sqls_for_sys) {
     login(host, port, sys_user, sys_password, sql, rs);
   }
 }
 
 TEST(MYSQL_AUTH, auth_ok)
 {
-  do_sha_password(password, password_sha1);
-  do_sha_password(sys_password, sys_password_sha1);
-
   Config::instance().ob_sys_username.set(sys_user);
   Config::instance().ob_sys_password.set(sys_password);
 
+  do_sha_password(password, password_sha1);
+  do_sha_password(sys_password, sys_password_sha1);
+
   OblogConfig config("first_start_timestamp=0 rootserver_list=" + host + ":2881:" + std::to_string(port) +
                      " cluster_user=" + user + " cluster_password=" + dumphex(password_sha1) +
                      " tb_white_list=" + tenant + ".*.*");