Merge pull request #216 from fabric-testbed/multi_threaded

Performance Improvements and more

Author: kthare10

Dockerfile-auth

@@ -1,22 +1,30 @@
 FROM python:3.9.0
 MAINTAINER Komal Thareja<[email protected]>
 
-ARG HANDLERS_VER=1.3.1
+ARG HANDLERS_VER=1.3.2
 
 RUN mkdir -p /usr/src/app
 WORKDIR /usr/src/app
 VOLUME ["/usr/src/app"]
 
 EXPOSE 11000
 
+RUN apt-get update
+RUN apt-get install cron -y
+
 COPY requirements.txt /usr/src/app/
+COPY docker-entrypoint.sh /usr/src/app/
 COPY fabric_cf /usr/src/app/fabric_cf
+COPY tools/cleanup.py /usr/src/app/fabric_cf/
+COPY tools/install.sh /usr/src/app/fabric_cf/
+
 RUN pip3 install --no-cache-dir -r requirements.txt
 RUN mkdir -p "/etc/fabric/message_bus/schema"
 RUN mkdir -p "/etc/fabric/actor/config"
 RUN mkdir -p "/var/log/actor"
 RUN cp /usr/local/lib/python3.9/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema
 RUN pip3 install fabric-am-handlers==${HANDLERS_VER}
+RUN sh /usr/src/app/fabric_cf/install.sh
 
-ENTRYPOINT ["python3"]
-CMD ["-m", "fabric_cf.authority"]
+ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]
+CMD ["fabric_cf.authority"]

Dockerfile-broker

@@ -7,14 +7,21 @@ VOLUME ["/usr/src/app"]
 
 EXPOSE 11000
 
+RUN apt-get update
+RUN apt-get install cron -y
+
 COPY requirements.txt /usr/src/app/
+COPY docker-entrypoint.sh /usr/src/app/
 COPY fabric_cf /usr/src/app/fabric_cf
+COPY tools/cleanup.py /usr/src/app/fabric_cf/
+COPY tools/install.sh /usr/src/app/fabric_cf/
+
 RUN pip3 install --no-cache-dir -r requirements.txt
 RUN mkdir -p "/etc/fabric/message_bus/schema"
 RUN mkdir -p "/etc/fabric/actor/config"
 RUN mkdir -p "/var/log/actor"
 RUN cp /usr/local/lib/python3.9/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema
+RUN sh /usr/src/app/fabric_cf/install.sh
 
-ENTRYPOINT ["python3"]
-
-CMD ["-m", "fabric_cf.broker"]
+ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]
+CMD ["fabric_cf.broker"]

Dockerfile-cf

@@ -7,12 +7,23 @@ VOLUME ["/usr/src/app"]
 
 EXPOSE 11000
 
+RUN apt-get update
+RUN apt-get install cron -y
+
 COPY requirements.txt /usr/src/app/
 COPY fabric_cf /usr/src/app/fabric_cf
+COPY tools/cleanup.py /usr/src/app/fabric_cf/
+COPY tools/install.sh /usr/src/app/fabric_cf/
+
 RUN pip3 install --no-cache-dir -r requirements.txt
 RUN mkdir -p "/etc/fabric/message_bus/schema"
 RUN mkdir -p "/etc/fabric/actor/config"
 RUN mkdir -p "/var/log/actor"
 RUN cp /usr/local/lib/python3.9/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema
 
-ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]
+RUN echo "0 2 * * * root /usr/local/bin/python3.9 /usr/src/app/cleanup.py -f /etc/fabric/actor/config/config.yaml -d 30 -c slices -o remove" >> /etc/crontab
+RUN service cron reload
+RUN service cron restart
+
+ENTRYPOINT ["sh"]
+CMD ["tail", "-f", "/dev/null"]

Dockerfile-orchestrator

@@ -8,15 +8,21 @@ VOLUME ["/usr/src/app"]
 EXPOSE 11000
 EXPOSE 8700
 
+RUN apt-get update
+RUN apt-get install cron -y
+
 COPY requirements.txt /usr/src/app/
+COPY docker-entrypoint.sh /usr/src/app/
 COPY fabric_cf /usr/src/app/fabric_cf
