planner: track number of migrations that have taken place (#358)

* planner: add client/server method to get the number of migrations and use in tests

* gh: bump minor code version

* planner-cli: return the number of migrations

* planner: add very simple client/server test

Author: csegarragonz

.env

@@ -1,4 +1,4 @@
-FAABRIC_VERSION=0.10.0
-FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.10.0
+FAABRIC_VERSION=0.11.0
+FAABRIC_CLI_IMAGE=faasm.azurecr.io/faabric:0.11.0
 COMPOSE_PROJECT_NAME=faabric-dev
 CONAN_CACHE_MOUNT_SOURCE=./conan-cache/

.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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       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.10.0
+      image: faasm.azurecr.io/faabric:0.11.0
       credentials:
         username: ${{ secrets.ACR_SERVICE_PRINCIPAL_ID }}
         password: ${{ secrets.ACR_SERVICE_PRINCIPAL_PASSWORD }}

VERSION

@@ -1 +1 @@
-0.10.0
+0.11.0

include/faabric/planner/Planner.h

@@ -81,6 +81,10 @@ class Planner
 
     faabric::batch_scheduler::InFlightReqs getInFlightReqs();
 
+    // Helper method to get the number of migrations that have happened since
+    // the planner was last reset
+    int getNumMigrations();
+
     // Main entrypoint to request the execution of batches
     std::shared_ptr<faabric::batch_scheduler::SchedulingDecision> callBatch(
       std::shared_ptr<BatchExecuteRequest> req);

include/faabric/planner/PlannerApi.h

@@ -15,7 +15,8 @@ enum PlannerCalls
     GetMessageResult = 9,
     GetBatchResults = 10,
     GetSchedulingDecision = 11,
-    CallBatch = 12,
-    PreloadSchedulingDecision = 13,
+    GetNumMigrations = 12,
+    CallBatch = 13,
+    PreloadSchedulingDecision = 14,
 };
 }

include/faabric/planner/PlannerClient.h

@@ -57,6 +57,7 @@ class PlannerClient final : public faabric::transport::MessageEndpointClient
 
     void ping();
 
+    // Clear the client-local cache
     void clearCache();
 
     // ------
@@ -92,6 +93,8 @@ class PlannerClient final : public faabric::transport::MessageEndpointClient
     faabric::batch_scheduler::SchedulingDecision getSchedulingDecision(
       std::shared_ptr<faabric::BatchExecuteRequest> req);
 
+    int getNumMigrations();
+
     void preloadSchedulingDecision(
       std::shared_ptr<faabric::batch_scheduler::SchedulingDecision> preloadDec);
 

include/faabric/planner/PlannerServer.h

@@ -40,6 +40,9 @@ class PlannerServer final : public faabric::transport::MessageEndpointServer
     std::unique_ptr<google::protobuf::Message> recvGetSchedulingDecision(
       std::span<const uint8_t> buffer);
 
+    std::unique_ptr<google::protobuf::Message> recvGetNumMigrations(
+      std::span<const uint8_t> buffer);
+
     std::unique_ptr<google::protobuf::Message> recvPreloadSchedulingDecision(
       std::span<const uint8_t> buffer);
 

include/faabric/planner/PlannerState.h

@@ -33,5 +33,8 @@ struct PlannerState
     // planner's scheduling
     std::map<int, std::shared_ptr<batch_scheduler::SchedulingDecision>>
       preloadedSchedulingDecisions;
+
+    // Helper coutner of the total number of migrations
+    std::atomic<int> numMigrations = 0;
 };
 }

src/endpoint/FaabricEndpointHandler.cpp

@@ -38,14 +38,19 @@ void FaabricEndpointHandler::onRequest(
     // TODO: for the moment we keep the endpoint handler, but we are not meant
     // to receive any requests here. Eventually we will delete it
     if (requestStr.empty()) {
-        SPDLOG_ERROR("Planner handler received empty request");
+        SPDLOG_WARN("Worker HTTP handler received empty request!");
         response.result(beast::http::status::bad_request);
         response.body() = std::string("Empty request");
         return ctx.sendFunction(std::move(response));
     }
 
-    SPDLOG_ERROR("Worker HTTP handler received non-empty request (body: {})",
-                 request.body());
-    throw std::runtime_error("Worker HTTP handler received non-empty request");
+    // We don't expect to receive any requests here, but we sometimes do (e.g.
+    // probing if exposed port is vulnerable to certain CVEs when deployed in
+    // AKS)
+    SPDLOG_WARN("Worker HTTP handler received non-empty request (body: {})",
+                request.body());
+    response.result(beast::http::status::bad_request);
+    response.body() = std::string("Unexpected request!");
+    return ctx.sendFunction(std::move(response));
 }
 }

src/planner/Planner.cpp

@@ -151,6 +151,7 @@ void Planner::flushSchedulingState()
     state.inFlightReqs.clear();
     state.appResults.clear();
     state.appResultWaiters.clear();
+    state.numMigrations = 0;
 }
 
 std::vector<std::shared_ptr<Host>> Planner::getAvailableHosts()
@@ -517,6 +518,11 @@ faabric::batch_scheduler::InFlightReqs Planner::getInFlightReqs()
     return inFlightReqsCopy;
 }
 
+int Planner::getNumMigrations()
+{
+    return state.numMigrations.load(std::memory_order_acquire);
+}
+
 static faabric::batch_scheduler::HostMap convertToBatchSchedHostMap(
   std::map<std::string, std::shared_ptr<Host>> hostMapIn)
 {
@@ -662,6 +668,7 @@ Planner::callBatch(std::shared_ptr<BatchExecuteRequest> req)
             oldDec->print("info");
             SPDLOG_INFO("New decision:");
             decision->print("info");
+            state.numMigrations += 1;
 
             // We want to let all hosts involved in the migration (not only
             // those in the new decision) that we are gonna migrate. For the

src/planner/PlannerClient.cpp

@@ -381,6 +381,16 @@ PlannerClient::getSchedulingDecision(
     return decision;
 }
 
+int PlannerClient::getNumMigrations()
+{
+    EmptyRequest request;
+    NumMigrationsResponse response;
+
+    syncSend(PlannerCalls::GetNumMigrations, &request, &response);
+
+    return response.nummigrations();
+}
+
 void PlannerClient::preloadSchedulingDecision(
   std::shared_ptr<faabric::batch_scheduler::SchedulingDecision> preloadDec)
 {

src/planner/PlannerEndpointHandler.cpp

@@ -185,6 +185,11 @@ void PlannerEndpointHandler::onRequest(
                 }
             }
 
+            // Also include the total number of migrations to-date
+            int numMigrations =
+              faabric::planner::getPlanner().getNumMigrations();
+            inFlightAppsResponse.set_nummigrations(numMigrations);
+
             response.result(beast::http::status::ok);
             response.body() =
               faabric::util::messageToJson(inFlightAppsResponse);

src/planner/PlannerServer.cpp

@@ -61,6 +61,9 @@ std::unique_ptr<google::protobuf::Message> PlannerServer::doSyncRecv(
         case PlannerCalls::GetSchedulingDecision: {
             return recvGetSchedulingDecision(message.udata());
         }
+        case PlannerCalls::GetNumMigrations: {
+            return recvGetNumMigrations(message.udata());
+        }
         case PlannerCalls::PreloadSchedulingDecision: {
             return recvPreloadSchedulingDecision(message.udata());
         }
@@ -195,6 +198,15 @@ PlannerServer::recvGetSchedulingDecision(std::span<const uint8_t> buffer)
     return std::make_unique<faabric::PointToPointMappings>(mappings);
 }
 
+std::unique_ptr<google::protobuf::Message> PlannerServer::recvGetNumMigrations(
+  std::span<const uint8_t> buffer)
+{
+    NumMigrationsResponse response;
+    response.set_nummigrations(planner.getNumMigrations());
+
+    return std::make_unique<NumMigrationsResponse>(response);
+}
+
 std::unique_ptr<google::protobuf::Message>
 PlannerServer::recvPreloadSchedulingDecision(std::span<const uint8_t> buffer)
 {

src/planner/planner.proto

@@ -42,9 +42,9 @@ message HttpMessage {
         GET_CONFIG = 6;
         GET_EXEC_GRAPH = 7;
         GET_IN_FLIGHT_APPS = 8;
-        EXECUTE_BATCH = 9;
-        EXECUTE_BATCH_STATUS = 10;
-        PRELOAD_SCHEDULING_DECISION = 11;
+        EXECUTE_BATCH = 10;
+        EXECUTE_BATCH_STATUS = 11;
+        PRELOAD_SCHEDULING_DECISION = 12;
     }
 
     Type type = 1 [json_name = "http_type"];
@@ -65,6 +65,11 @@ message GetInFlightAppsResponse {
     }
 
     repeated InFlightApp apps = 1;
+    int32 numMigrations = 2;
+}
+
+message NumMigrationsResponse {
+    int32 numMigrations = 1;
 }
 
 // ---------------------------------------------

tests/dist/dist_test_fixtures.h

@@ -21,6 +21,9 @@ class DistTestsFixture
   public:
     DistTestsFixture()
     {
+        // Reset the planner
+        resetPlanner();
+
         // Make sure the host list is up to date
         sch.addHostToGlobalSet(getMasterIP());
         sch.addHostToGlobalSet(getWorkerIP());

tests/dist/mpi/test_mpi_functions.cpp

@@ -189,6 +189,9 @@ TEST_CASE_METHOD(MpiDistTestsFixture, "Test MPI function migration", "[mpi]")
 
     // Wait for messages to be finished
     checkAllocationAndResult(req, hostsAfterMigration);
+
+    // Check that, indeed, one migration happened
+    REQUIRE(plannerCli.getNumMigrations() == 1);
 }
 
 TEST_CASE_METHOD(MpiDistTestsFixture, "Test MPI gather", "[mpi]")

tests/test/planner/test_planner_client_server.cpp

@@ -259,6 +259,16 @@ TEST_CASE_METHOD(PlannerClientServerExecTestFixture,
     }
 }
 
+TEST_CASE_METHOD(PlannerClientServerExecTestFixture,
+                 "Test getting the number of migrations",
+                 "[planner]")
+{
+    // We should be able to get the number of migrations from the planner
+    REQUIRE(plannerCli.getNumMigrations() == 0);
+
+    // To see a non-zero result, we need a distributed test
+}
+
 TEST_CASE_METHOD(PlannerClientServerExecTestFixture,
                  "Test preloading a scheduling decision from the client",
                  "[planner]")

tests/test/planner/test_planner_endpoint.cpp

@@ -630,6 +630,7 @@ TEST_CASE_METHOD(PlannerEndpointExecTestFixture,
     HttpMessage inFlightMsg;
     inFlightMsg.set_type(HttpMessage_Type_GET_IN_FLIGHT_APPS);
     GetInFlightAppsResponse expectedResponse;
+    expectedResponse.set_nummigrations(0);
     msgJsonStr = faabric::util::messageToJson(inFlightMsg);
     expectedResponseBody = faabric::util::messageToJson(expectedResponse);
     expectedReturnCode = beast::http::status::ok;