402 advance scheduling improvements

fabric_cf/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "1.7.0"
+__version__ = "1.8.0b1"
 __VERSION__ = __version__

fabric_cf/actor/core/apis/abc_actor_management_object.py

@@ -238,8 +238,8 @@ def get_sites(self, *, caller: AuthToken, site: str) -> ResultSitesAvro:
     def get_reservations(self, *, caller: AuthToken, states: List[int] = None,
                          slice_id: ID = None, rid: ID = None, oidc_claim_sub: str = None,
                          email: str = None, rid_list: List[str] = None, type: str = None,
-                         site: str = None, node_id: str = None,
-                         host: str = None, ip_subnet: str = None, full: bool = False) -> ResultReservationAvro:
+                         site: str = None, node_id: str = None, host: str = None, ip_subnet: str = None,
+                         full: bool = False, start: datetime = None, end: datetime = None) -> ResultReservationAvro:
         """
         Get Reservations
         @param states states
@@ -256,10 +256,30 @@ def get_reservations(self, *, caller: AuthToken, states: List[int] = None,
         @param host host
         @param ip_subnet ip subnet
         @param full
+        @param start: start time
+        @param end: end time
 
         @return returns list of the reservations
         """
 
+    def get_components(self, *, node_id: str, rsv_type: list[str], states: list[int],
+                       component: str = None, bdf: str = None, start: datetime = None,
+                       end: datetime = None, excludes: List[str] = None) -> Dict[str, List[str]]:
+        """
+        Returns components matching the search criteria
+        @param node_id: Worker Node ID to which components belong
+        @param states: list of states used to find reservations
+        @param rsv_type: type of reservations
+        @param component: component name
+        @param bdf: Component's PCI address
+        @param start: start time
+        @param end: end time
+        @param excludes: Excludes the list of reservations
+        NOTE# For P4 switches; node_id=node+renc-p4-sw  component=ip+192.168.11.8 bdf=p1
+
+        @return Dictionary with component name as the key and value as list of associated PCI addresses in use.
+        """
+
     def get_slices(self, *, slice_id: ID, caller: AuthToken, slice_name: str = None, email: str = None,
                    states: List[int] = None, project: str = None, limit: int = None,
                    offset: int = None, user_id: str = None, search: str = None,

fabric_cf/actor/core/apis/abc_mgmt_actor.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod
+from datetime import datetime
 from typing import TYPE_CHECKING, List, Tuple, Dict
 
 from fabric_mb.message_bus.messages.delegation_avro import DelegationAvro
@@ -151,7 +152,8 @@ def accept_update_slice(self, *, slice_id: ID) -> bool:
     def get_reservations(self, *, states: List[int] = None, slice_id: ID = None,
                          rid: ID = None, oidc_claim_sub: str = None, email: str = None, rid_list: List[str] = None,
                          type: str = None, site: str = None, node_id: str = None,
-                         host: str = None, ip_subnet: str = None, full: bool = False) -> List[ReservationMng]:
+                         host: str = None, ip_subnet: str = None, full: bool = False,
+                         start: datetime = None, end: datetime = None) -> List[ReservationMng]:
         """
         Get Reservations
         @param states states
@@ -166,10 +168,31 @@ def get_reservations(self, *, states: List[int] = None, slice_id: ID = None,
         @param ip_subnet ip subnet
         @param host host
         @param full
+        @param start: start time
+        @param end: end time
         Obtains all reservations
         @return returns list of the reservations
         """
 
