Merge pull request #361 from fabric-testbed/leaks

Network Service provisioning failures - interface already in use due to allocation algorithm deficiencies

Author: kthare10

Dockerfile-auth

@@ -1,7 +1,7 @@
 FROM python:3.11.0
 MAINTAINER Komal Thareja<[email protected]>
 
-ARG HANDLERS_VER=1.6.1
+ARG HANDLERS_VER=1.6.2
 
 RUN mkdir -p /usr/src/app
 WORKDIR /usr/src/app
@@ -25,7 +25,6 @@ RUN mkdir -p "/etc/fabric/actor/config"
 RUN mkdir -p "/var/log/actor"
 RUN cp /usr/local/lib/python3.11/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema
 RUN pip3 install fabric-am-handlers==${HANDLERS_VER}
-RUN pip3 install fabric-message-bus==1.6.1
 RUN sh /usr/src/app/install.sh
 
 ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]

fabric_cf/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "1.6.2"
+__version__ = "1.6.1"
 __VERSION__ = __version__

fabric_cf/actor/boot/configuration.py

@@ -460,7 +460,7 @@ def get_kafka_key_schema_location(self) -> str or None:
         return self.global_config.runtime.get(Constants.PROPERTY_CONF_KAFKA_KEY_SCHEMA, None)
 
     def get_kafka_consumer_auto_commit_interval(self) -> int:
-        return int(self.global_config.runtime.get(Constants.PROPERTY_CONF_KAFKA_AUTO_COMMIT_INTERVAL), 5)
+        return int(self.global_config.runtime.get(Constants.PROPERTY_CONF_KAFKA_AUTO_COMMIT_INTERVAL, 5))
 
     def get_kafka_consumer_commit_batch_size(self) -> int:
         return int(self.global_config.runtime.get(Constants.PROPERTY_CONF_KAFKA_BATCH_SIZE, 1))

fabric_cf/actor/core/apis/abc_actor_mixin.py

@@ -554,14 +554,15 @@ def fail_delegation(self, *, did: str, message: str):
         """
 
     @abstractmethod
-    def close_by_rid(self, *, rid: ID):
+    def close_by_rid(self, *, rid: ID, force: bool = False):
         """
         Closes the reservation. Note: the reservation must have already been registered with the actor.
         This method may involve either a client or a server side action or both. When called on a broker,
         this method will only close the broker reservation.
 
         Args:
             rid: reservation id
+            force: force close
         Raises:
             Exception in case of error
         """

fabric_cf/actor/core/apis/abc_database.py

@@ -27,7 +27,7 @@
 
 from abc import abstractmethod, ABC
 from datetime import datetime
-from typing import TYPE_CHECKING, List, Union
+from typing import TYPE_CHECKING, List, Union, Tuple, Dict
 
 from fabric_cf.actor.core.apis.abc_delegation import ABCDelegation
 from fabric_cf.actor.core.kernel.slice import SliceTypes
@@ -170,6 +170,17 @@ def get_reservations(self, *, slice_id: ID = None, graph_node_id: str = None, pr
         @throws Exception in case of error
         """
 
+    @abstractmethod
+    def get_components(self, *, node_id: str, states: list[int], rsv_type: list[str], component: str = None,
+                       bdf: str = None) -> Dict[str, List[str]]:
+        """
+        Retrieves the components.
+
+        @return list of components
+
+        @throws Exception in case of error
+        """
+
     @abstractmethod
     def get_client_reservations(self, *, slice_id: ID = None) -> List[ABCReservationMixin]:
         """

fabric_cf/actor/core/apis/abc_reservation_mixin.py

@@ -282,7 +282,7 @@ def can_renew(self) -> bool:
         """
 
     @abstractmethod
-    def close(self):
+    def close(self, force: bool = False):
         """
         Closes the reservation. Locked with the kernel lock.
         """
@@ -359,7 +359,7 @@ def probe_pending(self):
         """
 
     @abstractmethod
-    def reserve(self, *, policy: ABCPolicy):
+    def reserve(self, *, policy: ABCPolicy) -> bool:
         """
         Reserve resources: ticket() initiate or request, or redeem()
         request. New reservation.

fabric_cf/actor/core/core/actor.py

@@ -154,8 +154,8 @@ def fail_delegation(self, *, did: str, message: str):
     def close_delegation(self, *, did: str):
         self.wrapper.close_delegation(did=did)
 
