Support stats of all executor in graphd (#1451)

* Support all stats in graphd * Modify log level * address yixinglu's comment Co-authored-by: dutor <440396+dutor@users.noreply.github.com>
vesoft-inc · Dec 23, 2019 · e574db3 · e574db3
1 parent f3f5cbd
commit e574db3
Show file tree

Hide file tree

Showing 57 changed files with 370 additions and 619 deletions.
diff --git a/src/common/stats/StatsManager.cpp b/src/common/stats/StatsManager.cpp
@@ -40,9 +40,10 @@ int32_t StatsManager::registerStats(folly::StringPiece counterName) {
     auto& sm = get();
 
     std::string name = counterName.toString();
+    folly::RWSpinLock::WriteHolder wh(sm.nameMapLock_);
     auto it = sm.nameMap_.find(name);
     if (it != sm.nameMap_.end()) {
-        LOG(INFO) << "The counter \"" << name << "\" already exists";
+        VLOG(2) << "The counter \"" << name << "\" already exists";
         return it->second;
     }
 
@@ -66,15 +67,14 @@ int32_t StatsManager::registerHisto(folly::StringPiece counterName,
                                     StatsManager::VT bucketSize,
                                     StatsManager::VT min,
                                     StatsManager::VT max) {
-    LOG(INFO) << "registerHisto, bucketSize: " << bucketSize
-              << ", min: " << min << ", max: " << max;
     using std::chrono::seconds;
 
     auto& sm = get();
     std::string name = counterName.toString();
+    folly::RWSpinLock::WriteHolder wh(sm.nameMapLock_);
     auto it = sm.nameMap_.find(name);
     if (it != sm.nameMap_.end()) {
-        LOG(ERROR) << "The counter \"" << name << "\" already exists";
+        VLOG(2) << "The counter \"" << name << "\" already exists";
         return it->second;
     }
 
@@ -89,6 +89,9 @@ int32_t StatsManager::registerHisto(folly::StringPiece counterName,
                 StatsType(60, {seconds(60), seconds(600), seconds(3600)}))));
     int32_t index = - sm.histograms_.size();
     sm.nameMap_[name] = index;
+
+    LOG(INFO) << "registerHisto, bucketSize: " << bucketSize
+              << ", min: " << min << ", max: " << max;
     return index;
 }
 

