Merge pull request #2 from Aignosi/SIENTIAPDE-1095-realizar-testes-basicos-no-cluster-suse-linux

Sientiapde 1095 realizar testes basicos no cluster suse linux

orchestrator/activities/slot_manager.py

@@ -67,7 +67,15 @@ class SlotManager(Redis):
 
         self.logger.debug("Active ingestors: \n %s", active_ingestors)
 
-        return [ingestor.decode('utf-8') for ingestor in active_ingestors]
+        ingestors = []
+
+        for ingestor in active_ingestors:
+            if isinstance(ingestor, bytes):
+                ingestors.append(ingestor.decode('utf-8'))
+            else:
+                ingestors.append(ingestor)
+
+        return ingestors
 
     @activity.defn(name="update_slots")
     async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:

orchestrator/utils/orchestrator_functions.py

@@ -65,18 +65,18 @@ def process_path_priority(path_priority: list[str]):
 def predictions_batch(config: dict[str, Any]):
     tags = {}
     for tag in config['write_tags']:
-        if tag['server_name'] not in tags:
+        if tag['server_id'] not in tags:
-            tags[tag['server_name']] = {}
+            tags[tag['server_id']] = {}
 
         tag_type = tag['type']
 
         if tag_type == 'prediction' or tag_type == 'confidence':
             tag_type_str = f"{tag_type}_tags"
 
-            if tag_type_str not in tags[tag['server_name']]:
+            if tag_type_str not in tags[tag['server_id']]:
-                tags[tag['server_name']][tag_type_str] = {}
+                tags[tag['server_id']][tag_type_str] = {}
 
-            tags[tag['server_name']][tag_type_str][tag['addr']] = {
+            tags[tag['server_id']][tag_type_str][tag['addr']] = {
                 "data_type": tag.get('data_type', 'float'),
             }
 
@@ -128,7 +128,7 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]:
 
     # Get all read tags from pipelines
     for pipeline in pipelines:
-        for tag in pipeline['read_tags']:
+        for tag in pipeline.get('read_tags', []):
             tag_string = f"{tag['server_id']}:{tag['tag_address']}"
             if tag_string not in tags:
                 tags[tag_string] = {

samples.json

@@ -57,6 +57,65 @@
               }
             ],
             "tag_retention_minutes": 60
+          },
+        "2": {
+            "schedule_name": "laborious-orchestrated-pipeline",
+            "model_id": "1",
+            "workflow_type": "predictions_batch",
+            "frequency": "30s",
+            "max_retry_policy": 1,
+            "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
+            "retention_time": 60,
+            "write_tags": [
+              {
+                "server_id": "1",
+                "type": "prediction",
+                "addr": "ns=2;i=2",
+                "data_type": "float"
+              },
+              {
+                "server_id": "1",
+                "type": "confidence",
+                "addr": "ns=2;i=2",
+                "data_type": "float"
+              }
+            ],
+            "input_filters": [
+              {
+                "filter_name": "EMPTY_DATA",
+                "policy": "STOP"
+              },
+              {
+                "filter_name": "SPECIFIC_VARIABLES_NULL_VALUES",
+                "policy": "CONTINUE",
+                "config": {
+                  "variables": [
+                    "Counter"
+                  ]
+                }
+              }
+            ],
+            "mlflow_transform_filters": [
+              {
+                "filter_name": "API_ERROR",
+                "policy": "REPEAT"
+              },
+              {
+                "filter_name": "NAN_VALUES",
+                "policy": "STOP"
+              }
+            ],
+            "mlflow_predict_filters": [
+              {
+                "filter_name": "API_ERROR",
+                "policy": "CONTINUE"
+              }
+            ],
+            "path_priority": [
+              "STOP",
+              "CONTINUE",
+              "REPEAT"
+            ]
           }
     },
     "opc-servers": {

tests/orchestrator/activities/test_slot_manager.py

@@ -76,7 +76,7 @@ async def test_load_opc_slots_no_decode(slot_manager):
 @mark.asyncio
 async def test_load_active_ingestors(slot_manager):
     slot_manager.redis_client.keys.return_value = [
-        b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", b"heartbeat:ingestor:3"]
+        b"heartbeat:ingestor:1", b"heartbeat:ingestor:2", "heartbeat:ingestor:3"]
 
     response = await slot_manager.load_active_ingestors()
 

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -139,12 +139,12 @@ def test_predictions_batch(mock_process_path_priority,
         "query": "test_query",
         "write_tags": [
             {
-                "server_name": "test_server_name",
+                "server_id": "test_server_id",
                 "type": "prediction",
                 "addr": "test_addr"
             },
             {
-                "server_name": "test_server_name",
+                "server_id": "test_server_id",
                 "type": "confidence",
                 "addr": "test_addr"
             }
@@ -210,7 +210,7 @@ def test_predictions_batch(mock_process_path_priority,
         "table_name": "predictions",
         "retention_time": 60 * 60,
         "opc_output_config": {
-            "test_server_name": {
+            "test_server_id": {
                 "prediction_tags": {
                     "test_addr": {
                         "data_type": "float"

values.yaml

@@ -123,7 +123,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1030-implementar-os-workflows-mapeados-utilizando-as-workers-e-activities-apropriadas"
+    value: "main"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"
 
@@ -176,4 +176,4 @@ ssh:
 # kubectl create secret generic git-ssh-key-sientia-orchestrator-worker \
 #   --namespace sientia \
 #   --from-file=ssh-privatekey=git_key \
-#   --type=kubernetes.io/ssh-auth
+#   --type=kubernetes.io/ssh-auth