Merge pull request #109 from Ortec-Finance/bugfixes

[Carbon Aware HPC] - 5 - Bugfixes for Multi Cluster Controller

.vscode/launch.json

@@ -16,6 +16,7 @@
                 "-n rdlabs-experiment-cas-eu-west"
                 ],
                 "env": {
+                "PROMETHEUS_TOKEN": "GET ME FROM SERVICE ACCOUNT AND PORT FORWARD THANOS",
                 "KOPF_RUN_ARGS": "--verbose"
             },
         },

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## v0.40.0
+Bugfixes
+
 ## v0.39.0
 Demo for GSF
 

k8s/cluster-config/operators/sailfish/crd.yaml

@@ -106,6 +106,9 @@ spec:
                     activatedTargetCluster:
                       description: The current best destination based on the trigger variables and cost function
                       type: string
+                    status:
+                      description: If the operator's decision resulted into a fallback, this will be the reason why
+                      type: string
                 clusters:
                   type: array
                   description: "Defines the Remote Sailfish Clusters"
@@ -117,8 +120,6 @@ spec:
                       queue:
                         type: string
                         description: "To reference a remote Sailfish Cluster, use this, the Queues will be automatically generated"
-                      query:
-                        type: string
                       status:
                         type: string
                       reason: 

operator/multi-cluster-controller/app.py

@@ -6,6 +6,9 @@
 from tolerations import Tolerations
 from bridge import Bridge
 from score import Score
+import urllib3
+
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
 # Dynamically get the namespace the operator is running in
 if os.path.isfile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"):
@@ -95,8 +98,6 @@ def modify_activemq_artemis(spec, name, namespace, uid, logger, **kwargs):
                 name, namespace, cluster["name"], owner_reference, logger
             )
 
