Merge branch 'master' into fix-filter-expr

src/common/CMakeLists.txt

@@ -28,3 +28,4 @@ nebula_add_subdirectory(ssl)
 nebula_add_subdirectory(geo)
 nebula_add_subdirectory(memory)
 nebula_add_subdirectory(id)
+nebula_add_subdirectory(log)

src/common/log/CMakeLists.txt

@@ -0,0 +1,8 @@
+# Copyright (c) 2021 vesoft inc. All rights reserved.
+#
+# This source code is licensed under Apache 2.0 License.
+
+nebula_add_library(
+    log_monitor_obj OBJECT
+    LogMonitor.cpp
+)

src/common/log/LogMonitor.cpp

@@ -0,0 +1,69 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+#include "common/log/LogMonitor.h"
+
+namespace nebula {
+
+// default min_warn is 256M, disk freebytes < 256M will change LOG_LEVEL to WARNING
+DEFINE_uint64(log_min_reserved_bytes_to_warn,
+              256 * (1UL << 20),
+              "if freebytes in logdir less than this, will change log level to WARN");
+
+// default min_error is 64M, disk freebytes < 64M will change LOG_LEVEL to ERROR
+DEFINE_uint64(log_min_reserved_bytes_to_error,
+              64 * (1UL << 20),
+              "if freebytes in logdir less than this, will change log level to ERROR");
+
+// default min_fatal is 4M, disk freebytes < 4M will change LOG_LEVEL to FATAL
+DEFINE_uint64(log_min_reserved_bytes_to_fatal,
+              4 * (1UL << 20),
+              "if freebytes in logdir less than this, will change log level to FATAL");
+
+// default check log_disk interval is 10s
+DEFINE_int32(log_disk_check_interval_secs, 10, "interval to check free space of log path");
+
+void LogMonitor::getLogDiskFreeByte() {
+  boost::system::error_code ec;
+  auto info = boost::filesystem::space(FLAGS_log_dir, ec);
+  if (!ec) {
+    freeByte_ = info.available;
+  } else {
+    LOG(WARNING) << "Get filesystem info of logdir: " << FLAGS_log_dir << " failed";
+  }
+}
+
+void LogMonitor::checkAndChangeLogLevel() {
+  getLogDiskFreeByte();
+
+  if (FLAGS_log_min_reserved_bytes_to_fatal > FLAGS_log_min_reserved_bytes_to_error ||
+      FLAGS_log_min_reserved_bytes_to_fatal > FLAGS_log_min_reserved_bytes_to_warn ||
+      FLAGS_log_min_reserved_bytes_to_error > FLAGS_log_min_reserved_bytes_to_warn) {
+    LOG(ERROR) << "Get Invalid config in LogMonitor, the LogMonitor config should be "
+               << "FLAGS_log_min_reserved_bytes_to_warn >"
+               << "FLAGS_log_min_reserved_bytes_to_error > FLAGS_log_min_reserved_bytes_to_fatal;";
+    return;
+  }
+
+  if (freeByte_ < FLAGS_log_min_reserved_bytes_to_fatal) {
+    LOG(ERROR) << "log disk freebyte is less than " << FLAGS_log_min_reserved_bytes_to_fatal
+               << ", change log level to FATAL";
+    FLAGS_minloglevel = google::GLOG_FATAL;
+  } else if (freeByte_ < FLAGS_log_min_reserved_bytes_to_error) {
+    LOG(ERROR) << "log disk freebyte is less than " << FLAGS_log_min_reserved_bytes_to_error
+               << ", change log level to ERROR";
+    FLAGS_minloglevel = google::GLOG_ERROR;
+  } else if (freeByte_ < FLAGS_log_min_reserved_bytes_to_warn) {
+    LOG(ERROR) << "log disk freebyte is less than " << FLAGS_log_min_reserved_bytes_to_warn
+               << ", change log level to WARNING";
+    FLAGS_minloglevel = google::GLOG_WARNING;
+  } else {
+    // if freeByte_ is bigger than every min_log_flag, reset the FLAGS_minloglevel to old value
+    if (FLAGS_minloglevel != oldMinLogLevel_) {
+      FLAGS_minloglevel = oldMinLogLevel_;
+    }
+  }
+}
+
+}  // namespace nebula

src/common/log/LogMonitor.h

