refactor package (#42)

* refactor package

* omit rpm dep

* fix package

* fix reader coredump

* liboblog deps so finding

* liboblog deps so finding

* liboblog deps so finding

* fix package

CMakeLists.txt

@@ -21,6 +21,7 @@ option(WITH_US_TIMESTAMP "Enable microseconds start timestamp" ON)
 option(USE_OBCDC_NS "With libobcdc" ON)
 option(USE_LIBOBLOG "With precompiled liboblog" OFF)
 option(USE_CXX11_ABI "Build with C++11 ABI" OFF)
+option(USE_LIBOBLOG_3 "With liboblog 3.x" OFF)
 
 SET(OMS_PROJECT_SRC_PATH ${CMAKE_CURRENT_SOURCE_DIR})
 SET(OMS_PROJECT_BUILD_PATH ${CMAKE_CURRENT_BINARY_DIR})
@@ -36,49 +37,58 @@ SET(FIND_LIBOBLOG ON)
 if (USE_OBCDC_NS)
     SET(OBCDC_NAME "libobcdc")
     SET(OBCDC_NAME_VAR "-DUSE_OBCDC_NS")
+    if (USE_LIBOBLOG_3)
+        SET(OBCDC_NAME_VAR "${OBCDC_NAME_VAR} -DUSE_LIBOBLOG_3")
+    endif()
 else ()
     SET(OBCDC_NAME "liboblog")
 endif ()
 
 if (WITH_DEPS)
-    execute_process(
-            COMMAND bash deps/dep_create.sh ${OMS_PROJECT_BUILD_PATH}/deps
-            WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
-            COMMAND_ERROR_IS_FATAL ANY
-    )
     SET(DEP_VAR ${OMS_PROJECT_BUILD_PATH}/deps)
+    SET(THIRD_LIB_DIR ${DEP_VAR}/usr/local/oceanbase/deps/devel/lib)
 
-    SET(THIRD_LIB_DIR ${DEP_VAR}/usr/local/oceanbase/deps/devel/lib/)
-
-    # make liboblog found
+    SET(LIBOBLOG_RPM_NAME "oceanbase-ce-cdc")
     if (NOT USE_LIBOBLOG)
         SET(FIND_LIBOBLOG OFF)
-        SET(LIBOBLOG_INCLUDE_PATH "${DEP_VAR}/home/admin/oceanbase/include")
-        SET(LIBOBLOG_LIBRARIES ${DEP_VAR}/home/admin/oceanbase/lib64/${OBCDC_NAME}.so)
-        SET(LIBOBLOG_LIB_DIR ${DEP_VAR}/home/admin/oceanbase/lib64/)
 
-        FILE(GLOB LIBOBLOG_DEPS ${THIRD_LIB_DIR}/*.so ${THIRD_LIB_DIR}/*/*.so)
+        if (USE_LIBOBLOG_3)
+            # version before 4.0
+            SET(LIBOBLOG_RPM_NAME "oceanbase-ce-devel")
+            SET(LIBOBLOG_INCLUDE_PATH ${DEP_VAR}/usr/include)
+            SET(LIBOBLOG_LIBRARIES ${DEP_VAR}/usr/lib/${OBCDC_NAME}.so)
+        else()
+            SET(LIBOBLOG_INCLUDE_PATH "${DEP_VAR}/home/admin/oceanbase/include")
+            SET(LIBOBLOG_LIBRARIES ${DEP_VAR}/home/admin/oceanbase/lib64/${OBCDC_NAME}.so)
+            SET(LIBOBLOG_LIB_DIR ${DEP_VAR}/home/admin/oceanbase/lib64/)
+
+            SET(LIBOBLOG_DEPS
+                    ${THIRD_LIB_DIR}/libaio.so
+                    ${THIRD_LIB_DIR}/mariadb/libmariadb.so
+                    )
+        endif()
 
-        # FIXME... auto detected
-        # version before 4.0
-#        SET(LIBOBLOG_INCLUDE_PATH ${DEP_VAR}/usr/include/)
-#        SET(LIBOBLOG_LIBRARIES ${DEP_VAR}/usr/lib/${OBCDC_NAME}.so)
     endif ()
 
+    execute_process(
+            COMMAND bash deps/dep_create.sh ${OMS_PROJECT_BUILD_PATH}/deps ${LIBOBLOG_RPM_NAME}
+            WORKING_DIRECTORY ${PROJECT_SOURCE_DIR}
+            COMMAND_ERROR_IS_FATAL ANY
+    )
+
     # make openssl found
     CMAKE_POLICY(SET CMP0074 NEW)
     SET(OpenSSL_ROOT ${DEP_VAR}/usr/local/oceanbase/deps/devel/)
 
-
     # compiler
     SET(COMPILER_DIR ${DEP_VAR}/usr/local/oceanbase/devtools/bin/)
 else ()
-# compiler
-execute_process(
-        COMMAND which gcc
-        OUTPUT_VARIABLE GCC_BIN
-)
-GET_FILENAME_COMPONENT(COMPILER_DIR ${GCC_BIN} DIRECTORY)
+    # compiler
+    execute_process(
+            COMMAND which gcc
+            OUTPUT_VARIABLE GCC_BIN
+    )
+    GET_FILENAME_COMPONENT(COMPILER_DIR ${GCC_BIN} DIRECTORY)
 endif ()
 
 message(STATUS "COMPILER_DIR: ${COMPILER_DIR}")
@@ -235,7 +245,7 @@ if (WITH_US_TIMESTAMP)
 endif()
 
 message("oblogmsg: ${OBLOGMSG_INCLUDE_DIR}, ${OBLOGMSG_LIBRARIES}")
-message("liboblog: ${LIBOBLOG_INCLUDE_PATH}, ${LIBOBLOG_LIBRARIES}")
+message("liboblog: ${LIBOBLOG_INCLUDE_PATH}, ${LIBOBLOG_LIBRARIES}, ${LIBOBLOG_DEPS}")
 
 execute_process(
         COMMAND git log -1 --format=%H

cmake/rpm.cmake

@@ -33,8 +33,13 @@ set(CPACK_RPM_SPEC_MORE_DEFINE
 ## TIPS
 #
 # - PATH is relative to the **ROOT directory** of project other than the cmake directory.
+if (NOT ${OBLOGPROXY_INSTALL_PREFIX})
+    set(CPACK_PACKAGING_INSTALL_PREFIX ${OBLOGPROXY_INSTALL_PREFIX})
+endif()
+
+message("CPACK_PACKAGING_INSTALL_PREFIX: ${CPACK_PACKAGING_INSTALL_PREFIX}")
 
-list(APPEND OBLOGPROXY_BIN_FILES ${CMAKE_BINARY_DIR}/logproxy)
+list(APPEND OBLOGPROXY_BIN_FILES ${OMS_PROJECT_BUILD_PATH}/logproxy)
 list(APPEND OBLOGPROXY_CONF_FILES ${CMAKE_SOURCE_DIR}/conf/conf.json)
 list(APPEND OBLOGPROXY_SCRIPT_FILES ${CMAKE_SOURCE_DIR}/script/run.sh)
 
@@ -56,14 +61,44 @@ install(PROGRAMS
         COMPONENT oblogproxy
         )
 
+if (WITH_DEPS)
+    if (NOT USE_LIBOBLOG)
+        if (USE_LIBOBLOG_3)
+            file(GLOB LIBOBLOG_OUTPUT_LIBS
+                    ${DEP_VAR}/usr/lib/${OBCDC_NAME}*.so*
+                    ${DEP_VAR}/usr/local/oceanbase/deps/devel/lib/libaio.so*
+                    )
+            install(FILES
+                    ${LIBOBLOG_OUTPUT_LIBS}
+                    DESTINATION ${CPACK_PACKAGING_INSTALL_PREFIX}/liboblog
+                    COMPONENT oblogproxy
+                    )
+        else()
+            file(GLOB LIBOBLOG_OUTPUT_LIBS ${DEP_VAR}/home/admin/oceanbase/lib64/${OBCDC_NAME}*.so*
+                    ${DEP_VAR}/usr/local/oceanbase/deps/devel/lib/libaio.so*
+                    ${DEP_VAR}/usr/local/oceanbase/deps/devel/lib/mariadb/libmariadb.so*
+                    )
+            install(FILES
+                    ${LIBOBLOG_OUTPUT_LIBS}
+                    DESTINATION ${CPACK_PACKAGING_INSTALL_PREFIX}/liboblog
+                    COMPONENT oblogproxy
+                    )
+        endif()
+    endif()
+endif()
+
 file(WRITE ${CMAKE_CURRENT_BINARY_DIR}/utils_post.script "/sbin/ldconfig /usr/lib")
 set(CPACK_RPM_UTILS_POST_INSTALL_SCRIPT_FILE  ${CMAKE_CURRENT_BINARY_DIR}/utils_post.script)
 file(WRITE ${CMAKE_CURRENT_BINARY_DIR}/utils_postun.script "/sbin/ldconfig")
 set(CPACK_RPM_UTILS_POST_UNINSTALL_SCRIPT_FILE  ${CMAKE_CURRENT_BINARY_DIR}/utils_postun.script)
 if(USE_OBCDC_NS)
-    set(CPACK_RPM_PACKAGE_REQUIRES "devdeps-libaio >= 0.3.112, devdeps-openssl-static >= 1.0.1e, oceanbase-ce-devel >= 3.1.3")
+    if (USE_LIBOBLOG_3)
+        set(CPACK_RPM_PACKAGE_REQUIRES "devdeps-libaio >= 0.3.112")
+    else()
+        set(CPACK_RPM_PACKAGE_REQUIRES "devdeps-libaio >= 0.3.112")
+    endif()
 else()
-    set(CPACK_RPM_PACKAGE_REQUIRES "devdeps-libaio >= 0.3.112, devdeps-openssl-static >= 1.0.1e, oceanbase-ce-devel = 3.1.2")
+    set(CPACK_RPM_PACKAGE_REQUIRES "devdeps-libaio >= 0.3.112, oceanbase-ce-devel = 3.1.2")
 endif()
 # install cpack to make everything work
 include(CPack)

conf/conf.json

@@ -25,7 +25,7 @@
   "ob_sys_password": "",
   "counter_interval_s": 2,
   "metric_enable": true,
-  "metric_interval_s": 120,
+  "metric_interval_s": 10,
   "debug": false,
   "verbose": false,
   "verbose_packet": false,

deps/dep_create.sh

@@ -10,7 +10,9 @@ if [[ ! -z "$1" ]]; then
   mkdir -p ${TARGET_DIR}
 fi
 
-echo "dep_create.sh in ${PWD}, target dir: ${TARGET_DIR}"
+LIBOBLOG_RPM_NAME=$2
+
+echo "dep_create.sh in ${PWD}, target dir: ${TARGET_DIR}, liboblog rpm name: ${LIBOBLOG_RPM_NAME}"
 
 OS_ARCH="$(uname -p)" || exit 1
 OS_RELEASE="0"
@@ -121,15 +123,22 @@ RPMS="$(grep '\.rpm' "${DEP_FILE}" | grep -Pv '^#')"
 for pkg in $RPMS
 do
   if [[ -f "${TARGET_DIR}/pkg/${pkg}" ]]; then
-    echo "find package <${pkg}> in cache"
+    echo -e "find package <${pkg}> in cache... \c"
   else
     echo -e "download package <${pkg}>... \c"
     TEMP=$(mktemp -p "/" -u ".${pkg}.XXXX")
 
-    if [[ ! -z `echo "${pkg}" | grep 'oceanbase-ce-devel'` ]]; then
-      wget "$STABLE_REPO/${pkg}" -q -O "${TARGET_DIR}/pkg/${TEMP}"
-    elif [[ ! -z `echo "${pkg}" | grep 'oceanbase-ce-cdc'` ]]; then
-      wget "$STABLE_REPO/${pkg}" -q -O "${TARGET_DIR}/pkg/${TEMP}"
+    if [ "${LIBOBLOG_RPM_NAME}" == 'oceanbase-ce-devel' ] && [ ! -z `echo "${pkg}" | grep 'devdeps-mariadb-connector-c'` ]; then
+      echo "SKIP"
+      continue
+    fi
+    if [ ! -z `echo "${pkg}" | grep 'oceanbase-ce-devel'` ] || [ ! -z `echo "${pkg}" | grep 'oceanbase-ce-cdc'` ]; then
+      if [[ ! -z `echo "${pkg}" | grep ${LIBOBLOG_RPM_NAME}` ]]; then
+        wget "$STABLE_REPO/${pkg}" -q -O "${TARGET_DIR}/pkg/${TEMP}"
+      else
+        echo "SKIP"
+        continue
+      fi
     else
       wget "$REPO/${pkg}" -q -O "${TARGET_DIR}/pkg/${TEMP}"
     fi

deps/oblogproxy.el7.x86_64.deps

@@ -9,6 +9,7 @@ devdeps-openssl-static-1.0.1e-12022100422.el7.x86_64.rpm
 devdeps-libaio-0.3.112-12022092915.el7.x86_64.rpm
 devdeps-mariadb-connector-c-3.1.12-12022100422.el7.x86_64.rpm
 oceanbase-ce-cdc-4.0.0.0-10000052022110116.el7.x86_64.rpm
+oceanbase-ce-devel-3.1.4-10000092022071511.el7.x86_64.rpm
 
 [tools]
 obdevtools-gcc9-9.3.0-52022092914.el7.x86_64.rpm
\ No newline at end of file

deps/oblogproxy.el8.x86_64.deps

@@ -9,6 +9,7 @@ devdeps-openssl-static-1.0.1e-12022100422.el8.x86_64.rpm
 devdeps-libaio-0.3.112-12022092915.el8.x86_64.rpm
 devdeps-mariadb-connector-c-3.1.12-12022100422.el8.x86_64.rpm
 oceanbase-ce-cdc-4.0.0.0-10000052022110116.el8.x86_64.rpm
+oceanbase-ce-devel-3.1.4-10000092022071511.el8.x86_64.rpm
 
 [tools]
 obdevtools-gcc9-9.3.0-52022092914.el8.x86_64.rpm

package.sh

+#!/usr/bin/env
+
+ce=$1
+version=$2
+if [ -z "${ce}" ] || [ -z "${version}" ]; then
+    echo "Invalid input arguments"
+    exit -1
+fi
+
+file_path=$(dirname $0 | xargs readlink -f)
+
+function compile_ce() {
+  liboblog_3x_flag=$1
+  name=$2
+
+  echo "building and packaging ${name}"
+
+  cd ${file_path} && \
+  mkdir -p packenv && cd packenv && \
+  ls | grep -v *.tar.gz | xargs rm -rf && \
+  cmake -DOBLOGPROXY_INSTALL_PREFIX=`pwd`/oblogproxy -DUSE_LIBOBLOG_3=${liboblog_3x_flag} .. && \
+  make -j $(grep -c ^processor /proc/cpuinfo) install oblogproxy && \
+  tar -zcf ${name}.tar.gz oblogproxy
+}
+
+tm=$(date +%Y%m%d%H%M%S)
+
+compile_ce ON oblogproxy-ce-for-3x-${version}-${tm}
+
+compile_ce OFF oblogproxy-ce-for-4x-${version}-${tm}

script/run.sh

@@ -4,6 +4,8 @@ cd $(dirname $0)
 DEPLOY_PATH=$(pwd)
 echo "DEPLOY_PATH : "${DEPLOY_PATH}
 
+LIB_PATH=${DEPLOY_PATH}/liboblog
+
 BIN='logproxy'
 GPID=0
 function is_running() {
@@ -62,6 +64,9 @@ start() {
   fi
   log_path=$(readlink -f ${log_path})
 
+  if [ ! -z "${LIB_PATH}" ]; then
+    export LD_LIBRARY_PATH=${LIB_PATH}:${LD_LIBRARY_PATH}
+  fi
   chmod u+x ./bin/${BIN} && ./bin/${BIN} -f ./conf/conf.json &>${log_path}/out.log &
   if [ $? -ne 0 ]; then
     exit -1
@@ -82,6 +87,9 @@ do_config_sys() {
     exit -1
   fi
 
+  if [ ! -z "${LIB_PATH}" ]; then
+    export LD_LIBRARY_PATH=${LIB_PATH}:${LD_LIBRARY_PATH}
+  fi
   username_x=`./bin/${BIN} -x ${username}`
   password_x=`./bin/${BIN} -x ${password}`
 

src/communication/channel.h

@@ -43,7 +43,7 @@ public:
   {
     if (_owned_fd && _peer.fd != 0) {
       close(_peer.fd);
-      OMS_INFO << "Closed fd: " << _peer.fd;
+      OMS_DEBUG << "Closed fd: " << _peer.fd;
     }
     free(_read_event);
     free(_write_event);

src/communication/comm.cpp

@@ -194,7 +194,9 @@ int Comm::add(const Peer& peer)
   }
   ch.set_communicator(this);
 
-  OMS_INFO << "Add channel, peer: " << ch.peer().to_string();
+  if (_s_conf.verbose.val()) {
+    OMS_INFO << "Add channel, peer: " << ch.peer().to_string();
+  }
 
   // we use level trigger as simple, and expect READ first for handshake packet
   int ret = event_assign(ch._read_event, _event_base, peer.fd, EV_READ | EV_PERSIST, _s_evcb_on_event, &ch);
@@ -211,7 +213,9 @@ int Comm::add(const Peer& peer)
     return OMS_FAILED;
   }
 
-  OMS_INFO << "Add read channel to Communicator with peer: " << peer.to_string();
+  if (_s_conf.verbose.val()) {
+    OMS_INFO << "Add read channel to Communicator with peer: " << peer.to_string();
+  }
   return OMS_OK;
 }
 
@@ -241,7 +245,7 @@ void Comm::_s_evcb_on_event(int fd, short event, void* arg)
   } else {
 
     if ((event & EV_WRITE)) {
-      OMS_INFO << "On event about to write message: " << ch.peer().to_string();
+      OMS_DEBUG << "On event about to write message: " << ch.peer().to_string();
       if (!_s_msg_queue.empty()) {
         const Message* msg = _s_msg_queue.front();
         _s_msg_queue.pop_front();
@@ -263,7 +267,9 @@ void Comm::_s_evcb_on_event(int fd, short event, void* arg)
       } else if (result == PacketError::IGNORE) {
         // do nothing
       } else {
-        OMS_ERROR << "Failed to handle read message, ret: " << (int)result;
+        if (_s_conf.verbose.val()) {
+          OMS_ERROR << "Failed to handle read message, ret: " << (int)result;
+        }
         err = EventResult::ER_CLOSE_CHANNEL;
       }
       delete msg;

src/metric/sys_metric.cpp

@@ -23,7 +23,9 @@
 namespace oceanbase {
 namespace logproxy {
 
-static const int64_t MEM_DEFAULT = INT64_MAX;
+// The maximum value of cgoup memory limit, in KB. Greater than or equal to this value means that docker has no limit on
+// memory
+static const int64_t MEM_DEFAULT = 9223372036854771712;
 static const int64_t UNIT_GB = 1024 * 1024 * 1024L;
 static const int64_t UNIT_MB = 1024 * 1024L;
 
@@ -44,7 +46,7 @@ bool init_metric()
   }
   std::vector<std::string> nets;
   split_by_str(result, " ", nets);
-  if (nets.size() != 5 || nets[4].empty()) {
+  if (nets.size() < 5 || nets[4].empty()) {
     OMS_ERROR << "Failed to collect network for invalid content:" << result;
     return false;
   }

src/obaccess/clog_meta_routine.cpp

@@ -85,8 +85,13 @@ int ClogMetaRoutine::fetch_once(uint64_t& min_clog_timestamp_us)
 
   min_clog_timestamp_us = 0;
   for (const MySQLRow& row : rs.rows) {
-    uint64_t tmp = atoll(row.fields().front().c_str());
-    min_clog_timestamp_us = std::max(min_clog_timestamp_us, tmp);
+    int64_t tmp = atoll(row.fields().front().c_str());
+    // The query result will contain -1, indicating that there is currently no clog related information. So when the
+    // result value is less than 0, the clog start point is considered to be 0
+    if (tmp < 0) {
+      tmp = 0;
+    }
+    min_clog_timestamp_us = std::max(min_clog_timestamp_us, (uint64_t)tmp);
   }
   return OMS_OK;
 }

src/oblogreader/oblog_access.h

@@ -21,12 +21,22 @@
 #ifdef USE_OBCDC_NS
 #include "libobcdc.h"
 
+#ifdef USE_LIBOBLOG_3
+
+using oceanbase::liboblog::IObLog;
+using oceanbase::liboblog::ObLogError;
+using oceanbase::liboblog::ObLogFactory;
+
+#else
+
 typedef oceanbase::libobcdc::IObCDCInstance IObLog;
 typedef oceanbase::libobcdc::ObCDCFactory ObLogFactory;
 typedef oceanbase::libobcdc::ObCDCError ObLogError;
 
 #define construct_oblog construct_obcdc
 
+#endif
+
 #else
 #include "liboblog.h"
 

src/oblogreader/reader_routine.cpp

@@ -92,7 +92,9 @@ void ReaderRoutine::run()
       ::usleep(_s_config.read_fail_interval_us.val());
       continue;
     }
-
+    record_us = record->getTimestamp() * 1000000 + record->getRecordUsec();
+    int record_size = record->getRealSize();
+    // once put record to queue, never access it
     stage_tm.reset();
     while (!_queue.offer(record, _s_config.read_timeout_us.val())) {
       OMS_WARN << "reader transfer queue full(" << _queue.size(false) << "), retry...";
@@ -101,10 +103,8 @@ void ReaderRoutine::run()
 
     counter.count_key(Counter::READER_FETCH_US, fetch_us);
     counter.count_key(Counter::READER_OFFER_US, offer_us);
-    counter.count_read_io(record->getRealSize());
+    counter.count_read_io(record_size);
     counter.count_read(1);
-
-    record_us = record->getTimestamp() * 1000000 + record->getRecordUsec();
   }
 
   _reader.stop();