-    def close_by_rid(self, *, rid: ID):
-        self.wrapper.close(rid=rid)
+    def close_by_rid(self, *, rid: ID, force: bool = False):
+        self.wrapper.close(rid=rid, force=force)
 
     def close(self, *, reservation: ABCReservationMixin):
         if reservation is not None:

fabric_cf/actor/core/delegation/broker_delegation.py

@@ -218,6 +218,8 @@ def accept_delegation_update(self, *, incoming: ABCDelegation, update_data: Upda
                 self.delegation_update_satisfies(incoming=incoming, update_data=update_data)
                 self.absorb_delegation_update(incoming=incoming, update_data=update_data)
             except Exception as e:
+                if incoming.get_graph() is not None:
+                    incoming.get_graph().delete_graph()
                 success = False
                 update_data.error(message=str(e))
                 self.logger.error(traceback.format_exc())

fabric_cf/actor/core/kernel/authority_reservation.py

@@ -139,17 +139,20 @@ def prepare(self, *, callback: ABCCallbackProxy, logger):
 
         self.state = ReservationStates.Ticketed
 
-    def reserve(self, *, policy: ABCPolicy):
+    def reserve(self, *, policy: ABCPolicy) -> bool:
         self.nothing_pending()
         self.incoming_request()
         if self.is_active():
-            self.error(err="reservation already holds a lease")
+            #self.error(err="reservation already holds a lease")
+            self.logger.warning(f"Reservation: {self.get_reservation_id()} already holds a lease")
+            return False
 
         self.policy = policy
         self.approved = False
         self.bid_pending = True
         self.pending_recover = False
         self.map_and_update(extend=False)
+        return True
 
     def service_reserve(self):
         try:
@@ -210,7 +213,7 @@ def service_modify_lease(self):
             self.logger.error("authority failed servicing modifylease e: {}".format(e))
             self.fail_notify(message=str(e))
 
-    def close(self):
+    def close(self, force: bool = False):
         self.logger.debug("Processing  close for #{}".format(self.rid))
         self.transition(prefix="external close", state=self.state, pending=ReservationPendingStates.Closing)
 
@@ -418,8 +421,12 @@ def probe_pending(self):
 
         elif self.pending_state == ReservationPendingStates.Closing:
             if self.resources is None or self.resources.is_closed():
-                self.transition(prefix="close complete", state=ReservationStates.Closed,
-                                pending=ReservationPendingStates.None_)
+                if self.update_data.failed:
+                    self.transition(prefix="close complete failed", state=ReservationStates.CloseFail,
+                                    pending=ReservationPendingStates.None_)
+                else:
+                    self.transition(prefix="close complete", state=ReservationStates.Closed,
+                                    pending=ReservationPendingStates.None_)
                 self.pending_recover = False
                 self.generate_update()
 
@@ -515,7 +522,10 @@ def reap(self):
                 released = self.resources.collect_released()
                 if released is not None:
                     if not released.get_notices().is_empty():
-                        self.update_data.post(event=released.get_notices().get_notice())
+                        notice = released.get_notices().get_notice()
+                        self.update_data.post(event=notice)
+                        if "Exception" in notice:
+                            self.update_data.error(message=notice)
                     self.policy.release(resources=released)
         except Exception as e:
             self.logger.error("exception in authority reap e: {}".format(e))

fabric_cf/actor/core/kernel/broker_reservation.py

@@ -221,7 +221,7 @@ def prepare(self, *, callback: ABCCallbackProxy, logger):
 
         self.set_dirty()
 
-    def reserve(self, *, policy: ABCPolicy):
+    def reserve(self, *, policy: ABCPolicy) -> bool:
         # These handlers may need to be slightly more sophisticated, since a
         # client may bid multiple times on a ticket as part of an auction
         # protocol: so we may receive a reserve or extend when there is already
@@ -232,12 +232,13 @@ def reserve(self, *, policy: ABCPolicy):
                 self.pending_state != ReservationPendingStates.Ticketing:
             # We do not want to fail the reservation simply log a warning and exit from reserve
             self.logger.warning("Duplicate ticket request")
-            return
+            return False
 
         self.policy = policy
         self.approved = False
         self.bid_pending = True
         self.map_and_update(ticketed=False)
+        return True
 
     def service_reserve(self):
         # resources is null initially. It becomes non-null once the
@@ -276,13 +277,13 @@ def service_extend_ticket(self):
                                 pending=ReservationPendingStates.None_)
                 self.generate_update()
 
