feat: move the Noma custom line to upstream v1.103.0

.github/workflows/noma-build-custom-image.yml

@@ -0,0 +1,99 @@
+name: Noma - Build custom image
+
+# Builds the Noma litellm proxy image and pushes it to ECR tagged with the
+# branch name (e.g. v1.102.0-custom). Replaces the previous manual
+# `docker buildx ... --push` step. Deploy is still a separate values bump +
+# canary — this workflow only builds/pushes, it does NOT sync ArgoCD.
+
+on:
+  push:
+    branches:
+      - "v*-custom"          # v1.102.0-custom, v1.99.x-custom, ...
+  workflow_dispatch:
+    inputs:
+      ref:
+        description: "Branch/tag to build (defaults to the branch this is run from)"
+        required: false
+        type: string
+
+concurrency:
+  group: noma-build-custom-${{ inputs.ref || github.ref_name }}
+  cancel-in-progress: false
+
+permissions:
+  contents: read
+  id-token: write            # OIDC -> AWS role
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    environment: prod
+    env:
+      AWS_REGION: us-east-1
+      AWS_ACCOUNT_ID: "381491951875"
+      IMAGE_NAME: litellm
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v5
+        with:
+          ref: ${{ inputs.ref || github.ref_name }}
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Resolve image tag (branch/tag name)
+        id: tag
+        shell: bash
+        run: echo "value=${{ inputs.ref || github.ref_name }}" >> "$GITHUB_OUTPUT"
+
+      - name: Set up QEMU (arm64 emulation on the GitHub-hosted runner)
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v4
+
+      - name: Configure AWS credentials (OIDC)
+        uses: aws-actions/configure-aws-credentials@v6
+        with:
+          role-to-assume: arn:aws:iam::381491951875:role/github-actions-region-deploy
+          aws-region: ${{ env.AWS_REGION }}
+
+      - name: Login to Amazon ECR
+        uses: docker/login-action@v4
+        with:
+          registry: ${{ env.AWS_ACCOUNT_ID }}.dkr.ecr.${{ env.AWS_REGION }}.amazonaws.com
+          username: AWS
+
+      - name: Build args
+        id: vars
+        shell: bash
+        run: |
+          echo "sha_short=$(git rev-parse --short HEAD)" >> "$GITHUB_OUTPUT"
+          echo "build_time=$(date '+%Y-%m-%dT%H:%M:%S')" >> "$GITHUB_OUTPUT"
+
+      - name: Build and push (multi-arch)
+        env:
+          REGISTRY: ${{ env.AWS_ACCOUNT_ID }}.dkr.ecr.${{ env.AWS_REGION }}.amazonaws.com
+          TAG: ${{ steps.tag.outputs.value }}
+          SHA_SHORT: ${{ steps.vars.outputs.sha_short }}
+          BUILD_TIME: ${{ steps.vars.outputs.build_time }}
+        shell: bash
+        run: |
+          docker buildx build \
+            --push \
+            --platform linux/amd64,linux/arm64 \
+            --file docker/Dockerfile.non_root \
+            --tag "$REGISTRY/$IMAGE_NAME:$TAG" \
+            --tag "$REGISTRY/$IMAGE_NAME:$TAG-$SHA_SHORT" \
+            --build-arg GIT_VERSION_TAG="$TAG" \
+            --build-arg GIT_VERSION_HASH="$SHA_SHORT" \
+            --build-arg BUILD_TIME="$BUILD_TIME" \
+            --cache-from type=registry,ref="$REGISTRY/$IMAGE_NAME:cache-custom" \
+            --cache-to type=registry,mode=max,image-manifest=true,oci-mediatypes=true,ref="$REGISTRY/$IMAGE_NAME:cache-custom" \
+            .
+
+      - name: Summary
+        shell: bash
+        run: |
+          echo "### Pushed \`$IMAGE_NAME:${{ steps.tag.outputs.value }}\`" >> "$GITHUB_STEP_SUMMARY"
+          echo "Registry: ${{ env.AWS_ACCOUNT_ID }}.dkr.ecr.${{ env.AWS_REGION }}.amazonaws.com" >> "$GITHUB_STEP_SUMMARY"
+          echo "Next: bump the tag in argo-cd/charts values-litellm.yaml and canary." >> "$GITHUB_STEP_SUMMARY"

litellm/caching/evicted_client_closer.py

@@ -29,8 +29,11 @@
 close them, and each bucket is ordered by deadline, so a reap walks the entries
 that are due rather than the whole queue.
 
-The queue holds its clients weakly, so waiting out a grace window never keeps
-alive anything the collector would have reclaimed first.
+The queue holds its clients strongly. An SDK client is a reference cycle that a
+generational collection reclaims within minutes, long before the grace window
+ends, and a client the collector takes first is never closed: its aiohttp
+session is finalized by the garbage collector instead, unclosed. An entry whose
+event loop has been closed can no longer be closed, so it is dropped.
 """
 
 import asyncio
@@ -59,15 +62,13 @@
 class _PendingClose:
     """A queued close.
 
