Performance Cluster Management

Confirmed cluster process is wired in correctly for CLI.

Also found latent bug in data_store.  Forgot to include the data_store_port
variable.  Have included now in all connection strings.

Closes #11

Author: OpenDataAlex

dbscripts/mysql_process_tracker.sql

@@ -277,10 +277,10 @@ create table process_tracker.cluster_tracking
 	cluster_id int auto_increment
 		primary key,
 	cluster_name varchar(250) not null,
-	cluster_max_memory int not null,
+	cluster_max_memory int null,
 	cluster_max_memory_unit char(2) null,
-	cluster_max_processing int not null,
-	cluster_max_processing_unit varchar(3) not null,
+	cluster_max_processing int null,
+	cluster_max_processing_unit varchar(3) null,
 	cluster_current_memory_usage int null,
 	cluster_current_process_usage int null,
 	constraint cluster_tracking_cluster_name_uindex

dbscripts/postgresql_process_tracker.sql

@@ -423,10 +423,10 @@ create table cluster_tracking
 		constraint cluster_tracking_pk
 			primary key,
 	cluster_name varchar(250) not null,
-	cluster_max_memory integer not null,
-	cluster_max_memory_unit char(2) not null,
-	cluster_max_processing integer not null,
-	cluster_max_processing_unit varchar(3) not null,
+	cluster_max_memory integer null,
+	cluster_max_memory_unit char(2) null,
+	cluster_max_processing integer null,
+	cluster_max_processing_unit varchar(3) null,
 	cluster_current_memory_usage integer,
 	cluster_current_process_usage integer
 );

process_tracker/data_store.py

@@ -571,6 +571,7 @@ def topic_validator(self, topic):
         valid_topics = [
             "actor",
             "cluster",
+            "cluster process",
             "error type",
             "extract status",
             "process dependency",
@@ -656,6 +657,8 @@ def verify_and_connect_to_data_store(self):
                     + data_store_password
                     + "@"
                     + data_store_host
+                    + ":"
+                    + data_store_port
                     + "/"
                     + data_store_name
                 )
@@ -669,6 +672,8 @@ def verify_and_connect_to_data_store(self):
                     + data_store_password
                     + "@"
                     + data_store_host
+                    + ":"
+                    + data_store_port
                     + "/"
                     + data_store_name
                 )
@@ -681,6 +686,8 @@ def verify_and_connect_to_data_store(self):
                     + data_store_password
                     + "@"
                     + data_store_host
+                    + ":"
+                    + data_store_port
                     + "/"
                     + data_store_name
                 )

process_tracker/models/capacity.py

@@ -19,10 +19,10 @@ class Cluster(Base):
         primary_key=True,
     )
     cluster_name = Column(String(250), unique=True, nullable=False)
-    cluster_max_memory = Column(Integer, nullable=False)
-    cluster_max_memory_unit = Column(String(2), nullable=False)
-    cluster_max_processing = Column(Integer, nullable=False)
-    cluster_max_processing_unit = Column(String(3), nullable=False)
+    cluster_max_memory = Column(Integer, nullable=True)
+    cluster_max_memory_unit = Column(String(2), nullable=True)
+    cluster_max_processing = Column(Integer, nullable=True)
+    cluster_max_processing_unit = Column(String(3), nullable=True)
     cluster_current_memory_usage = Column(Integer)
     cluster_current_process_usage = Column(Integer)
 
@@ -50,7 +50,7 @@ class ClusterProcess(Base):
 
     def __repr__(self):
 
