[Bug]: Azure storage callback logs

[ma-armenta]

What happened?

Sending logs over to an Azure storage container doesn't seem to be working as expected / documented. I suspected connection or auth issues, but rule them out with a sample script below, to connect to the container and create / write a new file. It seems to be the client is not closed properly? Haven't dive in the code just yet.

import os
from azure.storage.blob import BlobServiceClient, ContainerClient
from azure.identity import DefaultAzureCredential

def validate_blob_storage_connectivity():
    try:
        # Retrieve the storage account name and key from environment variables
        account_name = os.getenv('AZURE_STORAGE_ACCOUNT_NAME')
        account_key = os.getenv('AZURE_STORAGE_ACCOUNT_KEY')
        if not account_name or not account_key:
            raise ValueError("AZURE_STORAGE_ACCOUNT_NAME or AZURE_STORAGE_ACCOUNT_KEY environment variable not set")

        # Create the BlobServiceClient object using the account name and key
        blob_service_client = BlobServiceClient(
            account_url=f"https://{account_name}.blob.core.windows.net",
            credential=account_key
        )

        # Retrieve the container name from environment variables
        container_name = os.getenv('AZURE_STORAGE_FILE_SYSTEM')
        if not container_name:
            raise ValueError("AZURE_STORAGE_FILE_SYSTEM environment variable not set")

        # Create the ContainerClient object
        container_client = blob_service_client.get_container_client(container_name)

        # Check if the container exists
        if container_client.exists():
            print(f"Successfully connected to the container: {container_name}")
            # Create a new blob called 'test-litellm'
            blob_client = container_client.get_blob_client("test-litellm")
            blob_client.upload_blob("This is a test blob.", overwrite=True)
            print("Created a new blob called 'test-litellm'")
            for b in container_client.list_blobs():
                print(b)
        else:
            print(f"Container {container_name} does not exist")

    except Exception as e:
        print(f"Failed to connect to Azure Blob Storage: {e}")

if __name__ == "__main__":
    validate_blob_storage_connectivity()

Relevant log output

14:08:11 - LiteLLM:ERROR: azure_storage.py:381 - Error occurred: You are attempting to read or stream the content from request <HttpRequest [HEAD], url: 'https://<account>.blob.core.windows.net/<container-name>/2025-03-18'>. You have likely already consumed this stream, so it can not be accessed anymore.
Traceback (most recent call last):
  File "/usr/lib/python3.13/site-packages/litellm/integrations/azure_storage/azure_storage.py", line 356, in upload_to_azure_data_lake_with_azure_account_key
    if not await directory_client.exists():
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/tracing/decorator_async.py", line 114, in wrapper_use_tracer
    return await func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/filedatalake/aio/_data_lake_directory_client_async.py", line 195, in exists
    return await self._exists(**kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/filedatalake/aio/_path_client_async.py", line 760, in _exists
    return await self._blob_client.exists(**kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/tracing/decorator_async.py", line 114, in wrapper_use_tracer
    return await func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/blob/aio/_blob_client_async.py", line 884, in exists
    await self._client.blob.get_properties(
    ...<2 lines>...
        **kwargs)
  File "/usr/lib/python3.13/site-packages/azure/core/tracing/decorator_async.py", line 114, in wrapper_use_tracer
    return await func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/blob/_generated/aio/operations/_blob_operations.py", line 508, in get_properties
    pipeline_response: PipelineResponse = await self._client._pipeline.run(  # pylint: disable=protected-access
                                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
        _request, stream=_stream, **kwargs
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    )
    ^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 229, in run
    return await first_node.send(pipeline_request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 77, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 77, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 77, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  [Previous line repeated 5 more times]
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/policies/_redirect_async.py", line 76, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 77, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/filedatalake/_shared/policies_async.py", line 140, in send
    raise err
  File "/usr/lib/python3.13/site-packages/azure/storage/filedatalake/_shared/policies_async.py", line 114, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/pipeline/_base_async.py", line 77, in send
    response = await self.next.send(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/storage/filedatalake/_shared/policies_async.py", line 67, in send
    await response.http_response.load_body()
  File "/usr/lib/python3.13/site-packages/azure/core/rest/_aiohttp.py", line 165, in _load_body
    self._content = await self.read()  # type: ignore
                    ^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.13/site-packages/azure/core/rest/_aiohttp.py", line 214, in read
    self._stream_download_check()
    ~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
  File "/usr/lib/python3.13/site-packages/azure/core/rest/_http_response_impl.py", line 336, in _stream_download_check
    raise StreamConsumedError(self)
azure.core.exceptions.StreamConsumedError: You are attempting to read or stream the content from request <HttpRequest [HEAD], url: 'https://<account>.blob.core.windows.net/<container-name>/2025-03-18'>. You have likely already consumed this stream, so it can not be accessed anymore.
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f1975a56ad0>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7f1938348410>, 72098.666)]']
connector: <aiohttp.connector.TCPConnector object at 0x7f19507cb360>

Are you a ML Ops Team?

Yes

What LiteLLM version are you on ?

v1.63.11-stable

Twitter / LinkedIn details

https://www.linkedin.com/in/maarmenta/

[rsam444]

@ma-armenta even i have the same issue, is it been fixed @ishaan-jaff ?

[ishaan-jaff]

cc @adrianlyjak any chance you can help take a look at this ?

[adrianlyjak]

@ma-armenta and @rsam444, under what conditions are you getting the error? Are you receiving no logs, or is this intermittent? Like, I'm somewhat wondering if its something that can occur if logs are being stored concurrently. This error looks fairly deep within the azure library

I was able to do a simple test using the litellm lib (not proxy), and am successfully getting logs stored.

import litellm
import asyncio
import dotenv

dotenv.load_dotenv()

async def main():
    litellm.logging_callback_manager.add_litellm_callback("azure_storage")

    response = await litellm.acompletion(model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello, world!"}])

    print(response)

    # Wait for logs to flush
    await asyncio.sleep(10)

if __name__ == "__main__":
    asyncio.run(main())

I'll look into the proxy next, but any additional info you have to recreate it would be helpful

[ishaan-jaff]

thanks for looking into this @adrianlyjak. It's most likely an issue with litellm proxy at higher traffic, around ~10-50 RPS

[adrianlyjak]

no luck here recreating the error with heavy load against the proxy server either

litellm --config config..yaml

config.yaml

model_list:
  - model_name: gpt-4o-mini
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY

litellm_settings:
  callbacks: ["azure_storage"]

script sending 100 requests simultaneously, completes in 2-3s

import asyncio
import aiohttp
import json
import time

async def make_request(session, request_number):
    url = 'http://localhost:4000/v1/chat/completions'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer sk-1234'
    }
    payload = {
        'model': 'gpt-4o-mini',
        'messages': [
            {
                'role': 'user',
                'content': f'Hello! This is request number {request_number}'
            }
        ]
    }
    
    try:
        start_time = time.time()
        async with session.post(url, headers=headers, json=payload) as response:
            response_text = await response.text()
            end_time = time.time()
            print(f"Request {request_number} completed in {end_time - start_time:.2f}s with status {response.status}")
            return response.status, response_text
    except Exception as e:
        print(f"Request {request_number} failed: {str(e)}")
        return None, str(e)

async def main():
    total_requests = 100
    parallelism = 100
    
    print(f"Starting {total_requests} requests with parallelism of {parallelism}")
    start_time = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(1, total_requests + 1):
            task = asyncio.create_task(make_request(session, i))
            tasks.append(task)
            
            # Control parallelism
            if len(tasks) >= parallelism:
                # Wait for some tasks to complete before adding more
                completed, pending = await asyncio.wait(
                    tasks, 
                    return_when=asyncio.FIRST_COMPLETED
                )
                tasks = list(pending)  # Convert set of pending tasks back to list
        
        # Wait for remaining tasks
        if tasks:
            await asyncio.gather(*tasks)
    
    end_time = time.time()
    print(f"All requests completed in {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

I see the request/response logs stored in azure

I see these litellm logs, but not the stack trace

Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x11cbefb90>, 9285.746156875)]', '[(<aiohttp.client_proto.ResponseHandler object at 0x11f651c10>, 9286.048692125)]']
connector: <aiohttp.connector.TCPConnector object at 0x11f6ed640>
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x11ef38110>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x11c479790>, 9286.246360333)]', '[(<aiohttp.client_proto.ResponseHandler object at 0x11c47b2f0>, 9286.563089333)]']
connector: <aiohttp.connector.TCPConnector object at 0x11f029850>

The unclosed connections do seem related to the azure integration, so perhaps fixing that issue will fix the error

[adrianlyjak]

No resolution yet, but I've found some things, so jotting notes down

• LiteLLM is creating a new azure client for each log request. Storing a single azure client on the logging handler makes the unclosed connection errors go away. Creating a new client each time seems non-optimal anyways, since that creates an underlying aio client which is somewhat expensive. This seems like a good thing to fix
• The azure stack trace log lines do not line up with the version of azure libraries I have installed. Perhaps there's a bug here that's being exacerbated.

This is what I was running my tests on

$ pip show azure-storage-blob | grep Version
Version: 12.25.1
$ pip show azure-storage-file-datalake | grep Version
Version: 12.20.0

This issue Azure/azure-sdk-for-python#40217 makes me speculate that this part of the azure client is going through an overhaul, and a newer version may not have the same bug, as I think the referenced load_body in your stack trace is no longer even part of the transport object.

I looked into the docker container (v1.63.11-stable), and its using azure-storage-file-datalake==12.15.0, and the file contents match up with the stack trace.

Next steps:

• I'll proceed with a change to share a single client, that seems more optimal and will reduce the log noise at the very least.
• I can try to recreate the error with the 12.15.0. I'm noticing that the latest stable docker container (v1.65.0-stable) is still on azure-storage-file-datalake==12.15.0,

  litellm/requirements.txt

  Line 32 in 8ecd9ed

  azure-storage-file-datalake==12.15.0 # for azure buck storage logging

  , I'll look into updating that

[ishaan-jaff]

re-using the client makes sense, we do this for OpenAI clients too, checkout in_memory_llm_clients_cache

FYI most of our loggers use httpx clients that are cached, check out datadog

IMO a good path forward would be to cache the azure client in memory for X minutes ttl. run a 100 RPS load test, verify there are no errors

[ishaan-jaff]

Another approach could be to move away from using their client and moving this logger entirely to httpx, (that's your call @adrianlyjak )