Performance Cluster Management

Confirmed cluster is wired in correctly for CLI.  Also hit circular reference
issue and had to tweak code to use just the data_store object from the process_tracker (since that is the primary object users use and everything ties back to process).

Working on #11

Author: OpenDataAlex

process_tracker/cli.py

@@ -54,9 +54,39 @@ def upgrade():
     "-p", "--parent", help="The parent process' name, if creating a process dependency"
 )
 @click.option(
-    "-c", "--child", help="The child process' name, if creating a process dependency"
+    "-c",
+    "--child",
+    help="The child process' name, if creating a process dependency.  For cluster/process relationships, the name of "
+    "the process.",
 )
-def create(topic, name, parent=None, child=None):
+@click.option(
+    "--max-processing",
+    help="For clusters, the max processing ability allocated to the cluster",
+)
+@click.option(
+    "--max-memory", help="For clusters, the max memory allocated to the cluster"
+)
+@click.option(
+    "--processing-unit",
+    help="For clusters, the unit of processing ability (Ghz, CPU, DPU, etc.",
+)
+@click.option(
+    "--memory-unit", help="For clusters, the unit of allocated memory (MB, GB, etc."
+)
+@click.option(
+    "--cluster", help="For cluster/process relationships, the name of the cluster"
+)
+def create(
+    topic,
+    name,
+    parent=None,
+    child=None,
+    max_processing=None,
+    processing_unit=None,
+    max_memory=None,
+    memory_unit=None,
+    cluster=None,
+):
     """
     Create an item that is within the valid topics list.
     :param topic: The name of the topic.
@@ -65,11 +95,32 @@ def create(topic, name, parent=None, child=None):
     :type name: string
     :param parent: The parent process' name, if creating a process dependency
     :type parent: string
-    :param child: The child process' name, if creating a process dependency
+    :param child: The child process' name, if creating a process dependency.  For cluster/process relationships, the
+                  name of the process.
     :type child: string
+    :param max_processing: For performance clusters, the maximum processing ability allocated to the cluster
+    :type max_processing: integer
+    :param max_memory: For performance clusters, the maximum memory allocated to the cluster
+    :type max_memory: integer
+    :param processing_unit: For performance clusters, the unit of processing ability allocated to the cluster
+    :type processing_unit: string
+    :param memory_unit: For performance clusters, the unit of allocated memory to the cluster
+    :type memory_unit: string
+    :param cluster: For cluster/process relationships, the name of the cluster.
+    :type cluster: string
     """
     click.echo("Attempting to create %s with name %s" % (topic, name))
-    data_store.topic_creator(topic=topic, name=name, parent=parent, child=child)
+    data_store.topic_creator(
+        topic=topic,
+        name=name,
+        parent=parent,
+        child=child,
+        max_processing=max_processing,
+        max_memory=max_memory,
+        processing_unit=processing_unit,
+        memory_unit=memory_unit,
+        cluster=cluster,
+    )
 
 
 @main.command()
@@ -81,7 +132,10 @@ def create(topic, name, parent=None, child=None):
 @click.option(
     "-c", "--child", help="The child process' name, if deleting a process dependency"
 )
-def delete(topic, name, parent=None, child=None):
+@click.option(
+    "--cluster", help="For cluster/process relationships, the name of the cluster"
+)
+def delete(topic, name, parent=None, child=None, cluster=None):
     """
     Delete an item that is within the valid topics list and not a pre-loaded item.
     :param topic: The name of the topic.
@@ -90,20 +144,71 @@ def delete(topic, name, parent=None, child=None):
     :type name: string
     :param parent: The parent process' name, if deleting a process dependency
     :type parent: string
-    :param child: The child process' name, if deleting a process dependency
+    :param child: The child process' name, if deleting a process dependency.  For cluster/process relationships, the
+                  name of the process.
     :type child: string
+    :param cluster: For cluster/process relationships, the name of the cluster.
+    :type cluster: string
     """
     click.echo("Attempting to delete %s with name %s" % (topic, name))
-    data_store.topic_deleter(topic=topic, name=name, parent=parent, child=child)
+    data_store.topic_deleter(
+        topic=topic, name=name, parent=parent, child=child, cluster=cluster
+    )
 
 
 @main.command()
 @click.option("-t", "--topic", help="The topic being created")
 @click.option("-i", "--initial-name", help="The name that needs to be changed.")
 @click.option("-n", "--name", help="The new name for the topic.")
