Merge pull request #429 from fabric-testbed/slice-status

Fix stuck slice status and reservation state handling

Author: kthare10

fabric_cf/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "1.9.2"
+__version__ = "1.10.0"
 __VERSION__ = __version__

fabric_cf/actor/core/kernel/reservation.py

@@ -144,6 +144,7 @@ def __init__(self, *, rid: ID = None, resources: ResourceSet = None, term: Term
         # Scratch element to trigger post-actions on a probe.
         self.service_pending = ReservationPendingStates.None_
         self.last_transition_time = None
+        self.closed_at = None
         self.last_pending_state = ReservationPendingStates.None_
         self.thread_lock = threading.Lock()
 
@@ -621,6 +622,10 @@ def transition(self, *, prefix: str, state: ReservationStates, pending: Reservat
         self.state_transition = True
         self.last_transition_time = datetime.now(timezone.utc)
 
+        if state in (ReservationStates.Closed, ReservationStates.Failed, ReservationStates.CloseFail):
+            if self.closed_at is None:
+                self.closed_at = datetime.now(timezone.utc)
+
         if change:
             sliver = None
             if self.get_resources() is not None:

fabric_cf/actor/core/kernel/reservation_client.py

@@ -553,7 +553,7 @@ def prepare_ticket(self, extend: bool = False):
 
             self.logger.trace(f"Updated Network Res# {self.get_reservation_id()} {sliver}")
 
-    def approve_extend_ticket(self) -> Tuple[bool, bool]:
+    def approve_extend_ticket(self) -> Tuple[bool, bool, List[str]]:
         """
         ExtendTicket predicate: invoked internally to determine if the reservation
         should be extended. This gives subclasses an opportunity sequence actions at the orchestrator side.
@@ -601,7 +601,7 @@ def approve_extend_ticket(self) -> Tuple[bool, bool]:
                 rollback = True
                 break
 
-        return approved, rollback
+        return approved, rollback, failed_preds
 
     def approve_ticket(self, extend: bool = False) -> Tuple[bool, List[str]]:
         """
@@ -649,7 +649,7 @@ def can_extend(self) -> bool:
         if self.get_type() is not None:
             resource_type_str = str(self.get_type())
             if resource_type_str in Constants.SUPPORTED_SERVICES_STR:
-                ret_val, rollback = self.approve_extend_ticket()
+                ret_val, rollback, _ = self.approve_extend_ticket()
             else:
                 ret_val = True
 
@@ -1058,7 +1058,7 @@ def probe_join_state(self):
             if self.requested_resources.sliver is not None:
                 status, failed_preds = self.approve_ticket(extend=True)
             else:
-                status, rollback = self.approve_extend_ticket()
+                status, rollback, failed_preds = self.approve_extend_ticket()
 
             if status:
                 if not rollback:
@@ -1077,20 +1077,27 @@ def probe_join_state(self):
                 # Update ASM with Reservation Info
                 self.update_slice_graph(sliver=self.resources.sliver)
             else:
-                # Modify scenario; interfaces to the newly added VMs cannot be attached
-                # as the VM failed to ticket at the broker
                 if len(failed_preds) > 0:
-                    msg = f"ignore modify, redeem predecessor reservation# {failed_preds[0]} is in a terminal state"
-                    self.transition_with_join(prefix=msg,
-                                              state=self.state, pending=ReservationPendingStates.None_,
-                                              join_state=JoinState.NoJoin)
-
-                    for rid in failed_preds:
-                        self.remove_redeem_predecessor(rid=ID(uid=rid))
-
-                    # Update ASM with Reservation Info
-                    self.update_slice_graph(sliver=self.resources.sliver)
-                    self.pending_recover = False
+                    if self.requested_resources.sliver is not None:
+                        # Modify scenario; interfaces to the newly added VMs cannot be attached
+                        # as the VM failed to ticket at the broker
+                        msg = f"ignore modify, redeem predecessor reservation# {failed_preds[0]} is in a terminal state"
+                        self.transition_with_join(prefix=msg,
+                                                  state=self.state, pending=ReservationPendingStates.None_,
+                                                  join_state=JoinState.NoJoin)
+
+                        for rid in failed_preds:
+                            self.remove_redeem_predecessor(rid=ID(uid=rid))
+
+                        # Update ASM with Reservation Info
+                        self.update_slice_graph(sliver=self.resources.sliver)
+                        self.pending_recover = False
+                    else:
+                        # Extend/renew scenario — close the reservation since predecessors are dead
+                        msg = f"Closing reservation# {self.get_reservation_id()} - " \
+                              f"redeem predecessor(s) {failed_preds} in terminal state during extend"
+                        self.logger.error(msg)
+                        self.fail(message=msg)
 
         elif self.joinstate == JoinState.BlockedRedeem:
             # this reservation has a ticket to redeem, and the redeem is

fabric_cf/actor/core/kernel/slice_state_machine.py

@@ -105,9 +105,9 @@ def translate(state_name: str):
         elif state_name.lower() == SliceState.Dead.name.lower():
             return SliceState.Dead
         elif state_name.lower() == SliceState.AllocatedOK.name.lower():
-            return SliceState.Closing
+            return SliceState.AllocatedOK
         elif state_name.lower() == SliceState.AllocatedError.name.lower():
-            return SliceState.Dead
+            return SliceState.AllocatedError
         else:
             return SliceState.All
 
@@ -296,26 +296,28 @@ def transition_slice(self, *, operation: SliceOperation, reservations: Reservati
 
             if self.state in [SliceState.Nascent, SliceState.Configuring]:
                 if not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Closed,
-                                                 ReservationStates.CloseFail):
+                                                 ReservationStates.CloseWait, ReservationStates.CloseFail):
                     if not has_error:
                         self.state = SliceState.StableOK
                     else:
                         self.state = SliceState.StableError
 
                 if (not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Failed,
-                                                  ReservationStates.Closed, ReservationStates.CloseFail)) and \
+                                                  ReservationStates.Closed, ReservationStates.CloseWait,
+                                                  ReservationStates.CloseFail)) and \
                         bins.has_state(s=ReservationStates.Failed):
                     self.state = SliceState.StableError
 
                 if not bins.has_state_other_than(ReservationStates.Ticketed, ReservationStates.Closed,
-                                                 ReservationStates.CloseFail):
+                                                 ReservationStates.CloseWait, ReservationStates.CloseFail):
                     if not has_error:
                         self.state = SliceState.AllocatedOK
                     else:
                         self.state = SliceState.AllocatedError
 
                 if (not bins.has_state_other_than(ReservationStates.Ticketed, ReservationStates.Failed,
-                                                  ReservationStates.Closed, ReservationStates.CloseFail)) and \
+                                                  ReservationStates.Closed, ReservationStates.CloseWait,
+                                                  ReservationStates.CloseFail)) and \
                         bins.has_state(s=ReservationStates.Failed):
                     self.state = SliceState.AllocatedError
 
@@ -325,14 +327,15 @@ def transition_slice(self, *, operation: SliceOperation, reservations: Reservati
 
             elif self.state in [SliceState.AllocatedOK, SliceState.AllocatedError]:
                 if not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Closed,
-                                                 ReservationStates.CloseFail):
+                                                 ReservationStates.CloseWait, ReservationStates.CloseFail):
                     if not has_error:
                         self.state = SliceState.StableOK
                     else:
                         self.state = SliceState.StableError
 
                 if (not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Failed,
-                                                  ReservationStates.Closed, ReservationStates.CloseFail)) and \
+                                                  ReservationStates.Closed, ReservationStates.CloseWait,
+                                                  ReservationStates.CloseFail)) and \
                         bins.has_state(s=ReservationStates.Failed):
                     self.state = SliceState.StableError
 
@@ -342,26 +345,28 @@ def transition_slice(self, *, operation: SliceOperation, reservations: Reservati
 
             elif self.state == SliceState.Modifying:
                 if not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Closed,
-                                                 ReservationStates.CloseFail):
+                                                 ReservationStates.CloseWait, ReservationStates.CloseFail):
                     if has_error:
                         self.state = SliceState.ModifyError
                     else:
                         self.state = SliceState.ModifyOK
 
                 if (not bins.has_state_other_than(ReservationStates.Active, ReservationStates.Failed,
-                                                  ReservationStates.Closed, ReservationStates.CloseFail)) and \
+                                                  ReservationStates.Closed, ReservationStates.CloseWait,
+                                                  ReservationStates.CloseFail)) and \
                         bins.has_state(s=ReservationStates.Failed):
                     self.state = SliceState.ModifyError
 
                 if not bins.has_state_other_than(ReservationStates.Ticketed, ReservationStates.Closed,
-                                                 ReservationStates.CloseFail):
+                                                 ReservationStates.CloseWait, ReservationStates.CloseFail):
                     if has_error:
                         self.state = SliceState.ModifyError
                     else:
                         self.state = SliceState.ModifyOK
 
                 if (not bins.has_state_other_than(ReservationStates.Ticketed, ReservationStates.Failed,
-                                                  ReservationStates.Closed, ReservationStates.CloseFail)) and \
+                                                  ReservationStates.Closed, ReservationStates.CloseWait,
+                                                  ReservationStates.CloseFail)) and \
                         bins.has_state(s=ReservationStates.Failed):
                     self.state = SliceState.ModifyError
 