-        label = [f"ortec-finance.com/sailfish-cluster: {name}"]
-
         logger.info(f"Applying BrokerConfiguration to {SAILFISH_BROKER_NAME}")
         bridgeConfig.update_activemq_artemis(
             SAILFISH_BROKER_NAME, namespace, patch, logger
@@ -106,28 +107,30 @@ def modify_activemq_artemis(spec, name, namespace, uid, logger, **kwargs):
         logger.warning(f"ActiveMQArtemis {name} not found in namespace {namespace}")
 
 
-@kopf.on.timer("ortec-finance.com", "v1alpha1", "sailfishclusters", interval=2)
+@kopf.on.update("ortec-finance.com", "v1alpha1", "sailfishclusters")
+@kopf.on.timer("ortec-finance.com", "v1alpha1", "sailfishclusters", interval=60)
 def poll_sailfish_clusters_status(spec, patch, logger, **kwargs):
     logger.info("Polling Sailfish Clusters Status")
     cluster_statuses = []
 
     ## Append Remote Sailfish Clusters
     for cluster in spec.get("clusters", []):
         queue_name = cluster.get("queue", f"sailfish{cluster['name']}")
-        query = next(
-            (
-                item
-                for item in spec.get("triggers")
-                if item["clusterRef"] == cluster["name"]
-            ),
-            None,
-        )["query"]
-        if query:
+        trigger = None
+        if "triggers" in spec:
+            trigger = next(
+                (
+                    item
+                    for item in spec.get("triggers")
+                    if item["clusterRef"] == cluster["name"]
+                ),
+                None,
+            )
+        if trigger and "query" in trigger:
             cluster_statuses.append(
                 {
                     "name": cluster["name"],
                     "queue": queue_name,
-                    "query": query,
                     "status": "active",
                 }
             )
@@ -158,29 +161,34 @@ def get_active_sailfish_clusters(sailfish_cluster):
     return activeClusters
 
 
-@kopf.on.timer("ortec-finance.com", "v1alpha1", "sailfishclusters", interval=2)
+@kopf.on.update("ortec-finance.com", "v1alpha1", "sailfishclusters")
+@kopf.on.timer("ortec-finance.com", "v1alpha1", "sailfishclusters", interval=60)
 def poll_sailfish_cluster_best_destination(spec, patch, status, logger, **kwargs):
     evaluator = PrometheusEvaluator()
     cluster_statuses = []
 
-    for cluster in spec.get("clusters", []):
+    for cluster in status.get("clusters", []):
+        if cluster["status"] == "inactive":
+            logger.info(f"Skipping Inactive cluster {cluster['name']}")
+            continue
         trigger_statuses = []
-        for trigger in spec.get("triggers"):
-            if trigger["clusterRef"] == cluster["name"]:
-                value = evaluator.evaluate_query(trigger["query"])
-                if value is None:
-                    logger.error(
-                        f"Was not able to get a value from the query: {trigger['query']}"
+        if "triggers" in spec:
+            for trigger in spec.get("triggers"):
+                if trigger["clusterRef"] == cluster["name"]:
+                    value = evaluator.evaluate_query(trigger["query"])
+                    if value is None:
+                        logger.error(
+                            f"Was not able to get a value from the query: {trigger['query']}"
+                        )
+                        raise kopf.PermanentError("Value is None. Cannot proceed.")
+                    scaled_value = variableScore.apply_scaler(value, trigger["scaler"])
+
+                    trigger_statuses.append(
+                        {
+                            "name": trigger["name"],
+                            "value": scaled_value,
+                        }
                     )
-                    raise kopf.PermanentError("Value is None. Cannot proceed.")
-                scaled_value = variableScore.apply_scaler(value, trigger["scaler"])
-
-                trigger_statuses.append(
-                    {
-                        "name": trigger["name"],
-                        "value": scaled_value,
-                    }
-                )
 
         score = variableScore.sum_trigger_values(trigger_statuses)
 
@@ -203,9 +211,22 @@ def poll_sailfish_cluster_best_destination(spec, patch, status, logger, **kwargs
             }
         )
 
-    reward = variableScore.cost_function(logger, cluster_statuses)
+    if cluster_statuses:
+        reward = variableScore.cost_function(logger, cluster_statuses)
+        scheduler = {
+            "clusters": cluster_statuses,
+            "activatedTargetCluster": reward["name"],
+            "status": "Healthy",
+        }
+
+    else:  # Fallback to Local cluster if no triggers are set.
+        for cluster in spec.get("clusters", []):
+            if "queue" in cluster:
+                scheduler = {
+                    "activatedTargetCluster": cluster["name"],
+                    "status": "Unhealthy configuration: No triggers set. Fallback to Local Cluster",
+                }
 
-    scheduler = {"clusters": cluster_statuses, "activatedTargetCluster": reward["name"]}
     if "scheduler" in status and status["scheduler"] == scheduler:
         logger.info("No change in scheduler status")
     else:

operator/multi-cluster-controller/prometheus.py

@@ -1,4 +1,6 @@
 import os
+import re
+
 from prometheus_api_client import PrometheusConnect
 
 
@@ -20,7 +22,7 @@ def __init__(self):
             ) as file:
                 self.namespace = file.read().strip()
         else:
-            TOKEN = "GET ME FROM SERVICE ACCOUNT AND PORT FORWARD THANOS"
+            TOKEN = os.getenv("PROMETHEUS_TOKEN")
 
             self.prom = PrometheusConnect(
                 url=url, headers={"Authorization": f"Bearer {TOKEN}"}, disable_ssl=True
@@ -60,8 +62,6 @@ def evaluate_expression(self, expression):
         return self.compare(result_value, operator, float(threshold))
 
     def parse_expression(self, expression):
-        import re
-
         """Parse the expression into a query, operator, and threshold."""
         # This regex allows spaces around the operator and handles complex queries
         match = re.search(

operator/multi-cluster-controller/score.py

@@ -11,12 +11,10 @@ def sum_trigger_values(self, trigger_statuses):
             result += trigger["value"]
         return result
 
-    def cost_function(self, logger, cluster_results, operator="MIN"):
-        if operator == "MIN":
-            winner = min(cluster_results, key=lambda x: x["score"])
-        elif operator == "MAX":
-            winner = max(cluster_results, key=lambda x: x["score"])
-        else:
+    def cost_function(self, logger, cluster_results, operator=min):
+        try:
+            winner = operator(cluster_results, key=lambda x: x["score"])
+        except Exception as e:
             logger.error("Operator not supported")
             raise kopf.PermanentError("Operator not supported")
         logger.info(