+    def get_components(self, *, node_id: str, rsv_type: list[str], states: list[int],
+                       component: str = None, bdf: str = None, start: datetime = None,
+                       end: datetime = None, excludes: List[str] = None) -> Dict[str, List[str]]:
+        """
+        Returns components matching the search criteria
+        @param node_id: Worker Node ID to which components belong
+        @param states: list of states used to find reservations
+        @param rsv_type: type of reservations
+        @param component: component name
+        @param bdf: Component's PCI address
+        @param start: start time
+        @param end: end time
+        @param excludes: Excludes the list of reservations
+        NOTE# For P4 switches; node_id=node+renc-p4-sw  component=ip+192.168.11.8 bdf=p1
+
+        @return Dictionary with component name as the key and value as list of associated PCI addresses in use.
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def get_sites(self, *, site: str) -> List[SiteAvro] or None:
         """

fabric_cf/actor/core/common/constants.py

@@ -308,7 +308,7 @@ class Constants:
 
     # Orchestrator Lease params
     TWO_WEEKS = timedelta(days=15)
-    DEFAULT_MAX_DURATION = TWO_WEEKS
+    DEFAULT_MAX_DURATION_IN_WEEKS = TWO_WEEKS
     LEASE_TIME_FORMAT = "%Y-%m-%d %H:%M:%S %z"
     DEFAULT_LEASE_IN_HOURS = 24
     LONG_LIVED_SLICE_TIME_WEEKS = timedelta(weeks=26)

fabric_cf/actor/core/core/controller.py

@@ -87,9 +87,6 @@ def __init__(self, *, identity: AuthToken = None, clock: ActorClock = None):
                                                                  thread_name_prefix=self.__class__.__name__)
         self.pluggable_registry = PluggableRegistry()
 
-        self.combined_broker_model = None
-        self.combined_broker_model_graph_id = None
-
     def __getstate__(self):
         state = self.__dict__.copy()
         del state['recovered']
@@ -116,8 +113,6 @@ def __getstate__(self):
         if hasattr(self, 'pluggable_registry'):
             del state['pluggable_registry']
 
-        del state['combined_broker_model']
-
         return state
 
     def __setstate__(self, state):

fabric_cf/actor/core/core/policy.py

@@ -23,6 +23,9 @@
 #
 #
 # Author: Komal Thareja (kthare10@renci.org)
+import enum
+from enum import Enum
+
 from fabric_cf.actor.boot.configuration import ActorConfig
 from fabric_cf.actor.core.apis.abc_actor_mixin import ABCActorMixin
 from fabric_cf.actor.core.apis.abc_delegation import ABCDelegation
@@ -36,6 +39,19 @@
 from fabric_cf.actor.core.kernel.resource_set import ResourceSet
 
 