-    def close(self):
+    def close(self, force: bool = False):
         send_notification = False
         if self.state == ReservationStates.Nascent or self.pending_state != ReservationPendingStates.None_:
             self.logger.warning("Closing a reservation in progress")
             send_notification = True
 
-        if self.state != ReservationStates.Closed:
+        if self.state not in [ReservationStates.Closed, ReservationStates.CloseFail] or force:
             if self.pending_state == ReservationPendingStates.Priming or \
                     (self.pending_state == ReservationPendingStates.Ticketing and not self.bid_pending):
                 # Close in Priming is a special case: when processing the close
@@ -296,7 +297,11 @@ def close(self):
                 self.logger.debug("closing reservation #{} while in Priming".format(self.rid))
                 self.closed_in_priming = True
 
-            self.transition(prefix="closed", state=ReservationStates.Closed, pending=ReservationPendingStates.None_)
+            if not self.update_data.is_failed():
+                self.transition(prefix="closed", state=ReservationStates.Closed, pending=ReservationPendingStates.None_)
+            else:
+                self.transition(prefix="closed-failed", state=ReservationStates.CloseFail,
+                                pending=ReservationPendingStates.None_)
             self.policy.closed(reservation=self)
 
         if send_notification:
@@ -536,7 +541,16 @@ def set_authority(self, *, authority: ABCAuthorityProxy):
     def update_lease(self, *, incoming: ABCReservationMixin, update_data):
         self.logger.info(f"Received Update Lease: {incoming} at Broker")
         # TODO add any processing if needed
-        self.logger.info(f"Do Nothing!")
+        if incoming.get_resources() and incoming.get_resources().get_sliver() and incoming.get_resources().get_sliver().get_reservation_info():
+            incoming_state = incoming.get_resources().get_sliver().get_reservation_info().reservation_state
+        else:
+            incoming_state = None
+        self.logger.info(f"Update Lease from authority in state: {self.get_state()} "
+                         f"Incoming: {incoming_state}|{incoming.get_notices()}  update_data: {update_data}!")
+        if incoming_state and incoming_state == str(ReservationStates.CloseFail):
+            self.update_data.absorb(other=update_data)
+            self.logger.info("Closing a reservation which failed to delete at the authority")
+            self.close()
 
     def handle_failed_rpc(self, *, failed: FailedRPC):
         if failed.get_request_type() == RPCRequestType.UpdateTicket and \

fabric_cf/actor/core/kernel/kernel.py

@@ -85,9 +85,9 @@ def amend_reserve(self, *, reservation: ABCReservationMixin):
         """
         try:
             reservation.lock()
-            reservation.reserve(policy=self.policy)
+            ignore = reservation.reserve(policy=self.policy)
             self.plugin.get_database().update_reservation(reservation=reservation)
-            if not reservation.is_failed():
+            if not reservation.is_failed() and not ignore:
                 reservation.service_reserve()
         except Exception as e:
             self.logger.error(traceback.format_exc())
@@ -201,21 +201,22 @@ def close_delegation(self, *, delegation: ABCDelegation):
         finally:
             delegation.unlock()
 
-    def close(self, *, reservation: ABCReservationMixin):
+    def close(self, *, reservation: ABCReservationMixin, force: bool = False):
         """
         Handles a close operation for the reservation.
         Client: perform local close operations and issue close request to
         authority.
         Broker: perform local close operations
         Authority: process a close request
         @param reservation reservation for which to perform close
