gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1150

Browse Source 
refactor: update logging in Couchbase activity to use f-strings for improved readability; modify orchestration workflow to conditionally execute report handlers and replace method calls for consistency
This commit is contained in:
vitor-aignosi
2025-07-15 10:03:04 -03:00
parent 499eae44d9
commit 2533ea882f
5 changed files with 114 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
orchestrator/activities/couchbase.py
View File

@@ -47,7 +47,7 @@ class Couchbase(BaseActivity):
try: try:
self.cluster.close() self.cluster.close()
except Exception as e: except Exception as e:
self.logger.error("Failed to close Couchbase connection: %s", e) self.logger.error(f"Failed to close Couchbase connection: {e}")
def __del__(self): def __del__(self):
self.shutdown() self.shutdown()
@@ -65,7 +65,7 @@ class Couchbase(BaseActivity):
""" """
query = input_data['query'] query = input_data['query']
self.logger.info("Executing couchbase query: %s", query) self.logger.info(f"Executing couchbase query: {query}")
try: try:
result = self.cluster.query(query) result = self.cluster.query(query)

9
orchestrator/workflows/orchestrator.py
View File

@@ -40,7 +40,7 @@ class Orchestrator:
} }
}, },
retry_policy=retry_policy, retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=600) start_to_close_timeout=timedelta(seconds=60)
) )
current_slot_config_handler = workflow.start_local_activity_method( current_slot_config_handler = workflow.start_local_activity_method(
@@ -236,8 +236,11 @@ class Orchestrator:
start_to_close_timeout=timedelta(seconds=60) start_to_close_timeout=timedelta(seconds=60)
) )
await schedule_report_handler if schedule_insertion_report or schedule_update_report or schedule_deletion_report:
await slot_report_handler await schedule_report_handler
if slot_insertion_report or slot_deletion_report:
await slot_report_handler
if schedule_update_report: if schedule_update_report:
await update_pipelines_timestamps_handler await update_pipelines_timestamps_handler

20
tests/orchestrator/activities/test_formatters.py
View File

@@ -18,9 +18,9 @@ def formatters():
@mark.asyncio @mark.asyncio
@patch("orchestrator.activities.formatters.scouter", @patch("orchestrator.activities.formatters.scouter",
return_value="test_scouter") return_value={"test_scouter": "test_scouter"})
@patch("orchestrator.activities.formatters.predictions_batch", @patch("orchestrator.activities.formatters.predictions_batch",
return_value="test_predictions_batch") return_value={"test_predictions_batch": "test_predictions_batch"})
async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters): async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters):
input_data = { input_data = {
"pipelines": [ "pipelines": [
@@ -28,13 +28,15 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
"schedule_name": "test_schedule_name", "schedule_name": "test_schedule_name",
"workflow_type": "scouter", "workflow_type": "scouter",
"model_name": "test_model_name", "model_name": "test_model_name",
"model_id": "test_model_id" "model_id": "test_model_id",
"updated_at": "2021-01-01"
}, },
{ {
"schedule_name": "test_schedule_name2", "schedule_name": "test_schedule_name2",
"workflow_type": "predictions_batch", "workflow_type": "predictions_batch",
"model_name": "test_model_name", "model_name": "test_model_name",
"model_id": "test_model_id" "model_id": "test_model_id",
"updated_at": "2021-01-02"
} }
] ]
} }
@@ -43,10 +45,16 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
assert result == { assert result == {
"scouter": { "scouter": {
"test_schedule_name": "test_scouter" "test_schedule_name": {
"test_scouter": "test_scouter",
"updated_at": "2021-01-01"
}
}, },
"laborious": { "laborious": {
"test_schedule_name2": "test_predictions_batch" "test_schedule_name2": {
"test_predictions_batch": "test_predictions_batch",
"updated_at": "2021-01-02"
}
} }
} }

6
tests/orchestrator/utils/test_connectors_config.py
View File

@@ -36,8 +36,8 @@ def test_build_redis_config_with_defaults():
assert build_redis_config() == { assert build_redis_config() == {
'host': 'localhost', 'host': 'localhost',
'port': 6379, 'port': 6379,
'username': None, 'username': 'default',
'password': None 'password': 'bdnZOpcyiL'
} }
@@ -71,7 +71,7 @@ def test_build_mongo_db_config_with_defaults():
environ.pop('MONGODB_URL', None) environ.pop('MONGODB_URL', None)
assert build_mongodb_config() == { assert build_mongodb_config() == {
'connection_string': 'mongodb://sientia:sientia@localhost:27017', 'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018',
'database_name': 'sientia' 'database_name': 'sientia'
} }

120
tests/orchestrator/workflows/test_orchestrator.py
View File