+COPY tools/cleanup.py /usr/src/app/fabric_cf/
+COPY tools/install.sh /usr/src/app/fabric_cf/
+
 RUN pip3 install --no-cache-dir -r requirements.txt
 RUN mkdir -p "/etc/fabric/message_bus/schema"
 RUN mkdir -p "/etc/fabric/actor/config"
 RUN mkdir -p "/var/log/actor"
 RUN cp /usr/local/lib/python3.9/site-packages/fabric_mb/message_bus/schema/*.avsc /etc/fabric/message_bus/schema
+RUN sh /usr/src/app/fabric_cf/install.sh
 
-
-ENTRYPOINT ["python3"]
-
-CMD ["-m", "fabric_cf.orchestrator"]
+ENTRYPOINT ["/usr/src/app/docker-entrypoint.sh"]
+CMD ["fabric_cf.orchestrator"]

docker-entrypoint.sh

@@ -1,2 +1,3 @@
 #!/bin/sh
-tail -f /dev/null
+service cron start
+python3.9 -m $1

fabric_cf/__init__.py

@@ -1 +1 @@
-__VERSION__ = "1.3.0"
+__VERSION__ = "1.3.1"

fabric_cf/actor/boot/configuration_processor.py

@@ -99,14 +99,12 @@ def process(self):
         try:
             self.create_actor()
             self.initialize_actor()
-            self.logger.info(f"There are {ActorMixin.actor_count} actors")
             self.register_actor()
             self.create_default_slice()
             self.populate_inventory_neo4j()
             self.recover_actor()
             self.enable_ticking()
             self.process_topology()
-            self.logger.info(f"Processing exports with actor count {ActorMixin.actor_count}")
             self.process_advertise()
             self.logger.info("Processing exports completed")
         except Exception as e:
@@ -211,7 +209,7 @@ def make_actor_policy(self, *, actor: ABCActorMixin, config: ActorConfig):
         """
         Creates Actor Policy instance
         @param actor actor
-        @param actor_config actor config
+        @param config actor config
         @raises ConfigurationException in case of error
         """
         policy = None
@@ -349,7 +347,6 @@ def initialize_actor(self):
         @raises ConfigurationException in case of error
         """
         try:
-            ActorMixin.actor_count += 1
             self.actor.initialize()
         except Exception as e:
             raise ConfigurationException(f"Actor failed to initialize: {self.actor.get_name()} {e}")

fabric_cf/actor/core/apis/abc_actor_mixin.py

@@ -300,6 +300,12 @@ def query(self, *, query: dict = None, caller: AuthToken = None,
         Returns:
             query response
         """
+    @abstractmethod
+    def execute_on_actor_thread(self, *, runnable: ABCActorRunnable):
+        """
+        Execute on Actor Thread and Wait until response is processed
+        @params runnable: reservation to be processed
+        """
 
     @abstractmethod
     def execute_on_actor_thread_and_wait(self, *, runnable: ABCActorRunnable):
@@ -545,6 +551,19 @@ def close_by_rid(self, *, rid: ID):
             Exception in case of error
         """
 
+    @abstractmethod
+    def close_delegation(self, *, did: str):
+        """
+        Closes the delegation. Note: the delegation 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 delegation.
+
+        Args:
+            did: delegation id
+        Raises:
+            Exception in case of error
+        """
+
     @abstractmethod
     def close(self, *, reservation: ABCReservationMixin):
         """
@@ -682,3 +701,19 @@ def unregister(self, *, reservation: ABCReservationMixin, rid: ID):
         Raises:
             Exception in case of error
         """
+
+    def get_asm_thread(self):
+        return None
+
+    @abstractmethod
+    def remove_delegation(self, *, did: str):
+        """
+        Removes the specified delegation. Note: the delegation must have already been registered with the actor.
+        This method will unregister the reservation and remove it from the underlying database.
+        Only closed and failed delegation can be removed.
+
+        Args:
+            did: delegation id
+        Raises:
+            Exception if an error occurs or when trying to remove a delegation that is neither failed or closed.
+        """

fabric_cf/actor/core/apis/abc_database.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 from abc import abstractmethod, ABC
+from datetime import datetime
 from typing import TYPE_CHECKING, List
 
 from fabric_cf.actor.core.apis.abc_delegation import ABCDelegation
@@ -179,7 +180,7 @@ def get_client_reservations(self, *, slice_id: ID = None) -> List[ABCReservation
     @abstractmethod
     def get_slices(self, *, slice_id: ID = None, slice_name: str = None, project_id: str = None, email: str = None,
                    state: list[int] = None, oidc_sub: str = None, slc_type: List[SliceTypes] = None,
-                   limit: int = None, offset: int = None) -> List[ABCSlice] or None:
+                   limit: int = None, offset: int = None, lease_end: datetime = None) -> List[ABCSlice] or None:
         """
         Retrieves the specified slices.
 
@@ -192,6 +193,7 @@ def get_slices(self, *, slice_id: ID = None, slice_name: str = None, project_id:
         @param slc_type slice type
         @param limit limit
         @param offset offset
+        @param lease_end lease_end
 
         @return list of slices
 

fabric_cf/actor/core/apis/abc_delegation.py

@@ -398,3 +398,15 @@ def fail(self, *, message: str, exception: Exception = None):
               message: error message
               exception: exception
         """
+
+    @abstractmethod
+    def lock(self):
+        """
+        Lock delegation
+        """
+
+    @abstractmethod
+    def unlock(self):
+        """
+        Unlock delegation
+        """

fabric_cf/actor/core/apis/abc_mgmt_actor.py

@@ -126,6 +126,14 @@ def close_reservation(self, *, rid: ID) -> bool:
         @return true for success; false otherwise
         """
 
+    @abstractmethod
+    def close_delegation(self, *, did: str) -> bool:
+        """
+        Closes the specified delegation
+        @param did delegation id
+        @return true for success; false otherwise
+        """
+
     @abstractmethod
     def close_reservations(self, *, slice_id: ID) -> bool:
         """
@@ -143,6 +151,15 @@ def remove_reservation(self, *, rid: ID) -> bool:
         @return true for success; false otherwise
         """
 
+    @abstractmethod
+    def remove_delegation(self, *, did: str) -> bool:
+        """
+        Removes the specified delegation.
+        Note only closed delegation can be removed.
+        @param did delegation id of the delegation to be removed
+        @return true for success; false otherwise
+        """
+
     @abstractmethod
     def get_reservation_state_for_reservations(self, *, reservation_list: List[str]) -> List[ReservationStateAvro]:
         """

fabric_cf/actor/core/apis/abc_reservation_mixin.py

@@ -508,3 +508,15 @@ def handle_failed_rpc(self, *, failed: FailedRPC):
         Processes a failed RPC request.
         @param failed failed
         """
+
+    @abstractmethod
+    def lock(self):
+        """
+        Lock the reservation
+        """
+
+    @abstractmethod
+    def unlock(self):
+        """
+        Unlock the reservation
+        """

fabric_cf/actor/core/apis/abc_reservation_resources.py

@@ -34,7 +34,6 @@
     from fabric_cf.actor.core.time.term import Term
     from fabric_cf.actor.core.util.resource_type import ResourceType
     from fabric_cf.actor.core.kernel.resource_set import ResourceSet
-    from fabric_cf.actor.core.util.resource_count import ResourceCount
 
 
 class ABCReservationResources(ABC):
@@ -48,19 +47,6 @@ class ABCReservationResources(ABC):
         previousResources: the previous resource set bound to the reservation
         leasedResources: the concrete resources bound to the reservation.
     """
-
-    @abstractmethod
-    def count(self, *, rc: ResourceCount, when: datetime):
-        """
-        Counts the number of resources in the reservation relative to the specified time.
-        The ResourceCount object is updated with the count of active, pending, expired, failed, etc. units.
-        Note: "just a hint" unless the kernel lock is held.
-
-        Args:
-            rc: holder for counts
-            time: when instance
-        """
-
     @abstractmethod
     def get_approved_resources(self) -> ResourceSet:
         """

fabric_cf/actor/core/apis/abc_slice.py

@@ -467,3 +467,15 @@ def unregister_delegation(self, *, delegation: ABCDelegation):
 
         @param delegation delegation to unregister
         """
+
+    @abstractmethod
+    def lock_slice(self):
+        """
+        Lock slice
+        """
+
+    @abstractmethod
+    def unlock_slice(self):
+        """
+        Unlock slice
+        """

fabric_cf/actor/core/common/constants.py

@@ -174,6 +174,8 @@ class Constants:
     PROPERTY_CONF_CLASS_NAME = 'class'
     PROPERTY_CONF_PROPERTIES_NAME = 'properties'
 
+    CONFIG_SECTION_ACTOR = "actor"
+
     PROTOCOL_LOCAL = "local"
     PROTOCOL_KAFKA = "kafka"
 

fabric_cf/actor/core/container/globals.py

@@ -476,7 +476,7 @@ def timer_loop(self):
         """
         Timer thread run function
         """
-        self.log.debug("Timer thread started")
+        self.log.debug(f"Timer thread started")
         while True:
             with self.timer_condition:
                 while self.timer_scheduler.empty() and self.started:
@@ -485,17 +485,17 @@ def timer_loop(self):
                         self.timer_condition.wait()
                     except InterruptedError as e:
                         self.log.error(traceback.format_exc())
-                        self.log.error("Timer thread interrupted. Exiting {}".format(e))
+                        self.log.error(f" Timer thread interrupted. Exiting {e}")
                         return
 
                     if not self.started:
-                        self.log.info("Timer thread exiting")
+                        self.log.info(f"Timer thread exiting")
                         return
 
                     self.timer_condition.notify_all()
 
                 if not self.timer_scheduler.empty():
-                    #self.log.debug("Executing Scheduled items")
+                    #self.log.debug(f"Executing Scheduled items")
                     self.timer_scheduler.run(blocking=False)
 
 

fabric_cf/actor/core/container/message_service.py

@@ -103,6 +103,8 @@ def handle_message(self, message: AbcMessageAvro):
                     message.get_message_name() == AbcMessageAvro.update_reservation or \
                     message.get_message_name() == AbcMessageAvro.remove_reservation or \
                     message.get_message_name() == AbcMessageAvro.extend_reservation or \
+                    message.get_message_name() == AbcMessageAvro.close_delegations or \
+                    message.get_message_name() == AbcMessageAvro.remove_delegation or \
                     message.get_message_name() == AbcMessageAvro.maintenance_request:
                 self.kafka_mgmt_service.process(message=message)
             else: