SIENTIAPDE-1273

Implement drift monitoring configuration and update formatters to support drift workflows. Add drift function to orchestrator utilities and enhance tests for drift functionality.

orchestrator/activities/formatters.py

@@ -19,6 +19,7 @@ with workflow.unsafe.imports_passed_through():
         minimal_retrain,
         predictions_batch,
         scouter,
+        drift,
     )
 
 topic_separator = '\n ========== \n'
@@ -34,7 +35,7 @@ class Formatters(SientiaMonitoring):
     scheduled reports.
 
     Key features:
-    - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain")
+    - Pipeline schedule configuration formatting ("scouter", "predictions_batch", "minimal_retrain", "drift")
     - OPC slot distribution across active ingestors
     - Notification filtering for comprehensive scheduled reports
     - Group-based report filtering with ignore list support
@@ -132,6 +133,13 @@ class Formatters(SientiaMonitoring):
                         'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
                     ),
                 }
+            elif pipeline['workflow_type'] == 'drift':
+                schedule_config[self.laborious_namespace][pipeline['schedule_name']] = {
+                    **drift(pipeline),
+                    'updated_at': pipeline.get(
+                        'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ)
+                    ),
+                }
 
         self.info('Processed schedules', metadata=metadata)
         self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata)

orchestrator/utils/orchestrator_functions.py

@@ -32,6 +32,25 @@ def common_config(config: dict[str, Any]):
     }
 
 
+def drift(config: dict[str, Any]):
+    """
+    Build drift configuration from pipeline config.
+    """
+    model = config['model']
+    return {
+        **common_config(config),
+        'schema': 'sientia_data',
+        'source_table_name': 'laborious_data',
+        'target_table_name': 'drift_metrics',
+        'interval': config.get('interval_minutes', 60),
+        'drift_metrics': config.get('drift_metrics', [
+            "kolmogorov_smirnov",
+            "jensen_shannon",
+            "wasserstein"
+        ])
+    }
+
+
 def minimal_retrain(config: dict[str, Any]):
     """
     Build minimal retrain configuration from pipeline config.
@@ -47,8 +66,6 @@ def minimal_retrain(config: dict[str, Any]):
     """
     return {
         **common_config(config),
-        'workflow_type': 'minimal_retrain',
-        'schedule_name': config['schedule_name'],
         'query': config['query'],
         'schema': 'sientia_data',
         'table_name': 'log_retrain',

tests/orchestrator/activities/test_formatters.py

@@ -44,8 +44,12 @@ metadata = {
     'orchestrator.activities.formatters.minimal_retrain',
     return_value={'test_minimal_retrain': 'test_minimal_retrain'},
 )
+@patch(
+    'orchestrator.activities.formatters.drift',
+    return_value={'test_drift': 'test_drift'},
+)
 async def test_process_schedules(
-    mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters
+    mock_drift, mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters
 ):
     input_data = {
         'pipelines': [
@@ -70,6 +74,13 @@ async def test_process_schedules(
                 'model_id': 'test_model_id',
                 'updated_at': '2021-01-03',
             },
+            {
+                'schedule_name': 'test_schedule_name4',
+                'workflow_type': 'drift',
+                'model_name': 'test_model_name',
+                'model_id': 'test_model_id',
+                'updated_at': '2021-01-04',
+            },
         ]
     }
 
@@ -88,11 +99,17 @@ async def test_process_schedules(
                 'test_minimal_retrain': 'test_minimal_retrain',
                 'updated_at': '2021-01-03',
             },
+            'test_schedule_name4': {
+                'test_drift': 'test_drift',
+                'updated_at': '2021-01-04',
+            },
         },
     }
 
     mock_scouter.assert_called_once_with(input_data['pipelines'][0])
     mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1])
+    mock_minimal_retrain.assert_called_once_with(input_data['pipelines'][2])
+    mock_drift.assert_called_once_with(input_data['pipelines'][3])
 
 
 @mark.asyncio

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -3,6 +3,7 @@ from unittest.mock import call, patch
 from orchestrator.utils.orchestrator_functions import (
     build_tag_config,
     common_config,
+    drift,
     gather_read_tags,
     minimal_retrain,
     overlap_filter_config,
@@ -35,6 +36,36 @@ def test_common_config():
     assert result == expected
 
 
+def test_drift():
+    config = {
+        'workflow_type': 'drift',
+        'schedule_name': 'test_schedule',
+        'model_id': 'test_model_id',
+        'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}},
+        'interval_minutes': 120,
+        'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
+    }
+    result = drift(config)
+    expected = {
+        'workflow_type': 'drift',
+        'schedule_name': 'test_schedule',
+        'frequency': '1m',
+        'offset': '0m',
+        'max_retry_policy': 1,
+        'model_id': 'test_model_id',
+        'model_name': 'test_model_name',
+        'model_config': {'test_config': 'test_config'},
+        'schema': 'sientia_data',
+        'source_table_name': 'laborious_data',
+        'target_table_name': 'drift_metrics',
+        'interval': 120,
+        'drift_metrics': ['kolmogorov_smirnov', 'jensen_shannon'],
+        'execution_timeout_seconds': 300,
+        'task_timeout_seconds': 300,
+    }
+    assert result == expected
+
+
 def test_minimal_retrain():
     config = {
         'workflow_type': 'minimal_retrain',