@@ -20,7 +20,7 @@ async def test_run(workflow_mock, orchestrator):
await orchestrator.run(input_data) await orchestrator.run(input_data)
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.aggregate_documents_in_mongodb, Activities.aggregate_documents_in_mongodb,
{ {
@@ -31,7 +31,7 @@ async def test_run(workflow_mock, orchestrator):
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.find_documents_in_mongodb, Activities.find_documents_in_mongodb,
{ {
@@ -42,7 +42,7 @@ async def test_run(workflow_mock, orchestrator):
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.find_documents_in_mongodb, Activities.find_documents_in_mongodb,
{ {
@@ -55,7 +55,7 @@ async def test_run(workflow_mock, orchestrator):
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.load_opc_slots, Activities.load_opc_slots,
retry_policy=ANY, retry_policy=ANY,
@@ -63,7 +63,7 @@ async def test_run(workflow_mock, orchestrator):
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.load_active_ingestors, Activities.load_active_ingestors,
retry_policy=ANY, retry_policy=ANY,
@@ -71,130 +71,188 @@ async def test_run(workflow_mock, orchestrator):
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.format_schedule_config, Activities.format_schedule_config,
{ {
'schedule_config': workflow_mock.execute_local_activity_method.return_value 'schedule_config': workflow_mock.start_local_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.process_schedules, Activities.process_schedules,
{ {
'pipelines': workflow_mock.execute_local_activity_method.return_value 'pipelines': workflow_mock.start_local_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.process_slots, Activities.process_slots,
{ {
'opc_servers': workflow_mock.execute_local_activity_method.return_value, 'opc_servers': workflow_mock.start_local_activity_method.return_value,
'active_ingestors': workflow_mock.execute_local_activity_method.return_value, 'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.execute_local_activity_method.return_value, 'pipelines': workflow_mock.start_local_activity_method.return_value,
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.create_schedule_config, Activities.create_schedule_config,
{ {
'current_schedule_config': workflow_mock.execute_local_activity_method.return_value, 'current_schedule_config': workflow_mock.start_local_activity_method.return_value,
'schedule_config': workflow_mock.execute_local_activity_method.return_value 'schedule_config': workflow_mock.start_local_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.create_slot_config, Activities.create_slot_config,
{ {
'current_slot_config': workflow_mock.execute_local_activity_method.return_value, 'current_slot_config': workflow_mock.start_local_activity_method.return_value,
'slot_config': workflow_mock.execute_local_activity_method.return_value 'slot_config': workflow_mock.start_local_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_local_activity_method.assert_has_calls([ workflow_mock.start_local_activity_method.assert_has_calls([
call( call(
Activities.normalize_schedules, Activities.normalize_schedules,
{ {
'orchestrated_schedules': workflow_mock.execute_local_activity_method.return_value 'orchestrated_schedules': workflow_mock.start_local_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.delete_slots, Activities.delete_slots,
{ {
'to_delete': 'to_delete':
workflow_mock.execute_local_activity_method.return_value['to_delete'] workflow_mock.start_local_activity_method.return_value['to_delete']
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.update_slots, Activities.update_slots,
{ {
'to_insert': 'to_insert':
workflow_mock.execute_local_activity_method.return_value['to_insert'] workflow_mock.start_local_activity_method.return_value['to_insert']
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.delete_schedules, Activities.delete_schedules,
{ {
'schedules': 'schedules':
workflow_mock.execute_local_activity_method.return_value['to_delete'] workflow_mock.start_local_activity_method.return_value['to_delete']
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.create_schedules, Activities.create_schedules,
{ {
'schedules': 'schedules':
workflow_mock.execute_local_activity_method.return_value['to_create'] workflow_mock.start_local_activity_method.return_value['to_create']
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )
]) ])
workflow_mock.execute_activity_method.assert_has_calls([ workflow_mock.start_activity_method.assert_has_calls([
call( call(
Activities.update_schedules, Activities.update_schedules,
{ {
'schedules': 'schedules':
workflow_mock.execute_local_activity_method.return_value['to_update'] workflow_mock.start_local_activity_method.return_value['to_update']
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.report_schedule_orchestration,
{
'created_schedules': workflow_mock.start_activity_method.return_value,
'updated_schedules': workflow_mock.start_activity_method.return_value,
'deleted_schedules': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.report_slot_orchestration,
{
'inserted_slots': workflow_mock.start_activity_method.return_value,
'deleted_slots': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.update_pipelines_timestamps,
{
'updated_pipelines': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.delete_pipelines_timestamps,
{
'deleted_pipelines': workflow_mock.start_activity_method.return_value
},
retry_policy=ANY,
start_to_close_timeout=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.create_pipelines_timestamps,
{
'created_pipelines': workflow_mock.start_activity_method.return_value
}, },
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY