Route invoke requests through planner (#330)

* wip * nits: run clang-format and fix basic endpoint test * tests: add more coverage * utils: move batch utils to different source file * util: fix compilation errors after moving batchExec to a different header file * tests: add more test coverage * more groundwork * tests: improve coverage * tests: small fix w/ scheduler * scheduler: clear clients upon scheduler reset * scheduler: also clear-up this host used slots * dist-tests: actually use separate planner * planner: move logging to trace * gh: bump code version * tmp fixes * nits * endpoint: add a test for endpoint * scheduler: make sure we set the timestamp for the first BER message too
faasm · Jul 18, 2023 · fa0d07d · fa0d07d
1 parent 9b90d3f
commit fa0d07d
Show file tree

Hide file tree

Showing 22 changed files with 632 additions and 560 deletions.
diff --git a/.env b/.env
@@ -1,4 +1,4 @@
-FAABRIC_VERSION=0.4.5
-FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.4.5
+FAABRIC_VERSION=0.4.6
+FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.4.6
 COMPOSE_PROJECT_NAME=faabric-dev
 CONAN_CACHE_MOUNT_SOURCE=./conan-cache/
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -23,7 +23,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -36,7 +36,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -50,7 +50,7 @@ jobs:
     if: github.event.pull_request.draft == false
     runs-on: ubuntu-latest
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -73,7 +73,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -113,7 +113,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}
@@ -167,7 +167,7 @@ jobs:
       REDIS_QUEUE_HOST: redis
       REDIS_STATE_HOST: redis
     container:
-      image: faasm.azurecr.io/faabric:0.4.5
+      image: faasm.azurecr.io/faabric:0.4.6
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}

diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-0.4.5
+0.4.6
diff --git a/include/faabric/planner/Planner.h b/include/faabric/planner/Planner.h
@@ -2,6 +2,7 @@
 
 #include <faabric/planner/PlannerState.h>
 #include <faabric/planner/planner.pb.h>
+#include <faabric/proto/faabric.pb.h>
 
 #include <shared_mutex>
 
@@ -59,6 +60,10 @@ class Planner
     std::shared_ptr<faabric::Message> getMessageResult(
       std::shared_ptr<faabric::Message> msg);
 
+    // Get all the results recorded for one batch
+    std::shared_ptr<faabric::BatchExecuteRequestStatus> getBatchResults(
+      int32_t appId);
+
   private:
     // There's a singleton instance of the planner running, but it must allow
     // concurrent requests

diff --git a/include/faabric/util/batch.h b/include/faabric/util/batch.h
@@ -3,6 +3,11 @@
 #include <faabric/proto/faabric.pb.h>
 
 namespace faabric::util {
+
+// ----------
+// Batch Execute Requests (BER)
+// ----------
+
 std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory();
 
 std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
@@ -11,4 +16,14 @@ std::shared_ptr<faabric::BatchExecuteRequest> batchExecFactory(
   int count = 1);
 
 bool isBatchExecRequestValid(std::shared_ptr<faabric::BatchExecuteRequest> ber);
+
+// ----------
+// Batch Execute Requests' Status
+// ----------
+
+std::shared_ptr<faabric::BatchExecuteRequestStatus> batchExecStatusFactory(
+  int32_t appId);
+
+std::shared_ptr<faabric::BatchExecuteRequestStatus> batchExecStatusFactory(
+  std::shared_ptr<faabric::BatchExecuteRequest> ber);
 }
diff --git a/src/endpoint/FaabricEndpointHandler.cpp b/src/endpoint/FaabricEndpointHandler.cpp
@@ -31,116 +31,11 @@ void FaabricEndpointHandler::onRequest(
     // Text response type
     response.set(header::content_type, "text/plain");
 
-    PROF_START(endpointRoundTrip)
-
-    // Parse message from JSON in request
-    const std::string& requestStr = request.body();
-
-    // Handle JSON
-    if (requestStr.empty()) {
-        SPDLOG_ERROR("Faabric handler received empty request");
-        response.result(beast::http::status::bad_request);
-        response.body() = std::string("Empty request");
-    } else {
-        auto req = faabric::util::batchExecFactory();
-        req->set_type(req->FUNCTIONS);
-        faabric::Message& msg = *req->add_messages();
-        faabric::util::jsonToMessage(requestStr, &msg);
-        faabric::scheduler::Scheduler& sched =
-          faabric::scheduler::getScheduler();
-
-        if (msg.isstatusrequest()) {
-            SPDLOG_DEBUG("Processing status request");
-            const faabric::Message result = sched.getFunctionResult(msg, 0);
-
-            if (result.type() == faabric::Message_MessageType_EMPTY) {
-                response.result(beast::http::status::ok);
-                response.body() = std::string("RUNNING");
-            } else if (result.returnvalue() == 0) {
-                response.result(beast::http::status::ok);
-                response.body() = faabric::util::messageToJson(result);
-            } else {
-                response.result(beast::http::status::internal_server_error);
-                response.body() = "FAILED: " + result.outputdata();
-            }
-        } else {
-            executeFunction(
-              std::move(ctx), std::move(response), std::move(req), 0);
-            return;
-        }
-    }
-
-    PROF_END(endpointRoundTrip)
+    // TODO: for the moment we keep the endpoint handler, but we are not meant
+    // to receive any requests here. Eventually we will delete it
+    SPDLOG_ERROR("Faabric handler received empty request");
+    response.result(beast::http::status::bad_request);
+    response.body() = std::string("Empty request");
     ctx.sendFunction(std::move(response));
 }
-
-void FaabricEndpointHandler::executeFunction(
-  HttpRequestContext&& ctx,
-  faabric::util::BeastHttpResponse&& response,
-  std::shared_ptr<faabric::BatchExecuteRequest> ber,
-  size_t messageIndex)
-{
-    auto& conf = faabric::util::getSystemConfig();
-
-    // Set app ID, message ID and master host on the first message of the BER
-    // TODO: eventually do it on the BER itself
-    faabric::util::setMessageId(*ber->mutable_messages(0));
-    ber->mutable_messages(0)->set_masterhost(conf.endpointHost);
-    int appId = ber->messages(0).appid();
-    int msgId = ber->messages(0).id();
-    assert(appId != 0);
-    assert(msgId != 0);
-
-    if (ber->messages(0).user().empty()) {
-        response.result(beast::http::status::bad_request);
-        response.body() = std::string("Empty user");
-        return ctx.sendFunction(std::move(response));
-    }
-
-    if (ber->messages(0).function().empty()) {
-        response.result(beast::http::status::bad_request);
-        response.body() = std::string("Empty function");
-        return ctx.sendFunction(std::move(response));
-    }
-
-    // This is set to false by the scheduler if the function ends up being sent
-    // elsewhere
-    if (!ber->messages(0).isasync()) {
-        ber->mutable_messages(0)->set_executeslocally(true);
-    }
-
-    auto tid = gettid();
-    const std::string funcStr =
-      faabric::util::funcToString(ber->messages(0), true);
-    SPDLOG_DEBUG("Worker HTTP thread {} scheduling {}", tid, funcStr);
-
-    // Schedule it
-    faabric::scheduler::Scheduler& sch = faabric::scheduler::getScheduler();
-    sch.callFunctions(ber);
-
-    // Await result on global bus (may have been executed on a different worker)
-    if (ber->messages(0).isasync()) {
-        response.result(beast::http::status::ok);
-        response.body() = faabric::util::messageToJson(ber->messages(0));
-        return ctx.sendFunction(std::move(response));
-    }
-
-    // TODO: temporarily make this HTTP call block one server thread.
-    // Eventually. we will route all HTTP requests through the planner instead
-    // of the worker, so we will be able to remove this blocking call
-    SPDLOG_DEBUG("Worker thread {} awaiting {}", tid, funcStr);
-    auto result =
-      sch.getFunctionResult(appId, msgId, conf.globalMessageTimeout);
-
-    beast::http::status statusCode =
-      (result.returnvalue() == 0) ? beast::http::status::ok
-                                  : beast::http::status::internal_server_error;
-    response.result(statusCode);
-    SPDLOG_DEBUG("Worker thread {} result {}",
-                 gettid(),
-                 faabric::util::funcToString(result, true));
-
-    response.body() = result.outputdata();
-    return ctx.sendFunction(std::move(response));
-}
 }
diff --git a/src/planner/Planner.cpp b/src/planner/Planner.cpp
@@ -1,5 +1,7 @@
 #include <faabric/planner/Planner.h>
+#include <faabric/proto/faabric.pb.h>
 #include <faabric/scheduler/Scheduler.h>
+#include <faabric/util/batch.h>
 #include <faabric/util/clock.h>
 #include <faabric/util/config.h>
 #include <faabric/util/environment.h>
@@ -267,6 +269,27 @@ std::shared_ptr<faabric::Message> Planner::getMessageResult(
     return nullptr;
 }
 
+std::shared_ptr<faabric::BatchExecuteRequestStatus> Planner::getBatchResults(
+  int32_t appId)
+{
+    auto berStatus = faabric::util::batchExecStatusFactory(appId);
+
+    // Acquire a read lock to copy all the results we have for this batch
+    {
+        faabric::util::SharedLock lock(plannerMx);
+
+        if (!state.appResults.contains(appId)) {
+            return nullptr;
+        }
+
+        for (auto msgResultPair : state.appResults.at(appId)) {
+            *berStatus->add_messageresults() = *(msgResultPair.second);
+        }
+    }
+
+    return berStatus;
+}
+
 Planner& getPlanner()
 {
     static Planner planner;