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
+        """