Merge pull request #2210 from BerriAI/litellm_use_clickhouse_viewing_logs

[FEAT] Admin UI  - View /spend/logs from clickhouse data

cookbook/misc/clickhouse_insert_logs.py

@@ -0,0 +1,39 @@
+# insert data into clickhouse
+# response = client.command(
+#     """
+#     CREATE TEMPORARY TABLE temp_spend_logs AS (
+#     SELECT
+#         generateUUIDv4() AS request_id,
+#         arrayElement(['TypeA', 'TypeB', 'TypeC'], rand() % 3 + 1) AS call_type,
+#         'ishaan' as api_key,
+#         rand() * 1000 AS spend,
+#         rand() * 100 AS total_tokens,
+#         rand() * 50 AS prompt_tokens,
+#         rand() * 50 AS completion_tokens,
+#         toDate('2024-02-01') + toIntervalDay(rand()%27) AS startTime,
+#         now() AS endTime,
+#         arrayElement(['azure/gpt-4', 'gpt-3.5', 'vertexai/gemini-pro', 'mistral/mistral-small', 'ollama/llama2'], rand() % 3 + 1) AS model,
+#         'ishaan-insert-rand' as user,
+#         'data' as metadata,
+#         'true'AS cache_hit,
+#         'ishaan' as cache_key,
+#         '{"tag1": "value1", "tag2": "value2"}' AS request_tags
+#     FROM numbers(1, 1000000)
+#     );
+#     """
+# )
+
+# client.command(
+#     """
+#     -- Insert data into spend_logs table
+#     INSERT INTO spend_logs
+#     SELECT * FROM temp_spend_logs;
+#     """
+# )
+
+
+# client.command(
+#     """
+#     DROP TABLE IF EXISTS temp_spend_logs;
+#     """
+# )

enterprise/utils.py

@@ -1,4 +1,5 @@
 # Enterprise Proxy Util Endpoints
+from litellm._logging import verbose_logger
 
 
 async def get_spend_by_tags(start_date=None, end_date=None, prisma_client=None):
@@ -14,3 +15,98 @@ async def get_spend_by_tags(start_date=None, end_date=None, prisma_client=None):
     )
 
     return response
+
+
+async def view_spend_logs_from_clickhouse(
+    api_key=None, user_id=None, request_id=None, start_date=None, end_date=None
+):
+    verbose_logger.debug("Reading logs from Clickhouse")
+    import os
+
+    # if user has setup clickhouse
+    # TODO: Move this to be a helper function
+    # querying clickhouse for this data
+    import clickhouse_connect
+    from datetime import datetime
+
+    port = os.getenv("CLICKHOUSE_PORT")
+    if port is not None and isinstance(port, str):
+        port = int(port)
+
+    client = clickhouse_connect.get_client(
+        host=os.getenv("CLICKHOUSE_HOST"),
+        port=port,
+        username=os.getenv("CLICKHOUSE_USERNAME", ""),
+        password=os.getenv("CLICKHOUSE_PASSWORD", ""),
+    )
+    if (
+        start_date is not None
+        and isinstance(start_date, str)
+        and end_date is not None
+        and isinstance(end_date, str)
+    ):
+        # Convert the date strings to datetime objects
+        start_date_obj = datetime.strptime(start_date, "%Y-%m-%d")
+        end_date_obj = datetime.strptime(end_date, "%Y-%m-%d")
+
+        # get top spend per day
+        response = client.query(
+            f"""
+                SELECT
+                    toDate(startTime) AS day,
+                    sum(spend) AS total_spend
+                FROM
+                    spend_logs
+                WHERE
+                    toDate(startTime) BETWEEN toDate('2024-02-01') AND toDate('2024-02-29')
+                GROUP BY
+                    day
+                ORDER BY
+                    total_spend
+                """
+        )
+
+        results = []
+        result_rows = list(response.result_rows)
+        for response in result_rows:
+            current_row = {}
+            current_row["users"] = {"example": 0.0}
+            current_row["models"] = {}
+
+            current_row["spend"] = float(response[1])
+            current_row["startTime"] = str(response[0])
+
+            # stubbed api_key
+            current_row[""] = 0.0  # type: ignore
+            results.append(current_row)
+
+        return results
+    else:
+        # check if spend logs exist, if it does then return last 10 logs, sorted in descending order of startTime
+        response = client.query(
+            """
+                SELECT
+                    *
+                FROM
+                    default.spend_logs
+                ORDER BY
+                    startTime DESC
+                LIMIT
+                    10
+            """
+        )
+
+        # get size of spend logs
+        num_rows = client.query("SELECT count(*) FROM default.spend_logs")
+        num_rows = num_rows.result_rows[0][0]
+
+        # safely access num_rows.result_rows[0][0]
+        if num_rows is None:
+            num_rows = 0
+
+        raw_rows = list(response.result_rows)
+        response_data = {
+            "logs": raw_rows,
+            "log_count": num_rows,
+        }
+        return response_data

litellm/integrations/clickhouse.py

@@ -85,12 +85,6 @@ def _start_clickhouse():
             # check if spend logs exist, if it does then return the schema
             response = client.query("DESCRIBE default.spend_logs")
             verbose_logger.debug(f"spend logs schema ={response.result_rows}")
-            # get all logs from spend logs
-            response = client.query("SELECT * FROM default.spend_logs")
-            verbose_logger.debug(f"spend logs ={response.result_rows}")
-            # get size of spend logs
-            response = client.query("SELECT count(*) FROM default.spend_logs")
-            verbose_logger.debug(f"spend logs count ={response.result_rows}")
 
 
 class ClickhouseLogger:

litellm/proxy/proxy_server.py

@@ -3782,6 +3782,17 @@ async def view_spend_logs(
 -H "Authorization: Bearer sk-1234"
     ```
     """
+    if os.getenv("CLICKHOUSE_HOST") is not None:
+        # gettting spend logs from clickhouse
+        from litellm.proxy.enterprise.utils import view_spend_logs_from_clickhouse
+
+        return await view_spend_logs_from_clickhouse(
+            api_key=api_key,
+            user_id=user_id,
+            request_id=request_id,
+            start_date=start_date,
+            end_date=end_date,
+        )
     global prisma_client
     try:
         verbose_proxy_logger.debug("inside view_spend_logs")
@@ -6005,7 +6016,6 @@ async def health_readiness():
             except Exception as e:
                 index_info = "index does not exist - error: " + str(e)
             cache_type = {"type": cache_type, "index_info": index_info}
-
     if prisma_client is not None:  # if db passed in, check if it's connected
         await prisma_client.health_check()  # test the db connection
         response_object = {"db": "connected"}