arangodb · neunhoef · Sep 9, 2021 · Sep 7, 2021 · Sep 7, 2021 · Sep 7, 2021
diff --git a/CHANGELOG b/CHANGELOG
@@ -1,6 +1,9 @@
 devel
 -----
 
+* No runtime limits for shard move and server cleanout jobs, instead
+  possibility to cancel them.
+
 * Updated ArangoDB Starter to 0.15.2.
 
 * SEARCH-238: Improved SortNodes placement optimization in cluster so 

diff --git a/arangod/Agency/CleanOutServer.cpp b/arangod/Agency/CleanOutServer.cpp
@@ -82,17 +82,8 @@ JOB_STATUS CleanOutServer::status() {
   }
 
   if (found > 0) {  // some subjob still running
-    // timeout here:
-    auto tmp_time =
-        _snapshot.hasAsString(pendingPrefix + _jobId + "/timeCreated");
-    std::string timeCreatedString = tmp_time.value();
-    Supervision::TimePoint timeCreated = stringToTimepoint(timeCreatedString);
-    Supervision::TimePoint now(std::chrono::system_clock::now());
-    if (now - timeCreated > std::chrono::duration<double>(86400.0)) { // 1 day
-      abort("job timed out");
-      return FAILED;
-    }
-    return PENDING;
+    // consider cancellation
+    return considerCancellation() ? FAILED : PENDING;
   }
 
   Node::Children const& failed = _snapshot.hasAsChildren(failedPrefix).value().get();
@@ -198,6 +189,11 @@ bool CleanOutServer::create(std::shared_ptr<VPackBuilder> envelope) {
 }
 
 bool CleanOutServer::start(bool& aborts) {
+
+  if (considerCancellation()) {
+    return false;
+  }
+
   // If anything throws here, the run() method catches it and finishes
   // the job.
 

diff --git a/arangod/Agency/Job.cpp b/arangod/Agency/Job.cpp
@@ -88,6 +88,17 @@ Job::~Job() = default;
 // this will be initialized in the AgencyFeature
 std::string Job::agencyPrefix = "arango";
 
+bool Job::considerCancellation() {
+  // Allow for cancellation of shard moves
+  auto [cancel,exists] = 
+    _snapshot.hasAsBool(std::string("/Target/") + jobStatus[_status] + "/" + _jobId + "/abort");
+  auto cancelled = exists && cancel;
+  if (cancelled) {
+    abort("Killed via API");
+  }
+  return cancelled;
+};
+
 bool Job::finish(std::string const& server, std::string const& shard,
                  bool success, std::string const& reason, query_t const payload) {
   try {  // protect everything, just in case

diff --git a/arangod/Agency/Job.h b/arangod/Agency/Job.h
@@ -48,6 +48,7 @@ namespace consensus {
 class Node;
 
 enum JOB_STATUS { TODO, PENDING, FINISHED, FAILED, NOTFOUND };
+const std::vector<std::string> jobStatus {"ToDo", "Pending", "Finished", "Failed"};
 const std::vector<std::string> pos({"/Target/ToDo/", "/Target/Pending/",
                                     "/Target/Finished/", "/Target/Failed/"});
 extern std::string const mapUniqueToShortID;
@@ -88,6 +89,8 @@ struct Job {
 
   virtual void run(bool& aborts) = 0;
 
+  bool considerCancellation();
+
   void runHelper(std::string const& server, std::string const& shard, bool& aborts) {
     if (_status == FAILED) {  // happens when the constructor did not work
       return;

diff --git a/arangod/Agency/MoveShard.cpp b/arangod/Agency/MoveShard.cpp
@@ -181,6 +181,11 @@ bool MoveShard::create(std::shared_ptr<VPackBuilder> envelope) {
 }
 
 bool MoveShard::start(bool&) {
+
+  if (considerCancellation()) {
+    return false;
+  }
+
   // If anything throws here, the run() method catches it and finishes
   // the job.
 
@@ -476,6 +481,13 @@ bool MoveShard::start(bool&) {
 }
 
 JOB_STATUS MoveShard::status() {
+
+  if (_status == PENDING || _status == TODO) {
+    if (considerCancellation()) {
+      return FAILED;
+    }
+  }
+
   if (_status != PENDING) {
     return _status;
   }
@@ -494,19 +506,6 @@ JOB_STATUS MoveShard::status() {
 
 JOB_STATUS MoveShard::pendingLeader() {
 
-  auto considerTimeout = [&]() -> bool {
-    // Not yet all in sync, consider timeout:
-    std::string timeCreatedString =
-        _snapshot.hasAsString(pendingPrefix + _jobId + "/timeCreated").value();
-    Supervision::TimePoint timeCreated = stringToTimepoint(timeCreatedString);
-    Supervision::TimePoint now(std::chrono::system_clock::now());
-    if (now - timeCreated > std::chrono::duration<double>(43200.0)) {  // 12h
-      abort("MoveShard timed out in pending leader");
-      return true;
-    }
-    return false;
-  };
-
   // Find the other shards in the same distributeShardsLike group:
   std::vector<Job::shard_t> shardsLikeMe =
       clones(_snapshot, _database, _collection, _shard);
@@ -555,12 +554,9 @@ JOB_STATUS MoveShard::pendingLeader() {
                      }
                    });
 
-    // Consider timeout:
+    // Consider cancellation:
     if (done < shardsLikeMe.size()) {
-      if (considerTimeout()) {
-        return FAILED;
-      }
-      return PENDING;  // do not act
+      return considerCancellation() ? FAILED : PENDING;
     }
 
     // We need to ask the old leader to retire:
@@ -622,12 +618,9 @@ JOB_STATUS MoveShard::pendingLeader() {
                      }
                    });
 
-    // Consider timeout:
+    // Consider cancellation:
     if (done < shardsLikeMe.size()) {
-      if (considerTimeout()) {
-        return FAILED;
-      }
-      return PENDING;  // do not act!
+      return considerCancellation() ? FAILED : PENDING;
     }
 
     // We need to switch leaders:
@@ -751,12 +744,9 @@ JOB_STATUS MoveShard::pendingLeader() {
                      }
                    });
 
-    // Consider timeout:
+    // Consider cancellation
     if (done < shardsLikeMe.size()) {
-      if (considerTimeout()) {
-        return FAILED;
-      }
-      return PENDING;  // do not act!
+      return considerCancellation() ? FAILED : PENDING;
     }
 
     // We need to end the job, Plan remains unchanged:
@@ -840,6 +830,7 @@ JOB_STATUS MoveShard::pendingLeader() {
 }
 
 JOB_STATUS MoveShard::pendingFollower() {
+
   // Check if any of the servers in the Plan are FAILED, if so,
   // we abort:
   std::string planPath =
@@ -867,16 +858,7 @@ JOB_STATUS MoveShard::pendingFollower() {
                  });
 
   if (done < shardsLikeMe.size()) {
-    // Not yet all in sync, consider timeout:
-    std::string timeCreatedString =
-        _snapshot.hasAsString(pendingPrefix + _jobId + "/timeCreated").value();
-    Supervision::TimePoint timeCreated = stringToTimepoint(timeCreatedString);
-    Supervision::TimePoint now(std::chrono::system_clock::now());
-    if (now - timeCreated > std::chrono::duration<double>(10000.0)) {
-      abort("MoveShard timed out in pending follower");
-      return FAILED;
-    }
-    return PENDING;
+    return considerCancellation() ? FAILED : PENDING;
   }
 
   // All in sync, so move on and remove the fromServer, for all shards,

diff --git a/arangod/RestHandler/RestAdminClusterHandler.cpp b/arangod/RestHandler/RestAdminClusterHandler.cpp
@@ -37,6 +37,7 @@
 #include "Agency/TransactionBuilder.h"
 #include "ApplicationFeatures/ApplicationServer.h"
 #include "Basics/ResultT.h"
+#include "Cluster/AgencyCache.h"
 #include "Cluster/ClusterFeature.h"
 #include "Cluster/ClusterHelpers.h"
 #include "Cluster/ClusterInfo.h"
@@ -270,6 +271,7 @@ std::string const RestAdminClusterHandler::ResignLeadership =
     "resignLeadership";
 std::string const RestAdminClusterHandler::MoveShard = "moveShard";
 std::string const RestAdminClusterHandler::QueryJobStatus = "queryAgencyJob";
+std::string const RestAdminClusterHandler::CancelJob = "cancelAgencyJob";
 std::string const RestAdminClusterHandler::RemoveServer = "removeServer";
 std::string const RestAdminClusterHandler::Rebalan
29C2
ceShards = "rebalanceShards";
 std::string const RestAdminClusterHandler::ShardStatistics = "shardStatistics";
@@ -331,6 +333,8 @@ RestStatus RestAdminClusterHandler::execute() {
       return handleResignLeadership();
     } else if (command == MoveShard) {
       return handleMoveShard();
+    } else if (command == CancelJob) {
+      return handleCancelJob();
     } else if (command == QueryJobStatus) {
       return handleQueryJobStatus();
     } else if (command == RemoveServer) {
@@ -881,6 +885,142 @@ RestStatus RestAdminClusterHandler::handleQueryJobStatus() {
           }));
 }
 
+RestStatus RestAdminClusterHandler::handleCancelJob() {
+  if (!ExecContext::current().isAdminUser()) {
+    generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_HTTP_FORBIDDEN);
+    return RestStatus::DONE;
+  }
+
+  if (!ServerState::instance()->isCoordinator()) {
+    generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_HTTP_FORBIDDEN,
+                  "only allowed on coordinators");
+    return RestStatus::DONE;
+  }
+
+  if (request()->requestType() != rest::RequestType::POST) {
+    generateError(rest::ResponseCode::METHOD_NOT_ALLOWED, TRI_ERROR_HTTP_METHOD_NOT_ALLOWED);
+    return RestStatus::DONE;
+  }
+
+  bool parseSuccess;
+  VPackSlice body = parseVPackBody(parseSuccess);
+  if (!parseSuccess) {
+    return RestStatus::DONE;
+  }
+
+  std::string jobId;
+  if (body.isObject() && body.hasKey("id") && body.get("id").isString()) {
+    jobId = body.get("id").copyString();
+  } else {
+    generateError(rest::ResponseCode::BAD, TRI_ERROR_BAD_PARAMETER, "object with key `id`");
+    return RestStatus::DONE;
+  }
+
+  auto targetPath = arangodb::cluster::paths::root()->arango()->target();
+  std::vector<std::string> paths = {
+      targetPath->pending()->job(jobId)->str(),
+      targetPath->failed()->job(jobId)->str(),
+      targetPath->finished()->job(jobId)->str(),
+      targetPath->toDo()->job(jobId)->str(),
+  };
+
+  try {
+    auto& agencyCache = server().getFeature<ClusterFeature>().agencyCache();
+    auto [acb, idx] = agencyCache.read(paths);
+    auto res = acb->slice();
+
+    auto paths = std::vector{
+      targetPath->pending()->job(jobId)->vec(),
+      targetPath->failed()->job(jobId)->vec(),
+      targetPath->finished()->job(jobId)->vec(),
+      targetPath->toDo()->job(jobId)->vec(),
+    };
+
+    for (auto const& path : paths) {
+      VPackSlice job = res.at(0).get(path);
+
+      if (job.isObject()) {
+        LOG_DEVEL << job.toJson();
+        LOG_TOPIC("eb139", INFO, Logger::SUPERVISION)
+          << "Attempting to abort supervision job " << job.toJson();
+        auto type = job.get("type").copyString();
+        if (type != "moveShard") { // only moveshard may be aborted
+          generateError(Result{400, "Only MoveShard jobs can be aborted"});
+          return RestStatus::DONE;
+        } else if (path[2] != "Pending" && path[2] != "ToDo") {
+          generateError(Result{400, path[2] + " jobs can no longer be cancelled."});
+          return RestStatus::DONE;
+        }
+
+        auto sendTransaction = [&] {
+          VPackBuffer<uint8_t> trxBody;
+          { VPackBuilder builder(trxBody);
+            { VPackArrayBuilder trxs(&builder);
+              if (path[2] == "ToDo") {
+                VPackArrayBuilder trx(&builder);
+                { VPackObjectBuilder op(&builder);
+                  builder.add("arango/Target/ToDo/" + jobId + "/abort", VPackValue(true)); }
+                { VPackObjectBuilder pre(&builder);
+                  builder.add(VPackValue("arango/Target/ToDo/" + jobId));
+                  { VPackObjectBuilder val(&builder);
+                    builder.add("oldEmpty", VPackValue(false)); }}
+              }
+              VPackArrayBuilder trx(&builder);
+              { VPackObjectBuilder op(&builder);
+                builder.add("arango/Target/Pending/" + jobId + "/abort", VPackValue(true)); }
+              { VPackObjectBuilder pre(&builder);
+                builder.add(VPackValue("arango/Target/Pending/" + jobId));
+                { VPackObjectBuilder val(&builder);
+                  builder.add("oldEmpty", VPackValue(false)); }}
+            }
+          }
+          return AsyncAgencyComm().sendWriteTransaction(60s, std::move(trxBody));
+        };
+
+        waitForFuture(
+          sendTransaction()
+          .thenValue([this](AsyncAgencyCommResult&& wr) {
+            if (!wr.ok()) {
+              if (wr.statusCode() == 412) {
+                generateError(Result{412, "Job is no longer running"});
+              } else {
+                generateError(wr.asResult());
+              }
+            }
+            return futures::makeFuture();
+          })
+          .thenError<VPackException>([this](VPackException const& e) {
+            generateError(Result{e.errorCode(), e.what()});
+          })
+          .thenError<std::exception>([this](std::exception const& e) {
+            generateError(rest::ResponseCode::SERVER_ERROR,
+                          TRI_ERROR_HTTP_SERVER_ERROR, e.what());
+          }));
+        VPackBuffer<uint8_t> payload;
+        {
+          VPackBuilder builder(payload);
+          VPackObjectBuilder ob(&builder);
+          builder.add("job", VPackValue(jobId));
+          builder.add("status", VPackValue("aborted"));
+          builder.add("error", VPackValue(false));
+        }
+        resetResponse(rest::ResponseCode::OK);
+        response()->setPayload(std::move(payload));
+        return RestStatus::DONE;
+      }
+    }
+
+    generateError(rest::ResponseCode::NOT_FOUND, TRI_ERROR_HTTP_NOT_FOUND);
+
+  } catch (VPackException const& e) {
+    generateError(Result{e.errorCode(), e.what()});
+  } catch (std::exception const& e) {
+    generateError(rest::ResponseCode::SERVER_ERROR, TRI_ERROR_HTTP_SERVER_ERROR, e.what());
+  }
+
+  return RestStatus::DONE;
+}
+
 RestStatus RestAdminClusterHandler::handleSingleServerJob(std::string const& job) {
   if (!ExecContext::current().isAdminUser()) {
     generateError(rest::ResponseCode::FORBIDDEN, TRI_ERROR_HTTP_FORBIDDEN);
@@ -1204,7 +1344,7 @@ RestStatus RestAdminClusterHandler::setMaintenance(bool wantToActivate) {
   auto sendTransaction = [&] {
     if (wantToActivate) {
       constexpr int timeout = 3600; // 1 hour
-      return AsyncAgencyComm().setValue(60s, maintenancePath, 
+      return AsyncAgencyComm().setValue(60s, maintenancePath,
           VPackValue(timepointToString(
               std::chrono::system_clock::now() + std::chrono::seconds(timeout))), 3600);
     } else {

diff --git a/arangod/RestHandler/RestAdminClusterHandler.h b/arangod/RestHandler/RestAdminClusterHandler.h
@@ -46,6 +46,7 @@ class RestAdminClusterHandler : public RestVocbaseBaseHandler {
   RequestLane lane() const override final { return RequestLane::CLIENT_SLOW; }
 
  private:
+  static std::string const CancelJob;
   static std::string const Health;
   static std::string const NumberOfServers;
   static std::string const Maintenance;
@@ -86,6 +87,7 @@ class RestAdminClusterHandler : public RestVocbaseBaseHandler {
   RestStatus handleCleanoutServer();
   RestStatus handleResignLeadership();
   RestStatus handleMoveShard();
+  RestStatus handleCancelJob();
   RestStatus handleQueryJobStatus();
 
   RestStatus handleRemoveServer();