fabric_cf/actor/core/manage/converter.py

@@ -126,6 +126,10 @@ def fill_reservation(*, reservation: ABCReservationMixin, full: bool) -> Reserva
 
         rsv_mng.set_notices(reservation.get_notices())
 
+        closed_at = getattr(reservation, 'closed_at', None)
+        if closed_at is not None:
+            rsv_mng.set_closed_at(ActorClock.to_milliseconds(when=closed_at))
+
         if full:
             rsv_mng = Converter.attach_res_properties(mng=rsv_mng, reservation=reservation)
 

fabric_cf/actor/core/plugins/db/actor_database.py

@@ -368,7 +368,8 @@ def add_reservation(self, *, reservation: ABCReservationMixin):
                                     components=components,
                                     lease_start=term.get_start_time() if term else None,
                                     lease_end=term.get_end_time() if term else None,
-                                    host=host, ip_subnet=ip_subnet, links=links)
+                                    host=host, ip_subnet=ip_subnet, links=links,
+                                    closed_at=getattr(reservation, 'closed_at', None))
             self.logger.debug(
                 "Reservation {} added to slice {}".format(reservation.get_reservation_id(), reservation.get_slice()))
         finally:
@@ -470,7 +471,8 @@ def update_reservation(self, *, reservation: ABCReservationMixin):
                                        site=site, rsv_type=rsv_type, components=components,
                                        lease_start=term.get_start_time() if term else None,
                                        lease_end=term.get_end_time() if term else None,
-                                       ip_subnet=ip_subnet, host=host, links=links)
+                                       ip_subnet=ip_subnet, host=host, links=links,
+                                       closed_at=getattr(reservation, 'closed_at', None))
             diff = int(time.time() - begin)
             if diff > 0:
                 self.logger.info(f"DB TIME: {diff}")

fabric_cf/actor/db/__init__.py

@@ -137,6 +137,7 @@ class Reservations(Base):
     rsv_joining = Column(Integer, nullable=False)
     lease_start = Column(TIMESTAMP(timezone=True), nullable=True)
     lease_end = Column(TIMESTAMP(timezone=True), nullable=True)
+    closed_at = Column(TIMESTAMP(timezone=True), nullable=True)
     properties = Column(LargeBinary)
     components = relationship('Components', back_populates='reservation')
     links = relationship('Links', back_populates='reservation')

fabric_cf/actor/db/psql_database.py

@@ -662,7 +662,7 @@ def add_reservation(self, *, slc_guid: str, rsv_resid: str, rsv_category: int, r
                         lease_end: datetime = None, rsv_graph_node_id: str = None, oidc_claim_sub: str = None,
                         email: str = None, project_id: str = None, site: str = None, rsv_type: str = None,
                         components: List[Tuple[str, str, str]] = None, host: str = None, ip_subnet: str = None,
-                        links: list[dict] = None):
+                        links: list[dict] = None, closed_at: datetime = None):
         """
         Add a reservation
         @param slc_guid slice guid
@@ -684,6 +684,7 @@ def add_reservation(self, *, slc_guid: str, rsv_resid: str, rsv_category: int, r
         @param host host
         @param ip_subnet ip_subnet
         @param links: list of dictionary objects representing link
+        @param closed_at timestamp when reservation was closed
         """
         session = self.get_session()
         try:
