diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 57a43be2..e5c35ba5 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -61,8 +61,17 @@ jobs: run: | git config --global --add safe.directory '*' cd $GITHUB_WORKSPACE + + # Derive branch name from tag: 1.2.2 -> rel_1_2_2_eloqkv + REL_BRANCH="rel_$(echo "${{ inputs.tag_name }}" | sed 's/^v//' | tr '.' '_')_eloqkv" + git clone --recurse-submodules git@github.com:eloqdata/eloq_log_service.git logservice_src + git -C logservice_src checkout $REL_BRANCH + git -C logservice_src submodule update --recursive + git clone git@github.com:eloqdata/raft_host_manager.git raft_host_manager_src + git -C raft_host_manager_src checkout $REL_BRANCH + git -C raft_host_manager_src submodule update --recursive rm -rf eloqkv_src/data_substrate/eloq_log_service rm -rf eloqkv_src/data_substrate/tx_service/raft_host_manager diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml deleted file mode 100644 index 0930bca1..00000000 --- a/.github/workflows/test-build.yml +++ /dev/null @@ -1,156 +0,0 @@ -name: Test Build - -on: - workflow_dispatch: - inputs: - test_tag: - description: 'Test version (e.g. 0.0.0-test)' - required: true - default: '0.0.0-test' - -jobs: - build: - runs-on: ubuntu-latest - container: - image: eloqdata/eloq-dev-ci-ubuntu2404:latest - options: --user root - env: - DATA_STORE_TYPE: ROCKSDB - LOG_STATE_TYPE: ROCKSDB - BUILD_TYPE: RelWithDebInfo - BUILD_LOG_SRV: "true" - DATA_STORE_ID: rocksdb - TAGGED: ${{ github.event.inputs.test_tag }} - - steps: - # 1. SSH for private repos - - uses: webfactory/ssh-agent@v0.9.0 - with: - ssh-private-key: ${{ secrets.SSH_PRIVATE_KEY }} - - # 2. Checkout eloqkv source - - uses: actions/checkout@v4 - with: - submodules: recursive - path: eloqkv_src - - # 3. Clone private deps + symlink - - name: Setup private dependencies - env: - GIT_SSH_COMMAND: "ssh -o StrictHostKeyChecking=accept-new" - run: | - git config --global --add safe.directory '*' - cd $GITHUB_WORKSPACE - git clone --recurse-submodules git@github.com:eloqdata/eloq_log_service.git logservice_src - git clone git@github.com:eloqdata/raft_host_manager.git raft_host_manager_src - - ln -s $GITHUB_WORKSPACE/logservice_src \ - $GITHUB_WORKSPACE/eloqkv_src/data_substrate/eloq_log_service - ln -s $GITHUB_WORKSPACE/raft_host_manager_src \ - $GITHUB_WORKSPACE/eloqkv_src/data_substrate/tx_service/raft_host_manager - - # 4. Install build deps - - name: Install dependencies - run: | - apt-get update -qq - DEBIAN_FRONTEND=noninteractive apt-get install -y -qq \ - postgresql-client rsync patchelf - - # 5. Download eloqkv-cli from CDN (public, no credentials needed) - - name: Download eloqkv-cli - run: | - curl -fSL -o redis-cli \ - "https://download.eloqdata.com/eloqkv/client/eloqkv-cli-7.2.5-ubuntu24-amd64" - chmod +x redis-cli - - # 6. CMake build - - name: Build eloqkv - working-directory: ${{ github.workspace }}/eloqkv_src - run: | - mkdir -p build && cd build - cmake .. \ - -DCMAKE_BUILD_TYPE=$BUILD_TYPE \ - -DWITH_DATA_STORE=$DATA_STORE_TYPE \ - -DWITH_LOG_SERVICE=ON \ - -DWITH_LOG_STATE=$LOG_STATE_TYPE \ - -DDISABLE_CKPT_REPORT=ON \ - -DDISABLE_CODE_LINE_IN_LOG=ON \ - -DWITH_ASAN=OFF \ - -DOPEN_LOG_SERVICE=OFF \ - -DFORK_HM_PROCESS=ON - cmake --build . -j$(nproc) - - # 7. Package eloqkv tarball (EloqKV/{bin,lib,conf}) - - name: Package eloqkv tarball - run: | - DEST_DIR=$HOME/EloqKV - mkdir -p $DEST_DIR/bin $DEST_DIR/lib $DEST_DIR/conf - - cp eloqkv_src/build/eloqkv $DEST_DIR/bin/ - cp eloqkv_src/build/data_substrate/host_manager $DEST_DIR/bin/ - cp redis-cli $DEST_DIR/bin/eloqkv-cli - - # Exact copy_libraries() logic from concourse/scripts/build_tarball.bash - for bin in $DEST_DIR/bin/*; do - ldd "$bin" 2>/dev/null | awk 'NF==4 {print $(NF-1)}' | while read lib; do - [ -e "$lib" ] || continue - lib_name=$(basename "$lib") - case "$lib_name" in - libc.so*|libm.so*|libpthread.so*|libdl.so*|librt.so*|libanl.so*|libresolv.so*|libutil.so*|libnsl.so*|ld-linux*.so*) - continue ;; - *) cp -n "$lib" $DEST_DIR/lib/ 2>/dev/null || true ;; - esac - done - done - - echo "=== libs copied ===" - ls -la $DEST_DIR/lib/ - - find $DEST_DIR/lib \ - -name 'libbrpc*' -o -name 'libbraft*' \ - -o -name 'libleveldb*' -o -name 'librocksdb*' | while read f; do - patchelf --set-rpath '$ORIGIN' "$f" 2>/dev/null || true - done - - tar -czvf $GITHUB_WORKSPACE/eloqkv-${TAGGED}-${DATA_STORE_ID}-ubuntu24-amd64.tar.gz -C $HOME EloqKV - - # 8. Build and package log-service tarball (LogServer/{bin,lib}, independent) - - name: Package log-service tarball - run: | - LOG_SRC=$GITHUB_WORKSPACE/eloqkv_src/data_substrate/eloq_log_service - mkdir -p $LOG_SRC/LogServer/bin $LOG_SRC/LogServer/lib - - cd $LOG_SRC - mkdir -p build && cd build - cmake .. -DCMAKE_BUILD_TYPE=$BUILD_TYPE -DWITH_LOG_STATE=${LOG_STATE_TYPE} -DDISABLE_CODE_LINE_IN_LOG=ON - cmake --build . -j$(nproc) - - mv launch_sv $LOG_SRC/LogServer/bin/ - - ldd $LOG_SRC/LogServer/bin/launch_sv 2>/dev/null | awk 'NF==4 {print $(NF-1)}' | while read lib; do - [ -e "$lib" ] || continue - lib_name=$(basename "$lib") - case "$lib_name" in - libc.so*|libm.so*|libpthread.so*|libdl.so*|librt.so*|libanl.so*|libresolv.so*|libutil.so*|libnsl.so*|ld-linux*.so*) - continue ;; - *) cp -n "$lib" $LOG_SRC/LogServer/lib/ 2>/dev/null || true ;; - esac - done - - echo "=== log libs copied ===" - ls -la $LOG_SRC/LogServer/lib/ - - tar -czvf $GITHUB_WORKSPACE/log-service-${TAGGED}-${DATA_STORE_ID}-ubuntu24-amd64.tar.gz -C $LOG_SRC LogServer - - # 9. Upload to draft release - - name: Create Release and Upload Assets - uses: softprops/action-gh-release@v2 - with: - tag_name: ${{ github.event.inputs.test_tag }} - name: ${{ github.event.inputs.test_tag }} - files: | - eloqkv-${{ github.event.inputs.test_tag }}-rocksdb-ubuntu24-amd64.tar.gz - log-service-${{ github.event.inputs.test_tag }}-rocksdb-ubuntu24-amd64.tar.gz - draft: true - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/data_substrate b/data_substrate index 64554e46..9857f213 160000 --- a/data_substrate +++ b/data_substrate @@ -1 +1 @@ -Subproject commit 64554e4658f030b5f3e07a2833f1bc8c354ee1a8 +Subproject commit 9857f213c0c05a14cb16adc8233c81b6313432f5 diff --git a/include/redis_command.h b/include/redis_command.h index 5d282cfd..e5742927 100644 --- a/include/redis_command.h +++ b/include/redis_command.h @@ -97,6 +97,11 @@ #define SLOWLOG_RESET (1 << 1) #define SLOWLOG_LEN (1 << 2) +#if defined(DATA_STORE_TYPE_ELOQDSS_ROCKSDB_CLOUD_S3) || \ + defined(DATA_STORE_TYPE_ELOQDSS_ROCKSDB_CLOUD_GCS) +#define ELOQKV_WITH_DSS_ROCKSDB_CLOUD 1 +#endif + namespace EloqKV { enum struct RedisCommandType @@ -320,6 +325,9 @@ enum struct RedisCommandType #endif UNLINK, +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD + COMPACT, +#endif }; enum RedisResultType @@ -800,6 +808,9 @@ struct InfoCommand : public DirectCommand int64_t data_memory_committed_{0}; uint64_t last_ckpt_ts_{0}; int event_dispatcher_num_{0}; + std::string os_info_; + std::string executable_path_; + int64_t total_system_memory_kb_{0}; const char *version_{nullptr}; int64_t conn_received_count_{0}; @@ -813,7 +824,23 @@ struct InfoCommand : public DirectCommand int64_t cmds_per_sec_{0}; double cmd_latency_{0}; std::vector dbsizes_; +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD + uint64_t store_disk_keys_{0}; +#endif +}; + +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +struct CompactCommand : public DirectCommand +{ + CompactCommand() = default; + + void Execute(RedisServiceImpl *redis_impl, + RedisConnectionContext *ctx) override; + void OutputResult(OutputHandler *reply) const override; + + bool success_{false}; }; +#endif struct CommandCommand : public DirectCommand { @@ -7223,7 +7250,9 @@ struct DumpCommand : public RedisCommand RedisStringResult result_; + // Retained for accepting payloads produced before Redis DUMP compatibility. static constexpr inline uint16_t dump_version_{0}; + static constexpr inline uint16_t redis_dump_version_{10}; }; struct RestoreCommand : public RedisCommand @@ -7829,6 +7858,11 @@ std::tuple ParseConfigCommand( std::tuple ParseInfoCommand( const std::vector &args, OutputHandler *output); +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +std::tuple ParseCompactCommand( + const std::vector &args, OutputHandler *output); +#endif + std::tuple> ParseCommandCommand( const std::vector &args, OutputHandler *output); diff --git a/include/redis_handler.h b/include/redis_handler.h index c87714a8..7af02d0c 100644 --- a/include/redis_handler.h +++ b/include/redis_handler.h @@ -240,6 +240,26 @@ class InfoCommandHandler : public RedisCommandHandler RedisServiceImpl *redis_impl_; }; +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +class CompactCommandHandler : public RedisCommandHandler +{ +public: + explicit CompactCommandHandler(RedisServiceImpl *redis_impl) + : redis_impl_(redis_impl) + { + } + + brpc::RedisCommandHandlerResult Run( + RedisConnectionContext *ctx, + const std::vector &args, + brpc::RedisReply *output, + bool /*flush_batched*/) override; + +private: + RedisServiceImpl *redis_impl_; +}; +#endif + class CommandCommandHandler : public RedisCommandHandler { public: diff --git a/include/redis_rdb_restore.h b/include/redis_rdb_restore.h index 8059360f..a1001b60 100644 --- a/include/redis_rdb_restore.h +++ b/include/redis_rdb_restore.h @@ -5,6 +5,8 @@ namespace EloqKV { +struct RedisEloqObject; + enum class RestorePayloadFormat { EloqKV, @@ -13,4 +15,7 @@ enum class RestorePayloadFormat bool ConvertRedisDumpPayloadToEloqPayload(std::string_view dump_payload, std::string &eloq_payload); + +bool ConvertEloqObjectToRedisDumpPayload(const RedisEloqObject &object, + std::string &dump_payload); } // namespace EloqKV diff --git a/include/redis_service.h b/include/redis_service.h index dcdcea66..89299cc2 100644 --- a/include/redis_service.h +++ b/include/redis_service.h @@ -265,6 +265,11 @@ class RedisServiceImpl : public brpc::RedisService const std::vector &args, brpc::RedisReply *output); + store::DataStoreHandler *GetStoreHandler() const + { + return store_hd_; + } + void GenericCommand(RedisConnectionContext *ctx, TransactionExecution *txm, const std::vector &cmd_arg_list, @@ -470,6 +475,18 @@ class RedisServiceImpl : public brpc::RedisService { return event_dispatcher_num_; } + const std::string &GetOsInfo() const + { + return os_info_; + } + const std::string &GetExecutablePath() const + { + return executable_path_; + } + int64_t GetTotalSystemMemoryKB() const + { + return total_system_memory_kb_; + } TxService *GetTxService() { return tx_service_; @@ -570,6 +587,9 @@ class RedisServiceImpl : public brpc::RedisService std::string config_file_; uint32_t node_memory_limit_mb_; int event_dispatcher_num_; + std::string os_info_; + std::string executable_path_; + int64_t total_system_memory_kb_{0}; const char *version_; // Isolation level and concurrency control protocol of MULTI/EXEC or lua // transactions. ReadCommitted and OccRead are always used for simple diff --git a/src/redis_command.cpp b/src/redis_command.cpp index 58aa02e6..72cd2d66 100644 --- a/src/redis_command.cpp +++ b/src/redis_command.cpp @@ -33,6 +33,7 @@ #include #include #include +#include #include #include #include @@ -55,7 +56,6 @@ #include "eloqkv_key.h" #include "local_cc_shards.h" #include "output_handler.h" -#include "redis/commands.h" #include "redis/server.h" #include "redis_connection_context.h" #include "redis_errors.h" @@ -99,6 +99,9 @@ const std::vector> command_types{{ {"info", RedisCommandType::INFO}, {"cluster", RedisCommandType::CLUSTER}, {"dbsize", RedisCommandType::DBSIZE}, +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD + {"compact", RedisCommandType::COMPACT}, +#endif {"time", RedisCommandType::TIME}, {"slowlog", RedisCommandType::SLOWLOG}, {"select", RedisCommandType::SELECT}, @@ -1569,6 +1572,9 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl, event_dispatcher_num_ = redis_impl->GetEventDispatcherNum(); version_ = redis_impl->GetVersion(); + os_info_ = redis_impl->GetOsInfo(); + executable_path_ = redis_impl->GetExecutablePath(); + total_system_memory_kb_ = redis_impl->GetTotalSystemMemoryKB(); max_connection_count_ = redis_impl->MaxConnectionCount(); if (redis_impl->IsEnableRedisStats()) @@ -1584,6 +1590,7 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl, // cmds_per_sec_ = RedisStats::GetCommandsPerSecond(); } +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD if (set_section_.size() == 0 || set_section_.find("keyspace") != set_section_.end()) { @@ -1597,8 +1604,33 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl, } dbsizes_ = DBSizeCommand::FetchDBSize(std::move(table_names)); + auto *store_hd = redis_impl->GetStoreHandler(); + store_disk_keys_ = + store_hd == nullptr ? 0 : store_hd->ApproxStoreKeyCount(); + } +#endif +} + +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +void CompactCommand::Execute(RedisServiceImpl *redis_impl, + RedisConnectionContext *ctx) +{ + auto *store_hd = redis_impl->GetStoreHandler(); + success_ = store_hd != nullptr && store_hd->CompactStore(); +} + +void CompactCommand::OutputResult(OutputHandler *reply) const +{ + if (success_) + { + reply->OnStatus(redis_get_error_messages(RD_OK)); + } + else + { + reply->OnError("ERR failed to compact data store"); } } +#endif void ClusterCommand::Execute(RedisServiceImpl *redis_impl, RedisConnectionContext *ctx) @@ -2095,36 +2127,32 @@ void PingCommand::OutputResult(OutputHandler *reply) const reply->OnString(reply_value_.StringView()); } -std::string ExecCommand(const std::string &cmd) +static uint64_t GetProcessVmRSSKb(pid_t pid) { - char line[1024]; - FILE *fp; - const char *sysCommand = cmd.data(); - if ((fp = popen(sysCommand, "r")) == NULL) + std::ifstream file("/proc/" + std::to_string(pid) + "/status"); + if (!file.is_open()) { - return "Failed to execute command '" + cmd + "'!"; + return 0; } - std::string rst; - while (fgets(line, sizeof(line) - 1, fp) != NULL) + std::string line; + while (std::getline(file, line)) { - if (rst.size() > 0 && *rst.rbegin() != '\n') + if (line.rfind("VmRSS:", 0) != 0) { - rst += "\n"; + continue; } - - rst += line; - } - pclose(fp); - - if (*rst.rbegin() == '\n') - { - return rst.substr(0, rst.size() - 1); - } - else - { - return rst; + std::istringstream iss(line); + std::string key; + uint64_t value = 0; + std::string unit; + if (iss >> key >> value >> unit) + { + return value; + } + return 0; } + return 0; } void InfoCommand::OutputResult(OutputHandler *reply) const @@ -2141,14 +2169,13 @@ void InfoCommand::OutputResult(OutputHandler *reply) const result += "\r\neloqkv_mode:"; result += (node_count_ == 1 ? "standalone" : "cluster"); result += "\r\nmultiplexing_api:epoll"; - result += "\r\nos:" + ExecCommand("uname -srm"); + result += "\r\nos:" + os_info_; result += "\r\nprocess_id:" + pid; result += "\r\ntcp_port:" + std::to_string(tcp_port_); result += "\r\nuptime_in_seconds:" + std::to_string(uptime_in_secs_); result += "\r\nuptime_in_days:" + std::to_string(uptime_in_secs_ / (3600 * 24)); - result += "\r\nexecutable:" + ExecCommand("ls -l /proc/" + pid + - "/exe | awk '{print $11}'"); + result += "\r\nexecutable:" + executable_path_; result += "\r\nconfig_file:" + config_file_; result += "\r\nbrpc_io_threads_active:" + std::to_string(event_dispatcher_num_); @@ -2182,11 +2209,9 @@ void InfoCommand::OutputResult(OutputHandler *reply) const result += "# Memory"; result += "\r\nused_memory_rss:" + - ExecCommand("cat /proc/" + pid + - "/status | grep VmRSS | awk '{print $2}'") + - " kb"; + std::to_string(GetProcessVmRSSKb(getpid())) + " kb"; result += "\r\ntotal_system_memory:" + - ExecCommand("free | grep Mem | awk '{print $2}'") + " kb"; + std::to_string(total_system_memory_kb_) + " kb"; result += "\r\ndata_memory_allocated:" + std::to_string(data_memory_allocated_) + " kb"; result += "\r\ndata_memory_committed:" + @@ -2282,6 +2307,7 @@ void InfoCommand::OutputResult(OutputHandler *reply) const } } +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD if (set_section_.size() == 0 || set_section_.find("keyspace") != set_section_.end()) { @@ -2299,7 +2325,9 @@ void InfoCommand::OutputResult(OutputHandler *reply) const ":keys=" + std::to_string(dbsizes_[idx]); } } + result += "\r\nstore:disk_keys=" + std::to_string(store_disk_keys_); } +#endif result += "\r\n"; reply->OnString(result); @@ -6522,7 +6550,9 @@ void SMoveCommand::OutputResult(OutputHandler *reply) const { if (cmd_dst_->result_.err_code_ == RD_OK) { - reply->OnInt(cmd_dst_->result_.ret_); + // SMOVE reports whether the member existed in the source. The + // destination may already contain it, making SADD a no-op. + reply->OnInt(cmd_src_->result_.ret_); } else { @@ -8034,6 +8064,19 @@ ParseMultiCommand(RedisServiceImpl *redis_impl, success, DirectRequest(ctx, std::make_unique(std::move(cmd)))}; } +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD + case RedisCommandType::COMPACT: + { + auto [success, cmd] = ParseCompactCommand(args, output); + if (!success) + { + return {false, DirectRequest{}}; + } + return {success, + DirectRequest( + ctx, std::make_unique(std::move(cmd)))}; + } +#endif case RedisCommandType::CLUSTER: { auto [success, cmd] = ParseClusterCommand(args, output); @@ -10287,6 +10330,20 @@ std::tuple ParseInfoCommand( return {true, InfoCommand(std::move(set_sct))}; } +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +std::tuple ParseCompactCommand( + const std::vector &args, OutputHandler *output) +{ + if (args.size() != 1) + { + output->OnError("ERR wrong number of arguments for 'compact' command"); + return {false, CompactCommand()}; + } + + return {true, CompactCommand()}; +} +#endif + std::tuple> ParseCommandCommand( const std::vector &args, OutputHandler *output) { @@ -15453,93 +15510,26 @@ void ScanCommand::OutputResult(OutputHandler *reply, } } -/** - * +-----------+--------------------------+-------------------+------------+ - * |1 byte type|Serialized RedisEloqObject|2 byte MONO version|8 byte CRC64| - * +-----------+--------------------------+-------------------+------------+ - * - * MONO version and CRC are both in big endian. - */ txservice::ExecResult DumpCommand::ExecuteOn(const txservice::TxObject &object) { const RedisEloqObject &eloq_obj = dynamic_cast(object); - RedisObjectType obj_type = eloq_obj.ObjectType(); - assert(obj_type != RedisObjectType::Unknown && - obj_type != RedisObjectType::Del); - - uint8_t obj_type_u8 = static_cast(obj_type); - result_.str_.append(reinterpret_cast(&obj_type_u8), - sizeof(uint8_t)); - - // If object has TTL, serialize as base type (without TTL) for DUMP - // This matches Redis semantics where DUMP doesn't include TTL info - // We call the base class Serialize() method directly instead of creating - // a new object - if (eloq_obj.HasTTL()) - { - // Cast to base class and call base class Serialize() method - // ObjectType() already returns the base class type, so we can safely - // cast to the corresponding base class - switch (obj_type) - { - case RedisObjectType::String: - { - // TTLString -> String: cast to base class and call base Serialize() - const RedisStringObject *base_obj = - static_cast(&eloq_obj); - base_obj->RedisStringObject::Serialize(result_.str_); - break; - } - case RedisObjectType::List: - { - // TTLList -> List: cast to base class and call base Serialize() - const RedisListObject *base_obj = - static_cast(&eloq_obj); - base_obj->RedisListObject::Serialize(result_.str_); - break; - } - case RedisObjectType::Hash: - { - // TTLHash -> Hash: cast to base class and call base Serialize() - const RedisHashObject *base_obj = - static_cast(&eloq_obj); - base_obj->RedisHashObject::Serialize(result_.str_); - break; - } - case RedisObjectType::Zset: - { - // TTLZset -> Zset: cast to base class and call base Serialize() - const RedisZsetObject *base_obj = - static_cast(&eloq_obj); - base_obj->RedisZsetObject::Serialize(result_.str_); - break; - } - case RedisObjectType::Set: - { - // TTLSet -> Set: cast to base class and call base Serialize() - const RedisHashSetObject *base_obj = - static_cast(&eloq_obj); - base_obj->RedisHashSetObject::Serialize(result_.str_); - break; - } - default: - assert(false && "Unsupported object type for TTL serialization"); - break; - } - } - else + if (!ConvertEloqObjectToRedisDumpPayload(eloq_obj, result_.str_)) { - eloq_obj.Serialize(result_.str_); + result_.err_code_ = RD_ERR_SYNTAX; + return txservice::ExecResult::Read; } - result_.str_.append(reinterpret_cast(&dump_version_), - sizeof(dump_version_)); + result_.str_.append(reinterpret_cast(&redis_dump_version_), + sizeof(redis_dump_version_)); - uint64_t crc64 = - crc64speed_big(0, result_.str_.data(), result_.str_.size()); - result_.str_.append(reinterpret_cast(&crc64), sizeof(crc64)); + uint64_t checksum = crc64(0, result_.str_.data(), result_.str_.size()); +#if __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__ + checksum = __builtin_bswap64(checksum); +#endif + result_.str_.append(reinterpret_cast(&checksum), + sizeof(checksum)); result_.err_code_ = RD_OK; diff --git a/src/redis_handler.cpp b/src/redis_handler.cpp index 5c968aba..6f1bdb04 100644 --- a/src/redis_handler.cpp +++ b/src/redis_handler.cpp @@ -629,6 +629,28 @@ brpc::RedisCommandHandlerResult InfoCommandHandler::Run( return brpc::REDIS_CMD_HANDLED; } +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD +brpc::RedisCommandHandlerResult CompactCommandHandler::Run( + RedisConnectionContext *ctx, + const std::vector &args, + brpc::RedisReply *output, + bool /*flush_batched*/) +{ + assert(args[0] == "compact"); + + RedisReplier reply(output); + std::vector cmd_arg_list = Transform(args); + + auto [success, cmd] = ParseCompactCommand(cmd_arg_list, &reply); + if (success) + { + redis_impl_->ExecuteCommand(ctx, &cmd, &reply); + } + + return brpc::REDIS_CMD_HANDLED; +} +#endif + brpc::RedisCommandHandlerResult FlushDBCommandHandler::Run( RedisConnectionContext *ctx, const std::vector &args, diff --git a/src/redis_rdb_restore.cpp b/src/redis_rdb_restore.cpp index 0c4863eb..a31497cf 100644 --- a/src/redis_rdb_restore.cpp +++ b/src/redis_rdb_restore.cpp @@ -71,6 +71,62 @@ constexpr uint8_t kQuickListNodeContainerPacked = 2; constexpr size_t kMaxLzfDecodedLength = 64ULL * 1024ULL * 1024ULL; constexpr size_t kMaxCollectionEntries = 1ULL << 20; +void AppendBE32(uint32_t value, std::string &output) +{ + output.push_back(static_cast((value >> 24) & 0xFF)); + output.push_back(static_cast((value >> 16) & 0xFF)); + output.push_back(static_cast((value >> 8) & 0xFF)); + output.push_back(static_cast(value & 0xFF)); +} + +void AppendBE64(uint64_t value, std::string &output) +{ + for (int shift = 56; shift >= 0; shift -= 8) + { + output.push_back(static_cast((value >> shift) & 0xFF)); + } +} + +void AppendRdbLen(uint64_t length, std::string &output) +{ + if (length < (1ULL << 6)) + { + output.push_back(static_cast(length)); + } + else if (length < (1ULL << 14)) + { + output.push_back(static_cast(0x40 | (length >> 8))); + output.push_back(static_cast(length & 0xFF)); + } + else if (length <= std::numeric_limits::max()) + { + output.push_back(static_cast(0x80)); + AppendBE32(static_cast(length), output); + } + else + { + output.push_back(static_cast(0x81)); + AppendBE64(length, output); + } +} + +void AppendRdbString(std::string_view value, std::string &output) +{ + AppendRdbLen(value.size(), output); + output.append(value.data(), value.size()); +} + +void AppendLEDouble(double value, std::string &output) +{ + uint64_t bits = 0; + static_assert(sizeof(bits) == sizeof(value)); + std::memcpy(&bits, &value, sizeof(bits)); + for (size_t i = 0; i < sizeof(bits); ++i) + { + output.push_back(static_cast((bits >> (i * 8)) & 0xFF)); + } +} + void PrefixEloqOuterType(RedisObjectType type, std::string &payload) { uint8_t obj_type = static_cast(type); @@ -1364,4 +1420,71 @@ bool ConvertRedisDumpPayloadToEloqPayload(std::string_view dump_payload, eloq_payload.clear(); return DecodeRedisObjectPayload(dump_payload, eloq_payload); } + +bool ConvertEloqObjectToRedisDumpPayload(const RedisEloqObject &object, + std::string &dump_payload) +{ + dump_payload.clear(); + + switch (object.ObjectType()) + { + case RedisObjectType::String: + { + const auto &string_object = + static_cast(object); + dump_payload.push_back(static_cast(kRdbTypeString)); + AppendRdbString(string_object.StringView(), dump_payload); + return true; + } + case RedisObjectType::List: + { + const auto &list_object = static_cast(object); + dump_payload.push_back(static_cast(kRdbTypeList)); + AppendRdbLen(list_object.Elements().size(), dump_payload); + for (const auto &element : list_object.Elements()) + { + AppendRdbString(element.StringView(), dump_payload); + } + return true; + } + case RedisObjectType::Set: + { + const auto &set_object = + static_cast(object); + dump_payload.push_back(static_cast(kRdbTypeSet)); + AppendRdbLen(set_object.Elements().size(), dump_payload); + for (const auto &element : set_object.Elements()) + { + AppendRdbString(element.StringView(), dump_payload); + } + return true; + } + case RedisObjectType::Hash: + { + const auto &hash_object = static_cast(object); + dump_payload.push_back(static_cast(kRdbTypeHash)); + AppendRdbLen(hash_object.Elements().size(), dump_payload); + for (const auto &[field, value] : hash_object.Elements()) + { + AppendRdbString(field.StringView(), dump_payload); + AppendRdbString(value.StringView(), dump_payload); + } + return true; + } + case RedisObjectType::Zset: + { + const auto &zset_object = static_cast(object); + dump_payload.push_back(static_cast(kRdbTypeZSet2)); + AppendRdbLen(zset_object.Elements().size(), dump_payload); + for (const auto &[member, score] : zset_object.Elements()) + { + AppendRdbString(member, dump_payload); + AppendLEDouble(score, dump_payload); + } + return true; + } + default: + return false; + } +} } // namespace EloqKV diff --git a/src/redis_service.cpp b/src/redis_service.cpp index 7b8bf9f9..6838ff8f 100644 --- a/src/redis_service.cpp +++ b/src/redis_service.cpp @@ -31,6 +31,7 @@ #include #include #include +#include #include #include @@ -39,9 +40,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -139,6 +142,35 @@ DEFINE_bool(enable_tls, false, "Enable TLS for brpc RPC connections"); DEFINE_string(tls_cert_file, "", "Path to TLS certificate file (PEM format)"); DEFINE_string(tls_key_file, "", "Path to TLS private key file (PEM format)"); +std::string ExecCommand(const std::string &cmd) +{ + char line[1024]; + FILE *fp; + const char *sysCommand = cmd.data(); + if ((fp = popen(sysCommand, "r")) == NULL) + { + return "Failed to execute command '" + cmd + "'!"; + } + + std::string rst; + while (fgets(line, sizeof(line) - 1, fp) != NULL) + { + if (rst.size() > 0 && *rst.rbegin() != '\n') + { + rst += "\n"; + } + + rst += line; + } + pclose(fp); + + if (!rst.empty() && rst.back() == '\n') + { + return rst.substr(0, rst.size() - 1); + } + return rst; +} + namespace EloqKV { const auto NUM_VCPU = std::thread::hardware_concurrency(); @@ -523,6 +555,51 @@ bool RedisServiceImpl::Init(brpc::Server &brpc_server) } } + // Capture static system info at startup (avoid fork/exec on every INFO). + { + struct utsname uts; + if (uname(&uts) == 0) + { + os_info_ = std::string(uts.sysname) + " " + uts.release + " " + + uts.machine; + } + } + { + char buf[4096]; + ssize_t len = readlink("/proc/self/exe", buf, sizeof(buf) - 1); + if (len > 0) + { + buf[len] = '\0'; + executable_path_ = buf; + } + } + { + long total_mem_kb = 0; + std::ifstream file("/proc/meminfo"); + if (file.is_open()) + { + std::string line; + while (std::getline(file, line)) + { + if (line.rfind("MemTotal:", 0) != 0) + { + continue; + } + std::istringstream iss(line); + std::string key; + uint64_t value = 0; + std::string unit; + if (iss >> key >> value >> unit) + { + total_mem_kb = static_cast(value); + } + break; + } + } + total_system_memory_kb_ = static_cast(total_mem_kb); + } + event_dispatcher_num_ = brpc::FLAGS_event_dispatcher_num; + return true; } @@ -1271,6 +1348,12 @@ void RedisServiceImpl::AddHandlers() hd_vec_.emplace_back(std::make_unique(this)); AddCommandHandler("info", info_hd.get()); +#ifdef ELOQKV_WITH_DSS_ROCKSDB_CLOUD + auto &compact_hd = + hd_vec_.emplace_back(std::make_unique(this)); + AddCommandHandler("compact", compact_hd.get()); +#endif + auto &command_hd = hd_vec_.emplace_back(std::make_unique(this)); AddCommandHandler("command", command_hd.get()); diff --git a/tests/unit/eloq/dump.tcl b/tests/unit/eloq/dump.tcl index f1d31721..6eb75067 100644 --- a/tests/unit/eloq/dump.tcl +++ b/tests/unit/eloq/dump.tcl @@ -14,6 +14,27 @@ start_server {tags {"dump"}} { assert_equal $stringval [r get stringobj] } + test {dump produces redis rdb version 10 payload} { + r set dump_version value + set dumpval [r dump dump_version] + binary scan [string range $dumpval end-9 end-8] H* version + assert_equal 0a00 $version + } + + test {dump restore preserves binary and numeric-looking strings} { + r del binary_string numeric_string + set binaryval [binary format H* "0030303100ff"] + r set binary_string $binaryval + r set numeric_string "001" + set binarydump [r dump binary_string] + set numericdump [r dump numeric_string] + r del binary_string numeric_string + r restore binary_string 0 $binarydump + r restore numeric_string 0 $numericdump + assert_equal $binaryval [r get binary_string] + assert_equal "001" [r get numeric_string] + } + test {dump restore hash} { r del hashobj r hset hashobj k1 v1 k2 v2 k3 v3 @@ -35,11 +56,12 @@ start_server {tags {"dump"}} { test {dump restore zset} { r del zsetobj - r zadd zsetobj 1 k1 2 k2 3 k3 + r zadd zsetobj 1 k1 2.123456789012345 k2 3 k3 set dumpval [r dump zsetobj] r del zsetobj r restore zsetobj 0 $dumpval assert_equal {k1 k2 k3} [r zrange zsetobj 0 -1] + assert_equal 2.123456789012345 [r zscore zsetobj k2] } test {RESTORE are able to serialize / unserialize a simple key} { @@ -59,6 +81,15 @@ start_server {tags {"dump"}} { r get foo } {bar} + test {DUMP does not carry the source ttl} { + r set ttl_source value px 5000 + set encoded [r dump ttl_source] + r del ttl_source + r restore ttl_source 0 $encoded + assert_equal -1 [r pttl ttl_source] + assert_equal value [r get ttl_source] + } + test {RESTORE can set an expire that overflows a 32 bit integer} { r set foo bar set encoded [r dump foo] diff --git a/tests/unit/eloq/set.tcl b/tests/unit/eloq/set.tcl index 857ba332..daefe0e0 100644 --- a/tests/unit/eloq/set.tcl +++ b/tests/unit/eloq/set.tcl @@ -1187,6 +1187,15 @@ foreach type {single multiple single_multiple} { assert_encoding intset myset3{t} } + test "SMOVE returns one when destination already contains the member" { + r del srcset{t} dstset{t} + r sadd srcset{t} a b + r sadd dstset{t} a c + assert_equal 1 [r smove srcset{t} dstset{t} a] + assert_equal {b} [lsort [r smembers srcset{t}]] + assert_equal {a c} [lsort [r smembers dstset{t}]] + } + test "SMOVE wrong src key type" { r set x{t} 10 assert_error "WRONGTYPE*" {r smove x{t} myset2{t} foo} @@ -1332,4 +1341,3 @@ test "SMOVE bug" { r smove myset myset a r srandmember myset }} -