arangodb · jsteemann · Dec 11, 2020 · Oct 30, 2020 · Nov 9, 2020 · Nov 9, 2020
diff --git a/3rdParty/fuerte/include/fuerte/message.h b/3rdParty/fuerte/include/fuerte/message.h
@@ -133,7 +133,7 @@ struct ResponseHeader final : public MessageHeader {
 // from (Response) a server.
 class Message {
  protec
6D47
ted:
-  Message() = default;
+  Message() : _timestamp(std::chrono::steady_clock::now()) {}
   virtual ~Message() = default;
 
  public:
@@ -174,6 +174,13 @@ class Message {
   bool isContentTypeVPack() const;
   bool isContentTypeHtml() const;
   bool isContentTypeText() const;
+
+  std::chrono::steady_clock::time_point timestamp() const { return _timestamp; }
+  // set timestamp when it was sent
+  void timestamp(std::chrono::steady_clock::time_point t) { _timestamp = t; }
+
+ private:
+  std::chrono::steady_clock::time_point _timestamp;
 };
 
 // Request contains the message send to a server in a request.

diff --git a/CHANGELOG b/CHANGELOG
@@ -1,6 +1,30 @@
 devel
 -----
 
+* Throttle work coming from low priority queue, according to a constant
+  and to an estimate taking into account fanout for multi-shard operations.
+
+* Move to 4 priority levels "low", "medium", "high" and "maintenance" in
+  scheduler to ensure that maintenance work and diagnostics is always
+  possible, even in the case of RocksDB throttles. Do not allow any
+  RocksDB work on "maintenance".
+
+* Commit replications on high priority queue.
+
+* Essentially get rid of timeout in replication to drop followers. This
+  is now entirely handled via reboot and failure tracking. The timeout
+  has now a default minimum of 15 minutes but can still be configured via
+  options.
+
+* Additional metrics for all queue lengths and low prio ongoing work.
+
+* New metric for number and total time of replication operations.
+
+* New metrics for number of internal requests in flight, internal request
+  duration, and internal request timeouts
+
+* Fix Gauge class' assignment operators.
+
 * Fix Windows directory creation error handling.
 
 * Add an AQL query kill check during early pruning. Fixes issue #13141.
@@ -109,11 +133,11 @@ devel
 * Added metrics for the system CPU usage:
   - `arangodb_server_statistics_user_percent`: Percentage of time that the
     system CPUs have spent in user mode
-  - `arangodb_server_statistics_system_percent`: Percentage of time that 
+  - `arangodb_server_statistics_system_percent`: Percentage of time that
     the system CPUs have spent in kernel mode
-  - `arangodb_server_statistics_idle_percent`: Percentage of time that the 
+  - `arangodb_server_statistics_idle_percent`: Percentage of time that the
     system CPUs have been idle
-  - `arangodb_server_statistics_iowait_percent`: Percentage of time that 
+  - `arangodb_server_statistics_iowait_percent`: Percentage of time that
     the system CPUs have been waiting for I/O
 
   These metrics resemble the overall CPU usage metrics in `top`.
@@ -127,19 +151,19 @@ devel
   configured value was effectively clamped to a value of `1`.
 
 * Improvements for the Pregel distributed graph processing feature:
-  - during the loading/startup phase, the in-memory edge cache is now 
-    intentionally bypassed. The reason for this is that any edges are 
+  - during the loading/startup phase, the in-memory edge cache is now
+    intentionally bypassed. The reason for this is that any edges are
     looked up exactly once, so caching them is not beneficial, but would
     only lead to cache pollution.
   - the loading/startup phase can now load multiple collections in parallel,
     whereas previously it was only loading multiple shards of the same
     collection in parallel. This change helps to reduce load times in case
     there are many collections with few shards, and on single server.
-  - the loading and result storage phases code has been overhauled so that 
+  - the loading and result storage phases code has been overhauled so that
     it runs slightly faster.
   - for Pregel runs that are based on named graphs (in contrast to explicit
     naming of the to-be-used vertex and edge collections), only those edge
-    collections are considered that, according to the graph definition, can 
+    collections are considered that, according to the graph definition, can
     have connections with the vertex. This change can reduce the loading
     time substantially in case the graph contains many edge definitions.
   - the number of executed rounds for the underlying Pregel algorithm now

diff --git a/arangod/Aql/Query.cpp b/arangod/Aql/Query.cpp
@@ -1179,7 +1179,8 @@ futures::Future<Result> finishDBServerParts(Query& query, int errorCode) {
   network::RequestOptions options;
   options.database = query.vocbase().name();
   options.timeout = network::Timeout(60.0);  // Picked arbitrarily
-//  options.skipScheduler = true;
+  options.continuationLane = RequestLane::CLUSTER_AQL_CONTINUATION;
+  //  options.skipScheduler = true;
 
   VPackBuffer<uint8_t> body;
   VPackBuilder builder(body);

diff --git a/arangod/Aql/SharedQueryState.cpp b/arangod/Aql/SharedQueryState.cpp
@@ -128,37 +128,36 @@ void SharedQueryState::queueHandler() {
     // We are shutting down
     return;
   }
-
-  bool queued = scheduler->queue(RequestLane::CLUSTER_AQL,
-                                 [self = shared_from_this(),
-                                  cb = _wakeupCb,
-                                  v = _cbVersion]() {
-
-    std::unique_lock<std::mutex> lck(self->_mutex, std::defer_lock);
-
-    do {
-      bool cntn = false;
-      try {
-        cntn = cb();
-      } catch (...) {}
-
-      lck.lock();
-      if (v == self->_cbVersion) {
-        unsigned c = self->_numWakeups--;
-        TRI_ASSERT(c > 0);
-        if (c == 1 || !cntn || !self->_valid) {
-          break;
-        }
-      } else {
-        return;
-      }
-      lck.unlock();
-    } while (true);
-
-    TRI_ASSERT(lck);
-    self->queueHandler();
-  });
-
+
+  bool queued =
+      scheduler->queue(RequestLane::CLUSTER_AQL_CONTINUATION,
+                       [self = shared_from_this(), cb = _wakeupCb, v = _cbVersion]() {
+                         std::unique_lock<std::mutex> lck(self->_mutex, std::defer_lock);
+
+                         do {
+                           bool cntn = false;
+                           try {
+                             cntn = cb();
+                           } catch (...) {
+                           }
+
+                           lck.lock();
+                           if (v == self->_cbVersion) {
+                             unsigned c = self->_numWakeups--;
+                             TRI_ASSERT(c > 0);
+                             if (c == 1 || !cntn || !self->_valid) {
+                               break;
+                             }
+                           } else {
+                             return;
+                           }
+                           lck.unlock();
+                         } while (true);
+
+                         TRI_ASSERT(lck);
+                         self->queueHandler();
+                       });
+
   if (!queued) { // just invalidate
      _wakeupCb = nullptr;
      _valid = false;

diff --git a/arangod/Cluster/ClusterTrxMethods.cpp b/arangod/Cluster/ClusterTrxMethods.cpp
@@ -217,6 +217,12 @@ Future<Result> commitAbortTransaction(transaction::Methods& trx, transaction::St
 
   TransactionId tidPlus = state->id().child();
   const std::string path = "/_api/transaction/" + std::to_string(tidPlus.id());
+  if (state->isDBServer()) {
+    // This is a leader replicating the transaction commit or abort and
+    // we should tell the follower that this is a replication operation.
+    // It will then execute the request with a higher priority.
+    reqOpts.param(StaticStrings::IsSynchronousReplicationString, ServerState::instance()->getId());
+  }
 
   fuerte::RestVerb verb;
   if (status == transaction::Status::COMMITTED) {

diff --git a/arangod/Cluster/MaintenanceRestHandler.h b/arangod/Cluster/MaintenanceRestHandler.h
@@ -43,7 +43,7 @@ class MaintenanceRestHandler : public RestBaseHandler {
   char const* name() const override { return "MaintenanceRestHandler"; }
 
   RequestLane lane() const override final {
-    return RequestLane::CLUSTER_INTERNAL;
+    return RequestLane::CLIENT_FAST;
   }
 
   /// @brief Performs routing of request to appropriate subroutines

diff --git a/arangod/Cluster/ReplicationTimeoutFeature.cpp b/arangod/Cluster/ReplicationTimeoutFeature.cpp
@@ -24,6 +24,7 @@
 #include "ReplicationTimeoutFeature.h"
 
 #include "FeaturePhases/DatabaseFeaturePhase.h"
+#include "RestServer/MetricsFeature.h"
 #include "StorageEngine/EngineSelectorFeature.h"
 #include "StorageEngine/StorageEngine.h"
 
@@ -33,7 +34,18 @@ namespace arangodb {
 
 double ReplicationTimeoutFeature::timeoutFactor = 1.0;
 double ReplicationTimeoutFeature::timeoutPer4k = 0.1;
-double ReplicationTimeoutFeature::lowerLimit = 30.0;  // longer than heartbeat timeout
+double ReplicationTimeoutFeature::lowerLimit = 900.0;  // used to be 30.0
+double ReplicationTimeoutFeature::upperLimit = 3600.0;  // used to be 120.0
+
+// We essentially stop using a meaningful timeout for this operation. 
+// This is achieved by setting the default for the minimal timeout to 1h or 3600s.
+// The reason behind this is the following: We have to live with RocksDB stalls
+// and write stops, which can happen in overload situations. Then, no meaningful
+// timeout helps and it is almost certainly better to keep trying to not have
+// to drop the follower and make matters worse. In case of an actual failure
+// (or indeed a restart), the follower is marked as failed and its reboot id is
+// increased. As a consequence, the connection is aborted and we run into an
+// error anyway. This is when a follower will be dropped.
 
 ReplicationTimeoutFeature::ReplicationTimeoutFeature(application_features::ApplicationServer& server)
     : ApplicationFeature(server, "ReplicationTimeout") {
@@ -44,10 +56,16 @@ ReplicationTimeoutFeature::ReplicationTimeoutFeature(application_features::Appli
 void ReplicationTimeoutFeature::collectOptions(std::shared_ptr<ProgramOptions> options) {
   options->addSection("cluster", "Configure the cluster");
 
-  options->addOption(
-      "--cluster.synchronous-replication-timeout-minimum",
-      "all synchronous replication timeouts will be at least this value (in seconds)",
-      new DoubleParameter(&lowerLimit));
+  options->addOption("--cluster.synchronous-replication-timeout-minimum",
+                     "all synchronous replication timeouts will be at least "
+                     "this value (in seconds)",
+                     new DoubleParameter(&lowerLimit));
+
+  options->addOption("--cluster.synchronous-replication-timeout-maximum",
+                     "all synchronous replication timeouts will be at most "
+                     "this value (in seconds)",
+                     new DoubleParameter(&upperLimit))
                     .setIntroducedIn(30800);
 
   options->addOption(
       "--cluster.synchronous-replication-timeout-factor",
@@ -62,4 +80,14 @@ void ReplicationTimeoutFeature::collectOptions(std::shared_ptr<ProgramOptions> o
       arangodb::options::makeDefaultFlags(arangodb::options::Flags::Hidden));
 }
 
+void ReplicationTimeoutFeature::validateOptions(std::shared_ptr<ProgramOptions> options) {
+  if (upperLimit < lowerLimit) {
+    LOG_TOPIC("8a9f3", WARN, Logger::CONFIG)
+        << "--cluster.synchronous-replication-timeout-maximum must be at least "
+        << "--cluster.synchronous-replication-timeout-minimum, setting max to "
+        << "min";
+    upperLimit = lowerLimit;
+  }
+}
+
 }  // namespace arangodb
diff --git a/arangod/Cluster/ReplicationTimeoutFeature.h b/arangod/Cluster/ReplicationTimeoutFeature.h
@@ -25,8 +25,10 @@
 #define ARANGOD_CLUSTER_REPLICATION_TIMEOUT_FEATURE_H 1
 
 #include "Basics/Common.h"
+#include "RestServer/Metrics.h"
 
 #include "ApplicationFeatures/ApplicationFeature.h"
+#include "ApplicationFeatures/ApplicationServer.h"
 
 namespace arangodb {
 
@@ -35,12 +37,14 @@ class ReplicationTimeoutFeature : public application_features::ApplicationFeatur
   explicit ReplicationTimeoutFeature(application_features::ApplicationServer& server);
 
   void collectOptions(std::shared_ptr<options::ProgramOptions>) override final;
+  void validateOptions(std::shared_ptr<options::ProgramOptions>) override final;
 
   static double timeoutFactor;
   static double timeoutPer4k;
   static double lowerLimit;
+  static double upperLimit;
 };
 
 }  // namespace arangodb
 
-#endif
+#endif
diff --git a/arangod/Cluster/RestClusterHandler.cpp b/arangod/Cluster/RestClusterHandler.cpp
@@ -153,7 +153,6 @@ void RestClusterHandler::handleClusterInfo() {
   auto& ci = server().getFeature<ClusterFeature>().clusterInfo();
   auto dump = ci.toVelocyPack();
 
-  LOG_DEVEL << dump.toJson();
   generateResult(rest::ResponseCode::OK, dump.slice());
 
 }

diff --git a/arangod/FeaturePhases/ServerFeaturePhase.cpp b/arangod/FeaturePhases/ServerFeaturePhase.cpp
@@ -26,6 +26,7 @@
 #include "FeaturePhases/AqlFeaturePhase.h"
 #include "GeneralServer/GeneralServerFeature.h"
 #include "GeneralServer/SslServerFeature.h"
+#include "Network/NetworkFeature.h"
 #include "RestServer/EndpointFeature.h"
 #include "RestServer/ServerFeature.h"
 #include "RestServer/UpgradeFeature.h"
@@ -41,6 +42,7 @@ ServerFeaturePhase::ServerFeaturePhase(ApplicationServer& server)
 
   startsAfter<EndpointFeature>();
   startsAfter<GeneralServerFeature>();
+  startsAfter<NetworkFeature>();
   startsAfter<ServerFeature>();
   startsAfter<SslServerFeature>();
   startsAfter<StatisticsFeature>();

diff --git a/arangod/GeneralServer/CommTask.cpp b/arangod/GeneralServer/CommTask.cpp
@@ -480,15 +480,23 @@ bool CommTask::handleRequestSync(std::shared_ptr<RestHandler> handler) {
   handler->statistics().SET_QUEUE_START(SchedulerFeature::SCHEDULER->queueStatistics()._queued);
 
   RequestLane lane = handler->getRequestLane();
+  RequestPriority prio = PriorityRequestLane(lane);
+
   ContentType respType = handler->request()->contentTypeResponse();
   uint64_t mid = handler->messageId();
 
   // queue the operation in the scheduler, and make it eligible for direct execution
   // only if the current CommTask type allows it (HttpCommTask: yes, CommTask: no)
   // and there is currently only a single client handled by the IoContext
-  auto cb = [self = shared_from_this(), handler = std::move(handler)]() mutable {
+  auto cb = [self = shared_from_this(), handler = std::move(handler), prio]() mutable {
+    if (prio == RequestPriority::LOW) {
+      SchedulerFeature::SCHEDULER->ongoingLowPriorityTasks() += 1;
+    }
     handler->statistics().SET_QUEUE_END();
-    handler->runHandler([self = std::move(self)](rest::RestHandler* handler) {
+    handler->runHandler([self = std::move(self), prio](rest::RestHandler* handler) {
+      if (prio == RequestPriority::LOW) {
+        SchedulerFeature::SCHEDULER->ongoingLowPriorityTasks() -= 1;
+      }
       try {
         // Pass the response to the io context
         self->sendResponse(handler->stealResponse(), handler->stealStatistics());