-    The client is held weakly, so queueing one never keeps alive anything the
-    collector would otherwise have reclaimed first.
-
     ``needs_loop`` is set for a client whose close is a coroutine; those can only
     be closed from the event loop they were evicted on, recorded in ``loop_id``.
     A client that closes synchronously carries neither constraint.
     """
 
-    client_ref: "weakref.ref[object]"
+    client: object
+    loop_ref: "weakref.ref[asyncio.AbstractEventLoop] | None"
     loop_id: int | None
     needs_loop: bool
     close_after: float
@@ -82,13 +83,25 @@ def _bucket_key(pending: _PendingClose) -> _BucketKey:
     return pending.loop_id
 
 
-def _running_loop_id() -> int | None:
+def _running_loop() -> asyncio.AbstractEventLoop | None:
     try:
-        return id(asyncio.get_running_loop())
+        return asyncio.get_running_loop()
     except RuntimeError:
         return None
 
 
+def _running_loop_id() -> int | None:
+    loop: Final = _running_loop()
+    return None if loop is None else id(loop)
+
+
+def _loop_is_gone(pending: _PendingClose) -> bool:
+    if pending.loop_ref is None:
+        return False
+    loop: Final = pending.loop_ref()
+    return loop is None or loop.is_closed()
+
+
 def _close_function(client: object) -> Callable[[], object] | None:
     close_fn: Final[Callable[[], object] | None] = getattr(client, "aclose", None) or getattr(client, "close", None)
     return close_fn
@@ -140,6 +153,8 @@ def _has_connection_in_flight(client: object) -> bool:
         pooled_busy: Final = _pool_has_busy_connection(transport)
         if pooled_busy is not None:
             return pooled_busy
+        if not getattr(transport, "_owns_session", True):
+            return False  # a shared session's leases belong to every client on it
         session: Final[object] = getattr(transport, "client", None)
         return bool(getattr(getattr(session, "connector", None), "_acquired", None))
     except Exception:  # noqa: BLE001 - a client that cannot report its state is treated as idle
@@ -197,10 +212,12 @@ def schedule(self, client: object) -> None:
             return
         if self._pending_count >= self._max_pending:
             return
+        loop: Final = _running_loop()
         self._enqueue(
             _PendingClose(
-                client_ref=weakref.ref(client),
-                loop_id=_running_loop_id(),
+                client=client,
+                loop_ref=None if loop is None else weakref.ref(loop),
+                loop_id=None if loop is None else id(loop),
                 needs_loop=inspect.iscoroutinefunction(close_fn),
                 close_after=self._clock() + self._grace_seconds,
             )
@@ -216,9 +233,7 @@ def reap(self) -> None:
             return
         now: Final = self._clock()
         for pending in self._take_due(_running_loop_id(), now):
-            client = pending.client_ref()
-            if client is None:
-                continue
+            client = pending.client
             if _has_connection_in_flight(client):
                 self._enqueue(replace(pending, close_after=now + self._grace_seconds))
                 continue
@@ -229,15 +244,14 @@ def pending_count(self) -> int:
         return self._pending_count
 
     def _enqueue(self, pending: _PendingClose) -> None:
-        """Append to the entry's bucket, dropping any dead entries it queues behind.
+        """Append to the entry's bucket, dropping entries whose loop has closed.
 
         Deadlines only ever move forward, so appending keeps each bucket ordered
-        by deadline, and entries whose client the collector already took sit at
-        the front rather than having to be searched for.
+        by deadline.
         """
         with self._queue_lock:
             bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())  # mutable-ok: FIFO by design
-            while bucket and bucket[0].client_ref() is None:
+            while bucket and _loop_is_gone(bucket[0]):
                 bucket.popleft()
                 self._pending_count -= 1
             bucket.append(pending)
@@ -246,8 +260,14 @@ def _enqueue(self, pending: _PendingClose) -> None:
     def _take_due(self, loop_id: int | None, now: float) -> tuple[_PendingClose, ...]:
         buckets = (_CLOSABLE_ANYWHERE,) if loop_id is None else (_CLOSABLE_ANYWHERE, _CLOSABLE_ON_ANY_LOOP, loop_id)
         with self._queue_lock:
+            self._drop_buckets_of_closed_loops_locked()
             return tuple(pending for key in buckets for pending in self._drain_locked(key, now))
 
