From 08cf77623bdbc1f684519f1e2b4f9579ca3102b0 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 14:01:13 -0700 Subject: [PATCH 1/6] fix(caching.py): remove url parsing logic - causing redis ssl connections to fail this reverts a change that was causing redis url w/ ssl to fail. this also adds unit testing for this sc enario, to prevent future regressions --- litellm/caching.py | 33 ++++++++++++---- litellm/integrations/prometheus.py | 2 +- litellm/proxy/_new_secret_config.yaml | 8 ++-- litellm/tests/test_router_caching.py | 55 +++++++++++++++++++++++++++ 4 files changed, 84 insertions(+), 14 deletions(-) diff --git a/litellm/caching.py b/litellm/caching.py index d73112d21ce..87d99587f82 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -154,13 +154,6 @@ class RedisCache(BaseCache): self.redis_kwargs = redis_kwargs self.async_redis_conn_pool = get_redis_connection_pool(**redis_kwargs) - if "url" in redis_kwargs and redis_kwargs["url"] is not None: - parsed_kwargs = redis.connection.parse_url(redis_kwargs["url"]) - redis_kwargs.update(parsed_kwargs) - self.redis_kwargs.update(parsed_kwargs) - # pop url - self.redis_kwargs.pop("url") - # redis namespaces self.namespace = namespace # for high traffic, we store the redis results in memory and then batch write to redis @@ -175,6 +168,12 @@ class RedisCache(BaseCache): ### HEALTH MONITORING OBJECT ### self.service_logger_obj = ServiceLogging() + ### ASYNC HEALTH PING ### + try: + asyncio.get_running_loop().create_task(self.ping()) + except Exception: + pass + def init_async_client(self): from ._redis import get_redis_async_client @@ -601,13 +600,31 @@ class RedisCache(BaseCache): print_verbose(f"Error occurred in pipeline read - {str(e)}") return key_value_dict - async def ping(self): + def sync_ping(self) -> bool: + """ + Tests if the sync redis client is correctly setup. + """ + print_verbose(f"Pinging Async Redis Cache") + try: + response = self.redis_client.ping() + print_verbose(f"Redis Cache PING: {response}") + return response + except Exception as e: + # NON blocking - notify users Redis is throwing an exception + print_verbose( + f"LiteLLM Redis Cache PING: - Got exception from REDIS : {str(e)}" + ) + traceback.print_exc() + raise e + + async def ping(self) -> bool: _redis_client = self.init_async_client() async with _redis_client as redis_client: print_verbose(f"Pinging Async Redis Cache") try: response = await redis_client.ping() print_verbose(f"Redis Cache PING: {response}") + return response except Exception as e: # NON blocking - notify users Redis is throwing an exception print_verbose( diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index 7943d5dba90..74632d49a06 100644 --- a/litellm/integrations/prometheus.py +++ b/litellm/integrations/prometheus.py @@ -67,7 +67,7 @@ class PrometheusLogger: # unpack kwargs model = kwargs.get("model", "") - response_cost = kwargs.get("response_cost", 0.0) + response_cost = kwargs.get("response_cost", 0.0) or 0 litellm_params = kwargs.get("litellm_params", {}) or {} proxy_server_request = litellm_params.get("proxy_server_request") or {} end_user_id = proxy_server_request.get("body", {}).get("user", None) diff --git a/litellm/proxy/_new_secret_config.yaml b/litellm/proxy/_new_secret_config.yaml index 0f7c24576ee..0f7fdd132d0 100644 --- a/litellm/proxy/_new_secret_config.yaml +++ b/litellm/proxy/_new_secret_config.yaml @@ -3,8 +3,8 @@ model_list: litellm_params: model: openai/my-fake-model api_key: my-fake-key - # api_base: https://openai-function-calling-workers.tasslexyz.workers.dev/ - api_base: http://0.0.0.0:8080 + api_base: https://openai-function-calling-workers.tasslexyz.workers.dev/ + # api_base: http://0.0.0.0:8080 stream_timeout: 0.001 rpm: 10 - litellm_params: @@ -33,9 +33,7 @@ litellm_settings: router_settings: routing_strategy: usage-based-routing-v2 - redis_host: os.environ/REDIS_HOST - redis_password: os.environ/REDIS_PASSWORD - redis_port: os.environ/REDIS_PORT + redis_url: "rediss://:073f655645b843c4839329aea8384e68@us1-great-lizard-40486.upstash.io:40486/0" enable_pre_call_checks: True general_settings: diff --git a/litellm/tests/test_router_caching.py b/litellm/tests/test_router_caching.py index 1fb699c1778..3bf68595eeb 100644 --- a/litellm/tests/test_router_caching.py +++ b/litellm/tests/test_router_caching.py @@ -15,6 +15,61 @@ from litellm import Router ## 2. 2 models - openai, azure - 2 diff model groups, 1 caching group +@pytest.mark.asyncio +async def test_router_async_caching_with_ssl_url(): + """ + Tests when a redis url is passed to the router, if caching is correctly setup + """ + try: + router = Router( + model_list=[ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "gpt-3.5-turbo-0613", + "api_key": os.getenv("OPENAI_API_KEY"), + }, + "tpm": 100000, + "rpm": 10000, + }, + ], + redis_url=os.getenv("REDIS_URL"), + ) + + response = await router.cache.redis_cache.ping() + print(f"response: {response}") + assert response == True + except Exception as e: + pytest.fail(f"An exception occurred - {str(e)}") + + +def test_router_sync_caching_with_ssl_url(): + """ + Tests when a redis url is passed to the router, if caching is correctly setup + """ + try: + router = Router( + model_list=[ + { + "model_name": "gpt-3.5-turbo", + "litellm_params": { + "model": "gpt-3.5-turbo-0613", + "api_key": os.getenv("OPENAI_API_KEY"), + }, + "tpm": 100000, + "rpm": 10000, + }, + ], + redis_url=os.getenv("REDIS_URL"), + ) + + response = router.cache.redis_cache.sync_ping() + print(f"response: {response}") + assert response == True + except Exception as e: + pytest.fail(f"An exception occurred - {str(e)}") + + @pytest.mark.asyncio async def test_acompletion_caching_on_router(): # tests acompletion + caching on router From 84685b5f3403605e8b256665bdaa52d376af64f9 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 15:27:11 -0700 Subject: [PATCH 2/6] fix(_redis.py): fix args passed to redis.from_url argument --- litellm/_redis.py | 46 ++++++++++++++++++++++++++++++++++++++-------- litellm/caching.py | 3 +++ 2 files changed, 41 insertions(+), 8 deletions(-) diff --git a/litellm/_redis.py b/litellm/_redis.py index 69ff6f3f2cb..e2688bf418c 100644 --- a/litellm/_redis.py +++ b/litellm/_redis.py @@ -32,6 +32,25 @@ def _get_redis_kwargs(): return available_args +def _get_redis_url_kwargs(client=None): + if client is None: + client = redis.Redis.from_url + arg_spec = inspect.getfullargspec(redis.Redis.from_url) + + # Only allow primitive arguments + exclude_args = { + "self", + "connection_pool", + "retry", + } + + include_args = ["url"] + + available_args = [x for x in arg_spec.args if x not in exclude_args] + include_args + + return available_args + + def _get_redis_env_kwarg_mapping(): PREFIX = "REDIS_" @@ -98,20 +117,31 @@ def _get_redis_client_logic(**env_overrides): def get_redis_client(**env_overrides): redis_kwargs = _get_redis_client_logic(**env_overrides) if "url" in redis_kwargs and redis_kwargs["url"] is not None: - redis_kwargs.pop( - "connection_pool", None - ) # redis.from_url doesn't support setting your own connection pool - return redis.Redis.from_url(**redis_kwargs) + args = _get_redis_url_kwargs() + url_kwargs = {} + for arg in redis_kwargs: + if arg in args: + url_kwargs[arg] = redis_kwargs[arg] + + return redis.Redis.from_url(**url_kwargs) return redis.Redis(**redis_kwargs) def get_redis_async_client(**env_overrides): redis_kwargs = _get_redis_client_logic(**env_overrides) if "url" in redis_kwargs and redis_kwargs["url"] is not None: - redis_kwargs.pop( - "connection_pool", None - ) # redis.from_url doesn't support setting your own connection pool - return async_redis.Redis.from_url(**redis_kwargs) + args = _get_redis_url_kwargs(client=async_redis.Redis.from_url) + url_kwargs = {} + for arg in redis_kwargs: + if arg in args: + url_kwargs[arg] = redis_kwargs[arg] + else: + litellm.print_verbose( + "REDIS: ignoring argument: {}. Not an allowed async_redis.Redis.from_url arg.".format( + arg + ) + ) + return async_redis.Redis.from_url(**url_kwargs) return async_redis.Redis( socket_timeout=5, **redis_kwargs, diff --git a/litellm/caching.py b/litellm/caching.py index 87d99587f82..22e53d6f935 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -174,6 +174,9 @@ class RedisCache(BaseCache): except Exception: pass + ### SYNC HEALTH PING ### + self.redis_client.ping() + def init_async_client(self): from ._redis import get_redis_async_client From 0d9c96bebf3d07703743eb123a54e148edd76aed Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 16:15:29 -0700 Subject: [PATCH 3/6] test(test_prometheus_services.py): fix testing to handle caching ping in init --- litellm/caching.py | 57 ++++++++++++++++++--- litellm/integrations/prometheus_services.py | 1 - litellm/tests/test_prometheus_service.py | 16 ++---- 3 files changed, 56 insertions(+), 18 deletions(-) diff --git a/litellm/caching.py b/litellm/caching.py index 22e53d6f935..a90a3941d98 100644 --- a/litellm/caching.py +++ b/litellm/caching.py @@ -149,6 +149,14 @@ class RedisCache(BaseCache): if password is not None: redis_kwargs["password"] = password + ### HEALTH MONITORING OBJECT ### + if kwargs.get("service_logger_obj", None) is not None and isinstance( + kwargs["service_logger_obj"], ServiceLogging + ): + self.service_logger_obj = kwargs.pop("service_logger_obj") + else: + self.service_logger_obj = ServiceLogging() + redis_kwargs.update(kwargs) self.redis_client = get_redis_client(**redis_kwargs) self.redis_kwargs = redis_kwargs @@ -165,12 +173,10 @@ class RedisCache(BaseCache): except Exception as e: pass - ### HEALTH MONITORING OBJECT ### - self.service_logger_obj = ServiceLogging() - ### ASYNC HEALTH PING ### try: - asyncio.get_running_loop().create_task(self.ping()) + # asyncio.get_running_loop().create_task(self.ping()) + result = asyncio.get_running_loop().create_task(self.ping()) except Exception: pass @@ -607,13 +613,31 @@ class RedisCache(BaseCache): """ Tests if the sync redis client is correctly setup. """ - print_verbose(f"Pinging Async Redis Cache") + print_verbose(f"Pinging Sync Redis Cache") + start_time = time.time() try: response = self.redis_client.ping() print_verbose(f"Redis Cache PING: {response}") + ## LOGGING ## + end_time = time.time() + _duration = end_time - start_time + self.service_logger_obj.service_success_hook( + service=ServiceTypes.REDIS, + duration=_duration, + call_type="sync_ping", + ) return response except Exception as e: # NON blocking - notify users Redis is throwing an exception + ## LOGGING ## + end_time = time.time() + _duration = end_time - start_time + self.service_logger_obj.service_failure_hook( + service=ServiceTypes.REDIS, + duration=_duration, + error=e, + call_type="sync_ping", + ) print_verbose( f"LiteLLM Redis Cache PING: - Got exception from REDIS : {str(e)}" ) @@ -622,14 +646,35 @@ class RedisCache(BaseCache): async def ping(self) -> bool: _redis_client = self.init_async_client() + start_time = time.time() async with _redis_client as redis_client: print_verbose(f"Pinging Async Redis Cache") try: response = await redis_client.ping() - print_verbose(f"Redis Cache PING: {response}") + ## LOGGING ## + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_success_hook( + service=ServiceTypes.REDIS, + duration=_duration, + call_type="async_ping", + ) + ) return response except Exception as e: # NON blocking - notify users Redis is throwing an exception + ## LOGGING ## + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_failure_hook( + service=ServiceTypes.REDIS, + duration=_duration, + error=e, + call_type="async_ping", + ) + ) print_verbose( f"LiteLLM Redis Cache PING: - Got exception from REDIS : {str(e)}" ) diff --git a/litellm/integrations/prometheus_services.py b/litellm/integrations/prometheus_services.py index 4171593bafe..5f4796c4828 100644 --- a/litellm/integrations/prometheus_services.py +++ b/litellm/integrations/prometheus_services.py @@ -30,7 +30,6 @@ class PrometheusServicesLogger: raise Exception( "Missing prometheus_client. Run `pip install prometheus-client`" ) - print("INITIALIZES PROMETHEUS SERVICE LOGGER!") self.Histogram = Histogram self.Counter = Counter diff --git a/litellm/tests/test_prometheus_service.py b/litellm/tests/test_prometheus_service.py index 63ff347d3a4..ec2ffb5eb64 100644 --- a/litellm/tests/test_prometheus_service.py +++ b/litellm/tests/test_prometheus_service.py @@ -67,21 +67,15 @@ async def test_completion_with_caching_bad_call(): litellm.set_verbose = True sl = ServiceLogging(mock_testing=True) try: - litellm.cache = Cache(type="redis", host="hello-world") + from litellm.caching import RedisCache + litellm.service_callback = ["prometheus_system"] - litellm.cache.cache.service_logger_obj = sl - - messages = [{"role": "user", "content": "Hey, how's it going?"}] - response1 = await acompletion( - model="gpt-3.5-turbo", messages=messages, caching=True - ) - response1 = await acompletion( - model="gpt-3.5-turbo", messages=messages, caching=True - ) + RedisCache(host="hello-world", **{"service_logger_obj": sl}) except Exception as e: - pass + print(f"Receives exception = {str(e)}") + await asyncio.sleep(5) assert sl.mock_testing_async_failure_hook > 0 assert sl.mock_testing_async_success_hook == 0 assert sl.mock_testing_sync_success_hook == 0 From 41c81f6335057a6fd063f0df3cd5038ac5b22146 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 16:21:15 -0700 Subject: [PATCH 4/6] fix: fix tests --- litellm/tests/test_prometheus_service.py | 5 +++-- litellm/tests/test_router_caching.py | 4 ++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/litellm/tests/test_prometheus_service.py b/litellm/tests/test_prometheus_service.py index ec2ffb5eb64..1f562aa71ba 100644 --- a/litellm/tests/test_prometheus_service.py +++ b/litellm/tests/test_prometheus_service.py @@ -65,13 +65,14 @@ async def test_completion_with_caching_bad_call(): - Assert failure callback gets called """ litellm.set_verbose = True - sl = ServiceLogging(mock_testing=True) + try: from litellm.caching import RedisCache litellm.service_callback = ["prometheus_system"] + sl = ServiceLogging(mock_testing=True) - RedisCache(host="hello-world", **{"service_logger_obj": sl}) + RedisCache(host="hello-world", service_logger_obj=sl) except Exception as e: print(f"Receives exception = {str(e)}") diff --git a/litellm/tests/test_router_caching.py b/litellm/tests/test_router_caching.py index 3bf68595eeb..ebace161c98 100644 --- a/litellm/tests/test_router_caching.py +++ b/litellm/tests/test_router_caching.py @@ -33,7 +33,7 @@ async def test_router_async_caching_with_ssl_url(): "rpm": 10000, }, ], - redis_url=os.getenv("REDIS_URL"), + redis_url=os.getenv("REDIS_SSL_URL"), ) response = await router.cache.redis_cache.ping() @@ -60,7 +60,7 @@ def test_router_sync_caching_with_ssl_url(): "rpm": 10000, }, ], - redis_url=os.getenv("REDIS_URL"), + redis_url=os.getenv("REDIS_SSL_URL"), ) response = router.cache.redis_cache.sync_ping() From 62aab9118659893309961878617c1126a6387b49 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 16:23:47 -0700 Subject: [PATCH 5/6] test(test_prometheus_service.py): remove duplicate test --- litellm/tests/test_prometheus_service.py | 61 ------------------------ 1 file changed, 61 deletions(-) diff --git a/litellm/tests/test_prometheus_service.py b/litellm/tests/test_prometheus_service.py index 1f562aa71ba..9e3441abb5f 100644 --- a/litellm/tests/test_prometheus_service.py +++ b/litellm/tests/test_prometheus_service.py @@ -139,64 +139,3 @@ async def test_router_with_caching(): except Exception as e: pytest.fail(f"An exception occured - {str(e)}") - - -@pytest.mark.asyncio -async def test_router_with_caching_bad_call(): - """ - - Run completion with caching (incorrect credentials) - - Assert failure callback gets called - """ - try: - - def get_azure_params(deployment_name: str): - params = { - "model": f"azure/{deployment_name}", - "api_key": os.environ["AZURE_API_KEY"], - "api_version": os.environ["AZURE_API_VERSION"], - "api_base": os.environ["AZURE_API_BASE"], - } - return params - - model_list = [ - { - "model_name": "azure/gpt-4", - "litellm_params": get_azure_params("chatgpt-v-2"), - "tpm": 100, - }, - { - "model_name": "azure/gpt-4", - "litellm_params": get_azure_params("chatgpt-v-2"), - "tpm": 1000, - }, - ] - - router = litellm.Router( - model_list=model_list, - set_verbose=True, - debug_level="DEBUG", - routing_strategy="usage-based-routing-v2", - redis_host="hello world", - redis_port=os.environ["REDIS_PORT"], - redis_password=os.environ["REDIS_PASSWORD"], - ) - - litellm.service_callback = ["prometheus_system"] - - sl = ServiceLogging(mock_testing=True) - sl.prometheusServicesLogger.mock_testing = True - router.cache.redis_cache.service_logger_obj = sl - - messages = [{"role": "user", "content": "Hey, how's it going?"}] - try: - response1 = await router.acompletion(model="azure/gpt-4", messages=messages) - response1 = await router.acompletion(model="azure/gpt-4", messages=messages) - except Exception as e: - pass - - assert sl.mock_testing_async_failure_hook > 0 - assert sl.mock_testing_async_success_hook == 0 - assert sl.mock_testing_sync_success_hook == 0 - - except Exception as e: - pytest.fail(f"An exception occured - {str(e)}") From 978c1a19767e7d59df01f7cf3872aa15fdab2ce5 Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Fri, 19 Apr 2024 17:02:15 -0700 Subject: [PATCH 6/6] test(test_caching.py): add sleep --- litellm/tests/test_caching.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/litellm/tests/test_caching.py b/litellm/tests/test_caching.py index ae7bb428930..178d060bd74 100644 --- a/litellm/tests/test_caching.py +++ b/litellm/tests/test_caching.py @@ -390,6 +390,7 @@ async def test_embedding_caching_azure_individual_items_reordered(): @pytest.mark.asyncio async def test_embedding_caching_base_64(): """ """ + litellm.set_verbose = True litellm.cache = Cache( type="redis", host=os.environ["REDIS_HOST"], @@ -408,6 +409,8 @@ async def test_embedding_caching_base_64(): caching=True, encoding_format="base64", ) + await asyncio.sleep(5) + print("\n\nCALL2\n\n") embedding_val_2 = await aembedding( model="azure/azure-embedding-model", input=inputs, @@ -1094,10 +1097,6 @@ def test_custom_redis_cache_params(): port=os.environ["REDIS_PORT"], password=os.environ["REDIS_PASSWORD"], db=0, - ssl=True, - ssl_certfile="./redis_user.crt", - ssl_keyfile="./redis_user_private.key", - ssl_ca_certs="./redis_ca.pem", ) print(litellm.cache.cache.redis_client) @@ -1105,7 +1104,7 @@ def test_custom_redis_cache_params(): litellm.success_callback = [] litellm._async_success_callback = [] except Exception as e: - pytest.fail(f"Error occurred:", e) + pytest.fail(f"Error occurred: {str(e)}") def test_get_cache_key():