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 \