@@ -0,0 +1,43 @@
+/* Copyright (c) 2021 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+#pragma once
+
+#include <boost/filesystem.hpp>
+#include <boost/system/error_code.hpp>
+
+#include "common/thread/GenericWorker.h"
+
+namespace nebula {
+
+DECLARE_uint64(log_min_reserved_bytes_to_warn);
+DECLARE_uint64(log_min_reserved_bytes_to_error);
+DECLARE_uint64(log_min_reserved_bytes_to_fatal);
+DECLARE_int32(log_disk_check_interval_secs);
+
+class LogMonitor {
+ public:
+  LogMonitor() : oldMinLogLevel_(FLAGS_minloglevel), freeByte_(1UL << 60) {
+    worker_ = std::make_shared<thread::GenericWorker>();
+    CHECK(worker_->start());
+    worker_->addRepeatTask(
+        FLAGS_log_disk_check_interval_secs * 1000, &LogMonitor::checkAndChangeLogLevel, this);
+  }
+
+  ~LogMonitor() {
+    worker_->stop();
+    worker_->wait();
+  }
+
+  void getLogDiskFreeByte();
+
+  void checkAndChangeLogLevel();
+
+ private:
+  int32_t oldMinLogLevel_;
+  std::shared_ptr<thread::GenericWorker> worker_;
+  std::atomic_uint64_t freeByte_;
+};
+
+}  // namespace nebula

src/daemons/CMakeLists.txt

@@ -34,6 +34,7 @@ set(common_deps
     $<TARGET_OBJECTS:version_obj>
     $<TARGET_OBJECTS:ssl_obj>
     $<TARGET_OBJECTS:geo_index_obj>
+    $<TARGET_OBJECTS:log_monitor_obj>
 )
 
 set(storage_meta_deps

src/meta/processors/parts/CreateSpaceProcessor.cpp

@@ -193,7 +193,7 @@ void CreateSpaceProcessor::process(const cpp2::CreateSpaceReq& req) {
       auto key = MetaKeyUtils::hostKey(host.host, host.port);
       auto ret = doGet(key);
       if (!nebula::ok(ret)) {
-        code = nebula::error(zoneValueRet);
+        code = nebula::error(ret);
         LOG(ERROR) << "Get host " << host << " failed.";
         break;
       }

src/parser/parser.yy

@@ -1635,47 +1635,19 @@ unwind_clause
 
 with_clause
     : KW_WITH match_return_items match_order_by match_skip match_limit where_clause {
-        if ($6 && graph::ExpressionUtils::findAny($6->filter(),{Expression::Kind::kAggregate})) {
-            delete($2);
-            delete($3);
-            delete($4);
-            delete($5);
-            delete($6);
-            throw nebula::GraphParser::syntax_error(@6, "Invalid use of aggregating function in this context.");
-        }
         $$ = new WithClause($2, $3, $4, $5, $6, false/*distinct*/);
     }
     | KW_WITH KW_DISTINCT match_return_items match_order_by match_skip match_limit where_clause {
-        if ($7 && graph::ExpressionUtils::findAny($7->filter(),{Expression::Kind::kAggregate})) {
-            delete($3);
-            delete($4);
-            delete($5);
-            delete($6);
-            delete($7);
-            throw nebula::GraphParser::syntax_error(@7, "Invalid use of aggregating function in this context.");
-        }
         $$ = new WithClause($3, $4, $5, $6, $7, true);
     }
     ;
 
 match_clause
     : KW_MATCH match_path_list where_clause {
-        if ($3 && graph::ExpressionUtils::findAny($3->filter(),{Expression::Kind::kAggregate})) {
-            delete($2);
-            delete($3);
-            throw nebula::GraphParser::syntax_error(@3, "Invalid use of aggregating function in this context.");
-        } else {
-            $$ = new MatchClause($2, $3, false/*optional*/);
-        }
+        $$ = new MatchClause($2, $3, false/*optional*/);
     }
     | KW_OPTIONAL KW_MATCH match_path_list where_clause {
-        if ($4 && graph::ExpressionUtils::findAny($4->filter(),{Expression::Kind::kAggregate})) {
-            delete($3);
-            delete($4);
-            throw nebula::GraphParser::syntax_error(@4, "Invalid use of aggregating function in this context.");
-        } else {
-            $$ = new MatchClause($3, $4, true);
-        }
+        $$ = new MatchClause($3, $4, true);
     }
     ;
 

src/storage/StorageServer.cpp

@@ -198,6 +198,9 @@ bool StorageServer::start() {
   LOG(INFO) << "Init kvstore";
   kvstore_ = getStoreInstance();
 
+  LOG(INFO) << "Init LogMonitor";
+  logMonitor_ = std::make_unique<LogMonitor>();
+
   if (nullptr == kvstore_) {
     LOG(ERROR) << "Init kvstore failed";
     return false;

src/storage/StorageServer.h

@@ -11,6 +11,7 @@
 #include "clients/meta/MetaClient.h"
 #include "common/base/Base.h"
 #include "common/hdfs/HdfsHelper.h"
+#include "common/log/LogMonitor.h"
 #include "common/meta/IndexManager.h"
 #include "common/meta/SchemaManager.h"
 #include "kvstore/NebulaStore.h"
@@ -97,6 +98,8 @@ class StorageServer final {
   // used for communicate between one storaged to another
   std::unique_ptr<InternalStorageClient> interClient_;
 
+  std::unique_ptr<LogMonitor> logMonitor_;
+
   ServiceStatus serverStatus_{STATUS_UNINITIALIZED};
   std::mutex muStop_;
   std::condition_variable cvStop_;