+        @param force force close
         @throws Exception
         """
         try:
             reservation.lock()
             if not reservation.is_closed() and not reservation.is_closing():
                 self.policy.close(reservation=reservation)
-                reservation.close()
+                reservation.close(force=force)
                 self.plugin.get_database().update_reservation(reservation=reservation)
                 reservation.service_close()
         except Exception as e:
@@ -932,7 +933,13 @@ def register_slice(self, *, slice_object: ABCSlice):
             self.slices.add(slice_object=slice_object)
 
             try:
-                self.plugin.get_database().add_slice(slice_object=slice_object)
+                # Only add slice if it doesn't exist
+                # Slice may exist in cases where it only one sliver which was removed by a modify
+                exists = self.plugin.get_database().get_slices(slice_id=slice_object.get_slice_id())
+                if not len(exists):
+                    self.plugin.get_database().add_slice(slice_object=slice_object)
+                else:
+                    self.plugin.get_database().update_slice(slice_object=slice_object)
             except Exception as e:
                 self.slices.remove(slice_id=slice_object.get_slice_id())
                 self.logger.error(traceback.format_exc())
@@ -989,7 +996,8 @@ def remove_reservation(self, *, rid: ID):
         if real is not None:
             try:
                 real.lock()
-                if real.is_closed() or real.is_failed() or real.get_state() == ReservationStates.CloseWait:
+                if real.is_closed() or real.is_failed() or \
+                        real.get_state() in [ReservationStates.CloseWait, ReservationStates.CloseFail]:
                     self.unregister_reservation(rid=rid)
                 else:
                     raise KernelException("Only reservations in failed, closed, or closewait state can be removed.")
@@ -1312,7 +1320,8 @@ def __unregister(self, *, reservation: ABCReservationMixin, slice_object: ABCSli
         @param slice_object local slice object
         @throws Exception
         """
-        if reservation.is_closed() or reservation.is_failed() or reservation.get_state() == ReservationStates.CloseWait:
+        if reservation.is_closed() or reservation.is_failed() or \
+                reservation.get_state() in [ReservationStates.CloseWait, ReservationStates.CloseFail]:
             try:
                 slice_object.lock_slice()
                 slice_object.unregister(reservation=reservation)
@@ -1521,6 +1530,8 @@ def validate_delegation(self, *, delegation: ABCDelegation = None, did: str = No
         if delegation is not None:
             local = self.soft_validate_delegation(delegation=delegation)
             if local is None:
+                if delegation.get_graph() is not None:
+                    delegation.get_graph().delete_graph()
                 self.error(err="delegation not found", e=DelegationNotFoundException(did=did))
             return local
 

fabric_cf/actor/core/kernel/kernel_wrapper.py

@@ -152,21 +152,22 @@ def fail_delegation(self, *, did: str, message: str):
         target = self.kernel.validate_delegation(did=did)
         self.kernel.fail_delegation(delegation=target, message=message)
 
-    def close(self, *, rid: ID):
+    def close(self, *, rid: ID, force: bool = False):
         """
         Closes the reservation, potentially initiating a close request to another
         actor. If the reservation has concrete resources bound to it, this method
         may return before all close operations have completed. Check the
         reservation state to determine when close completes.
         @param rid identifier of reservation to close
+        @param force forced close
         @throws Exception in case of error
         """
         if rid is None:
             raise KernelException(Constants.INVALID_ARGUMENT)
         target = self.kernel.validate(rid=rid)
         # NOTE: this call does not require access control check, since
         # it is executed in the context of the actor represented by KernelWrapper.
-        self.kernel.close(reservation=target)
+        self.kernel.close(reservation=target, force=force)
 
     def close_delegation(self, *, did: str):
         """

fabric_cf/actor/core/kernel/reservation.py

@@ -237,7 +237,7 @@ def clear_notice(self, clear_fail: bool=False):
         Clears all event notices associated with the reservation.
         """
 
-    def close(self):
+    def close(self, force: bool = False):
         """
         Close a reservation
         """
@@ -416,7 +416,7 @@ def is_bid_pending(self) -> bool:
         return self.bid_pending
 
     def is_closed(self) -> bool:
-        return self.state == ReservationStates.Closed
+        return self.state == ReservationStates.Closed or self.state == ReservationStates.CloseFail
 
     def is_closing(self) -> bool:
         return self.state == ReservationStates.CloseWait or self.pending_state == ReservationPendingStates.Closing
@@ -495,11 +495,11 @@ def ready(self):
 
         @throws Exception thrown if the state is closed or failed
         """
-        if self.state == ReservationStates.Closed or self.state == ReservationStates.Failed:
+        if self.state in [ReservationStates.Closed, ReservationStates.Failed, ReservationStates.CloseFail]:
             self.error(err="invalid Reservation")
 
-    def reserve(self, *, policy: ABCPolicy):
-        return
+    def reserve(self, *, policy: ABCPolicy) -> bool:
+        return True
 
     def setup(self):
         """