Merge pull request #409 from fabric-testbed/404-allocation-control-resource-quotas-permissions

404 allocation control resource quotas permissions

Author: kthare10

fabric_cf/actor/boot/configuration.py

@@ -45,6 +45,10 @@ def __init__(self, *, config: dict):
         if Constants.CONFIG_SECTION_O_AUTH in config:
             self.oauth = config.get(Constants.CONFIG_SECTION_O_AUTH)
 
+        self.core_api = {}
+        if Constants.CONFIG_SECTION_CORE_API in config:
+            self.core_api = config.get(Constants.CONFIG_SECTION_CORE_API)
+
         self.smtp = {}
         if Constants.CONFIG_SECTION_SMTP in config:
             self.smtp = config.get(Constants.CONFIG_SECTION_SMTP)
@@ -91,6 +95,12 @@ def get_oauth(self) -> dict:
         """
         return self.oauth
 
+    def get_core_api(self) -> dict:
+        """
+        Return core api
+        """
+        return self.core_api
+
     def get_smtp(self) -> dict:
         """
         Return smtp config
@@ -425,15 +435,20 @@ def get_runtime_config(self) -> dict:
         """
         if self.global_config is not None:
             return self.global_config.get_runtime()
-        return None
 
     def get_oauth_config(self) -> dict:
         """
         Return OAuth Config
         """
         if self.global_config is not None:
             return self.global_config.get_oauth()
-        return None
+
+    def get_core_api_config(self) -> dict:
+        """
+        Return Core API Config
+        """
+        if self.global_config is not None:
+            return self.global_config.get_core_api()
 
     def get_smtp_config(self) -> dict:
         if self.global_config:

fabric_cf/actor/core/apis/abc_database.py

@@ -477,4 +477,4 @@ def get_poas(self, *, poa_id: str = None, email: str = None, sliver_id: ID = Non
         @param offset offset
         @param states states
         @param last_update_time last update time
-        """
+        """

fabric_cf/actor/core/common/constants.py

@@ -173,6 +173,9 @@ class Constants:
     PROPERTY_CONF_O_AUTH_TRL_REFRESH = "trl-refresh"
     PROPERTY_CONF_O_AUTH_VERIFY_EXP = "verify-exp"
 
+    CONFIG_SECTION_CORE_API = "core_api"
+    PROPERTY_CONF_HOST = "host"
+
     CONFIG_SECTION_SMTP = "smtp"
 
     CONFIG_SECTION_DATABASE = "database"
@@ -275,6 +278,7 @@ class Constants:
     CLAIMS_PROJECTS = "projects"
     UUID = "uuid"
     TAGS = "tags"
+    TOKEN = "token"
     TOKEN_HASH = "token_hash"
     PROJECT_ID = "project_id"
     USERS = "users"

fabric_cf/actor/core/container/globals.py

@@ -35,6 +35,7 @@
 import logging
 import os
 
+from fabric_cf.actor.core.util.quota_mgr import QuotaMgr
 from fim.graph.neo4j_property_graph import Neo4jGraphImporter
 from fim.graph.resources.abc_arm import ABCARMPropertyGraph
 from fss_utils.jwt_validate import JWTValidator
@@ -74,6 +75,7 @@ def __init__(self):
         self.lock = threading.Lock()
         self.jwt_validator = None
         self.token_validator = None
+        self.quota_mgr = None
 
     def make_logger(self):
         """
@@ -193,6 +195,12 @@ def load_validators(self):
                                               refresh_period=timedelta(hours=t.hour, minutes=t.minute, seconds=t.second),
                                               jwt_validator=self.jwt_validator)
 
+        core_api = self.config.get_core_api_config()
+        if core_api.get("enable", False):
+            self.quota_mgr = QuotaMgr(core_api_host=core_api.get(Constants.PROPERTY_CONF_HOST),
+                                      token=core_api.get(Constants.TOKEN, ""),
+                                      logger=self.log)
+
     def load_config(self):
         """
         Load the configuration
@@ -211,6 +219,9 @@ def get_jwt_validator(self) -> JWTValidator:
     def get_token_validator(self) -> TokenValidator:
         return self.token_validator
 
+    def get_quota_mgr(self) -> QuotaMgr:
+        return self.quota_mgr
+
     def get_container(self) -> ABCActorContainer:
         """
         Get the container

fabric_cf/actor/core/kernel/kernel.py

@@ -26,6 +26,7 @@
 import threading
 import time
 import traceback
+from datetime import datetime, timezone
 
 from typing import List, Dict
 
@@ -34,7 +35,7 @@
 from fabric_cf.actor.core.apis.abc_delegation import ABCDelegation
 from fabric_cf.actor.core.apis.abc_policy import ABCPolicy
 from fabric_cf.actor.core.common.constants import Constants
-from fabric_cf.actor.core.common.event_logger import EventLogger, EventLoggerSingleton
+from fabric_cf.actor.core.common.event_logger import EventLoggerSingleton
 from fabric_cf.actor.core.common.exceptions import ReservationNotFoundException, DelegationNotFoundException, \
     KernelException
 from fabric_cf.actor.core.kernel.authority_reservation import AuthorityReservation

