INNER CODE UNIT · Python

list_pods

robusta-dev/krr · robusta_krr/core/integrations/kubernetes/__init__.py:144

    async def list_pods(self, object: K8sObjectData) -> list[PodData]:
        loop = asyncio.get_running_loop()

        if object.kind == "CronJob":
            namespace_jobs = await self._list_jobs_for_cronjobs(object.namespace)
            ownered_jobs_uids = [
                job.metadata.uid
                for job in namespace_jobs
                if any(
                    owner.kind == "CronJob" and owner.uid == object._api_resource.metadata.uid
                    for owner in job.metadata.owner_references or []
                )
            ]
            selector = f"batch.kubernetes.io/controller-uid in ({','.join(ownered_jobs_uids)})"

        elif object.kind == "GroupedJob":
            if not hasattr(object._api_resource, "_label_filter") or not object._api_resource._label_filter:
                return []

View source record →