diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index af7468ba14..6ca28cd7e2 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -43,6 +43,7 @@ else: async_redis_cluster_client = Any Span = Any +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER def _get_call_stack_info(num_frames: int = 2) -> str: """ @@ -307,8 +308,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_scan_iter <- {_get_call_stack_info()}", @@ -322,8 +323,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_failure_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, error=e, @@ -373,8 +374,8 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_failure_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, error=e, @@ -410,8 +411,8 @@ class RedisCache(BaseCache): ) end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_set_cache <- {_get_call_stack_info()}", @@ -499,8 +500,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_set_cache_pipeline <- {_get_call_stack_info()}", @@ -590,8 +591,8 @@ class RedisCache(BaseCache): ) end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_set_cache_sadd <- {_get_call_stack_info()}", @@ -656,8 +657,8 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_increment <- {_get_call_stack_info()}", @@ -826,8 +827,8 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_get_cache <- {_get_call_stack_info()}", @@ -887,8 +888,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_batch_get_cache <- {_get_call_stack_info()}", @@ -974,8 +975,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_ping <- {_get_call_stack_info()}", @@ -1087,8 +1088,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_increment_pipeline <- {_get_call_stack_info()}", @@ -1167,8 +1168,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_rpush <- {_get_call_stack_info()}", @@ -1238,8 +1239,8 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_lpop <- {_get_call_stack_info()}", diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index e077d0ee92..8f98627542 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -1,4 +1,3 @@ -import asyncio import copy import time from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union @@ -45,6 +44,8 @@ else: ProxyConfig = Any +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER + def parse_cache_control(cache_control): cache_dict = {} directives = cache_control.split(", ") @@ -1011,9 +1012,9 @@ async def add_litellm_data_to_request( # noqa: PLR0915 premium_user=premium_user, ) - end_time = time.time() - asyncio.create_task( - service_logger_obj.async_service_success_hook( + end_time = time.time() + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=service_logger_obj.async_service_success_hook( service=ServiceTypes.PROXY_PRE_CALL, duration=end_time - start_time, call_type="add_litellm_data_to_request", diff --git a/litellm/router.py b/litellm/router.py index a1cfc22e7d..6703ca0ad1 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -34,6 +34,7 @@ from typing import ( cast, ) +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER import httpx import openai from openai import AsyncOpenAI @@ -1059,9 +1060,9 @@ class Router: else: response = await self.async_function_with_fallbacks(**kwargs) end_time = time.time() - _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + _duration = end_time - start_time + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.ROUTER, duration=_duration, call_type="acompletion", @@ -1247,8 +1248,8 @@ class Router: _timeout_debug_deployment_dict = deployment end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.ROUTER, duration=_duration, call_type="async_get_available_deployment", @@ -7042,8 +7043,8 @@ class Router: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( - self.service_logger_obj.async_service_success_hook( + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( + async_coroutine=self.service_logger_obj.async_service_success_hook( service=ServiceTypes.ROUTER, duration=_duration, call_type=".async_get_available_deployments",