+    def _drop_buckets_of_closed_loops_locked(self) -> None:
+        """A loop's bucket is never drained once that loop is closed, so release it."""
+        for key in [k for k, b in self._buckets.items() if b and _loop_is_gone(b[0])]:
+            self._pending_count -= len(self._buckets.pop(key))
+
     def _drain_locked(self, key: _BucketKey, now: float) -> Iterator[_PendingClose]:
         bucket: Final = self._buckets.get(key)
         if bucket is None:

litellm/constants.py

@@ -1860,6 +1860,7 @@
 SPEND_LOG_KEY_METADATA_CACHE_TTL: Final = 600
 SPEND_LOG_KEY_METADATA_MISS_CACHE_TTL: Final = 30
 SPEND_LOG_KEY_METADATA_CACHE_MAX_ITEMS: Final = 10000
+SPEND_COUNTER_CACHE_MAX_SIZE: Final = 10000
 SPEND_LOG_KEY_METADATA_QUERY_TIMEOUT_MS: Final = 5000
 # Short TTL for negative MCP access-group existence lookups. Keeps unauthenticated
 # callers from forcing a DB query per request for unknown names, while bounding
@@ -2084,16 +2085,12 @@
 PTU_SENTINEL_API_KEY: Final[str] = "__ptu_flat_cost__"
 PTU_ROLLUP_JOB_ID: Final[str] = "ptu_flat_cost_rollup_job"
 PTU_ROLLUP_LOCK_TTL_SECONDS: Final[int] = 900
-USAGE_TOP_API_KEYS_LIMIT: Final[int] = int(os.getenv("USAGE_TOP_API_KEYS_LIMIT", "100"))
 # Furthest back the catch-up pass looks for unpriced PTU days when a deployment
 # declares no ptu_effective_from, bounding the scan for an open-ended window.
 PTU_ROLLUP_MAX_BACKFILL_DAYS: Final[int] = 90
 # Deployments named in the lapsed-window alert before it is truncated, so a fleet-wide
 # expiry cannot produce an alert too large for the channel delivering it.
 PTU_LAPSED_ALERT_LIMIT: Final[int] = 10
-DAILY_GLOBAL_SPEND_RECONCILE_JOB_ID: Final[str] = "daily_global_spend_reconcile_job"
-DAILY_GLOBAL_SPEND_RECONCILE_LOCK_TTL_SECONDS: Final[int] = 3600
-DAILY_GLOBAL_SPEND_RECONCILED_THROUGH_PARAM: Final[str] = "daily_global_spend_reconciled_through"
 # Slack allowed when deciding a sentinel row is stale. The row's updated_at and the
 # run's cutoff are stamped by different hosts, so clock skew between them must not let
 # one run delete a charge another just wrote. A stale row is hours old and a concurrent

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -114,6 +114,7 @@ def __init__(
         self.default_webhook_url = default_webhook_url
         self.flush_lock = asyncio.Lock()
         self.periodic_started = False
+        self._periodic_flush_task: asyncio.Task[None] | None = None
         self.hanging_request_check = AlertingHangingRequestCheck(
             slack_alerting_object=self,
         )
@@ -125,6 +126,12 @@ def __init__(
         self.digest_lock = asyncio.Lock()
         super().__init__(**kwargs, flush_lock=self.flush_lock)
 
+    def _ensure_periodic_flush_task(self) -> None:
+        if self.periodic_started and (self._periodic_flush_task is None or not self._periodic_flush_task.done()):
+            return
+        self._periodic_flush_task = asyncio.create_task(self.periodic_flush())
+        self.periodic_started = True
+
     def update_values(
         self,
         alerting: list | None = None,
@@ -137,17 +144,14 @@ def update_values(
     ):
         if alerting is not None:
             self.alerting = alerting
-            asyncio.create_task(self.periodic_flush())
-            self.periodic_started = True
+            self._ensure_periodic_flush_task()
         if alerting_threshold is not None:
             self.alerting_threshold = alerting_threshold
         if alert_types is not None:
             self.alert_types = alert_types
         if alerting_args is not None:
             self.alerting_args = SlackAlertingArgs(**alerting_args)
-            if not self.periodic_started:
-                asyncio.create_task(self.periodic_flush())
-                self.periodic_started = True
+            self._ensure_periodic_flush_task()
         if alert_type_config is not None:
             for key, val in alert_type_config.items():
                 self.alert_type_config[key] = AlertTypeConfig(**val) if isinstance(val, dict) else val
@@ -1442,9 +1446,8 @@ async def send_alert(
             return
 
         # Start periodic flush if not already started
-        if not self.periodic_started and self.alerting is not None and len(self.alerting) > 0:
-            asyncio.create_task(self.periodic_flush())
-            self.periodic_started = True
+        if self.alerting is not None and len(self.alerting) > 0:
+            self._ensure_periodic_flush_task()
 
         if "webhook" in self.alerting and alert_type == "budget_alerts" and user_info is not None:
             await self.send_webhook_alert(webhook_event=user_info)