Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Make balance works #881

Merged
merged 14 commits into from
Sep 27, 2019
Merged

Make balance works #881

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
5dbb7a8
Implement member change
Jul 15, 2019
ce7375a
Address wadeliuyi's comments
Sep 3, 2019
5470a0f
Add Balance UT
Sep 4, 2019
4c4d0bb
Fix failed Meta http Uts
Sep 6, 2019
af6902f
Add balance data and balance data id command in console
Sep 9, 2019
76a58fe
Fix bugs
Sep 10, 2019
10b3d79
Fix stale peers when add part
Sep 11, 2019
1f2eb7b
Fix bugs
Sep 17, 2019
29df182
Remove retry logic and fix stale log response
Sep 20, 2019
bd0146c
Address cirtical27's comments
Sep 21, 2019
4549b22
Fix snapshot_test
Sep 22, 2019
3dadda8
Fix bug in snapshot
Sep 25, 2019
1a0cf01
Remove check in commitTransferLeader
Sep 25, 2019
2ac1b6e
Merge branch 'master' into member-change
dangleptr Sep 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/common/base/Status.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,7 @@ class Status final {
STATUS_GENERATOR(CfgErrorType);
STATUS_GENERATOR(CfgImmutable);
STATUS_GENERATOR(LeaderChanged);
STATUS_GENERATOR(Balanced);

#undef STATUS_GENERATOR

Expand Down Expand Up @@ -147,6 +148,7 @@ class Status final {
kCfgErrorType = 411,
kCfgImmutable = 412,
kLeaderChanged = 413,
kBalanced = 414,
};

Code code() const {
Expand Down
106 changes: 106 additions & 0 deletions src/graph/BalanceExecutor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,12 @@ void BalanceExecutor::execute() {
case BalanceSentence::SubType::kLeader:
balanceLeader();
break;
case BalanceSentence::SubType::kData:
balanceData();
break;
case BalanceSentence::SubType::kShowBalancePlan:
showBalancePlan();
break;
case BalanceSentence::SubType::kUnknown:
onError_(Status::Error("Type unknown"));
break;
Expand Down Expand Up @@ -60,5 +66,105 @@ void BalanceExecutor::balanceLeader() {
std::move(future).via(runner).thenValue(cb).thenError(error);
}

void BalanceExecutor::balanceData() {
auto future = ectx()->getMetaClient()->balance();
auto *runner = ectx()->rctx()->runner();

auto cb = [this] (auto &&resp) {
if (!resp.ok()) {
DCHECK(onError_);
onError_(std::move(resp).status());
return;
}
auto balanceId = std::move(resp).value();
resp_ = std::make_unique<cpp2::ExecutionResponse>();
std::vector<std::string> header{"ID"};
resp_->set_column_names(std::move(header));

std::vector<cpp2::RowValue> rows;
std::vector<cpp2::ColumnValue> row;
row.resize(1);
row[0].set_integer(balanceId);
rows.emplace_back();

rows.back().set_columns(std::move(row));

resp_->set_rows(std::move(rows));

DCHECK(onFinish_);
onFinish_();
};

auto error = [this] (auto &&e) {
LOG(ERROR) << "Exception caught: " << e.what();
DCHECK(onError_);
onError_(Status::Error("Internal error"));
return;
};

std::move(future).via(runner).thenValue(cb).thenError(error);
}

void BalanceExecutor::showBalancePlan() {
auto id = sentence_->balanceId();
auto future = ectx()->getMetaClient()->showBalance(id);
auto *runner = ectx()->rctx()->runner();
auto cb = [this] (auto &&resp) {
if (!resp.ok()) {
DCHECK(onError_);
onError_(std::move(resp).status());
return;
}
auto tasks = std::move(resp).value();
resp_ = std::make_unique<cpp2::ExecutionResponse>();
std::vector<std::string> header{"balanceId, spaceId:partId, src->dst", "status"};
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can create header in client directly?

resp_->set_column_names(std::move(header));

std::vector<cpp2::RowValue> rows;
rows.reserve(tasks.size());
for (auto& task : tasks) {
std::vector<cpp2::ColumnValue> row;
row.resize(2);
row[0].set_str(std::move(task.get_id()));
switch (task.get_result()) {
case meta::cpp2::TaskResult::SUCCEEDED:
row[1].set_str("succeeded");
break;
case meta::cpp2::TaskResult::FAILED:
row[1].set_str("failed");
break;
case meta::cpp2::TaskResult::IN_PROGRESS:
row[1].set_str("in progress");
break;
case meta::cpp2::TaskResult::INVALID:
row[1].set_str("invalid");
break;
}
rows.emplace_back();
rows.back().set_columns(std::move(row));
}
resp_->set_rows(std::move(rows));
DCHECK(onFinish_);
onFinish_();
};

auto error = [this] (auto &&e) {
LOG(ERROR) << "Exception caught: " << e.what();
DCHECK(onError_);
onError_(Status::Error("Internal error"));
return;
};

std::move(future).via(runner).thenValue(cb).thenError(error);
}

void BalanceExecutor::setupResponse(cpp2::ExecutionResponse &resp) {
if (resp_) {
resp = std::move(*resp_);
} else {
Executor::setupResponse(resp);
}
}

} // namespace graph
} // namespace nebula
6 changes: 6 additions & 0 deletions src/graph/BalanceExecutor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,12 @@ class BalanceExecutor final : public Executor {

void balanceLeader();

void balanceData();

void showBalancePlan();

void setupResponse(cpp2::ExecutionResponse &resp) override;

private:
BalanceSentence *sentence_{nullptr};
std::unique_ptr<cpp2::ExecutionResponse> resp_;
Expand Down
17 changes: 16 additions & 1 deletion src/interface/meta.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,10 @@ enum ErrorCode {
E_CONFLICT = -29,
E_WRONGCLUSTER = -30,

// KV Failure
E_STORE_FAILURE = -31,
E_STORE_SEGMENT_ILLEGAL = -32,
E_BAD_BALANCE_PLAN = -33,
E_BALANCED = -34,

E_INVALID_PASSWORD = -41,
E_INPROPER_ROLE = -42,
Expand Down Expand Up @@ -432,11 +433,25 @@ struct BalanceReq {
2: optional i64 id,
}

enum TaskResult {
SUCCEEDED = 0x00,
FAILED = 0x01,
IN_PROGRESS = 0x02,
INVALID = 0x03,
} (cpp.enum_strict)


struct BalanceTask {
1: string id,
2: TaskResult result,
}

struct BalanceResp {
1: ErrorCode code,
2: i64 id,
// Valid if code equals E_LEADER_CHANGED.
3: common.HostAddr leader,
4: list<BalanceTask> tasks,
}

struct LeaderBalanceReq {
Expand Down
2 changes: 2 additions & 0 deletions src/interface/raftex.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ enum ErrorCode {
E_TOO_MANY_REQUESTS = -15;
E_PERSIST_SNAPSHOT_FAILED = -16;

E_BAD_ROLE = -17,

E_EXCEPTION = -20; // An thrift internal exception was thrown
}

Expand Down
10 changes: 10 additions & 0 deletions src/interface/storage.thrift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ enum ErrorCode {
E_KEY_HAS_EXISTS = -12,
E_SPACE_NOT_FOUND = -13,
E_PART_NOT_FOUND = -14,
E_CONSENSUS_ERROR = -15,

// meta failures
E_EDGE_PROP_NOT_FOUND = -21,
Expand All @@ -32,6 +33,12 @@ enum ErrorCode {
// Invalid request
E_INVALID_FILTER = -31,
E_INVALID_UPDATER = -32,
E_INVALID_STORE = -33,
E_INVALID_PEER = -34,
E_RETRY_EXHAUSTED = -35,

// meta client failed
E_LOAD_META_FAILED = -41,
E_UNKNOWN = -100,
} (cpp.enum_strict)

Expand Down Expand Up @@ -221,6 +228,9 @@ struct RemovePartReq {
struct MemberChangeReq {
1: common.GraphSpaceID space_id,
2: common.PartitionID part_id,
3: common.HostAddr peer,
// true means add a peer, false means remove a peer.
4: bool add,
}

struct TransLeaderReq {
Expand Down
46 changes: 8 additions & 38 deletions src/kvstore/LogEncoder.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,59 +164,29 @@ std::vector<folly::StringPiece> decodeMultiValues(folly::StringPiece encoded) {
return values;
}

std::string encodeLearner(const HostAddr& learner) {
std::string encodeHost(LogType type, const HostAddr& host) {
std::string encoded;
encoded.reserve(kHeadLen + sizeof(HostAddr));
encoded.reserve(sizeof(int64_t) + 1 + sizeof(HostAddr));
// Timestamp (8 bytes)
int64_t ts = time::WallClock::fastNowInMilliSec();
encoded.append(reinterpret_cast<char*>(&ts), sizeof(int64_t));
// Log type
auto type = LogType::OP_ADD_LEARNER;
encoded.append(reinterpret_cast<char*>(&type), 1);
// Value length
uint32_t len = static_cast<uint32_t>(sizeof(HostAddr));
encoded.append(reinterpret_cast<char*>(&len), sizeof(len));
// Learner addr
encoded.append(reinterpret_cast<const char*>(&learner), sizeof(HostAddr));
encoded.append(reinterpret_cast<const char*>(&host), sizeof(HostAddr));
return encoded;
}

HostAddr decodeLearner(const std::string& encoded) {
HostAddr decodeHost(LogType type, const folly::StringPiece& encoded) {
HostAddr addr;
CHECK_EQ(kHeadLen + sizeof(HostAddr), encoded.size());
memcpy(&addr.first, encoded.data() + kHeadLen, sizeof(addr.first));
CHECK_EQ(sizeof(int64_t) + 1 + sizeof(HostAddr), encoded.size());
CHECK(encoded[sizeof(int64_t)] == type);
memcpy(&addr.first, encoded.begin() + sizeof(int64_t) + 1, sizeof(addr.first));
memcpy(&addr.second,
encoded.data() + kHeadLen + sizeof(addr.first),
encoded.begin() + sizeof(int64_t) + 1 + sizeof(addr.first),
sizeof(addr.second));
return addr;
}

std::string encodeTransLeader(const HostAddr& targetAddr) {
std::string encoded;
encoded.reserve(kHeadLen + sizeof(HostAddr));
// Timestamp (8 bytes)
int64_t ts = time::WallClock::fastNowInMilliSec();
encoded.append(reinterpret_cast<char*>(&ts), sizeof(int64_t));
// Log type
auto type = LogType::OP_TRANS_LEADER;
encoded.append(reinterpret_cast<char*>(&type), 1);
// Value length
uint32_t len = static_cast<uint32_t>(sizeof(HostAddr));
encoded.append(reinterpret_cast<char*>(&len), sizeof(len));
// Target addr
encoded.append(reinterpret_cast<const char*>(&targetAddr), sizeof(HostAddr));
return encoded;
}

HostAddr decodeTransLeader(folly::StringPiece encoded) {
HostAddr addr;
CHECK_EQ(kHeadLen + sizeof(HostAddr), encoded.size());
memcpy(&addr.first, encoded.begin() + kHeadLen, sizeof(addr.first));
memcpy(&addr.second,
encoded.begin() + kHeadLen + sizeof(addr.first),
sizeof(addr.second));
return addr;
}
} // namespace kvstore
} // namespace nebula

9 changes: 5 additions & 4 deletions src/kvstore/LogEncoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@ enum LogType : char {
OP_REMOVE_RANGE = 0x6,
OP_ADD_LEARNER = 0x07,
OP_TRANS_LEADER = 0x08,
OP_ADD_PEER = 0x09,
OP_REMOVE_PEER = 0x10,
};

std::string encodeKV(const folly::StringPiece& key,
Expand All @@ -38,11 +40,10 @@ std::string encodeMultiValues(LogType type,
folly::StringPiece v2);
std::vector<folly::StringPiece> decodeMultiValues(folly::StringPiece encoded);

std::string encodeLearner(const HostAddr& learner);
HostAddr decodeLearner(const std::string& encoded);

std::string encodeTransLeader(const HostAddr& targetAddr);
HostAddr decodeTransLeader(folly::StringPiece encoded);
std::string encodeHost(LogType type, const HostAddr& learner);
HostAddr decodeHost(LogType type, const folly::StringPiece& encoded);

} // namespace kvstore
} // namespace nebula
#endif // KVSTORE_LOGENCODER_H_
18 changes: 11 additions & 7 deletions src/kvstore/NebulaStore.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,8 @@ bool NebulaStore::init() {
spaceIt->second->parts_.emplace(partId,
newPart(spaceId,
partId,
enginePtr.get()));
enginePtr.get(),
false));
}
}
} catch (std::exception& e) {
Expand All @@ -118,7 +119,7 @@ bool NebulaStore::init() {
}
std::sort(partIds.begin(), partIds.end());
for (auto& partId : partIds) {
addPart(spaceId, partId);
addPart(spaceId, partId, false);
}
}

Expand Down Expand Up @@ -172,7 +173,7 @@ void NebulaStore::addSpace(GraphSpaceID spaceId) {
}


void NebulaStore::addPart(GraphSpaceID spaceId, PartitionID partId) {
void NebulaStore::addPart(GraphSpaceID spaceId, PartitionID partId, bool asLearner) {
folly::RWSpinLock::WriteHolder wh(&lock_);
auto spaceIt = this->spaces_.find(spaceId);
CHECK(spaceIt != this->spaces_.end()) << "Space should exist!";
Expand All @@ -199,13 +200,15 @@ void NebulaStore::addPart(GraphSpaceID spaceId, PartitionID partId) {
targetEngine->addPart(partId);
spaceIt->second->parts_.emplace(
partId,
newPart(spaceId, partId, targetEngine.get()));
LOG(INFO) << "Space " << spaceId << ", part " << partId << " has been added!";
newPart(spaceId, partId, targetEngine.get(), asLearner));
LOG(INFO) << "Space " << spaceId << ", part " << partId
<< " has been added, asLearner " << asLearner;
}

std::shared_ptr<Part> NebulaStore::newPart(GraphSpaceID spaceId,
PartitionID partId,
KVEngine* engine) {
KVEngine* engine,
bool asLearner) {
auto part = std::make_shared<Part>(spaceId,
partId,
raftAddr_,
Expand All @@ -226,7 +229,7 @@ std::shared_ptr<Part> NebulaStore::newPart(GraphSpaceID spaceId,
}
}
raftService_->addPartition(part);
part->start(std::move(peers));
part->start(std::move(peers), asLearner);
return part;
}

Expand Down Expand Up @@ -256,6 +259,7 @@ void NebulaStore::removePart(GraphSpaceID spaceId, PartitionID partId) {
auto* e = partIt->second->engine();
CHECK_NOTNULL(e);
raftService_->removePartition(partIt->second);
partIt->second->reset();
spaceIt->second->parts_.erase(partId);
e->removePart(partId);
}
Expand Down
Loading