diff --git a/src/common/stats/StatsManager.h b/src/common/stats/StatsManager.h
@@ -107,6 +107,7 @@ class StatsManager final {
     // <counter_name> => index
     // when index > 0, (index - 1) is the index of stats_ list
     // when index < 0, [- (index + 1)] is the index of histograms_ list
+    folly::RWSpinLock nameMapLock_;
     std::unordered_map<std::string, int32_t> nameMap_;
 
     // All time series stats

diff --git a/src/graph/AlterEdgeExecutor.cpp b/src/graph/AlterEdgeExecutor.cpp
@@ -12,7 +12,8 @@ namespace nebula {
 namespace graph {
 
 AlterEdgeExecutor::AlterEdgeExecutor(Sentence *sentence,
-                                     ExecutionContext *ectx) : Executor(ectx) {
+                                     ExecutionContext *ectx)
+    : Executor(ectx, "alter_edge") {
     sentence_ = static_cast<AlterEdgeSentence*>(sentence);
 }
 
@@ -39,8 +40,7 @@ Status AlterEdgeExecutor::getSchema() {
 void AlterEdgeExecutor::execute() {
     auto status = getSchema();
     if (!status.ok()) {
-        DCHECK(onError_);
-        onError_(std::move(status));
+        doError(std::move(status));
         return;
     }
     auto *mc = ectx()->getMetaClient();
@@ -51,18 +51,16 @@ void AlterEdgeExecutor::execute() {
     auto *runner = ectx()->rctx()->runner();
     auto cb = [this] (auto &&resp) {
         if (!resp.ok()) {
-            DCHECK(onError_);
-            onError_(resp.status());
+            doError(resp.status());
             return;
         }
 
-        DCHECK(onFinish_);
-        onFinish_(Executor::ProcessControl::kNext);
+        doFinish(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
         LOG(ERROR) << "Exception caught: " << e.what();
-        onError_(Status::Error("Internal error"));
+        doError(Status::Error("Internal error"));
     };
 
     std::move(future).via(runner).thenValue(cb).thenError(error);

diff --git a/src/graph/AlterTagExecutor.cpp b/src/graph/AlterTagExecutor.cpp
@@ -12,7 +12,8 @@ namespace nebula {
 namespace graph {
 
 AlterTagExecutor::AlterTagExecutor(Sentence *sentence,
-                                   ExecutionContext *ectx) : Executor(ectx) {
+                                   ExecutionContext *ectx)
+    : Executor(ectx, "alter_tag") {
     sentence_ = static_cast<AlterTagSentence*>(sentence);
 }
 
@@ -39,8 +40,7 @@ Status AlterTagExecutor::getSchema() {
 void AlterTagExecutor::execute() {
     auto status = getSchema();
     if (!status.ok()) {
-        DCHECK(onError_);
-        onError_(std::move(status));
+        doError(std::move(status));
         return;
     }
 
@@ -52,18 +52,16 @@ void AlterTagExecutor::execute() {
     auto *runner = ectx()->rctx()->runner();
     auto cb = [this] (auto &&resp) {
         if (!resp.ok()) {
-            DCHECK(onError_);
-            onError_(resp.status());
+            doError(resp.status());
             return;
         }
 
-        DCHECK(onFinish_);
-        onFinish_(Executor::ProcessControl::kNext);
+        doFinish(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
         LOG(ERROR) << "Exception caught: " << e.what();
-        onError_(Status::Error("Internal error"));
+        doError(Status::Error("Internal error"));
     };
 
     std::move(future).via(runner).thenValue(cb).thenError(error);

diff --git a/src/graph/AssignmentExecutor.cpp b/src/graph/AssignmentExecutor.cpp
@@ -14,7 +14,8 @@ namespace graph {
 
 
 AssignmentExecutor::AssignmentExecutor(Sentence *sentence,
-                                       ExecutionContext *ectx) : Executor(ectx) {
+                                       ExecutionContext *ectx)
+    : Executor(ectx, "assignment") {
     sentence_ = static_cast<AssignmentSentence*>(sentence);
 }
 
@@ -24,12 +25,10 @@ Status AssignmentExecutor::prepare() {
     executor_ = TraverseExecutor::makeTraverseExecutor(sentence_->sentence(), ectx());
 
     auto onError = [this] (Status s) {
-        DCHECK(onError_);
-        onError_(std::move(s));
+        doError(std::move(s));
     };
     auto onFinish = [this] (Executor::ProcessControl ctr) {
-        DCHECK(onFinish_);
-        onFinish_(ctr);
+        doFinish(ctr);
     };
     auto onResult = [this] (std::unique_ptr<InterimResult> result) {
         ectx()->variableHolder()->add(*var_, std::move(result));
@@ -52,8 +51,7 @@ Status AssignmentExecutor::prepare() {
 void AssignmentExecutor::execute() {
     auto status = checkIfGraphSpaceChosen();
     if (!status.ok()) {
-        DCHECK(onError_);
-        onError_(std::move(status));
+        doError(std::move(status));
         return;
     }
     executor_->execute();

diff --git a/src/graph/BalanceExecutor.cpp b/src/graph/BalanceExecutor.cpp
@@ -10,7 +10,8 @@ namespace nebula {
 namespace graph {
 
 BalanceExecutor::BalanceExecutor(Sentence *sentence,
-                                 ExecutionContext *ectx) : Executor(ectx) {
+                                 ExecutionContext *ectx)
+    : Executor(ectx, "balance") {
     sentence_ = static_cast<BalanceSentence*>(sentence);
 }
 
@@ -34,7 +35,7 @@ void BalanceExecutor::execute() {
             showBalancePlan();
             break;
         case BalanceSentence::SubType::kUnknown:
-            onError_(Status::Error("Type unknown"));
+            doError(Status::Error("Type unknown"));
             break;
     }
 }
@@ -45,24 +46,20 @@ void BalanceExecutor::balanceLeader() {
 
     auto cb = [this] (auto &&resp) {
         if (!resp.ok()) {
-            DCHECK(onError_);
-            onError_(std::move(resp).status());
+            doError(std::move(resp).status());
             return;
         }
         auto ret = std::move(resp).value();
         if (!ret) {
-            DCHECK(onError_);
-            onError_(Status::Error("Balance leader failed"));
+            doError(Status::Error("Balance leader failed"));
             return;
         }
-        DCHECK(onFinish_);
-        onFinish_(Executor::ProcessControl::kNext);
+        doFinish(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
         LOG(ERROR) << "Exception caught: " << e.what();
-        DCHECK(onError_);
-        onError_(Status::Error("Internal error"));
+        doError(Status::Error("Internal error"));
         return;
     };
 
@@ -81,8 +78,7 @@ void BalanceExecutor::balanceData(bool isStop) {
 
     auto cb = [this] (auto &&resp) {
         if (!resp.ok()) {
-            DCHECK(onError_);
-            onError_(std::move(resp).status());
+            doError(std::move(resp).status());
             return;
         }
         auto balanceId = std::move(resp).value();
@@ -100,14 +96,12 @@ void BalanceExecutor::balanceData(bool isStop) {
 
         resp_->set_rows(std::move(rows));
 
-        DCHECK(onFinish_);
-        onFinish_(Executor::ProcessControl::kNext);
+        doFinish(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
         LOG(ERROR) << "Exception caught: " << e.what();
-        DCHECK(onError_);
-        onError_(Status::Error("Internal error"));
+        doError(Status::Error("Internal error"));
         return;
     };
 
@@ -120,8 +114,7 @@ void BalanceExecutor::showBalancePlan() {
     auto *runner = ectx()->rctx()->runner();
     auto cb = [this] (auto &&resp) {
         if (!resp.ok()) {
-            DCHECK(onError_);
-            onError_(std::move(resp).status());
+            doError(std::move(resp).status());
             return;
         }
         auto tasks = std::move(resp).value();
@@ -171,14 +164,12 @@ void BalanceExecutor::showBalancePlan() {
         rows.emplace_back();
         rows.back().set_columns(std::move(row));
         resp_->set_rows(std::move(rows));
-        DCHECK(onFinish_);
-        onFinish_(Executor::ProcessControl::kNext);
+        doFinish(Executor::ProcessControl::kNext);
     };
 
     auto error = [this] (auto &&e) {
         LOG(ERROR) << "Exception caught: " << e.what();
-        DCHECK(onError_);
-        onError_(Status::Error("Internal error"));
+        doError(Status::Error("Internal error"));
         return;
     };