Merge pull request #397 from fabric-testbed/al2s-slice-state

Al2s slice state

Author: kthare10

fabric_cf/actor/core/kernel/slice_state_machine.py

@@ -323,7 +323,7 @@ def transition_slice(self, *, operation: SliceOperation, reservations: Reservati
                                                  ReservationStates.Failed, ReservationStates.CloseFail):
                     self.state = SliceState.Closing
 
-            if self.state in [SliceState.AllocatedOK, SliceState.AllocatedError]:
+            elif self.state in [SliceState.AllocatedOK, SliceState.AllocatedError]:
                 if not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Closed,
                                                  ReservationStates.CloseFail):
                     if not has_error:

fabric_cf/actor/core/policy/network_node_inventory.py

@@ -120,9 +120,6 @@ def __update_shared_nic_labels_and_capacities(self, *, available_component: Comp
             raise BrokerException(error_code=ExceptionErrorCode.INSUFFICIENT_RESOURCES,
                                   msg=f"{message}")
 
-        # Assign the first PCI Id from the list of available PCI slots
-        requested_component.label_allocations = Labels(bdf=delegated_label.bdf[0], numa=delegated_label.numa[0])
-
         # Find the VLAN from the BQM Component
         if available_component.network_service_info is None or \
                 len(available_component.network_service_info.network_services) != 1:
@@ -145,14 +142,25 @@ def __update_shared_nic_labels_and_capacities(self, *, available_component: Comp
 
         delegation_id, ifs_delegated_labels = self.get_delegations(lab_cap_delegations=ifs.get_label_delegations())
 
-        # Determine the index which points to the same PCI id as assigned above
-        # This index points to the other relevant information such as MAC Address,
-        # VLAN tag for that PCI device
-        i = 0
-        for pci_id in ifs_delegated_labels.bdf:
-            if pci_id == delegated_label.bdf[0]:
-                break
-            i += 1
+        assigned_bdf = delegated_label.bdf[0]
+        assigned_numa = delegated_label.numa[0]
+
+        # Check if the requested component's VLAN exists in the delegated labels
+        if requested_component.labels and requested_component.labels.vlan and \
+                requested_component.labels.vlan in ifs_delegated_labels.vlan:
+            vlan_index = ifs_delegated_labels.vlan.index(requested_component.labels.vlan)
+            bdf_for_requested_vlan = ifs_delegated_labels.bdf[vlan_index]
+            
+            if bdf_for_requested_vlan in delegated_label.bdf:
+                bdf_index = delegated_label.bdf.index(bdf_for_requested_vlan)
+                assigned_bdf = bdf_for_requested_vlan
+                assigned_numa = delegated_label.numa[bdf_index]
+
+        # Assign the first PCI Id from the list of available PCI slots
+        requested_component.label_allocations = Labels(bdf=assigned_bdf, numa=assigned_numa)
+
+        # Find index of assigned BDF in the interface delegated labels
+        assigned_index = ifs_delegated_labels.bdf.index(assigned_bdf)
 
         # Updated the Requested component with VLAN, BDF, MAC
         req_ns_name = next(iter(requested_component.network_service_info.network_services))
@@ -162,11 +170,12 @@ def __update_shared_nic_labels_and_capacities(self, *, available_component: Comp
 
         # Do not copy VLAN for OpenStack-vNIC
         if requested_component.get_model() == Constants.OPENSTACK_VNIC_MODEL:
-            lab = Labels(bdf=ifs_delegated_labels.bdf[i], mac=ifs_delegated_labels.mac[i],
-                         local_name=ifs_delegated_labels.local_name[i])
+            lab = Labels(bdf=ifs_delegated_labels.bdf[assigned_index], mac=ifs_delegated_labels.mac[assigned_index],
+                         local_name=ifs_delegated_labels.local_name[assigned_index])
         else:
-            lab = Labels(bdf=ifs_delegated_labels.bdf[i], mac=ifs_delegated_labels.mac[i],
-                         vlan=ifs_delegated_labels.vlan[i], local_name=ifs_delegated_labels.local_name[i])
+            lab = Labels(bdf=ifs_delegated_labels.bdf[assigned_index], mac=ifs_delegated_labels.mac[assigned_index],
+                         vlan=ifs_delegated_labels.vlan[assigned_index],
+                         local_name=ifs_delegated_labels.local_name[assigned_index])
 
         # For the Layer 2 copying the IP address to the label allocations
         # This is to be used by AM Handler to configure Network Interface

fabric_cf/actor/fim/asm_update_thread.py

@@ -43,14 +43,17 @@ class AsmUpdateException(Exception):
 
 class AsmEvent:
     def __init__(self, *, graph_id: str, sliver: BaseSliver, reservation_id: str,
-                 state: str, error_message: str):
+                 state: str, error_message: str, logger: logging.Logger = None):
         self.graph_id = graph_id
         self.sliver = sliver
         self.reservation_id = reservation_id
         self.state = state
         self.error_message = error_message
+        self.logger = logger
 
     def process(self):
+        if self.logger:
+            self.logger.debug(f"AsmEvent for Res# {self.reservation_id} State: {self.state} Graph: {self.graph_id}")
         FimHelper.update_node(graph_id=self.graph_id, sliver=self.sliver, reservation_id=self.reservation_id,
                               state=self.state, error_message=self.error_message)
 
@@ -124,7 +127,7 @@ def enqueue(self, *, graph_id: str, sliver: BaseSliver, rid: str, reservation_st
                 error_message: str):
         try:
             event = AsmEvent(graph_id=graph_id, sliver=sliver, reservation_id=rid,
-                             state=reservation_state, error_message=error_message)
+                             state=reservation_state, error_message=error_message, logger=self.logger)
             self.event_queue.put_nowait(event)
             with self.condition:
                 self.condition.notify_all()

fabric_cf/actor/fim/fim_helper.py

@@ -316,21 +316,26 @@ def get_delegation(delegated_capacities: Delegations, delegation_name: str) -> C
         return delegation.get_details() if delegation is not None else None
 
     @staticmethod
-    def update_node(*, graph_id: str, sliver: BaseSliver, reservation_id: str,
-                 state: str, error_message: str):
+    def update_node(*, sliver: BaseSliver, reservation_id: str, state: str, error_message: str, graph_id: str = None,
+                    asm_graph: ABCASMPropertyGraph = None):
         """
         Update Sliver Node in ASM
         :param graph_id:
         :param sliver:
         :param reservation_id:
         :param state:
         :param error_message:
+        :param asm_graph:
         :return:
         """
         if sliver is None:
             return
-        graph = FimHelper.get_graph(graph_id=graph_id)
-        asm_graph = Neo4jASMFactory.create(graph=graph)
+        if graph_id is None and asm_graph is None:
+            return
+        if graph_id:
+            graph = FimHelper.get_graph(graph_id=graph_id)
+            asm_graph = Neo4jASMFactory.create(graph=graph)
+
         neo4j_topo = ExperimentTopology()
         neo4j_topo.cast(asm_graph=asm_graph)
 

fabric_cf/orchestrator/core/orchestrator_handler.py

@@ -503,7 +503,7 @@ def modify_slice(self, *, token: str, slice_id: str, slice_graph: str) -> List[d
                                                     slice_obj=slice_obj, logger=self.logger)
 
             # Compute the reservations
-            computed_reservations = slice_object.modify(new_slice_graph=asm_graph)
+            topology_diff, computed_reservations = slice_object.modify(new_slice_graph=asm_graph)
             slice_object.update_topology(topology=topology)
 
             # Check if Test Bed or site is in maintenance
@@ -512,12 +512,12 @@ def modify_slice(self, *, token: str, slice_id: str, slice_graph: str) -> List[d
             # Add any new reservations to the database
             slice_object.add_reservations()
 
-            FimHelper.delete_graph(graph_id=slice_obj.get_graph_id())
-
             # Slice has sliver modifications - add/remove/update for slivers requiring AM updates
             modify_state = slice_object.has_sliver_updates_at_authority()
+            FimHelper.delete_graph(graph_id=slice_obj.get_graph_id())
+            graph_id = asm_graph.get_graph_id()
 
-            slice_obj.graph_id = asm_graph.get_graph_id()
+            slice_obj.graph_id = graph_id
             config_props = slice_obj.get_config_properties()
             config_props[Constants.PROJECT_ID] = project
             config_props[Constants.TAGS] = ','.join(tags)

fabric_cf/orchestrator/core/orchestrator_slice_wrapper.py

@@ -41,7 +41,7 @@
 from fim.slivers.instance_catalog import InstanceCatalog
 from fim.slivers.network_node import NodeSliver, NodeType
 from fim.slivers.network_service import NetworkServiceSliver
-from fim.slivers.topology_diff import WhatsModifiedFlag
+from fim.slivers.topology_diff import WhatsModifiedFlag, TopologyDiff
 from fim.user import ServiceType, ExperimentTopology, InterfaceType
 
 from fabric_cf.actor.core.common.constants import ErrorCodes, Constants
@@ -333,7 +333,7 @@ def __build_ns_sliver_reservation(self, *, slice_graph: ABCASMPropertyGraph, nod
                         if sliver.labels is None:
                             sliver.labels = Labels()
                         sliver.labels = Labels.update(sliver.labels,
-                                                      local_name=f"{self.slice_obj.get_slice_name()}-{ifs.peer_labels.account_id}")
+                                                      local_name=f"{self.slice_obj.get_slice_name()}")
 
                 # Generate reservation for the sliver
                 reservation = self.reservation_converter.generate_reservation(sliver=sliver,
@@ -460,7 +460,7 @@ def __build_network_node_reservations(self, *, slice_graph: ABCASMPropertyGraph)
             sliver_to_res_mapping[nn_id] = reservation.get_reservation_id()
         return reservations, sliver_to_res_mapping
 
-    def modify(self, *, new_slice_graph: ABCASMPropertyGraph) -> List[LeaseReservationAvro]:
+    def modify(self, *, new_slice_graph: ABCASMPropertyGraph) -> Tuple[TopologyDiff, List[LeaseReservationAvro]]:
         """
         Modify an existing slice
         :param new_slice_graph New Slice Graph
@@ -478,8 +478,8 @@ def modify(self, *, new_slice_graph: ABCASMPropertyGraph) -> List[LeaseReservati
         ns_peered_reservations = []
         ns_mapping = {}
 
-        if topology_diff is None:
-            return reservations
+        if not topology_diff:
+            return topology_diff, reservations
 
         node_res_mapping = {}
 
@@ -624,7 +624,7 @@ def modify(self, *, new_slice_graph: ABCASMPropertyGraph) -> List[LeaseReservati
         for x in modified_reservations:
             self.computed_reservations.append(x)
 
-        return self.computed_reservations
+        return topology_diff, self.computed_reservations
 
     def __check_modify_on_fabnetv4ext(self, *, rid: str, req_sliver: NetworkServiceSliver) -> NetworkServiceSliver:
         if req_sliver.get_type() != ServiceType.FABNetv4Ext:
@@ -704,21 +704,47 @@ def __check_modify_on_fabnetv4ext(self, *, rid: str, req_sliver: NetworkServiceS
 
         return req_sliver
 
-    def update_topology(self, *, topology: ExperimentTopology):
-        for x in self.computed_reservations:
-            sliver = x.get_sliver()
-            node_name = sliver.get_name()
-            if isinstance(sliver, NodeSliver) and node_name in topology.nodes:
-                node = topology.nodes[node_name]
-                node.set_properties(labels=sliver.labels,
-                                    label_allocations=sliver.label_allocations,
-                                    capacity_allocations=sliver.capacity_allocations,
-                                    reservation_info=sliver.reservation_info,
-                                    node_map=sliver.node_map,
-                                    management_ip=sliver.management_ip,
-                                    capacity_hints=sliver.capacity_hints,
-                                    capacities=sliver.capacities)
+    def update_topology(self, *, topology: ExperimentTopology = None,
+                        asm_graph: ABCASMPropertyGraph = None):
+        if topology:
+            for x in self.computed_reservations:
+                sliver = x.get_sliver()
+                node_name = sliver.get_name()
+                if isinstance(sliver, NodeSliver) and node_name in topology.nodes:
+                    node = topology.nodes[node_name]
+                    node.set_properties(labels=sliver.labels,
+                                        label_allocations=sliver.label_allocations,
+                                        capacity_allocations=sliver.capacity_allocations,
+                                        reservation_info=sliver.reservation_info,
+                                        node_map=sliver.node_map,
+                                        management_ip=sliver.management_ip,
+                                        capacity_hints=sliver.capacity_hints,
+                                        capacities=sliver.capacities)
 
     def has_sliver_updates_at_authority(self):
         return len(self.computed_reservations) or len(self.computed_remove_reservations) or \
                len(self.computed_modify_reservations) or len(self.computed_modify_properties_reservations)
+
+    def has_topology_diffs(self, *, topology_diff: TopologyDiff) -> bool:
+        """
+        Check if there any updates in topology
+        :param topology_diff: topology difference object
+        """
+        ret_val = False
+        if not topology_diff:
+            ret_val = False
+
+        if len(topology_diff.added.nodes) or len(topology_diff.added.components) or \
+                len(topology_diff.added.services) or len(topology_diff.added.interfaces):
+            ret_val = True
+
+        if len(topology_diff.removed.nodes) or len(topology_diff.removed.components) or \
+                len(topology_diff.removed.services) or len(topology_diff.removed.interfaces):
+            ret_val = True
+
+        if len(topology_diff.modified.nodes) or len(topology_diff.modified.components) or \
+                len(topology_diff.modified.services) or len(topology_diff.modified.interfaces):
+            ret_val = True
+
+        self.logger.debug(f"Topology diff found: {ret_val}")
+        return ret_val

tools/db_cli.py

@@ -46,15 +46,15 @@ class MainClass:
     GlobalsSingleton.get().load_config()
     GlobalsSingleton.get().initialized = True
 
-    def __init__(self, user: str, password: str, db: str, host: str = '127.0.0.1:5432'):
+    def __init__(self, user: str, password: str, db: str, host: str = 'orchestrator-db:5432'):
         self.logger = logging.getLogger("db-cli")
         file_handler = RotatingFileHandler('./db_cli.log', backupCount=5, maxBytes=50000)
         logging.basicConfig(level=logging.DEBUG,
                             format="%(asctime)s [%(filename)s:%(lineno)d] [%(levelname)s] %(message)s",
                             handlers=[logging.StreamHandler(), file_handler])
 
         self.db = ActorDatabase(user=user, password=password, database=db, db_host=host, logger=self.logger)
-        self.neo4j_config = {"url": "neo4j://0.0.0.0:9687",
+        self.neo4j_config = {"url": "neo4j://orchestrator-neo4j:9687",
                              "user": "neo4j",
                              "pass": "password",
                              "import_host_dir": "/Users/kthare10/renci/code/fabric/ControlFramework/neo4j1/imports/",
@@ -63,10 +63,10 @@ def __init__(self, user: str, password: str, db: str, host: str = '127.0.0.1:543
     def get_slices(self, email: str = None, slice_id: str = None, slice_name: str = None):
         try:
             if slice_id is not None:
-                slice_obj = self.db.get_slice(slice_id=ID(uid=slice_id))
+                slice_obj = self.db.get_slices(slice_id=ID(uid=slice_id))
                 slice_list = [slice_obj]
             elif email is not None:
-                slice_list = self.db.get_slice_by_email(email=email)
+                slice_list = self.db.get_slices(email=email)
             else:
                 slice_list = self.db.get_slices()