@@ -692,7 +693,8 @@ def add_reservation(self, *, slc_guid: str, rsv_resid: str, rsv_category: int, r
                                    rsv_state=rsv_state, rsv_pending=rsv_pending, rsv_joining=rsv_joining,
                                    lease_start=lease_start, lease_end=lease_end,
                                    properties=properties, oidc_claim_sub=oidc_claim_sub, email=email,
-                                   project_id=project_id, site=site, rsv_type=rsv_type, host=host, ip_subnet=ip_subnet)
+                                   project_id=project_id, site=site, rsv_type=rsv_type, host=host, ip_subnet=ip_subnet,
+                                   closed_at=closed_at)
             if rsv_graph_node_id is not None:
                 rsv_obj.rsv_graph_node_id = rsv_graph_node_id
 
@@ -761,7 +763,8 @@ def update_reservation(self, *, slc_guid: str, rsv_resid: str, rsv_category: int
                            rsv_pending: int, rsv_joining: int, properties, lease_start: datetime = None,
                            lease_end: datetime = None, rsv_graph_node_id: str = None, site: str = None,
                            rsv_type: str = None, components: List[Tuple[str, str, str]] = None,
-                           host: str = None, ip_subnet: str = None, links: list[dict] = None):
+                           host: str = None, ip_subnet: str = None, links: list[dict] = None,
+                           closed_at: datetime = None):
         session = self.get_session()
         try:
             rsv_obj = session.query(Reservations).filter_by(rsv_resid=rsv_resid).one()
@@ -776,6 +779,7 @@ def update_reservation(self, *, slc_guid: str, rsv_resid: str, rsv_category: int
             rsv_obj.properties = properties
             rsv_obj.lease_start = lease_start
             rsv_obj.lease_end = lease_end
+            rsv_obj.closed_at = closed_at
             if host:
                 rsv_obj.host = host
             if ip_subnet:

fabric_cf/actor/fim/plugins/broker/aggregate_bqm_plugin.py

@@ -605,7 +605,7 @@ def plug_produce_bqm(self, *, cbm: ABCCBMPropertyGraph, **kwargs) -> ABCBQMPrope
                                     }
 
                     if not self.DEBUG_FLAG and kwargs['query_level'] != 0:
-                        allocated_vlans = self.occupied_vlans(db=db, node_id=fac_sliver.resource_name,
+                        allocated_vlans = self.occupied_vlans(db=db, node_id=fac_sliver.node_id,
                                                               component_name=fac_cp_node_id, start=start, end=end)
 
                         if allocated_vlans and len(allocated_vlans):
@@ -1025,7 +1025,7 @@ def _extract_port(cp_props):
                     allocated_vlans = None
                     if not self.DEBUG_FLAG and query_level != 0:
                         alloc_vlan_list = self.occupied_vlans(
-                            db=db, node_id=fac_sliver.resource_name,
+                            db=db, node_id=fac_sliver.node_id,
                             component_name=fac_cp_node_id, start=start, end=end
                         )
                         if alloc_vlan_list:

fabric_cf/orchestrator/core/response_builder.py

@@ -50,6 +50,7 @@ class ResponseBuilder:
     PROP_GRAPH_ID = "graph_id"
     PROP_LEASE_START_TIME = "lease_start_time"
     PROP_LEASE_END_TIME = "lease_end_time"
+    PROP_CLOSED_AT = "closed_at"
 
     PROP_SLIVER_ID = "sliver_id"
     PROP_PENDING_STATE = "pending_state"
@@ -101,6 +102,11 @@ def get_reservation_summary(*, res_list: List[ReservationMng]) -> List[dict]:
                 if reservation.get_notices() is not None:
                     res_dict[ResponseBuilder.PROP_NOTICE] = reservation.get_notices()
 
+                closed_at_ms = reservation.get_closed_at()
+                if closed_at_ms is not None:
+                    closed_at_time = ActorClock.from_milliseconds(milli_seconds=closed_at_ms)
+                    res_dict[ResponseBuilder.PROP_CLOSED_AT] = closed_at_time.strftime(Constants.LEASE_TIME_FORMAT)
+
                 reservations.append(res_dict)
 
         return reservations