-def update(topic, initial_name, name):
-
+@click.option(
+    "--max-processing",
+    help="For clusters, the max processing ability allocated to the cluster",
+)
+@click.option(
+    "--max-memory", help="For clusters, the max memory allocated to the cluster"
+)
+@click.option(
+    "--processing-unit",
+    help="For clusters, the unit of processing ability (Ghz, CPU, DPU, etc.",
+)
+@click.option(
+    "--memory-unit", help="For clusters, the unit of allocated memory (MB, GB, etc."
+)
+def update(
+    topic,
+    initial_name,
+    name,
+    max_processing=None,
+    max_memory=None,
+    processing_unit=None,
+    memory_unit=None,
+):
+    """
+    Update an item that is within the valid topics list and not a pre-loaded item.
+    :param topic: The name of the topic.
+    :type topic: string
+    :param initial_name: The revised name of the topic item to be updated.
+    :type initial_name: string
+    :param name: The original name of the topic item to be updated.
+    :type name: string
+    :param max_processing: For performance clusters, the maximum processing ability allocated to the cluster
+    :type max_processing: string
+    :param max_memory: For performance clusters, the maximum memory allocated to the cluster
+    :type max_memory: string
+    :param processing_unit: For performance clusters, the unit of processing ability allocated to the cluster
+    :type processing_unit: string
+    :param memory_unit: For performance clusters, the unit of allocated memory to the cluster
+    :type memory_unit: string
+    """
     click.echo(
         "Attempting to update %s with name %s to %s" % (topic, initial_name, name)
     )
-    data_store.topic_updater(topic=topic, initial_name=initial_name, name=name)
+    data_store.topic_updater(
+        topic=topic,
+        initial_name=initial_name,
+        name=name,
+        max_processing=max_processing,
+        max_memory=max_memory,
+        processing_unit=processing_unit,
+        memory_unit=memory_unit,
+    )

process_tracker/data_store.py

@@ -10,6 +10,7 @@
 
 from process_tracker.models.model_base import Base
 from process_tracker.models.actor import Actor
