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

SIENTIAPDE-1199

Browse Source 
Refactor logging in worker.py to use custom_info and custom_error methods, enhancing log metadata with pod and workflow details for improved observability.
This commit is contained in:
vitor-aignosi
2025-08-21 11:35:26 -03:00
parent 76d96c2c13
commit 799aa5af3e
1 changed files with 20 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
scouter/worker/worker.py
View File

@@ -30,12 +30,20 @@ async def main():
logger = get_logger(__name__) logger = get_logger(__name__)
logger.info(f"Starting Worker with pod_id: {POD_ID}") metadata = {
'pod_id': POD_ID,
'model_name': '-',
'model_id': '-',
'workflow_name': '-',
'schedule_name': '-',
}
logger.info("Starting prometheus client...") logger.custom_info(f"Starting Worker with pod_id: {POD_ID}", metadata)
logger.custom_info("Starting prometheus client...", metadata)
start_prometheus_server() start_prometheus_server()
logger.info('Starting Notification Handler...') logger.custom_info('Starting Notification Handler...', metadata)
mongo_config = build_mongodb_config() mongo_config = build_mongodb_config()
notification_handler = NotificationHandler( notification_handler = NotificationHandler(
@@ -45,7 +53,7 @@ async def main():
project_name=os.getenv('PROJECT_NAME', 'scouter') project_name=os.getenv('PROJECT_NAME', 'scouter')
) )
logger.info('Starting Activities...') logger.custom_info('Starting Activities...', metadata)
activities = Activities( activities = Activities(
logger=logger, logger=logger,
@@ -55,7 +63,7 @@ async def main():
mongodb_config=build_mongodb_config() mongodb_config=build_mongodb_config()
) )
logger.info('Starting Faker Activities...') logger.custom_info('Starting Faker Activities...', metadata)
faker_activities = Faker( faker_activities = Faker(
logger=logger, logger=logger,
@@ -64,7 +72,8 @@ async def main():
'KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092') 'KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
) )
logger.info(f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...') logger.custom_info(
f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata)
new_runtime = Runtime( new_runtime = Runtime(
telemetry=TelemetryConfig( telemetry=TelemetryConfig(
@@ -73,7 +82,7 @@ async def main():
) )
) )
logger.info('Starting Temporal Client...') logger.custom_info('Starting Temporal Client...', metadata)
temporal_client = await client.Client.connect( temporal_client = await client.Client.connect(
target_host=host, target_host=host,
@@ -81,7 +90,7 @@ async def main():
runtime=new_runtime runtime=new_runtime
) )
logger.info('Starting Workers...') logger.custom_info('Starting Workers...', metadata)
workers = [ workers = [
Worker( Worker(
@@ -120,13 +129,14 @@ async def main():
for w in workers: for w in workers:
handlers.append(w.run()) handlers.append(w.run())
logger.info('Workers started successfully') logger.custom_info('Workers started successfully', metadata)
try: try:
await asyncio.gather(*handlers) await asyncio.gather(*handlers)
except BaseException as e: # NOSONAR except BaseException as e: # NOSONAR
logger.error("An unhandled exception occurred: %s", e, exc_info=True) logger.custom_error("An unhandled exception occurred: %s",
e, exc_info=True, metadata=metadata)
finally: finally:
if notification_handler: if notification_handler:
notification_handler.shutdown() notification_handler.shutdown()