fabric_cf/actor/core/kernel/reservation_client.py

@@ -1557,16 +1557,16 @@ def validate_redeem(self):
 
     def add_redeem_predecessor(self, *, reservation: ABCReservationMixin, filters: dict = None):
         if reservation.get_reservation_id() not in self.redeem_predecessors:
-            state = PredecessorState(reservation=reservation)
+            state = PredecessorState(reservation=reservation, filters=filters)
             self.redeem_predecessors[reservation.get_reservation_id()] = state
 
     def remove_redeem_predecessor(self, *, rid: ID):
         if rid in self.redeem_predecessors:
             self.redeem_predecessors.pop(rid)
 
-    def add_join_predecessor(self, *, predecessor):
+    def add_join_predecessor(self, *, predecessor: ABCReservationMixin, filters: dict = None):
         if predecessor.get_reservation_id() not in self.redeem_predecessors:
-            state = PredecessorState(reservation=predecessor)
+            state = PredecessorState(reservation=predecessor, filters=filters)
             self.join_predecessors[predecessor.get_reservation_id()] = state
 
     def get_redeem_predecessors(self) -> List[PredecessorState]:

fabric_cf/actor/core/manage/actor_management_object.py

@@ -29,6 +29,8 @@
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, List, Dict, Tuple
 
+from fabric_mb.message_bus.messages.lease_reservation_avro import LeaseReservationAvro
+
 from fabric_cf.actor.fim.fim_helper import FimHelper
 from fabric_mb.message_bus.messages.reservation_mng import ReservationMng
 from fabric_mb.message_bus.messages.result_delegation_avro import ResultDelegationAvro
@@ -902,4 +904,4 @@ def build_broker_query_model(self, level_0_broker_query_model: str, level: int,
                                                       end=end, includes=includes, excludes=excludes)
         except Exception as e:
             self.logger.error(f"Exception occurred build_broker_query_model e: {e}")
-            self.logger.error(traceback.format_exc())
+            self.logger.error(traceback.format_exc())

fabric_cf/actor/core/manage/client_actor_management_object_helper.py

@@ -43,8 +43,7 @@
 
 from fabric_cf.actor.core.apis.abc_actor_runnable import ABCActorRunnable
 from fabric_cf.actor.core.apis.abc_controller_reservation import ABCControllerReservation
-from fabric_cf.actor.core.apis.abc_reservation_mixin import ABCReservationMixin
-from fabric_cf.actor.core.common.constants import Constants, ErrorCodes
+from fabric_cf.actor.core.common.constants import ErrorCodes
 from fabric_cf.actor.core.common.exceptions import ManageException
 from fabric_cf.actor.core.kernel.reservation_client import ClientReservationFactory
 from fabric_cf.actor.core.kernel.reservation_states import ReservationStates, ReservationPendingStates
@@ -635,4 +634,4 @@ def run(self):
             result.set_message(ErrorCodes.ErrorInternalError.interpret(exception=e))
             result = ManagementObject.set_exception_details(result=result, e=e)
 
-        return result
+        return result

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

@@ -28,7 +28,8 @@
 import time
 import traceback
 from datetime import datetime
-from typing import List, Union, Tuple, Dict
+from typing import List, Union, Dict
+
 
 from fabric_cf.actor.core.apis.abc_actor_mixin import ABCActorMixin, ActorType
 from fabric_cf.actor.core.apis.abc_broker_proxy import ABCBrokerProxy

fabric_cf/actor/core/policy/broker_simpler_units_policy.py

