diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 85326df..7f668ed 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -31,12 +31,20 @@ async def main(): host = os.getenv('TEMPORAL_HOST', 'localhost:7233') 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() - logger.info('Starting Notification Handler...') + logger.custom_info('Starting Notification Handler...', metadata) mongo_config = build_mongodb_config() notification_handler = NotificationHandler( @@ -46,7 +54,7 @@ async def main(): project_name=os.getenv('PROJECT_NAME', 'laborious') ) - logger.info('Starting Activities...') + logger.custom_info('Starting Activities...', metadata) activities = Activities( postgres_config=build_postgres_config(), @@ -56,7 +64,8 @@ async def main(): notification_handler=notification_handler ) - 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( telemetry=TelemetryConfig( @@ -65,7 +74,7 @@ async def main(): ) ) - logger.info('Starting Temporal Client...') + logger.custom_info('Starting Temporal Client...', metadata) temporal_client = await client.Client.connect( target_host=host, @@ -73,7 +82,7 @@ async def main(): runtime=new_runtime ) - logger.info('Starting Workers...') + logger.custom_info('Starting Workers...', metadata) workers = [ Worker( @@ -130,14 +139,14 @@ async def main(): for w in workers: handlers.append(w.run()) - logger.info('Workers started successfully') + logger.custom_info('Workers started successfully', metadata) try: # This will run the workers and wait for them to complete. # If an exception occurs in any of the worker handlers, it will be propagated here. await asyncio.gather(*handlers) except BaseException as e: # NOSONAR - logger.error(f"An unhandled exception occurred: {e}") + logger.custom_error(f"An unhandled exception occurred: {e}", metadata) finally: if notification_handler: notification_handler.shutdown()