+from process_tracker.models.capacity import Cluster, ClusterProcess
 from process_tracker.models.extract import ExtractStatus
 from process_tracker.models.process import (
     ErrorType,
@@ -182,7 +183,18 @@ def initialize_data_store(self, overwrite=False):
 
         self.logger.debug("Finished the initialization check.")
 
-    def topic_creator(self, topic, name, parent=None, child=None):
+    def topic_creator(
+        self,
+        topic,
+        name,
+        parent=None,
+        child=None,
+        max_processing=None,
+        processing_unit=None,
+        max_memory=None,
+        memory_unit=None,
+        cluster=None,
+    ):
         """
         For the command line tool, validate the topic and create the new instance.
         :param topic: The name of the topic.
@@ -191,8 +203,19 @@ def topic_creator(self, topic, name, parent=None, child=None):
         :type name: string
         :param parent: The parent process' name, if creating a process dependency
         :type parent: string
-        :param child: The child process' name, if creating a process dependency
+        :param child: The child process' name, if creating a process dependency.For cluster/process relationships, the
+                      name of the process.
         :type child: string
+        :param max_processing: For performance clusters, the maximum processing ability allocated to the cluster
+        :type max_processing: string
+        :param max_memory: For performance clusters, the maximum memory allocated to the cluster
+        :type max_memory: string
+        :param processing_unit: For performance clusters, the unit of processing ability allocated to the cluster
+        :type processing_unit: string
+        :param memory_unit: For performance clusters, the unit of allocated memory to the cluster
+        :type memory_unit: string
+        :param cluster: For cluster/process relationships, the name of the cluster.
+        :type cluster: string
         :return:
         """
         self.logger.info("Attempting to create %s item: %s" % (topic, name))
@@ -203,6 +226,33 @@ def topic_creator(self, topic, name, parent=None, child=None):
                 item = self.get_or_create_item(model=Actor, actor_name=name)
                 self.logger.info("Actor created: %s" % item.__repr__)
 
+            elif topic == "cluster":
+                item = self.get_or_create_item(
+                    model=Cluster,
+                    cluster_name=name,
+                    cluster_max_memory=max_memory,
+                    cluster_max_memory_unit=memory_unit,
+                    cluster_max_processing=max_processing,
+                    cluster_max_processing_unit=processing_unit,
+                )
+                self.logger.info("Cluster created: %s" % item.__repr__)
+
+            elif topic == "cluster process":
+                cluster = self.get_or_create_item(
+                    model=Cluster, create=False, cluster_name=cluster
+                )
+                process = self.get_or_create_item(
+                    model=Process, create=False, process_name=child
+                )
+
+                item = self.get_or_create_item(
+                    model=ClusterProcess,
+                    cluster_id=cluster.cluster_id,
+                    process_id=process.process_id,
+                )
+
+                self.logger.info("Cluster Process created: %s" % item.__repr__)
+
             elif topic == "extract status":
                 item = self.get_or_create_item(
                     model=ExtractStatus, extract_status_name=name
@@ -260,7 +310,7 @@ def topic_creator(self, topic, name, parent=None, child=None):
 
         return item
 
-    def topic_deleter(self, topic, name, parent=None, child=None):
+    def topic_deleter(self, topic, name, parent=None, child=None, cluster=None):
         """
         For the command line tool, validate that the topic name is not a default value and if not, delete it.
         :param topic: The SQLAlchemy object type
@@ -269,8 +319,9 @@ def topic_deleter(self, topic, name, parent=None, child=None):
         :type name: string
         :param parent: The parent process' name, if deleting a process dependency
         :type parent: string
-        :param child: The child process' name, if deleting a process dependency
+        :param child: The child process' name, if deleting a process dependency.  For cluster/process relationship, the name of the process.
         :type child: string
+        :param cluster: For cluster/process relationship, the name of the cluster.
         :return:
         """
         item_delete = False
@@ -284,6 +335,33 @@ def topic_deleter(self, topic, name, parent=None, child=None):
                 self.session.query(Actor).filter(Actor.actor_name == name).delete()
                 self.logger.info("%s %s deleted." % (topic, name))
 
+            elif topic == "cluster":
+                item_delete = True
+                self.session.query(Cluster).filter(
+                    Cluster.cluster_name == name
+                ).delete()
+                self.logger.info("%s %s deleted." % (topic, name))
+
+            elif topic == "cluster process":
+                item_delete = True
+                cluster = self.get_or_create_item(
+                    model=Cluster, create=False, cluster_name=cluster
+                )
+                process = self.get_or_create_item(
+                    model=Process, create=False, process_name=child
+                )
+
+                item = self.get_or_create_item(
+                    model=ClusterProcess,
+                    create=False,
+                    cluster_id=cluster.cluster_id,
+                    process_id=process.process_id,
+                )
+
+                self.session.delete(item)
+
+                self.logger.info("%s %s - %s deleted." % (topic, cluster, child))
+
             elif topic == "extract status" and name not in preload_extract_status_types:
                 item_delete = True
                 self.session.query(ExtractStatus).filter(
@@ -359,7 +437,16 @@ def topic_deleter(self, topic, name, parent=None, child=None):
 
         return "blarg"
 
-    def topic_updater(self, topic, initial_name, name):
+    def topic_updater(
+        self,
+        topic,
+        initial_name,
+        name,
+        max_processing=None,
+        processing_unit=None,
+        max_memory=None,
+        memory_unit=None,
+    ):
         """
         For the command line tool, validate that the topic name is not a default value and if not, update it.
         :param topic: name of the SQLAlchemy object
@@ -368,6 +455,14 @@ def topic_updater(self, topic, initial_name, name):
         :type initial_name: string
         :param name: The updated name of the object to be updated.
         :type name: string
+        :param max_processing: For performance clusters, the maximum processing ability allocated to the cluster
+        :type max_processing: string
+        :param max_memory: For performance clusters, the maximum memory allocated to the cluster
+        :type max_memory: string
+        :param processing_unit: For performance clusters, the unit of processing ability allocated to the cluster
+        :type processing_unit: string
+        :param memory_unit: For performance clusters, the unit of allocated memory to the cluster
+        :type memory_unit: string
         :return:
         """
         if self.topic_validator(topic=topic):
@@ -378,6 +473,26 @@ def topic_updater(self, topic, initial_name, name):
                 item.actor_name = name
                 self.logger.info("%s %s updated." % (topic, name))
 
+            elif topic == "cluster":
+                item = self.get_or_create_item(
+                    model=Cluster, create=False, cluster_name=initial_name
+                )
+
+                item.cluster_name = name
+                if max_memory is not None:
+                    item.cluster_max_memory = max_memory
+
+                if memory_unit is not None:
+                    item.cluster_max_memory_unit = memory_unit
+
+                if max_processing is not None:
+                    item.cluster_max_processing = max_processing
+
+                if processing_unit is not None:
+                    item.cluster_max_processing_unit = processing_unit
+
+                self.logger.info("%s %s updated." % (topic, name))
+
             elif (
                 topic == "extract status"
                 and initial_name not in preload_extract_status_types
@@ -455,6 +570,7 @@ def topic_validator(self, topic):
         # Only data store topics that should be allowed to be created from the command line tool.
         valid_topics = [
             "actor",
+            "cluster",
             "error type",
             "extract status",
             "process dependency",

process_tracker/extract_tracker.py

@@ -6,7 +6,6 @@
 
 from sqlalchemy.orm import aliased
 
-from process_tracker.data_store import DataStore
 from process_tracker.location_tracker import LocationTracker
 from process_tracker.utilities.settings import SettingsManager
 from process_tracker.utilities import utilities
@@ -62,7 +61,9 @@ def __init__(
         elif location_path is not None:
             self.logger.info("Location path provided.  Creating Location object.")
             self.location = LocationTracker(
-                location_name=location_name, location_path=location_path
+                location_name=location_name,
+                location_path=location_path,
+                data_store=self.data_store,
             )
         else:
             raise Exception("A location object or location_path must be provided.")