+class AllocationAlgorithm(Enum):
+    FirstFit = enum.auto()
+    BestFit = enum.auto()
+    WorstFit = enum.auto()
+    Random = enum.auto()
+
+    def __repr__(self):
+        return self.name
+
+    def __str__(self):
+        return self.name
+
+
 class Policy(ABCPolicy):
     """
     Base class for all policy implementations.

fabric_cf/actor/core/manage/actor_management_object.py

@@ -428,11 +428,17 @@ def get_sites(self, *, caller: AuthToken, site: str) -> ResultSitesAvro:
 
         return result
 
+    def get_components(self, *, node_id: str, rsv_type: list[str], states: list[int],
+                       component: str = None, bdf: str = None, start: datetime = None,
+                       end: datetime = None, excludes: List[str] = None) -> Dict[str, List[str]]:
+        return self.db.get_components(node_id=node_id, rsv_type=rsv_type, states=states,
+                                      component=component, bdf=bdf, start=start, end=end, excludes=excludes)
+
     def get_reservations(self, *, caller: AuthToken, states: List[int] = None,
                          slice_id: ID = None, rid: ID = None, oidc_claim_sub: str = None,
                          email: str = None, rid_list: List[str] = None, type: str = None,
-                         site: str = None, node_id: str = None, host: str = None,
-                         ip_subnet: str = None, full: bool = False) -> ResultReservationAvro:
+                         site: str = None, node_id: str = None, host: str = None, ip_subnet: str = None,
+                         full: bool = False, start: datetime = None, end: datetime = None) -> ResultReservationAvro:
         result = ResultReservationAvro()
         result.status = ResultAvro()
 
@@ -452,7 +458,8 @@ def get_reservations(self, *, caller: AuthToken, states: List[int] = None,
                 else:
                     res_list = self.db.get_reservations(slice_id=slice_id, rid=rid, email=email,
                                                         states=states, rsv_type=rsv_type, site=site,
-                                                        graph_node_id=node_id, host=host, ip_subnet=ip_subnet)
+                                                        graph_node_id=node_id, host=host, ip_subnet=ip_subnet,
+                                                        start=start, end=end)
             except Exception as e:
                 self.logger.error("getReservations:db access {}".format(e))
                 result.status.set_code(ErrorCodes.ErrorDatabaseError.value)

fabric_cf/actor/core/manage/kafka/kafka_actor.py

@@ -25,6 +25,7 @@
 # Author: Komal Thareja (kthare10@renci.org)
 from __future__ import annotations
 
+from datetime import datetime
 from typing import List
 
 from fabric_mb.message_bus.messages.close_delegations_avro import CloseDelegationsAvro
@@ -132,7 +133,8 @@ def delete_slice(self, *, slice_id: ID) -> bool:
     def get_reservations(self, *, states: List[int] = None, slice_id: ID = None,
                          rid: ID = None, oidc_claim_sub: str = None, email: str = None, rid_list: List[str] = None,
                          type: str = None, site: str = None, node_id: str = None,
-                         host: str = None, ip_subnet: str = None, full: bool = False) -> List[ReservationMng]:
+                         host: str = None, ip_subnet: str = None, full: bool = False,
+                         start: datetime = None, end: datetime = None) -> List[ReservationMng]:
         request = GetReservationsRequestAvro()
         request = self.fill_request_by_id_message(request=request, slice_id=slice_id,
                                                   states=states, email=email, rid=rid,

fabric_cf/actor/core/manage/local/local_actor.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 import traceback
+from datetime import datetime
 from typing import TYPE_CHECKING, List, Tuple, Dict
 
 from fabric_mb.message_bus.messages.delegation_avro import DelegationAvro
@@ -44,12 +45,11 @@
     from fabric_mb.message_bus.messages.reservation_mng import ReservationMng
     from fabric_mb.message_bus.messages.reservation_state_avro import ReservationStateAvro
     from fabric_mb.message_bus.messages.slice_avro import SliceAvro
-    from fabric_cf.actor.core.manage.management_object import ManagementObject
     from fabric_cf.actor.security.auth_token import AuthToken
 
 
 class LocalActor(LocalProxy, ABCMgmtActor):
-    def __init__(self, *, manager: ManagementObject, auth: AuthToken):
+    def __init__(self, *, manager: ActorManagementObject, auth: AuthToken):
         super().__init__(manager=manager, auth=auth)
 
         if not isinstance(manager, ActorManagementObject):
@@ -111,13 +111,14 @@ def remove_slice(self, *, slice_id: ID) -> bool:
     def get_reservations(self, *, states: List[int] = None, slice_id: ID = None,
                          rid: ID = None, oidc_claim_sub: str = None, email: str = None, rid_list: List[str] = None,
                          type: str = None, site: str = None, node_id: str = None,
-                         host: str = None, ip_subnet: str = None, full: bool = False) -> List[ReservationMng]:
+                         host: str = None, ip_subnet: str = None, full: bool = False,
+                         start: datetime = None, end: datetime = None) -> List[ReservationMng]:
         self.clear_last()
         try:
             result = self.manager.get_reservations(caller=self.auth, states=states, slice_id=slice_id, rid=rid,
                                                    oidc_claim_sub=oidc_claim_sub, email=email, rid_list=rid_list,
                                                    type=type, site=site, node_id=node_id, host=host,
-                                                   ip_subnet=ip_subnet, full=full)
+                                                   ip_subnet=ip_subnet, full=full, start=start, end=end)
             self.last_status = result.status
 
             if result.status.get_code() == 0:
@@ -126,6 +127,16 @@ def get_reservations(self, *, states: List[int] = None, slice_id: ID = None,
         except Exception as e:
             self.on_exception(e=e, traceback_str=traceback.format_exc())
 
+    def get_components(self, *, node_id: str, rsv_type: list[str], states: list[int],
+                       component: str = None, bdf: str = None, start: datetime = None,
+                       end: datetime = None, excludes: List[str] = None) -> Dict[str, List[str]]:
+        try:
+            return self.manager.get_components(node_id=node_id, rsv_type=rsv_type, states=states,
+                                               component=component, bdf=bdf, start=start,
+                                               end=end, excludes=excludes)
+        except Exception as e:
+            self.on_exception(e=e, traceback_str=traceback.format_exc())
+
     def get_sites(self, *, site: str) -> List[SiteAvro] or None:
         self.clear_last()
         try:

fabric_cf/actor/core/manage/management_utils.py

@@ -67,6 +67,7 @@ class ManagementUtils:
     @staticmethod
     def update_slice(*, slice_obj: ABCSlice, slice_mng: SliceAvro) -> ABCSlice:
         slice_obj.set_graph_id(graph_id=slice_mng.get_graph_id())
+        slice_obj.set_lease_start(lease_start=slice_mng.get_lease_start())
         slice_obj.set_lease_end(lease_end=slice_mng.get_lease_end())
         slice_obj.set_config_properties(value=slice_mng.get_config_properties())
         return slice_obj

fabric_cf/actor/core/policy/broker_simpler_units_policy.py

@@ -39,7 +39,7 @@
 from fim.slivers.attached_components import ComponentSliver, ComponentType
 from fim.slivers.base_sliver import BaseSliver
 from fim.slivers.capacities_labels import Labels, Capacities
-from fim.slivers.interface_info import InterfaceSliver, InterfaceType
+from fim.slivers.interface_info import InterfaceType
 from fim.slivers.network_node import NodeSliver, NodeType
 from fim.slivers.network_service import NetworkServiceSliver, ServiceType, NSLayer
 from fim.slivers.path_info import Path
@@ -50,6 +50,7 @@
 from fabric_cf.actor.core.apis.abc_reservation_mixin import ABCReservationMixin
 from fabric_cf.actor.core.common.constants import Constants
 from fabric_cf.actor.core.container.maintenance import Maintenance
+from fabric_cf.actor.core.core.policy import AllocationAlgorithm
 from fabric_cf.actor.core.delegation.resource_ticket import ResourceTicketFactory
 from fabric_cf.actor.core.common.exceptions import BrokerException, ExceptionErrorCode
 from fabric_cf.actor.core.kernel.reservation_states import ReservationStates, ReservationOperation
@@ -75,19 +76,6 @@
     from fabric_cf.actor.core.apis.abc_broker_mixin import ABCBrokerMixin
 
 
-class BrokerAllocationAlgorithm(Enum):
-    FirstFit = enum.auto()
-    BestFit = enum.auto()
-    WorstFit = enum.auto()
-    Random = enum.auto()
-
-    def __repr__(self):
-        return self.name
-
-    def __str__(self):
-        return self.name
-
-
 class BrokerSimplerUnitsPolicy(BrokerCalendarPolicy):
     """
         BrokerSimplerUnitsPolicy is a simple implementation of the broker policy interface.
