scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 89

emit lock acquired and released events

Browse source
This commit is contained in:
Ishaan Jaff 2025-04-02 17:40:25 -07:00
parent e09ef4afc7
commit 73bbd0a446
1 changed files with 36 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
litellm/proxy/db/db_transaction_queue/pod_lock_manager.py
View file

@ -1,15 +1,20 @@
import asyncio
import uuid
from typing import TYPE_CHECKING, Any, Optional
from litellm._logging import verbose_proxy_logger
from litellm._service_logger import ServiceLogging
from litellm.caching.redis_cache import RedisCache
from litellm.constants import DEFAULT_CRON_JOB_LOCK_TTL_SECONDS
from litellm.types.services import ServiceTypes
if TYPE_CHECKING:
ProxyLogging = Any
else:
ProxyLogging = Any
service_logger_obj = ServiceLogging() # used for tracking current pod lock status
class PodLockManager:
"""
@ -57,6 +62,7 @@ class PodLockManager:
self.pod_id,
self.cronjob_id,
)
return True
else:
# Check if the current pod already holds the lock
@ -70,6 +76,7 @@ class PodLockManager:
self.pod_id,
self.cronjob_id,
)
self._emit_acquired_lock_event(self.cronjob_id, self.pod_id)
return True
return False
except Exception as e:
@ -104,6 +111,7 @@ class PodLockManager:
self.pod_id,
self.cronjob_id,
)
self._emit_released_lock_event(self.cronjob_id, self.pod_id)
else:
verbose_proxy_logger.debug(
"Pod %s failed to release Redis lock for cronjob_id=%s",
@ -127,3 +135,31 @@ class PodLockManager:
verbose_proxy_logger.error(
f"Error releasing Redis lock for {self.cronjob_id}: {e}"
)
@staticmethod
def _emit_acquired_lock_event(cronjob_id: str, pod_id: str):
asyncio.create_task(
service_logger_obj.async_service_success_hook(
service=ServiceTypes.POD_LOCK_MANAGER,
duration=DEFAULT_CRON_JOB_LOCK_TTL_SECONDS,
call_type="_emit_acquired_lock_event",
event_metadata={
"gauge_labels": f"{cronjob_id}:{pod_id}",
"gauge_value": 1,
},
)
)
@staticmethod
def _emit_released_lock_event(cronjob_id: str, pod_id: str):
asyncio.create_task(
service_logger_obj.async_service_success_hook(
service=ServiceTypes.POD_LOCK_MANAGER,
duration=DEFAULT_CRON_JOB_LOCK_TTL_SECONDS,
call_type="_emit_released_lock_event",
event_metadata={
"gauge_labels": f"{cronjob_id}:{pod_id}",
"gauge_value": 0,
},
)
)