CM Add support 0 num instances in launcher

src/core/cm/launcher/CMakeLists.txt

@@ -24,6 +24,7 @@ set(SOURCES
     networkmanager.cpp
     node.cpp
     nodemanager.cpp
+    runrequestsloader.cpp
     storagestate.cpp
 )
 
@@ -34,6 +35,7 @@ set(SOURCES
 set(HEADERS
     gidpool.hpp
     imageinfoprovider.hpp
+    runrequestsloader.hpp
     itf/instancerunner.hpp
     itf/instancestatusreceiver.hpp
     itf/launcher.hpp

src/core/cm/launcher/balancer.cpp

@@ -168,6 +168,11 @@
 
 Error Balancer::SelectNodes(Instance& instance, Array<Node*>& nodes)
 {
+    FilterNodesByID(instance, nodes);
+    if (nodes.IsEmpty()) {
+        return AOS_ERROR_WRAP(Error(ErrorEnum::eNotFound, "no nodes with given ID"));
+    }
+
     FilterNodesByLabels(instance, nodes);
     if (nodes.IsEmpty()) {
         return AOS_ERROR_WRAP(Error(ErrorEnum::eNotFound, "no nodes with instance labels"));
@@ -181,6 +186,11 @@
     return ErrorEnum::eNone;
 }
 
+void Balancer::FilterNodesByID(Instance& instance, Array<Node*>& nodes)
+{
+    nodes.RemoveIf([&instance](const Node* node) { return !instance.IsNodeIDOk(node->GetInfo().mNodeID); });
+}
+
 void Balancer::FilterNodesByLabels(Instance& instance, Array<Node*>& nodes)
 {
     nodes.RemoveIf([&instance](const Node* node) { return !instance.AreNodeLabelsOk(node->GetConfig().mLabels); });

src/core/cm/launcher/balancer.hpp

@@ -17,7 +17,7 @@
 
 namespace aos::cm::launcher {
 
-/** @addtogroup CM Launcher
+/** @addtogroup cm Communication Manager
  *  @{
  */
 
@@ -74,6 +74,7 @@ class Balancer {
 
     // Selects nodes
     Error SelectNodes(Instance& instance, Array<Node*>& nodes);
+    void  FilterNodesByID(Instance& instance, Array<Node*>& nodes);
     void  FilterNodesByLabels(Instance& instance, Array<Node*>& nodes);
     void  FilterNodesByResources(Instance& instance, Array<Node*>& nodes);
 

src/core/cm/launcher/instance.cpp

@@ -164,6 +164,11 @@
     return true;
 }
 
+bool Instance::IsNodeIDOk(const String& nodeID)
+{
+    return !mInfo.mDisableRebalancing || mInfo.mNodeID == nodeID;
+}
+
 bool Instance::AreNodeLabelsOk(const LabelsArray& nodeLabels)
 {
     for (const auto& label : mInfo.mLabels) {

src/core/cm/launcher/instance.hpp

@@ -25,7 +25,7 @@
 
 namespace aos::cm::launcher {
 
-/** @addtogroup CM Launcher
+/** @addtogroup cm Communication Manager
  *  @{
  */
 
@@ -191,6 +191,14 @@ class Instance {
      */
     bool IsPlatformOk(const PlatformInfo& platformInfo);
 
+    /**
+     * Checks whether node ID fits instance requirements.
+     *
+     * @param nodeID node ID.
+     * @return bool.
+     */
+    bool IsNodeIDOk(const String& nodeID);
+
     /**
      * Checks whether node resources fit instance requirements.
      *

src/core/cm/launcher/instancemanager.cpp

@@ -177,15 +177,34 @@
     return instance->UpdateStatus(status);
 }
 
-RetWithError<SharedPtr<Instance>> InstanceManager::CreateInstance(
-    const InstanceIdent& id, const RunInstanceRequest& request)
+RetWithError<SharedPtr<Instance>> InstanceManager::CreateInstance(const RunInstanceRequest& request, uint64_t index)
 {
+    auto id = InstanceIdent {request.mItemID, request.mSubjectInfo.mSubjectID, index, request.mUpdateItemType};
     if (auto instance = FindReadyInstance(id); instance) {
         return {instance, ErrorEnum::eNone};
     }
 
-    auto instanceInfo = MakeUnique<InstanceInfo>(&mAllocator);
-    CreateInfo(id, request, *instanceInfo);
+    auto instanceInfo = CreateInfo(id, "", request);
+
+    if (auto err = mStorage->AddInstance(*instanceInfo); !err.IsNone()) {
+        return {nullptr, AOS_ERROR_WRAP(err)};
+    }
+
+    return CreateInstance(*instanceInfo);
+}
+
+RetWithError<SharedPtr<Instance>> InstanceManager::CreateInstance(
+    const RunInstanceRequest& request, const String& nodeID, const Array<SharedPtr<Instance>>& newInstances)
+{
+    if (auto instance = FindReadyInstance(request.mItemID, request.mSubjectInfo.mSubjectID, nodeID); instance) {
+        return {instance, ErrorEnum::eNone};
+    }
+
+    auto index = FindIndexForNewInstance(request.mItemID, request.mSubjectInfo.mSubjectID);
+    index      = Max(index, FindIndexForNewInstance(newInstances, request.mItemID, request.mSubjectInfo.mSubjectID));
+
+    auto id = InstanceIdent {request.mItemID, request.mSubjectInfo.mSubjectID, index, request.mUpdateItemType};
+    auto instanceInfo = CreateInfo(id, nodeID, request);
 
     if (auto err = mStorage->AddInstance(*instanceInfo); !err.IsNone()) {
         return {nullptr, AOS_ERROR_WRAP(err)};
@@ -241,26 +260,17 @@
 
 SharedPtr<Instance> InstanceManager::FindActiveInstance(const InstanceIdent& id)
 {
-    auto it = mActiveInstances.FindIf(
-        [&id](SharedPtr<Instance>& instance) { return instance->GetInfo().mInstanceIdent == id; });
-
-    return it != mActiveInstances.end() ? *it : SharedPtr<Instance>();
+    return FindInstance(mActiveInstances, id);
 }
 
 SharedPtr<Instance> InstanceManager::FindScheduledInstance(const InstanceIdent& id)
 {
-    auto it = mScheduledInstances.FindIf(
-        [&id](SharedPtr<Instance>& instance) { return instance->GetInfo().mInstanceIdent == id; });
-
-    return it != mScheduledInstances.end() ? *it : SharedPtr<Instance>();
+    return FindInstance(mScheduledInstances, id);
 }
 
 SharedPtr<Instance> InstanceManager::FindCachedInstance(const InstanceIdent& id)
 {
-    auto it = mCachedInstances.FindIf(
-        [&id](SharedPtr<Instance>& instance) { return instance->GetInfo().mInstanceIdent == id; });
-
-    return it != mCachedInstances.end() ? *it : SharedPtr<Instance>();
+    return FindInstance(mCachedInstances, id);
 }
 
 InstanceStatus* InstanceManager::FindPreinstalledComponent(const InstanceIdent& id)
@@ -291,7 +301,7 @@
     mCachedInstances.Clear();
 
     auto instances = MakeUnique<StaticArray<InstanceInfo, cMaxNumInstances>>(&mAllocator);
-    if (auto err = mStorage->GetActiveInstances(*instances); !err.IsNone()) {
+    if (auto err = mStorage->LoadActiveInstances(*instances); !err.IsNone()) {
         return AOS_ERROR_WRAP(err);
     }
 
@@ -567,23 +577,93 @@
     return nullptr;
 }
 
-void InstanceManager::CreateInfo(const InstanceIdent& id, const RunInstanceRequest& request, InstanceInfo& info)
-{
-    info.mInstanceIdent  = id;
-    info.mManifestDigest = "";
-    info.mNodeID         = "";
-    info.mPrevNodeID     = "";
-    info.mRuntimeID      = "";
-    info.mUID            = 0;
-    info.mGID            = 0;
-    info.mTimestamp      = Time::Now();
-    info.mState          = InstanceStateEnum::eActive;
-    info.mIsUnitSubject  = request.mSubjectInfo.mIsUnitSubject;
-    info.mVersion        = request.mVersion;
-    info.mOwnerID        = request.mOwnerID;
-    info.mSubjectType    = request.mSubjectInfo.mSubjectType;
-    info.mLabels         = request.mLabels;
-    info.mPriority       = request.mPriority;
+SharedPtr<Instance> InstanceManager::FindReadyInstance(
+    const String& itemID, const String& subjectID, const String& nodeID)
+{
+    auto instance = FindInstance(mScheduledInstances, itemID, subjectID, nodeID);
+    if (instance) {
+        return instance;
+    }
+
+    instance = FindInstance(mActiveInstances, itemID, subjectID, nodeID);
+    if (instance) {
+        return instance;
+    }
+
+    instance = FindInstance(mCachedInstances, itemID, subjectID, nodeID);
+    if (instance) {
+        return instance;
+    }
+
+    return nullptr;
+}
+
+uint64_t InstanceManager::FindIndexForNewInstance(const String& itemID, const String& subjectID)
+{
+    uint64_t schedInstance  = FindIndexForNewInstance(mScheduledInstances, itemID, subjectID);
+    uint64_t activeInstance = FindIndexForNewInstance(mActiveInstances, itemID, subjectID);
+    uint64_t cachedInstance = FindIndexForNewInstance(mCachedInstances, itemID, subjectID);
+
+    return Max(schedInstance, activeInstance, cachedInstance);
+}
+
+UniquePtr<InstanceInfo> InstanceManager::CreateInfo(
+    const InstanceIdent& id, const String& nodeID, const RunInstanceRequest& request)
+{
+    auto info = MakeUnique<InstanceInfo>(&mAllocator);
+
+    info->mInstanceIdent      = id;
+    info->mManifestDigest     = "";
+    info->mNodeID             = nodeID;
+    info->mPrevNodeID         = "";
+    info->mRuntimeID          = "";
+    info->mUID                = 0;
+    info->mGID                = 0;
+    info->mTimestamp          = Time::Now();
+    info->mState              = InstanceStateEnum::eActive;
+    info->mIsUnitSubject      = request.mSubjectInfo.mIsUnitSubject;
+    info->mVersion            = request.mVersion;
+    info->mOwnerID            = request.mOwnerID;
+    info->mSubjectType        = request.mSubjectInfo.mSubjectType;
+    info->mLabels             = request.mLabels;
+    info->mPriority           = request.mPriority;
+    info->mDisableRebalancing = !nodeID.IsEmpty();
+
+    return info;
+}
+
+SharedPtr<Instance> InstanceManager::FindInstance(const Array<SharedPtr<Instance>>& instances, const InstanceIdent& id)
+{
+    auto it = instances.FindIf(
+        [&id](const SharedPtr<Instance>& instance) { return instance->GetInfo().mInstanceIdent == id; });
+
+    return it != instances.end() ? *it : SharedPtr<Instance>();
+}
+
+SharedPtr<Instance> InstanceManager::FindInstance(
+    const Array<SharedPtr<Instance>>& instances, const String& itemID, const String& subjectID, const String& nodeID)
+{
+    auto it = instances.FindIf([&itemID, &subjectID, &nodeID](const SharedPtr<Instance>& instance) {
+        return instance->GetInfo().mInstanceIdent.mItemID == itemID
+            && instance->GetInfo().mInstanceIdent.mSubjectID == subjectID && instance->GetInfo().mNodeID == nodeID;
+    });
+
+    return it != instances.end() ? *it : SharedPtr<Instance>();
+}
+
+uint64_t InstanceManager::FindIndexForNewInstance(
+    const Array<SharedPtr<Instance>>& instances, const String& itemID, const String& subjectID)
+{
+    uint64_t newIndex = 0;
+
+    for (auto it = instances.begin(); it != instances.end(); ++it) {
+        if ((*it)->GetInfo().mInstanceIdent.mItemID == itemID
+            && (*it)->GetInfo().mInstanceIdent.mSubjectID == subjectID) {
+            newIndex = Max(newIndex, (*it)->GetInfo().mInstanceIdent.mInstance + 1);
+        }
+    }
+
+    return newIndex;
 }
 
 } // namespace aos::cm::launcher

src/core/cm/launcher/instancemanager.hpp

@@ -23,14 +23,14 @@
 
 namespace aos::cm::launcher {
 
-/** @addtogroup CM Launcher
+/** @addtogroup cm Communication Manager
  *  @{
  */
 
 /**
  * Auxiliary class accumulating data about running service instances.
  */
 class InstanceManager {
 public:
     /**
      * Initializes the instance manager with configuration and required interfaces.
@@ -145,13 +145,24 @@
     Error UpdateStatus(const InstanceStatus& status);
 
     /**
-     * Creates new instance object or returns existing one.
+     * Creates new instance object.
      *
-     * @param id instance identifier.
      * @param request run instance request.
-     * @return SharedPtr<Instance>.
+     * @param index index of the instance.
+     * @return RetWithError<SharedPtr<Instance>>.
+     */
+    RetWithError<SharedPtr<Instance>> CreateInstance(const RunInstanceRequest& request, uint64_t index);
+
+    /**
+     * Creates new instance object.
+     *
+     * @param request run instance request.
+     * @param nodeID bound node ID.
+     * @param newInstances array of new instances.
+     * @return RetWithError<SharedPtr<Instance>>.
      */
-    RetWithError<SharedPtr<Instance>> CreateInstance(const InstanceIdent& id, const RunInstanceRequest& request);
+    RetWithError<SharedPtr<Instance>> CreateInstance(
+        const RunInstanceRequest& request, const String& nodeID, const Array<SharedPtr<Instance>>& newInstances);
 
     /**
      * Submits all stashed instances for execution.
@@ -252,7 +263,17 @@
     RetWithError<SharedPtr<Instance>> CreateInstance(const InstanceInfo& info);
 
     SharedPtr<Instance> FindReadyInstance(const InstanceIdent& id);
-    void                CreateInfo(const InstanceIdent& id, const RunInstanceRequest& request, InstanceInfo& info);
+    SharedPtr<Instance> FindReadyInstance(const String& itemID, const String& subjectID, const String& nodeID);
+    uint64_t            FindIndexForNewInstance(const String& itemID, const String& subjectID);
+
+    UniquePtr<InstanceInfo> CreateInfo(
+        const InstanceIdent& id, const String& nodeID, const RunInstanceRequest& request);
+
+    SharedPtr<Instance> FindInstance(const Array<SharedPtr<Instance>>& instances, const InstanceIdent& id);
+    SharedPtr<Instance> FindInstance(const Array<SharedPtr<Instance>>& instances, const String& itemID,
+        const String& subjectID, const String& nodeID);
+    uint64_t            FindIndexForNewInstance(
+                   const Array<SharedPtr<Instance>>& instances, const String& itemID, const String& subjectID);
 
     Config          mConfig;
     StorageItf*     mStorage {};
@@ -280,6 +301,24 @@
     OverrideEnvVarsRequest mEnvVarsOverrides;
 };
 
+/**
+ * Pushes a value to an array if it is not already present.
+ *
+ * @tparam T type of the value to push.
+ * @param array array to push the value to.
+ * @param value value to push.
+ * @return Error.
+ */
+template <typename Container, typename T>
+Error PushUnique(Container& array, const T& value)
+{
+    if (array.Contains(value)) {
+        return ErrorEnum::eNone;
+    }
+
+    return array.PushBack(value);
+}
+
 /** @}*/
 
 } // namespace aos::cm::launcher

src/core/cm/launcher/itf/envvarhandler.hpp

@@ -34,6 +34,8 @@ class EnvVarHandlerItf {
     virtual Error OverrideEnvVars(const OverrideEnvVarsRequest& envVars) = 0;
 };
 
+/** @}*/
+
 } // namespace aos::cm::launcher
 
 #endif

src/core/cm/launcher/itf/instancestatusreceiver.hpp

@@ -44,6 +44,8 @@ class InstanceStatusReceiverItf {
     virtual Error OnNodeInstancesStatusesReceived(const String& nodeID, const Array<InstanceStatus>& statuses) = 0;
 };
 
+/** @}*/
+
 } // namespace aos::cm::launcher
 
 #endif