add some data related counters

vesoft-inc · Dec 15, 2021 · ef77118 · ef77118
1 parent afa0894
commit ef77118
Show file tree

Hide file tree

Showing 18 changed files with 47 additions and 5 deletions.
diff --git a/src/graph/service/QueryInstance.cpp b/src/graph/service/QueryInstance.cpp
@@ -135,7 +135,8 @@ void QueryInstance::onError(Status status) {
       rctx->resp().errorCode = ErrorCode::E_BAD_PERMISSION;
       break;
     case Status::Code::kLeaderChanged:
-      stats::StatsManager::addValue(kNumQueryErrosLeaderChanges);
+      stats::StatsManager::addValue(kNumQueryErrorsLeaderChanges);
+      [[fallthrough]];
     case Status::Code::kBalanced:
     case Status::Code::kEdgeNotFound:
     case Status::Code::kError:

diff --git a/src/graph/stats/StatsDef.cpp b/src/graph/stats/StatsDef.cpp
@@ -19,7 +19,7 @@ stats::CounterId kNumQueries;
 stats::CounterId kNumActiveQueries;
 stats::CounterId kNumSlowQueries;
 stats::CounterId kNumQueryErrors;
-stats::CounterId kNumQueryErrosLeaderChanges;
+stats::CounterId kNumQueryErrorsLeaderChanges;
 stats::CounterId kNumSentences;
 stats::CounterId kQueryLatencyUs;
 stats::CounterId kSlowQueryLatencyUs;
@@ -48,7 +48,7 @@ void initCounters() {
   kNumSlowQueries = stats::StatsManager::registerStats("num_slow_queries", "rate, sum");
   kNumSentences = stats::StatsManager::registerStats("num_sentences", "rate, sum");
   kNumQueryErrors = stats::StatsManager::registerStats("num_query_errors", "rate, sum");
-  kNumQueryErrosLeaderChanges =
+  kNumQueryErrorsLeaderChanges =
       stats::StatsManager::registerStats("num_query_errors_leader_changes", "rate, sum");
   kQueryLatencyUs = stats::StatsManager::registerHisto(
       "query_latency_us", 1000, 0, 2000, "avg, p75, p95, p99, p999");

diff --git a/src/graph/stats/StatsDef.h b/src/graph/stats/StatsDef.h
@@ -18,7 +18,7 @@ extern stats::CounterId kNumQueries;
 extern stats::CounterId kNumActiveQueries;
 extern stats::CounterId kNumSlowQueries;
 extern stats::CounterId kNumQueryErrors;
-extern stats::CounterId kNumQueryErrosLeaderChanges;
+extern stats::CounterId kNumQueryErrorsLeaderChanges;
 extern stats::CounterId kNumSentences;
 extern stats::CounterId kQueryLatencyUs;
 extern stats::CounterId kSlowQueryLatencyUs;

diff --git a/src/meta/MetaServiceHandler.h b/src/meta/MetaServiceHandler.h
@@ -11,6 +11,8 @@
 #include "kvstore/KVStore.h"
 #include "meta/processors/admin/AdminClient.h"
 #include "meta/processors/admin/HBProcessor.h"
+#include "meta/processors/job/AdminJobProcessor.h"
+#include "meta/processors/job/JobManager.h"
 
 namespace nebula {
 namespace meta {
@@ -23,6 +25,8 @@ class MetaServiceHandler final : public cpp2::MetaServiceSvIf {
 
     // Initialize counters
     kHBCounters.init();
+    kNumActiveJobs = stats::StatsManager::registerStats("num_active_jobs", "sum");
+    kNumRunningJobs = stats::StatsManager::registerStats("num_running_jobs", "sum");
   }
 
   /**

diff --git a/src/meta/processors/job/AdminJobProcessor.cpp b/src/meta/processors/job/AdminJobProcessor.cpp
@@ -6,11 +6,13 @@
 #include "meta/processors/job/AdminJobProcessor.h"
 
 #include "common/base/StatusOr.h"
+#include "common/stats/StatsManager.h"
 #include "meta/processors/job/JobDescription.h"
 #include "meta/processors/job/JobManager.h"
 
 namespace nebula {
 namespace meta {
+stats::CounterId kNumActiveJobs;
 
 void AdminJobProcessor::process(const cpp2::AdminJobReq& req) {
   cpp2::AdminJobResult result;
@@ -59,6 +61,7 @@ void AdminJobProcessor::process(const cpp2::AdminJobReq& req) {
       errorCode = jobMgr->addJob(jobDesc, adminClient_);
       if (errorCode == nebula::cpp2::ErrorCode::SUCCEEDED) {
         result.set_job_id(nebula::value(jobId));
+        stats::StatsManager::addValue(kNumActiveJobs);
       }
       break;
     }
@@ -121,6 +124,7 @@ void AdminJobProcessor::process(const cpp2::AdminJobReq& req) {
       auto ret = jobMgr->recoverJob(spaceName, adminClient_, jobIds);
       if (nebula::ok(ret)) {
         result.set_recovered_job_num(nebula::value(ret));
+        stats::StatsManager::addValue(kNumActiveJobs);
       } else {
         errorCode = nebula::error(ret);
       }

diff --git a/src/meta/processors/job/AdminJobProcessor.h b/src/meta/processors/job/AdminJobProcessor.h
@@ -6,11 +6,13 @@
 #ifndef META_ADMINJOBPROCESSOR_H_
 #define META_ADMINJOBPROCESSOR_H_
 
+#include "common/stats/StatsManager.h"
 #include "meta/processors/BaseProcessor.h"
 #include "meta/processors/admin/AdminClient.h"
 
 namespace nebula {
 namespace meta {
+extern stats::CounterId kNumActiveJobs;
 
 class AdminJobProcessor : public BaseProcessor<cpp2::AdminJobResp> {
  public:

diff --git a/src/meta/processors/job/JobManager.cpp b/src/meta/processors/job/JobManager.cpp
@@ -34,7 +34,7 @@ using nebula::kvstore::KVIterator;
 
 namespace nebula {
 namespace meta {
-stats::CounterId kNumRunningJobs = stats::StatsManager::registerStats("num_running_jobs", "sum");
+stats::CounterId kNumRunningJobs;
 
 JobManager* JobManager::getInstance() {
   static JobManager inst;

diff --git a/src/storage/GraphStorageServiceHandler.cpp b/src/storage/GraphStorageServiceHandler.cpp
@@ -63,6 +63,12 @@ GraphStorageServiceHandler::GraphStorageServiceHandler(StorageEnv* env) : env_(e
   kPutCounters.init("kv_put");
   kGetCounters.init("kv_get");
   kRemoveCounters.init("kv_remove");
+
+  kNumVerticesInserted = stats::StatsManager::registerStats("num_vertices_inserted", "rate, sum");
+  kNumEdgesInserted = stats::StatsManager::registerStats("num_edges_inserted", "rate, sum");
+  kNumEdgesDeleted = stats::StatsManager::registerStats("num_edges_deleted", "rate, sum");
+  kNumTagsDeleted = stats::StatsManager::registerStats("num_tags_deleted", "rate, sum");
+  kNumVerticesDeleted = stats::StatsManager::registerStats("num_vertices_deleted", "rate, sum");
 }
 
 // Vertice section

diff --git a/src/storage/mutate/AddEdgesProcessor.cpp b/src/storage/mutate/AddEdgesProcessor.cpp
@@ -8,6 +8,7 @@
 #include <algorithm>
 
 #include "codec/RowWriterV2.h"
+#include "common/stats/StatsManager.h"
 #include "common/time/WallClock.h"
 #include "common/utils/IndexKeyUtils.h"
 #include "common/utils/NebulaKeyUtils.h"
@@ -17,6 +18,7 @@ namespace nebula {
 namespace storage {
 
 ProcessorCounters kAddEdgesCounters;
+stats::CounterId kNumEdgesInserted;
 
 void AddEdgesProcessor::process(const cpp2::AddEdgesRequest& req) {
   spaceId_ = req.get_space_id();
@@ -140,6 +142,7 @@ void AddEdgesProcessor::doProcess(const cpp2::AddEdgesRequest& req) {
             });
       } else {
         doPut(spaceId_, partId, std::move(data));
+        stats::StatsManager::addValue(kNumEdgesInserted, data.size());
       }
     }
   }
@@ -297,6 +300,7 @@ void AddEdgesProcessor::doProcessWithIndex(const cpp2::AddEdgesRequest& req) {
         break;
       }
       batchHolder->put(std::move(key), std::move(retEnc.value()));
+      stats::StatsManager::addValue(kNumEdgesInserted);
     }
     if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
       env_->edgesML_->unlockBatch(dummyLock);

diff --git a/src/storage/mutate/AddEdgesProcessor.h b/src/storage/mutate/AddEdgesProcessor.h
@@ -7,6 +7,7 @@
 #define STORAGE_MUTATE_ADDEDGESPROCESSOR_H_
 
 #include "common/base/Base.h"
+#include "common/stats/StatsManager.h"
 #include "kvstore/LogEncoder.h"
 #include "storage/BaseProcessor.h"
 #include "storage/StorageFlags.h"
@@ -15,6 +16,7 @@ namespace nebula {
 namespace storage {
 
 extern ProcessorCounters kAddEdgesCounters;
+extern stats::CounterId kNumEdgesInserted;
 
 class AddEdgesProcessor : public BaseProcessor<cpp2::ExecResponse> {
   friend class TransactionManager;

diff --git a/src/storage/mutate/AddVerticesProcessor.cpp b/src/storage/mutate/AddVerticesProcessor.cpp
@@ -8,6 +8,7 @@
 #include <algorithm>
 
 #include "codec/RowWriterV2.h"
+#include "common/stats/StatsManager.h"
 #include "common/time/WallClock.h"
 #include "common/utils/IndexKeyUtils.h"
 #include "common/utils/NebulaKeyUtils.h"
@@ -18,6 +19,7 @@ namespace nebula {
 namespace storage {
 
 ProcessorCounters kAddVerticesCounters;
+stats::CounterId kNumVerticesInserted;
 
 void AddVerticesProcessor::process(const cpp2::AddVerticesRequest& req) {
   spaceId_ = req.get_space_id();
@@ -126,6 +128,7 @@ void AddVerticesProcessor::doProcess(const cpp2::AddVerticesRequest& req) {
       handleAsync(spaceId_, partId, code);
     } else {
       doPut(spaceId_, partId, std::move(data));
+      stats::StatsManager::addValue(kNumVerticesInserted, data.size());
     }
   }
 }
@@ -280,6 +283,7 @@ void AddVerticesProcessor::doProcessWithIndex(const cpp2::AddVerticesRequest& re
          * step 3 , Insert new vertex data
          */
         batchHolder->put(std::move(key), std::move(retEnc.value()));
+        stats::StatsManager::addValue(kNumVerticesInserted);
       }
       if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
         break;

diff --git a/src/storage/mutate/AddVerticesProcessor.h b/src/storage/mutate/AddVerticesProcessor.h
@@ -16,6 +16,7 @@ namespace nebula {
 namespace storage {
 
 extern ProcessorCounters kAddVerticesCounters;
+extern stats::CounterId kNumVerticesInserted;
 
 class AddVerticesProcessor : public BaseProcessor<cpp2::ExecResponse> {
  public:

diff --git a/src/storage/mutate/DeleteEdgesProcessor.cpp b/src/storage/mutate/DeleteEdgesProcessor.cpp
@@ -7,6 +7,7 @@
 
 #include <algorithm>
 
+#include "common/stats/StatsManager.h"
 #include "common/utils/IndexKeyUtils.h"
 #include "common/utils/NebulaKeyUtils.h"
 #include "common/utils/OperationKeyUtils.h"
@@ -15,6 +16,7 @@ namespace nebula {
 namespace storage {
 
 ProcessorCounters kDelEdgesCounters;
+stats::CounterId kNumEdgesDeleted;
 
 void DeleteEdgesProcessor::process(const cpp2::DeleteEdgesRequest& req) {
   spaceId_ = req.get_space_id();
@@ -76,6 +78,7 @@ void DeleteEdgesProcessor::process(const cpp2::DeleteEdgesRequest& req) {
         continue;
       }
       doRemove(spaceId_, partId, std::move(keys));
+      stats::StatsManager::addValue(kNumEdgesDeleted, keys.size());
     }
   } else {
     for (auto& part : partEdges) {
@@ -185,6 +188,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::string> DeleteEdgesProcessor::deleteEdges(
         }
       }
       batchHolder->remove(std::move(key));
+      stats::StatsManager::addValue(kNumEdgesDeleted);
     } else if (ret == nebula::cpp2::ErrorCode::E_KEY_NOT_FOUND) {
       continue;
     } else {

diff --git a/src/storage/mutate/DeleteEdgesProcessor.h b/src/storage/mutate/DeleteEdgesProcessor.h
@@ -14,6 +14,7 @@ namespace nebula {
 namespace storage {
 
 extern ProcessorCounters kDelEdgesCounters;
+extern stats::CounterId kNumEdgesDeleted;
 
 class DeleteEdgesProcessor : public BaseProcessor<cpp2::ExecResponse> {
  public:

diff --git a/src/storage/mutate/DeleteTagsProcessor.cpp b/src/storage/mutate/DeleteTagsProcessor.cpp
@@ -5,6 +5,7 @@
 
 #include "storage/mutate/DeleteTagsProcessor.h"
 
+#include "common/stats/StatsManager.h"
 #include "common/utils/IndexKeyUtils.h"
 #include "common/utils/NebulaKeyUtils.h"
 #include "common/utils/OperationKeyUtils.h"
@@ -14,6 +15,7 @@ namespace nebula {
 namespace storage {
 
 ProcessorCounters kDelTagsCounters;
+stats::CounterId kNumTagsDeleted;
 
 void DeleteTagsProcessor::process(const cpp2::DeleteTagsRequest& req) {
   spaceId_ = req.get_space_id();
@@ -60,6 +62,7 @@ void DeleteTagsProcessor::process(const cpp2::DeleteTagsRequest& req) {
         }
       }
       doRemove(spaceId_, partId, std::move(keys));
+      stats::StatsManager::addValue(kNumTagsDeleted, keys.size());
     }
   } else {
     for (const auto& part : parts) {
@@ -147,6 +150,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::string> DeleteTagsProcessor::deleteTags(
         }
       }
       batchHolder->remove(std::move(key));
+      stats::StatsManager::addValue(kNumTagsDeleted);
     }
   }
   return encodeBatchValue(batchHolder->getBatch());

diff --git a/src/storage/mutate/DeleteTagsProcessor.h b/src/storage/mutate/DeleteTagsProcessor.h
@@ -15,6 +15,7 @@ namespace nebula {
 namespace storage {
 
 extern ProcessorCounters kDelTagsCounters;
+extern stats::CounterId kNumTagsDeleted;
 
 class DeleteTagsProcessor : public BaseProcessor<cpp2::ExecResponse> {
  public:

diff --git a/src/storage/mutate/DeleteVerticesProcessor.cpp b/src/storage/mutate/DeleteVerticesProcessor.cpp
@@ -14,6 +14,7 @@ namespace nebula {
 namespace storage {
 
 ProcessorCounters kDelVerticesCounters;
+stats::CounterId kNumVerticesDeleted;
 
 void DeleteVerticesProcessor::process(const cpp2::DeleteVerticesRequest& req) {
   spaceId_ = req.get_space_id();
@@ -80,6 +81,7 @@ void DeleteVerticesProcessor::process(const cpp2::DeleteVerticesRequest& req) {
         continue;
       }
       doRemove(spaceId_, partId, std::move(keys));
+      stats::StatsManager::addValue(kNumVerticesDeleted, keys.size());
     }
   } else {
     for (auto& pv : partVertices) {
@@ -171,6 +173,7 @@ ErrorOr<nebula::cpp2::ErrorCode, std::string> DeleteVerticesProcessor::deleteVer
         }
       }
       batchHolder->remove(key.str());
+      stats::StatsManager::addValue(kNumVerticesDeleted);
       iter->next();
     }
   }

diff --git a/src/storage/mutate/DeleteVerticesProcessor.h b/src/storage/mutate/DeleteVerticesProcessor.h
@@ -15,6 +15,7 @@ namespace nebula {
 namespace storage {
 
 extern ProcessorCounters kDelVerticesCounters;
+extern stats::CounterId kNumVerticesDeleted;
 
 class DeleteVerticesProcessor : public BaseProcessor<cpp2::ExecResponse> {
  public: