Lots of fixes to poor daggyr implementation, added debugging messages

daggyd/daggyd/daggyd.cpp

@@ -119,7 +119,9 @@ namespace dl = daggy::loggers::dag_run;
 
 std::unique_ptr<dl::DAGRunLogger> loggerFactory(const rj::Value &config)
 {
-  if (config.HasMember("logger")) {
+  if (!config.HasMember("logger"))
+    return std::make_unique<dl::OStreamLogger>(std::cout);
+
   const auto &logConf = config["logger"];
   if (!logConf.IsObject())
     throw std::runtime_error("logger config is not an object");
@@ -156,17 +158,16 @@ std::unique_ptr<dl::DAGRunLogger> loggerFactory(const rj::Value &config)
     return std::make_unique<dl::RedisLogger>(prefix, host, port);
   }
 #endif
-    else
   throw std::runtime_error("Unknown logger type: " + name);
 }
-  return std::make_unique<dl::OStreamLogger>(std::cout);
-}
 
 namespace de = daggy::executors::task;
 
 std::unique_ptr<de::TaskExecutor> executorFactory(const rj::Value &config)
 {
-  if (config.HasMember("executor")) {
+  if (!config.HasMember("executor"))
+    return std::make_unique<de::ForkingTaskExecutor>(10);
+
   const auto &execConf = config["executor"];
   if (!execConf.IsObject())
     throw std::runtime_error("Executor config is not an object");
@@ -196,7 +197,7 @@ std::unique_ptr<de::TaskExecutor> executorFactory(const rj::Value &config)
     auto exe = std::make_unique<de::DaggyRunnerTaskExecutor>();
 
     const auto &runners = execConfig["runners"];
-      if (!runners.IsArray()) {
+    if (!runners.IsArray())
       throw std::runtime_error(
           "DaggyRunnerExecutor runners must be an array of urls");
 
@@ -206,14 +207,11 @@ std::unique_ptr<de::TaskExecutor> executorFactory(const rj::Value &config)
             "DaggyRunnerExecutor runners must be an array of urls");
       exe->addRunner(runners[i].GetString());
     }
+
     return exe;
   }
-    }
-    else
-      throw std::runtime_error("Unknown executor type: " + name);
-  }
 
-  return std::make_unique<de::ForkingTaskExecutor>(10);
+  throw std::runtime_error("Unknown executor type: " + name);
 }
 
 int main(int argc, char **argv)

daggyr/libdaggyr/src/Server.cpp

@@ -159,6 +159,8 @@ namespace daggy::daggyr {
                      .resourcesUsed = resourcesUsed});
     }
 
+    std::cout << "Enqueued " << runID << " / " << taskName << std::endl;
+
     response.send(Pistache::Http::Code::Ok, "");
   }
 
@@ -175,6 +177,7 @@ namespace daggy::daggyr {
 
     std::string payload;
 
+    // Check to see if it's pending
     bool found = false;
     {
       std::lock_guard<std::mutex> lock(pendingGuard_);
@@ -193,13 +196,15 @@ namespace daggy::daggyr {
             curCapacity_.cores += it->second.resourcesUsed.cores;
             curCapacity_.memoryMB += it->second.resourcesUsed.memoryMB;
           }
+          std::cout << "Resolved " << it->first.first << " / "
+                    << it->first.second << std::endl;
+        }
       }
       else {
         payload = R"({ "state": "RUNNING" })";
         found   = true;
       }
     }
-    }
 
     if (!found) {
       std::lock_guard<std::mutex> lock(resultsGuard_);
@@ -211,6 +216,7 @@ namespace daggy::daggyr {
       payload = R"({ "state": "COMPLETED", "attempt": )" +
                 attemptRecordToJSON(it->second) + "}";
     }
+
     response.send(Pistache::Http::Code::Ok, payload);
   }
 

libdaggy/include/daggy/executors/task/DaggyRunnerTaskExecutor.hpp

@@ -55,6 +55,7 @@ namespace daggy::executors::task {
       DAGRunID runID;
       std::string taskName;
       std::string runnerURL;
+      uint32_t retries;
     };
 
     // Resolves jobs through polling

libdaggy/src/DAGRunner.cpp

@@ -106,8 +106,12 @@ namespace daggy {
       taskAttemptCounts_[taskName] = 1;
 
       logger_.updateTaskState(runID_, taskName, RunState::RUNNING);
-      runningTasks_.emplace(taskName,
+      try {
-                            executor_.execute(runID_, taskName, task));
+        auto fut = executor_.execute(runID_, taskName, task);
+        runningTasks_.emplace(taskName, std::move(fut));
+      }
+      catch (std::exception &e) {
+      }
       ++nRunningTasks_;
 
       auto nextTask = dag_.visitNext();

libdaggy/src/Utilities.cpp

@@ -234,6 +234,7 @@ namespace daggy {
       curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
       curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curlWriter);
       curl_easy_setopt(curl, CURLOPT_WRITEDATA, &buffer);
+      curl_easy_setopt(curl, CURLOPT_TIMEOUT, 2);
 
       if (trace) {
         curl_easy_setopt(curl, CURLOPT_DEBUGFUNCTION, http_trace);

libdaggy/src/executors/task/DaggyRunnerTaskExecutor.cpp

@@ -135,6 +135,7 @@ std::future<AttemptRecord> DaggyRunnerTaskExecutor::execute(
   // Capacities for a runner can be negative, meaning that they're currently
   // oversubscribed.
   std::vector<std::pair<std::string, double>> impacts;
+
   for (const auto &runner : runners_) {
     try {
       const auto &[code, doc] = JSON_HTTP_REQUEST(runner + "/v1/capacity");
@@ -158,8 +159,14 @@ std::future<AttemptRecord> DaggyRunnerTaskExecutor::execute(
     }
   }
 
-  if (impacts.empty())
+  if (impacts.empty()) {
-    throw std::runtime_error("No runners available for execution");
+    std::promise<AttemptRecord> prom;
+    auto fut = prom.get_future();
+    AttemptRecord record{.rc          = -1,
+                         .executorLog = "No runners available for execution"};
+    prom.set_value(std::move(record));
+    return fut;
+  }
 
   auto cit = impacts.begin();
   for (auto it = impacts.begin(); it != impacts.end(); ++it) {
@@ -167,8 +174,19 @@ std::future<AttemptRecord> DaggyRunnerTaskExecutor::execute(
       cit = it;
   }
 
-  RunningTask rt{
+  std::stringstream ss;
-      .prom{}, .runID = runID, .taskName = taskName, .runnerURL = cit->first};
+  ss << cit->first << "/v1/task/" << runID << "/" << taskName;
+  auto url = ss.str();
+
+  const auto response = HTTP_REQUEST(url, taskToJSON(task), "POST");
+  if (response.code != HTTPCode::Ok)
+    throw std::runtime_error("Unable to submit task: " + response.body);
+
+  RunningTask rt{.prom{},
+                 .runID     = runID,
+                 .taskName  = taskName,
+                 .runnerURL = cit->first,
+                 .retries   = 3};
 
   auto fut = rt.prom.get_future();
 
@@ -194,6 +212,7 @@ void DaggyRunnerTaskExecutor::monitor()
     {
       std::vector<std::pair<DAGRunID, std::string>> resolvedJobs;
 
+      {
         std::lock_guard<std::mutex> lock(rtGuard_);
         for (auto &[taskID, task] : runningTasks_) {
           try {
@@ -201,27 +220,34 @@ void DaggyRunnerTaskExecutor::monitor()
                 task.runnerURL + "/v1/task/" + std::to_string(taskID.first) +
                 "/" + taskID.second);
             if (code != HTTPCode::Ok) {
+              --task.retries;
+
+              if (task.retries == 0) {
                 AttemptRecord record{
-                .rc = -1, .executorLog = "Unable to query runner for progress"};
+                    .rc          = -1,
+                    .executorLog = "Unable to query runner for progress"};
                 task.prom.set_value(std::move(record));
                 resolvedJobs.emplace_back(taskID);
+              }
               continue;
             }
 
             if (json["state"] == "COMPLETED") {
-            task.prom.set_value(attemptRecordFromJSON(json["attempt"]));
+              auto attempt = attemptRecordFromJSON(json["attempt"]);
+              task.prom.set_value(std::move(attempt));
               resolvedJobs.emplace_back(taskID);
             }
           }
           catch (std::runtime_error &e) {
             continue;
           }
-
+        }
         for (const auto &tid : resolvedJobs) {
           runningTasks_.extract(tid);
         }
       }
-      std::this_thread::sleep_for(std::chrono::seconds(1));
+
+      std::this_thread::sleep_for(std::chrono::milliseconds(250));
     }
   }
 }