Move version info outside of HB (vesoft-inc#344)

#### What type of PR is this?
- [ ] bug
- [ ] feature
- [x] enhancement

#### What does this PR do?
1. Move the version info outside the heartbeat
2. Save the version info when establishing the connection to metad server.

#### Which issue(s)/PR(s) this PR relates to?
vesoft-inc#3055 

#### Special notes for your reviewer, ex. impact of this fix, etc:


#### Additional context:


#### Checklist：
- [ ] Documentation affected （Please add the label if documentation needs to be modified.)
- [x] Incompatible （If it is incompatible, please describe it and add corresponding label.）
- [ ] Need to cherry-pick （If need to cherry-pick to some branches, please label the destination version(s).)
- [ ] Performance impacted: Consumes more CPU/Memory


#### Release notes：
Please confirm whether to reflect in release notes and how to describe:
>                                                                 `


Migrated from vesoft-inc#3378

Co-authored-by: endy.li <25311962+heroicNeZha@users.noreply.github.com>

src/clients/meta/MetaClient.cpp

@@ -2435,7 +2435,6 @@ folly::Future<StatusOr<bool>> MetaClient::heartbeat() {
  req.set_host(options_.localHost_);
  req.set_role(options_.role_);
  req.set_git_info_sha(options_.gitInfoSHA_);
- req.set_version(getOriginVersion());
  if (options_.role_ == cpp2::HostRole::STORAGE ||
  options_.role_ == cpp2::HostRole::META_LISTENER ||
  options_.role_ == cpp2::HostRole::STORAGE_LISTENER) {
@@ -3674,6 +3673,7 @@ bool MetaClient::checkIsPlanKilled(SessionID sessionId, ExecutionPlanID planId)
 
 Status MetaClient::verifyVersion() {
  auto req = cpp2::VerifyClientVersionReq();
+ req.set_host(options_.localHost_);
  folly::Promise<StatusOr<cpp2::VerifyClientVersionResp>> promise;
  auto future = promise.getFuture();
  getResponse(

src/common/utils/MetaKeyUtils.cpp

@@ -20,6 +20,7 @@ namespace nebula {
 static const std::unordered_map<std::string, std::pair<std::string, bool>> systemTableMaps = {
  {"users", {"__users__", true}},
  {"hosts", {"__hosts__", false}},
+ {"versions", {"__versions__", false}},
  {"snapshots", {"__snapshots__", false}},
  {"configs", {"__configs__", true}},
  {"groups", {"__groups__", true}},
@@ -60,6 +61,7 @@ static const std::unordered_map<
 static const std::string kSpacesTable = tableMaps.at("spaces").first; // NOLINT
 static const std::string kPartsTable = tableMaps.at("parts").first; // NOLINT
 static const std::string kHostsTable = systemTableMaps.at("hosts").first; // NOLINT
+static const std::string kVersionsTable = systemTableMaps.at("versions").first; // NOLINT
 static const std::string kTagsTable = tableMaps.at("tags").first; // NOLINT
 static const std::string kEdgesTable = tableMaps.at("edges").first; // NOLINT
 static const std::string kIndexesTable = tableMaps.at("indexes").first; // NOLINT
@@ -273,6 +275,26 @@ HostAddr MetaKeyUtils::parseHostKeyV2(folly::StringPiece key) {
  return MetaKeyUtils::deserializeHostAddr(key);
 }
 
+std::string MetaKeyUtils::versionKey(const HostAddr& h) {
+ std::string key;
+ key.append(kVersionsTable.data(), kVersionsTable.size())
+ .append(MetaKeyUtils::serializeHostAddr(h));
+ return key;
+}
+
+std::string MetaKeyUtils::versionVal(const std::string& version) {
+ std::string val;
+ auto versionLen = version.size();
+ val.reserve(sizeof(int64_t) + versionLen);
+ val.append(reinterpret_cast<const char*>(&version), sizeof(int64_t)).append(version);
+ return val;
+}
+
+std::string MetaKeyUtils::parseVersion(folly::StringPiece val) {
+ auto len = *reinterpret_cast<const size_t*>(val.data());
+ return val.subpiece(sizeof(size_t), len).str();
+}
+
 std::string MetaKeyUtils::leaderKey(std::string addr, Port port) {
  LOG(ERROR) << "deprecated function\n" << boost::stacktrace::stacktrace();
  return leaderKeyV2(addr, port);

src/common/utils/MetaKeyUtils.h

@@ -122,6 +122,12 @@ class MetaKeyUtils final {
 
  static HostAddr parseHostKeyV2(folly::StringPiece key);
 
+ static std::string versionKey(const HostAddr& h);
+
+ static std::string versionVal(const std::string& version);
+
+ static std::string parseVersion(folly::StringPiece val);
+
  static std::string leaderKey(std::string ip, Port port);
 
  static std::string leaderKeyV2(std::string addr, Port port);

src/interface/meta.thrift

@@ -570,9 +570,7 @@ struct HBReq {
  4: optional map<common.GraphSpaceID, list<LeaderInfo>>
  (cpp.template = "std::unordered_map") leader_partIds;
  5: binary git_info_sha,
- // version of binary
- 6: optional binary version,
- 7: optional map<common.GraphSpaceID, map<binary, PartitionList>
+ 6: optional map<common.GraphSpaceID, map<binary, PartitionList>
  (cpp.template = "std::unordered_map")>
  (cpp.template = "std::unordered_map") disk_parts;
 }
@@ -1172,6 +1170,7 @@ struct VerifyClientVersionResp {
 
 struct VerifyClientVersionReq {
  1: required binary version = common.version;
+ 2: common.HostAddr host;
 }
 
 service MetaService {

src/meta/ActiveHostsMan.h

@@ -33,8 +33,6 @@ struct HostInfo {
  int64_t lastHBTimeInMilliSec_ = 0;
  cpp2::HostRole role_{cpp2::HostRole::UNKNOWN};
  std::string gitInfoSha_;
- // version of binary
- folly::Optional<std::string> version_;
 
  static HostInfo decode(const folly::StringPiece& data) {
  if (data.size() == sizeof(int64_t)) {
@@ -71,12 +69,6 @@ struct HostInfo {
  if (!info.gitInfoSha_.empty()) {
  encode.append(info.gitInfoSha_.data(), len);
  }
-
- if (info.version_.has_value()) {
- len = info.version_.value().size();
- encode.append(reinterpret_cast<const char*>(&len), sizeof(std::size_t));
- encode.append(info.version_.value().data(), len);
- }
  return encode;
  }
 
@@ -104,20 +96,6 @@ struct HostInfo {
  }
 
  info.gitInfoSha_ = std::string(data.data() + offset, len);
- offset += len;
-
- if (offset == data.size()) {
- return info;
- }
-
- len = *reinterpret_cast<const size_t*>(data.data() + offset);
- offset += sizeof(size_t);
-
- if (offset + len > data.size()) {
- FLOG_FATAL("decode out of range, offset=%zu, actual=%zu", offset, data.size());
- }
-
- info.version_ = std::string(data.data() + offset, len);
  return info;
  }
 };

src/meta/processors/admin/HBProcessor.cpp

@@ -69,9 +69,6 @@ void HBProcessor::process(const cpp2::HBReq& req) {
  }
 
  HostInfo info(time::WallClock::fastNowInMilliSec(), req.get_role(), req.get_git_info_sha());
- if (req.version_ref().has_value()) {
- info.version_ = *req.version_ref();
- }
  if (req.leader_partIds_ref().has_value()) {
  ret = ActiveHostsMan::updateHostInfo(kvstore_, host, info, &*req.leader_partIds_ref());
  } else {

src/meta/processors/admin/VerifyClientVersionProcessor.cpp

@@ -25,6 +25,12 @@ void VerifyClientVersionProcessor::process(const cpp2::VerifyClientVersionReq& r
  req.get_version().c_str(),
  FLAGS_client_white_list.c_str()));
  } else {
+ auto host = req.get_host();
+ auto versionKey = MetaKeyUtils::versionKey(host);
+ auto versionVal = MetaKeyUtils::versionVal(req.get_version().c_str());
+ std::vector<kvstore::KV> versionData;
+ versionData.emplace_back(std::move(versionKey), std::move(versionVal));
+ doSyncPut(versionData);
  resp_.set_code(nebula::cpp2::ErrorCode::SUCCEEDED);
  }
  onFinished();

src/meta/processors/parts/ListHostsProcessor.cpp

@@ -130,9 +130,14 @@ nebula::cpp2::ErrorCode ListHostsProcessor::allHostsWithStatus(cpp2::HostRole ro
 
  item.set_role(info.role_);
  item.set_git_info_sha(info.gitInfoSha_);
- if (info.version_.has_value()) {
- item.set_version(info.version_.value());
+
+ auto versionKey = MetaKeyUtils::versionKey(item.get_hostAddr());
+ auto versionRet = doGet(versionKey);
+ if (nebula::ok(versionRet)) {
+ auto versionVal = MetaKeyUtils::parseVersion(value(versionRet));
+ item.set_version(versionVal);
  }
+
  if (now - info.lastHBTimeInMilliSec_ < FLAGS_removed_threshold_sec * 1000) {
  // If meta didn't receive heartbeat with 2 periods, regard hosts as
  // offline. Same as ActiveHostsMan::getActiveHosts

src/meta/test/CMakeLists.txt

@@ -269,3 +269,18 @@ nebula_add_test(
  wangle
  gtest
 )
+
+nebula_add_test(
+ NAME
+ verify_client_version_test
+ SOURCES
+ VerifyClientVersionTest.cpp
+ OBJECTS
+ ${meta_test_deps}
+ LIBRARIES
+ ${ROCKSDB_LIBRARIES}
+ ${THRIFT_LIBRARIES}
+ ${PROXYGEN_LIBRARIES}
+ wangle
+ gtest
+)

src/meta/test/VerifyClientVersionTest.cpp

@@ -0,0 +1,75 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#include <gtest/gtest.h>
+
+#include "common/base/Base.h"
+#include "common/fs/TempDir.h"
+#include "meta/processors/admin/HBProcessor.h"
+#include "meta/processors/admin/VerifyClientVersionProcessor.h"
+#include "meta/processors/parts/ListHostsProcessor.h"
+#include "meta/test/TestUtils.h"
+
+namespace nebula {
+namespace meta {
+
+TEST(VerifyClientVersionTest, VersionTest) {
+ fs::TempDir rootPath("/tmp/VersionTest.XXXXXX");
+ std::unique_ptr<kvstore::KVStore> kv(MockCluster::initMetaKV(rootPath.path()));
+ {
+ auto req = cpp2::VerifyClientVersionReq();
+ req.set_version("1.0.1");
+ auto* processor = VerifyClientVersionProcessor::instance(kv.get());
+ auto f = processor->getFuture();
+ processor->process(req);
+ auto resp = std::move(f).get();
+ ASSERT_EQ(nebula::cpp2::ErrorCode::E_CLIENT_SERVER_INCOMPATIBLE, resp.get_code());
+ }
+ {
+ for (auto i = 0; i < 5; i++) {
+ auto req = cpp2::VerifyClientVersionReq();
+ req.set_host(HostAddr(std::to_string(i), i));
+ auto* processor = VerifyClientVersionProcessor::instance(kv.get());
+ auto f = processor->getFuture();
+ processor->process(req);
+ auto resp = std::move(f).get();
+ ASSERT_EQ(nebula::cpp2::ErrorCode::SUCCEEDED, resp.get_code());
+ }
+ }
+ {
+ const ClusterID kClusterId = 10;
+ for (auto i = 0; i < 5; i++) {
+ auto req = cpp2::HBReq();
+ req.set_role(cpp2::HostRole::GRAPH);
+ req.set_host(HostAddr(std::to_string(i), i));
+ req.set_cluster_id(kClusterId);
+ auto* processor = HBProcessor::instance(kv.get(), nullptr, kClusterId);
+ auto f = processor->getFuture();
+ processor->process(req);
+ auto resp = std::move(f).get();
+ ASSERT_EQ(nebula::cpp2::ErrorCode::SUCCEEDED, resp.get_code());
+ }
+ }
+ {
+ auto req = cpp2::ListHostsReq();
+ req.set_type(cpp2::ListHostType::GRAPH);
+ auto* processor = ListHostsProcessor::instance(kv.get());
+ auto f = processor->getFuture();
+ processor->process(req);
+ auto resp = std::move(f).get();
+ ASSERT_EQ(nebula::cpp2::ErrorCode::SUCCEEDED, resp.get_code());
+ ASSERT_EQ(resp.get_hosts().size(), 5);
+ }
+}
+
+} // namespace meta
+} // namespace nebula
+
+int main(int argc, char** argv) {
+ testing::InitGoogleTest(&argc, argv);
+ folly::init(&argc, &argv, true);
+ google::SetStderrLogging(google::INFO);
+ return RUN_ALL_TESTS();
+}