@@ -1075,8 +1075,20 @@ def ticket_inventory(self, *, reservation: ABCBrokerReservation, inv: InventoryF
         try:
             if operation == ReservationOperation.Extend:
                 rset = reservation.get_resources()
+                duration = Term.delta(reservation.get_term().get_end_time(), term.get_end_time())
             else:
                 rset = reservation.get_requested_resources()
+                duration = term.get_full_length()
+
+            from fabric_cf.actor.core.container.globals import GlobalsSingleton
+            if GlobalsSingleton.get().get_quota_mgr():
+                status, error_msg = GlobalsSingleton.get().get_quota_mgr().enforce_quota_limits(reservation=reservation,
+                                                                                                duration=duration)
+                self.logger.info(f"Quota enforcement status: {status}, error: {error_msg}")
+                # TODO: enable enforcement action later
+                #if not status:
+                #    return status, node_id_to_reservations, error_msg
+
             needed = rset.get_units()
 
             # for network node slivers
@@ -1115,6 +1127,11 @@ def ticket_inventory(self, *, reservation: ABCBrokerReservation, inv: InventoryF
                 if node_id_to_reservations.get(node_id, None) is None:
                     node_id_to_reservations[node_id] = ReservationSet()
                 node_id_to_reservations[node_id].add(reservation=reservation)
+
+                from fabric_cf.actor.core.container.globals import GlobalsSingleton
+                if GlobalsSingleton.get().get_quota_mgr():
+                    GlobalsSingleton.get().get_quota_mgr().update_quota(reservation=reservation, duration=duration)
+
                 self.logger.debug(f"Ticket Inventory returning: True {error_msg}")
                 return True, node_id_to_reservations, error_msg
         except Exception as e:
@@ -1204,6 +1221,13 @@ def issue_ticket(self, *, reservation: ABCBrokerReservation, units: int, rtype:
         return reservation
 
     def release(self, *, reservation):
+        duration = reservation.get_term().get_remaining_length()
+        if duration > 0:
+            from fabric_cf.actor.core.container.globals import GlobalsSingleton
+            if GlobalsSingleton.get().get_quota_mgr():
+                GlobalsSingleton.get().get_quota_mgr().update_quota(reservation=reservation,
+                                                                    duration=duration)
+
         if isinstance(reservation, ABCBrokerReservation):
             self.logger.debug("Broker reservation")
             super().release(reservation=reservation)

fabric_cf/actor/core/policy/inventory_for_type.py

@@ -59,7 +59,7 @@ def free(self, *, count: int, request: dict = None, resource: dict = None) -> di
         """
 
     @staticmethod
-    def _get_allocated_sliver(reservation: ABCReservationMixin) -> BaseSliver:
+    def get_allocated_sliver(reservation: ABCReservationMixin) -> BaseSliver:
         """
         Retrieve the allocated sliver from the reservation.
 
@@ -70,3 +70,5 @@ def _get_allocated_sliver(reservation: ABCReservationMixin) -> BaseSliver:
             return reservation.get_approved_resources().get_sliver()
         if (reservation.is_active() or reservation.is_ticketed()) and reservation.get_resources() is not None:
             return reservation.get_resources().get_sliver()
+        if (reservation.is_closed()) and reservation.get_resources() is not None:
+            return reservation.get_resources().get_sliver()

fabric_cf/actor/core/policy/network_node_inventory.py

@@ -69,7 +69,7 @@ def check_capacities(*, rid: ID, requested_capacities: Capacities, delegated: De
                 if rid == reservation.get_reservation_id():
                     continue
                 # For Active or Ticketed or Ticketing reservations; reduce the counts from available
-                resource_sliver = InventoryForType._get_allocated_sliver(reservation=reservation)
+                resource_sliver = InventoryForType.get_allocated_sliver(reservation=reservation)
 
                 if resource_sliver is not None and isinstance(resource_sliver, NodeSliver):
                     logger.debug(
@@ -380,7 +380,7 @@ def __exclude_components_for_existing_reservations(*, rid: ID, graph_node: NodeS
                     (operation == ReservationOperation.Extend or not reservation.is_ticketed()):
                 continue
             # For Active or Ticketed or Ticketing reservations; reduce the counts from available
-            allocated_sliver = InventoryForType._get_allocated_sliver(reservation=reservation)
+            allocated_sliver = InventoryForType.get_allocated_sliver(reservation=reservation)
 
             if reservation.is_extending_ticket() and reservation.get_requested_resources() is not None and \
                     reservation.get_requested_resources().get_sliver() is not None:

fabric_cf/actor/core/policy/network_service_inventory.py

@@ -76,7 +76,7 @@ def __exclude_allocated_vlans(self, *, rid: ID, available_vlan_range: List[int],
                 continue
 
             # For Active or Ticketed or Ticketing reservations; reduce the counts from available
-            allocated_sliver = self._get_allocated_sliver(reservation=reservation)
+            allocated_sliver = self.get_allocated_sliver(reservation=reservation)
 
             self.logger.debug(
                 f"Existing res# {reservation.get_reservation_id()} state:{reservation.get_state()} "
@@ -278,7 +278,7 @@ def allocate_vnic(self, *, rid: ID, requested_ns: NetworkServiceSliver, owner_ns
                     continue
 
                 # For Active or Ticketed or Ticketing reservations; reduce the counts from available
-                allocated_sliver = self._get_allocated_sliver(reservation=reservation)
+                allocated_sliver = self.get_allocated_sliver(reservation=reservation)
 
                 self.logger.debug(f"Existing res# {reservation.get_reservation_id()} "
                                   f"allocated: {allocated_sliver}")
@@ -413,7 +413,7 @@ def _exclude_allocated_subnets(self, *, subnet_list: List, requested_ns_type: st
             if rid == reservation.get_reservation_id():
                 continue
 
-            allocated_sliver = self._get_allocated_sliver(reservation)
+            allocated_sliver = self.get_allocated_sliver(reservation)
             if allocated_sliver is None:
                 continue
 

fabric_cf/actor/core/time/term.py

@@ -331,6 +331,18 @@ def get_full_length(self) -> int:
 
         return end_ms - start_ms + 1
 
+    def get_remaining_length(self) -> int:
+        """
+        Returns the length of remaining term in milliseconds. The length of a term is the
+        number of milliseconds in the closed interval [now, end]
+        @returns term length
+        """
+        now = datetime.now(timezone.utc)
+        current_ms = ActorClock.to_milliseconds(when=now)
+        end_ms = ActorClock.to_milliseconds(when=self.end_time)
+
+        return end_ms - current_ms + 1
+
     def get_length(self) -> int:
         """
         Returns the length of a term in milliseconds. The length of a term is the