@@ -669,8 +657,9 @@ def __allocate_nodes(self, *, reservation: ABCBrokerReservation, inv: NetworkNod
         @return tuple containing delegation id, sliver, error message if any
         """
         delegation_id = None
-        node_id_list = self.__candidate_nodes(sliver=sliver)
-        if self.get_algorithm_type(site=sliver.site) == BrokerAllocationAlgorithm.Random:
+        node_id_list = FimHelper.candidate_nodes(combined_broker_model=self.combined_broker_model,
+                                                 sliver=sliver)
+        if self.get_algorithm_type(site=sliver.site) == AllocationAlgorithm.Random:
             random.shuffle(node_id_list)
         else:
             # Reshuffle Nodes based on CPU Threshold only for VMs when no specific host is specified
@@ -857,8 +846,8 @@ def __allocate_services(self, *, rid: ID, inv: NetworkServiceInventory, sliver:
                 device_name = owner_switch.get_name()
 
                 if device_name == Constants.AL2S:
-                    delegation_id, delegated_label = InventoryForType.get_delegations(lab_cap_delegations=
-                                                                                       net_cp.get_label_delegations())
+                    delegation_id, delegated_label = FimHelper.get_delegations(delegations=
+                                                                               net_cp.get_label_delegations())
                     device_name = delegated_label.device_name
                     local_name = delegated_label.local_name
 
@@ -938,11 +927,11 @@ def __allocate_services(self, *, rid: ID, inv: NetworkServiceInventory, sliver:
                             owner_mpls_ns = ns
                             break
                 if owner_ns and ServiceType.MPLS == owner_ns.get_type():
-                    delegation_id, delegated_label = InventoryForType.get_delegations(lab_cap_delegations=
-                                                                                      owner_switch.get_label_delegations())
+                    delegation_id, delegated_label = FimHelper.get_delegations(delegations=
+                                                                               owner_switch.get_label_delegations())
                 else:
-                    delegation_id, delegated_label = InventoryForType.get_delegations(lab_cap_delegations=
-                                                                                      owner_ns.get_label_delegations())
+                    delegation_id, delegated_label = FimHelper.get_delegations(delegations=
+                                                                               owner_ns.get_label_delegations())
 
             # Set the Subnet and gateway from the Owner Switch (a)
             existing_reservations = self.get_existing_reservations(node_id=owner_ns_id,
@@ -1718,17 +1707,17 @@ def unmerge_adm(self, *, graph_id: str):
                 self.combined_broker_model.rollback(graph_id=snapshot_graph_id)
             raise e
 
-    def get_algorithm_type(self, site: str) -> BrokerAllocationAlgorithm:
+    def get_algorithm_type(self, site: str) -> AllocationAlgorithm:
         if self.properties is not None:
             algorithms = self.properties.get(Constants.ALGORITHM, None)
-            random_algo = algorithms.get(str(BrokerAllocationAlgorithm.Random))
+            random_algo = algorithms.get(str(AllocationAlgorithm.Random))
             if random_algo and random_algo.get('enabled') and random_algo.get('sites') and \
                     site in random_algo.get('sites'):
-                return BrokerAllocationAlgorithm.Random
-            first_fit_algo = algorithms.get(BrokerAllocationAlgorithm.Random.name)
+                return AllocationAlgorithm.Random
+            first_fit_algo = algorithms.get(AllocationAlgorithm.Random.name)
             if first_fit_algo and first_fit_algo.get('enabled'):
-                return BrokerAllocationAlgorithm.FirstFit
-        return BrokerAllocationAlgorithm.FirstFit
+                return AllocationAlgorithm.FirstFit
+        return AllocationAlgorithm.FirstFit
 
     def get_core_capacity_threshold(self) -> Tuple[bool, int]:
         if self.properties is not None:
@@ -1754,8 +1743,8 @@ def get_node_capacities(self, node_id: str, node_id_to_reservations: dict,
                                                                    start=term.get_start_time(),
                                                                    end=term.get_end_time())
 
-            delegation_id, delegated_capacity = NetworkNodeInventory.get_delegations(
-                lab_cap_delegations=graph_node.get_capacity_delegations())
+            delegation_id, delegated_capacity = FimHelper.get_delegations(
+                delegations=graph_node.get_capacity_delegations())
 
             allocated_capacity = Capacities()