-        return (
-            "<ClusterProcess (cluster_id=%s, process_id=%s)" % self.cluster_id,
+        return "<ClusterProcess (cluster_id=%s, process_id=%s)" % (
+            self.cluster_id,
             self.process_id,
         )

tests/test_cli.py

@@ -16,6 +16,7 @@
 from process_tracker.models.extract import ExtractStatus
 from process_tracker.models.process import (
     ErrorType,
+    Process,
     ProcessDependency,
     ProcessStatus,
     ProcessType,
@@ -128,6 +129,54 @@ def test_create_cluster(self):
 
         self.assertEqual("Test Cluster", given_name)
 
+    def test_create_cluster_process(self):
+        """
+        Testing that when creating a cluster process relationship record, it is added.
+        :return:
+        """
+        cluster = self.data_store.get_or_create_item(
+            model=Cluster, cluster_name="Test Cluster"
+        )
+
+        process = self.data_store.get_or_create_item(
+            model=Process, process_name="Test Cluster Process"
+        )
+
+        result = self.runner.invoke(
+            main,
+            'create -t "cluster process" --cluster "%s" --child "%s"'
+            % (cluster.cluster_name, process.process_name),
+        )
+
+        instance = (
+            self.session.query(ClusterProcess)
+            .filter(ClusterProcess.cluster_id == cluster.cluster_id)
+            .filter(ClusterProcess.process_id == process.process_id)
+            .first()
+        )
+
+        given_result = [instance.cluster_id, instance.process_id]
+        expected_result = [cluster.cluster_id, process.process_id]
+
+        self.runner.invoke(
+            main,
+            [
+                "delete",
+                "-t",
+                "cluster process",
+                "--cluster",
+                "Test Cluster",
+                "-c",
+                "Test Cluster Process",
+            ],
+        )
+        self.runner.invoke(main, ["delete", "-t", "cluster", "-n", "Test Cluster"])
+        self.runner.invoke(
+            main, ["delete", "-t", "process", "-n", "Test Cluster Process"]
+        )
+
+        self.assertEqual(expected_result, given_result)
+
     def test_create_extract_status(self):
         """
         Testing that when creating an extract status record it is added.
@@ -356,6 +405,51 @@ def test_delete_cluster(self):
         self.assertEqual(None, instance)
         self.assertEqual(0, result.exit_code)
 
+    def test_create_cluster_process(self):
+        """
+        Testing that when creating a cluster process relationship record, it is added.
+        :return:
+        """
+        cluster = self.data_store.get_or_create_item(
+            model=Cluster, cluster_name="Test Cluster"
+        )
+
+        process = self.data_store.get_or_create_item(
+            model=Process, process_name="Test Cluster Process"
+        )
+
+        result = self.runner.invoke(
+            main,
+            'create -t "cluster process" --cluster "%s" --child "%s"'
+            % (cluster.cluster_name, process.process_name),
+        )
+
+        self.runner.invoke(
+            main,
+            [
+                "delete",
+                "-t",
+                "cluster process",
+                "--cluster",
+                "Test Cluster",
+                "-c",
+                "Test Cluster Process",
+            ],
+        )
+        self.runner.invoke(main, ["delete", "-t", "cluster", "-n", "Test Cluster"])
+        self.runner.invoke(
+            main, ["delete", "-t", "process", "-n", "Test Cluster Process"]
+        )
+
+        instance = (
+            self.session.query(ClusterProcess)
+            .filter(ClusterProcess.cluster_id == cluster.cluster_id)
+            .filter(ClusterProcess.process_id == process.process_id)
+            .first()
+        )
+
+        self.assertEqual(None, instance)
+
     def test_delete_extract_status(self):
         """
         Testing that when deleting an extract status record not on the protected list, it is deleted.

tests/test_extract_tracker.py

@@ -2,6 +2,7 @@
 from datetime import datetime, timedelta
 import unittest
 
+from process_tracker.models.capacity import ClusterProcess
 from process_tracker.models.extract import Extract, ExtractProcess, Location
 from process_tracker.models.process import (
     Process,
@@ -37,6 +38,7 @@ def setUpClass(cls):
 
     @classmethod
     def tearDownClass(cls):
+        cls.session.query(ClusterProcess).delete()
         cls.session.query(ErrorTracking).delete()
         cls.session.query(ExtractProcess).delete()
         cls.session.query(ProcessTracking).delete()