feat(router): discover kubernetes pods behind a headless service api_base (#44342)

Co-authored-by: yassin <yassin@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/constants.py

@@ -40,6 +40,13 @@ class ServerStreamingClassification(str, Enum):
 SERVER_STREAMING_CLASSIFICATION_MARKER: Final = ServerStreamingClassification.MARKER
 DEFER_PYDANTIC_BUILD: Final = os.getenv("DEFER_PYDANTIC_BUILD", "true") in ("true", "1", "on")
 DEFAULT_HEALTH_CHECK_PROMPT: Final = str(os.getenv("DEFAULT_HEALTH_CHECK_PROMPT", "test from litellm"))
+KUBERNETES_POD_DISCOVERY_REFRESH_INTERVAL_SECONDS: Final = get_env_float(
+    "KUBERNETES_POD_DISCOVERY_REFRESH_INTERVAL_SECONDS", 5.0
+)
+KUBERNETES_POD_DISCOVERY_IDLE_EVICTION_SECONDS: Final = get_env_float(
+    "KUBERNETES_POD_DISCOVERY_IDLE_EVICTION_SECONDS", 300.0
+)
+KUBERNETES_POD_DISCOVERY_DNS_TIMEOUT_SECONDS: Final = 2.0
 AZURE_DEFAULT_RESPONSES_API_VERSION: Final = str(os.getenv("AZURE_DEFAULT_RESPONSES_API_VERSION", "preview"))
 AZURE_OPENAI_AUDIO_PROVIDERS: Final = frozenset({"azure", "azure_ai"})
 ROUTER_MAX_FALLBACKS: Final = get_env_int("ROUTER_MAX_FALLBACKS", 5)

litellm/router.py

@@ -266,6 +266,13 @@ from litellm.router_utils.handle_error import (
     send_llm_exception_alert,
 )
 from litellm.router_utils.health_state_cache import DeploymentHealthCache
+from litellm.router_utils.kubernetes_pod_discovery import (
+    KubernetesPodDiscovery,
+    async_resolve_pods_after,
+    async_resolve_pods_after_bound,
+    resolve_pods_after,
+    resolve_pods_after_bound,
+)
 from litellm.router_utils.pre_call_checks.continuation_prefill_check import (
     MID_STREAM_CONTINUATION_MARKER,
     ContinuationOutput,
@@ -1289,6 +1296,7 @@ class Router:
         cache_config: Final[dict[str, Any]] = {}
 
         self.client_ttl = client_ttl
+        self.kubernetes_pod_discovery: Final = KubernetesPodDiscovery()
         if redis_url is not None or (redis_host is not None and redis_port is not None):
             cache_type = "redis"
 
@@ -13512,6 +13520,13 @@ class Router:
                 client = self.cache.get_cache(key=cache_key, local_only=True, parent_otel_span=parent_otel_span)
             return client
         elif client_type == "async":
+            if (
+                deployment["litellm_params"].get(  # pyright: ignore[reportUnknownMemberType]  # legacy mapping
+                    "kubernetes_pod_discovery"
+                )
+                is True
+            ):
+                return None
             if kwargs.get("stream") is True:
                 cache_key = f"{model_id}_stream_async_client"
                 client = self.cache.get_cache(key=cache_key, local_only=True, parent_otel_span=parent_otel_span)
@@ -13521,6 +13536,13 @@ class Router:
                 client = self.cache.get_cache(key=cache_key, local_only=True, parent_otel_span=parent_otel_span)
                 return client
         else:
+            if (
+                deployment["litellm_params"].get(  # pyright: ignore[reportUnknownMemberType]  # legacy mapping
+                    "kubernetes_pod_discovery"
+                )
+                is True
+            ):
+                return None
             if kwargs.get("stream") is True:
                 cache_key = f"{model_id}_stream_client"
                 client = self.cache.get_cache(key=cache_key, parent_otel_span=parent_otel_span)
@@ -14550,7 +14572,7 @@ class Router:
         Router._pop_effort_from_nested_carrier(request_kwargs, "output_config")
         Router._pop_effort_from_nested_carrier(request_kwargs, "reasoning")
 
-    async def async_get_available_deployment(
+    async def _async_get_available_deployment_unresolved(  # noqa: ANN202  # legacy selector result remains inferred
         self,
         model: str,
         request_kwargs: dict,
@@ -14587,7 +14609,7 @@ class Router:
             and self.routing_strategy != "latency-based-routing"
             and self.routing_strategy != "least-busy"
         ):  # prevent regressions for other routing strategies, that don't have async get available deployments implemented.
-            return self.get_available_deployment(
+            return self._get_available_deployment_unresolved(
                 model=model,
                 messages=messages,
                 input=input,
@@ -14726,6 +14748,12 @@ class Router:
                     )
             raise e
 
+    async_get_available_deployment = (  # pyright: ignore[reportUnknownVariableType]  # legacy selector return
+        async_resolve_pods_after(
+            _async_get_available_deployment_unresolved  # pyright: ignore[reportUnknownArgumentType]  # legacy type
+        )
+    )
+
     async def async_get_available_deployment_for_pass_through(
         self,
         model: str,
@@ -15492,7 +15520,7 @@ class Router:
         }
         return cast(StandardLoggingRoutingDecision, kept)  # cast-ok: dropping optional keys preserves the type
 
-    def get_available_deployment(
+    def _get_available_deployment_unresolved(  # noqa: ANN202  # legacy selector result remains inferred
         self,
         model: str,
         messages: list[dict[str, str]] | None = None,
@@ -15665,6 +15693,12 @@ class Router:
         phase_event(_DEPLOYMENT_SELECTED_EVENT, pick_attributes)
         return deployment
 
+    get_available_deployment = (  # pyright: ignore[reportUnknownVariableType]  # legacy selector return
+        resolve_pods_after(
+            _get_available_deployment_unresolved  # pyright: ignore[reportUnknownArgumentType]  # legacy annotations
+        )
+    )
+
     def get_available_deployment_for_pass_through(
         self,
         model: str,
@@ -16086,15 +16120,22 @@ class Router:
             CustomRoutingStrategy: litellm.router.CustomRoutingStrategyBase
         """
 
+        strategy: Final = CustomRoutingStrategy
         setattr(
             self,
             "get_available_deployment",
-            CustomRoutingStrategy.get_available_deployment,
+            resolve_pods_after_bound(
+                strategy.get_available_deployment,  # pyright: ignore[reportUnknownArgumentType]  # legacy stub
+                self.kubernetes_pod_discovery,
+            ),
         )
         setattr(
             self,
             "async_get_available_deployment",
-            CustomRoutingStrategy.async_get_available_deployment,
+            async_resolve_pods_after_bound(
+                strategy.async_get_available_deployment,  # pyright: ignore[reportUnknownArgumentType]  # legacy stub
+                self.kubernetes_pod_discovery,
+            ),
         )
 
     def _reset_custom_routing_strategy(self) -> None:

litellm/router_utils/kubernetes_pod_discovery.py

@@ -0,0 +1,462 @@
+import asyncio
+import concurrent.futures
+import ipaddress
+import socket
+import threading
+import time
+import urllib.request
+from collections.abc import Awaitable, Callable, Coroutine, Mapping, Sequence
+from concurrent.futures import Future
+from dataclasses import dataclass, replace
+from functools import wraps
+from types import MappingProxyType
+from typing import (
+    Concatenate,
+    Final,
+    ParamSpec,
+    Protocol,
+    TypeAlias,
+    TypeVar,
+    cast,  # noqa: TID251  # typeshed lacks proxy_bypass_environment
+)
+
+import httpx
+
+from litellm._logging import verbose_router_logger
+from litellm.constants import (
+    KUBERNETES_POD_DISCOVERY_DNS_TIMEOUT_SECONDS,
+    KUBERNETES_POD_DISCOVERY_IDLE_EVICTION_SECONDS,
+    KUBERNETES_POD_DISCOVERY_REFRESH_INTERVAL_SECONDS,
+)
+
+_SocketAddress: TypeAlias = tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]
+_AddressInfo: TypeAlias = tuple[socket.AddressFamily, socket.SocketKind, int, str, _SocketAddress]
+_SyncLookup: TypeAlias = Future[list[_AddressInfo]]
+_AsyncLookup: TypeAlias = asyncio.Future[list[_AddressInfo]]
+_CacheKey: TypeAlias = tuple[str, int | None]
+_DeploymentT: Final = TypeVar("_DeploymentT")
+
+
+@dataclass(frozen=True, slots=True)
+class _PodSet:
+    ips: tuple[str, ...]
+    resolved_at: float
+    last_used_at: float
+    cursor: int = 0
+
+
+def _lookup_outcome(
+    done: _SyncLookup | _AsyncLookup,
+) -> Sequence[_AddressInfo] | OSError:
+    try:
+        return done.result()
+    except OSError as error:
+        return error
+    except BaseException as error:  # noqa: BLE001  # a cancelled or crashed lookup is a failed refresh, never a crash
+        return OSError(str(error))
+
+
+class KubernetesPodDiscovery:
+    def __init__(
+        self,
+        refresh_interval_seconds: float = KUBERNETES_POD_DISCOVERY_REFRESH_INTERVAL_SECONDS,
+        idle_eviction_seconds: float = KUBERNETES_POD_DISCOVERY_IDLE_EVICTION_SECONDS,
+        proxy_environment: Callable[[], Mapping[str, str]] = urllib.request.getproxies_environment,
+        clock: Callable[[], float] = time.monotonic,
+        dns_timeout_seconds: float = KUBERNETES_POD_DISCOVERY_DNS_TIMEOUT_SECONDS,
+        dns_executor: concurrent.futures.Executor | None = None,
+    ) -> None:
+        self.refresh_interval_seconds = refresh_interval_seconds
+        self.idle_eviction_seconds = idle_eviction_seconds
+        self.dns_timeout_seconds = dns_timeout_seconds
+        self._dns_executor = dns_executor or concurrent.futures.ThreadPoolExecutor(
+            thread_name_prefix="litellm-k8s-pod-dns"
+        )
+        self.proxy_environment = proxy_environment
+        self.clock = clock
+        self._cache: Mapping[_CacheKey, _PodSet] = MappingProxyType({})
+        self._refreshing: frozenset[_CacheKey] = frozenset()
+        self._timed_out: frozenset[_CacheKey] = frozenset()
+        self._https_warning_hosts: frozenset[str] = frozenset()
+        self._proxy_warning_hosts: frozenset[str] = frozenset()
+        self._lock = threading.Lock()
+
+    def _begin_refresh(self, key: _CacheKey, now: float) -> tuple[_PodSet | None, bool]:
+        with self._lock:
+            self._cache = MappingProxyType(
+                {
+                    cache_key: pod_set
+                    for cache_key, pod_set in self._cache.items()
+                    if cache_key in self._refreshing or now - pod_set.last_used_at <= self.idle_eviction_seconds
+                }
+            )
+            cached: Final = self._cache.get(key)
+            refreshing: Final = key in self._refreshing
+            if cached is not None and (now - cached.resolved_at < self.refresh_interval_seconds or refreshing):
+                return cached, False
+            if refreshing:
+                return None, False
+            self._refreshing = self._refreshing | {key}
+            return None, True
+
+    def resolve_deployment(self, deployment: _DeploymentT) -> _DeploymentT:
+        eligible: Final = self._eligible(deployment)
+        if eligible is None:
+            return deployment
+        url, key, deployment_mapping = eligible
+        host, port = key
+        now: Final = self.clock()
+        cached, should_refresh = self._begin_refresh(key, now)
+        if cached is not None:
+            return self._deployment_with_cached_ip(deployment, deployment_mapping, key, url, now)
+        if not should_refresh:
+            return deployment
+
+        lookup: Final = self._dns_executor.submit(
+            socket.getaddrinfo, host, port, type=socket.SOCK_STREAM, flags=socket.AI_ADDRCONFIG
+        )
+        lookup.add_done_callback(lambda done: self._finish_refresh(key, done))
+        try:
+            return self._apply_lookup(
+                deployment, deployment_mapping, key, url, self._await_lookup(key, host, lookup), now
+            )
+        finally:
+            self._end_refresh(key)
+
+    async def async_resolve_deployment(self, deployment: _DeploymentT) -> _DeploymentT:
+        eligible: Final = self._eligible(deployment)
+        if eligible is None:
+            return deployment
+        url, key, deployment_mapping = eligible
+        host, port = key
+        now: Final = self.clock()
+        cached, should_refresh = self._begin_refresh(key, now)
+        if cached is not None:
+            return self._deployment_with_cached_ip(deployment, deployment_mapping, key, url, now)
+        if not should_refresh:
+            return deployment
+
+        lookup: Final = asyncio.ensure_future(
+            asyncio.get_running_loop().getaddrinfo(host, port, type=socket.SOCK_STREAM, flags=socket.AI_ADDRCONFIG)
+        )
+        lookup.add_done_callback(lambda done: self._finish_refresh(key, done))
+        try:
+            result: Final = await self._await_async_lookup(key, host, lookup)
+            return self._apply_lookup(deployment, deployment_mapping, key, url, result, now)
+        finally:
+            self._end_refresh(key)
+
+    def _eligible(self, deployment: object) -> tuple[httpx.URL, _CacheKey, Mapping[str, object]] | None:
+        if not isinstance(deployment, Mapping):
+            return None
+        typed_deployment: Final = cast(  # cast-ok: runtime Mapping check precedes typed lookup
+            Mapping[str, object], deployment
+        )
+        target: Final = self._target(typed_deployment)
+        if target is None:
+            return None
+        url, host, port, target_deployment = target
+        if url.scheme == "https":
+            self._warn_https_once(host)
+            return None
+        if url.scheme != "http" or self._is_ip_literal(host):
+            return None
+        return (url, (host, port), target_deployment)
+
+    @staticmethod
+    def _target(
+        deployment_mapping: Mapping[str, object],
+    ) -> tuple[httpx.URL, str, int | None, Mapping[str, object]] | None:
+        raw_params: Final = deployment_mapping.get("litellm_params")
+        if not isinstance(raw_params, Mapping):
+            return None
+        params: Final = cast(  # cast-ok: runtime Mapping validation precedes the string field checks
+            Mapping[str, object], raw_params
+        )
+        if params.get("kubernetes_pod_discovery") is not True:
+            return None
+        raw_api_base: Final = params.get("api_base")
+        if not isinstance(raw_api_base, str):
+            return None
+        try:
+            url: Final = httpx.URL(raw_api_base)
+        except httpx.InvalidURL:
+            return None
+        return (url, url.host, url.port, deployment_mapping)
+
+    def _await_lookup(self, key: _CacheKey, host: str, lookup: _SyncLookup) -> Sequence[_AddressInfo] | OSError:
+        try:
+            return lookup.result(timeout=self.dns_timeout_seconds)
+        except (TimeoutError, concurrent.futures.TimeoutError):
+            return self._abandon_lookup(key, host, lookup)
+        except OSError as error:
+            return error
+
+    async def _await_async_lookup(
+        self, key: _CacheKey, host: str, lookup: _AsyncLookup
+    ) -> Sequence[_AddressInfo] | OSError:
+        try:
+            return await asyncio.wait_for(asyncio.shield(lookup), timeout=self.dns_timeout_seconds)
+        except (TimeoutError, asyncio.TimeoutError):
+            return self._abandon_lookup(key, host, lookup)
+        except asyncio.CancelledError:
+            self._abandon_lookup(key, host, lookup)
+            raise
+        except OSError as error:
+            return error
+
+    def _abandon_lookup(
+        self,
+        key: _CacheKey,
+        host: str,
+        lookup: _SyncLookup | _AsyncLookup,
+    ) -> Sequence[_AddressInfo] | OSError:
+        with self._lock:
+            if lookup.done():
+                return _lookup_outcome(lookup)
+            self._timed_out = self._timed_out | {key}
+        return socket.gaierror(socket.EAI_AGAIN, f"DNS lookup for {host} exceeded {self.dns_timeout_seconds}s")
+
+    def _end_refresh(self, key: _CacheKey) -> None:
+        with self._lock:
+            if key not in self._timed_out:
+                self._refreshing = self._refreshing - {key}
+
+    def _finish_refresh(
+        self,
+        key: _CacheKey,
+        done: _SyncLookup | _AsyncLookup,
+    ) -> None:
+        with self._lock:
+            timed_out: Final = key in self._timed_out
+            self._timed_out = self._timed_out - {key}
+            self._refreshing = self._refreshing - {key} if timed_out else self._refreshing
+        late_result: Final = _lookup_outcome(done) if timed_out else None
+        if late_result is None:
+            return
+        if not isinstance(late_result, OSError):
+            self._store(key, self._pod_ips(late_result), self.clock())
+        elif isinstance(late_result, socket.gaierror) and self._is_authoritative_no_pods(late_result):
+            self._store(key, (), self.clock())
+
+    def _apply_lookup(
+        self,
+        deployment: _DeploymentT,
+        deployment_mapping: Mapping[str, object],
+        key: _CacheKey,
+        url: httpx.URL,
+        result: Sequence[_AddressInfo] | OSError,
+        now: float,
+    ) -> _DeploymentT:
+        if isinstance(result, OSError):
+            if isinstance(result, socket.gaierror) and self._is_authoritative_no_pods(result):
+                self._store(key, (), self.clock())
+                return deployment
+            self._stamp_failed_refresh(key, self.clock())
+            verbose_router_logger.debug("Kubernetes pod discovery DNS refresh failed for %s: %s", key[0], result)
+            return self._deployment_with_cached_ip(deployment, deployment_mapping, key, url, now)
+        ips: Final = self._pod_ips(result)
+        self._store(key, ips, self.clock())
+        if not ips:
+            return deployment
+        return self._deployment_with_cached_ip(deployment, deployment_mapping, key, url, now)
+
+    @staticmethod
+    def _is_ip_literal(host: str) -> bool:
+        try:
+            ipaddress.ip_address(host)
+        except ValueError:
+            return False
+        return True
+
+    @staticmethod
+    def _is_authoritative_no_pods(error: socket.gaierror) -> bool:
+        no_pod_errors: Final = (socket.EAI_NONAME, socket.EAI_NODATA)
+        return error.errno in no_pod_errors
+
+    @staticmethod
+    def _pod_ip(record: _AddressInfo) -> str | None:
+        address: Final = record[4][0]
+        return address if isinstance(address, str) else None
+
+    @classmethod
+    def _pod_ips(cls, records: Sequence[_AddressInfo]) -> tuple[str, ...]:
+        addresses: Final = tuple(cls._pod_ip(record) for record in records)
+        return tuple(sorted({address for address in addresses if address is not None}))
+
+    def _store(self, key: _CacheKey, ips: tuple[str, ...], resolved_at: float) -> None:
+        with self._lock:
+            cached: Final = self._cache.get(key)
+            cursor: Final = cached.cursor % len(ips) if cached is not None and cached.ips and ips else 0
+            self._cache = MappingProxyType(
+                {
+                    **self._cache,
+                    key: _PodSet(ips=ips, resolved_at=resolved_at, last_used_at=resolved_at, cursor=cursor),
+                }
+            )
+
+    def _stamp_failed_refresh(self, key: _CacheKey, resolved_at: float) -> None:
+        with self._lock:
+            cached: Final = self._cache.get(key)
+            stamped: Final = (
+                replace(cached, resolved_at=resolved_at)
+                if cached is not None
+                else _PodSet(ips=(), resolved_at=resolved_at, last_used_at=resolved_at)
+            )
+            self._cache = MappingProxyType({**self._cache, key: stamped})
+
+    def _next_ip(self, key: _CacheKey, now: float) -> str | None:
+        with self._lock:
+            cached: Final = self._cache.get(key)
+            if cached is None or not cached.ips:
+                return None
+            ip: Final = cached.ips[cached.cursor]
+            next_cursor: Final = (cached.cursor + 1) % len(cached.ips)
+            self._cache = MappingProxyType(
+                {
+                    **self._cache,
+                    key: replace(cached, cursor=next_cursor, last_used_at=max(cached.last_used_at, now)),
+                }
+            )
+            return ip
+
+    def _deployment_with_cached_ip(
+        self,
+        deployment: _DeploymentT,
+        deployment_mapping: Mapping[str, object],
+        key: _CacheKey,
+        url: httpx.URL,
+        now: float,
+    ) -> _DeploymentT:
+        ip: Final = self._next_ip(key, now)
+        if ip is None:
+            return deployment
+        if self._proxy_bypasses_only_service_host(key, ip):
+            self._warn_proxy_once(key[0])
+            return deployment
+        params: Final = deployment_mapping.get("litellm_params")
+        if not isinstance(params, Mapping):
+            return deployment
+        typed_params: Final = cast(  # cast-ok: runtime Mapping validation precedes the copied provider parameters
+            Mapping[str, object], params
+        )
+        return cast(  # cast-ok: resolving produces a copied dict for the selector's Mapping type
+            _DeploymentT,
+            {
+                **deployment_mapping,
+                "litellm_params": {**typed_params, "api_base": str(url.copy_with(host=ip))},
+            },
+        )
+
+    def _proxy_bypasses_only_service_host(self, key: _CacheKey, ip: str) -> bool:
+        proxies: Final = self.proxy_environment()
+        if not (proxies.get("http") or proxies.get("all")):
+            return False
+        proxy_bypass_environment: Final = cast(  # cast-ok: stdlib helper accepts the supplied proxy mapping
+            Callable[[str, dict[str, str]], bool],
+            urllib.request.proxy_bypass_environment,  # pyright: ignore[reportAttributeAccessIssue]  # typeshed gap
+        )
+        no_proxy: Final = proxies.get("no", "")
+        entries: Final = tuple(entry.strip() for entry in no_proxy.split(","))
+        host: Final = key[0]
+        port: Final = key[1] if key[1] is not None else 80
+        bracketed_ip: Final = f"[{ip}]" if ipaddress.ip_address(ip).version == 6 else ip
+        service_bypassed: Final = "*" in entries or any(
+            proxy_bypass_environment(candidate, dict(proxies)) for candidate in (host, f"{host}:{port}")
+        )
+        pod_bypassed: Final = any(
+            proxy_bypass_environment(candidate, dict(proxies)) for candidate in (ip, f"{bracketed_ip}:{port}")
+        )
+        return service_bypassed and not pod_bypassed
+
+    def _warn_https_once(self, host: str) -> None:
+        with self._lock:
+            should_warn: Final = host not in self._https_warning_hosts
+            if should_warn:
+                self._https_warning_hosts = self._https_warning_hosts | {host}
+        if should_warn:
+            verbose_router_logger.warning(
+                "Kubernetes pod discovery is disabled for HTTPS host %s because substituting an IP "
+                "breaks TLS hostname verification",
+                host,
+            )
+
+    def _warn_proxy_once(self, host: str) -> None:
+        with self._lock:
+            should_warn: Final = host not in self._proxy_warning_hosts
+            if should_warn:
+                self._proxy_warning_hosts = self._proxy_warning_hosts | {host}
+        if should_warn:
+            verbose_router_logger.warning(
+                "Kubernetes pod discovery keeps the service hostname %s: NO_PROXY exempts it but not its pod IPs",
+                host,
+            )
+
+
+class _HasPodDiscovery(Protocol):
+    kubernetes_pod_discovery: KubernetesPodDiscovery
+
+
+_P: Final = ParamSpec("_P")
+_R: Final = TypeVar("_R")
+_S: Final = TypeVar("_S", bound=_HasPodDiscovery)
+
+
+def resolve_pods_after(
+    fn: Callable[Concatenate[_S, _P], _R],
+) -> Callable[Concatenate[_S, _P], _R]:
+    @wraps(fn)
+    def wrapped(
+        self: _S,
+        *args: _P.args,
+        **kwargs: _P.kwargs,  # kwargs-ok: ParamSpec forwards the wrapped call signature
+    ) -> _R:
+        deployment: Final = fn(self, *args, **kwargs)
+        return self.kubernetes_pod_discovery.resolve_deployment(deployment)
+
+    return wrapped
+
+
+def async_resolve_pods_after(
+    fn: Callable[Concatenate[_S, _P], Awaitable[_R]],
+) -> Callable[Concatenate[_S, _P], Coroutine[object, object, _R]]:
+    @wraps(fn)
+    async def wrapped(
+        self: _S,
+        *args: _P.args,
+        **kwargs: _P.kwargs,  # kwargs-ok: ParamSpec forwards the wrapped call signature
+    ) -> _R:
+        deployment: Final = await fn(self, *args, **kwargs)
+        return await self.kubernetes_pod_discovery.async_resolve_deployment(deployment)
+
+    return wrapped
+
+
+def resolve_pods_after_bound(
+    fn: Callable[_P, _R],
+    discovery: KubernetesPodDiscovery,
+) -> Callable[_P, _R]:
+    @wraps(fn)
+    def wrapped(
+        *args: _P.args,
+        **kwargs: _P.kwargs,  # kwargs-ok: ParamSpec forwards the wrapped call signature
+    ) -> _R:
+        deployment: Final = fn(*args, **kwargs)
+        return discovery.resolve_deployment(deployment)
+
+    return wrapped
+
+
+def async_resolve_pods_after_bound(
+    fn: Callable[_P, Awaitable[_R]],
+    discovery: KubernetesPodDiscovery,
+) -> Callable[_P, Coroutine[object, object, _R]]:
+    @wraps(fn)
+    async def wrapped(
+        *args: _P.args,
+        **kwargs: _P.kwargs,  # kwargs-ok: ParamSpec forwards the wrapped call signature
+    ) -> _R:
+        deployment: Final = await fn(*args, **kwargs)
+        return await discovery.async_resolve_deployment(deployment)
+
+    return wrapped

litellm/types/litellm_params.py

@@ -183,6 +183,7 @@ class DeploymentOptions:
     tpm: int | None = None
     itpm: int | None = None
     otpm: int | None = None
+    kubernetes_pod_discovery: bool | None = None
     default_api_key_rpm_limit: int | None = None
     default_api_key_tpm_limit: int | None = None
     max_parallel_requests: int | None = None

litellm/types/router.py

@@ -720,6 +720,7 @@ class LiteLLMParamsTypedDict(TypedDict, total=False):
     use_chat_completions_api: bool | None
     ## PASS-THROUGH ENDPOINTS ##
     use_in_pass_through: bool | None
+    kubernetes_pod_discovery: ReadOnly[bool | None]
     litellm_credential_name: str | None
     ## UNIFIED PROJECT/REGION ##
     region_name: str | None

tests/code_coverage_tests/router_code_coverage.py

@@ -42,9 +42,11 @@ def get_all_functions_called_in_tests(base_dir):
         print("dir_path: ", dir_path)
         for root, _, files in os.walk(dir_path):
             for file in files:
-                if file.endswith(".py") and ("router" in file.lower() or test_dir == "unit"):
+                file_path = os.path.join(root, file)
+                if file.endswith(".py") and (
+                    "router" in os.path.relpath(file_path, dir_path).lower() or test_dir == "unit"
+                ):
                     print("file: ", file)
-                    file_path = os.path.join(root, file)
                     with open(file_path, "r", encoding="utf-8") as f:
                         try:
                             tree = ast.parse(f.read())

tests/unit/types/test_litellm_params.py

@@ -131,6 +131,7 @@ OPTION_NAMES: Final = (
     "use_litellm_proxy",
     "use_chat_completions_api",
     "use_in_pass_through",
+    "kubernetes_pod_discovery",
     "allowed_openai_params",
     "fallbacks",
     "context_window_fallback_dict",

ui/litellm-dashboard/src/app/(dashboard)/models-and-endpoints/panels/AddModelPanel.integration.test.tsx

@@ -132,6 +132,7 @@ const alwaysMounted = {
 
 const advancedOpenExtras = {
   guardrails: undefined,
+  kubernetes_pod_discovery: undefined,
   tags: undefined,
   use_in_pass_through: undefined,
   vector_store_ids: undefined,
@@ -304,6 +305,24 @@ describe("AddModelPanel submit payload contract", () => {
     });
   });
 
+  it("submits Kubernetes pod discovery in litellm_params", async () => {
+    const { user, openAdvanced, fillRequired, submit } = await setup();
+    await fillRequired();
+    await openAdvanced();
+    await user.click(screen.getByRole("switch", { name: "Kubernetes pod discovery" }));
+    await submit();
+
+    expect(lastCreatedModel()).toStrictEqual({
+      model_name: "gpt-4o",
+      litellm_params: {
+        ...alwaysMounted,
+        ...advancedOpenExtras,
+        kubernetes_pod_discovery: true,
+      },
+      model_info: { ...baseModelInfo },
+    });
+  });
+
   it("drops a collapsed section's keys and the value typed into it", async () => {
     const { user, openAdvanced, closeAdvanced, fillRequired, submit } = await setup();
     await fillRequired();

ui/litellm-dashboard/src/components/add_model/advanced_settings.test.tsx

@@ -1,6 +1,8 @@
 import { act, fireEvent, render, waitFor, screen } from "@testing-library/react";
+import { useFormContext } from "react-hook-form";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { MountedFormHost } from "../../../tests/mounted-form-host";
+import type { MountedFormValues } from "../common_components/MountedFormField";
 import AdvancedSettings from "./advanced_settings";
 
 const mockUsePtuCostAttributionEnabled = vi.fn();
@@ -11,7 +13,16 @@ vi.mock("@/app/(dashboard)/hooks/uiSettings/usePtuCostAttributionEnabled", () =>
 
 const PTU_LABELS = ["PTU Count", "Calculated Cost per PTU / Hour (USD)", "PTU Effective From (UTC)"];
 
-const renderAdvancedSettings = () =>
+const KubernetesPodDiscoveryValue = () => {
+  const form = useFormContext<MountedFormValues>();
+  return (
+    <output aria-label="Kubernetes pod discovery form value" role="status">
+      {String(form.watch("kubernetes_pod_discovery") === true)}
+    </output>
+  );
+};
+
+const renderAdvancedSettings = (includeDiscoveryValue = false) =>
   render(
     <MountedFormHost>
       <AdvancedSettings
@@ -21,6 +32,7 @@ const renderAdvancedSettings = () =>
         tagsList={{}}
         accessToken="test-token"
       />
+      {includeDiscoveryValue && <KubernetesPodDiscoveryValue />}
     </MountedFormHost>,
   );
 
@@ -34,6 +46,23 @@ describe("AdvancedSettings", () => {
     renderAdvancedSettings();
   });
 
+  it("updates the Kubernetes pod discovery form value when enabled", async () => {
+    renderAdvancedSettings(true);
+    act(() => {
+      fireEvent.click(screen.getByText("Advanced Settings"));
+    });
+
+    const toggle = await screen.findByRole("switch", { name: "Kubernetes pod discovery" });
+    const formValue = screen.getByRole("status", { name: "Kubernetes pod discovery form value" });
+    expect(toggle).not.toBeChecked();
+    expect(formValue).toHaveTextContent("false");
+
+    fireEvent.click(toggle);
+
+    expect(toggle).toBeChecked();
+    expect(formValue).toHaveTextContent("true");
+  });
+
   it("should render tags list", async () => {
     renderAdvancedSettings();
     fireEvent.click(screen.getByText("Advanced Settings"));

ui/litellm-dashboard/src/components/add_model/advanced_settings.tsx

@@ -457,6 +457,20 @@ const AdvancedSettings: React.FC<AdvancedSettingsProps> = ({
               )}
             </MountedFormField>
 
+            <MountedFormField
+              name="kubernetes_pod_discovery"
+              label={labelWithHint(
+                "Kubernetes pod discovery",
+                "Treat the API Base hostname as a Kubernetes headless service: resolve it to its ready pod IPs " +
+                  "and spread requests across the pods. LiteLLM must run inside the same cluster.",
+              )}
+              className="mb-4"
+            >
+              {(control) => (
+                <Switch id={control.id} checked={control.value === true} onCheckedChange={control.onChange} />
+              )}
+            </MountedFormField>
+
             <MountedFormField
               name="cache_control"
               label={labelWithHint(CACHE_CONTROL_LABEL, CACHE_CONTROL_TOOLTIP)}