INNER CODE UNIT · Python

_list_rollouts

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

    def _list_rollouts(self) -> list[K8sObjectData]:
        async def _extract_containers(item: Any) -> list[V1Container]:
            if item.spec.template is not None:
                return item.spec.template.spec.containers

            loop = asyncio.get_running_loop()

            logging.debug(
                f"Rollout has workloadRef, fetching template for {item.metadata.name} in {item.metadata.namespace}"
            )

            # Template can be None and object might have workloadRef
            workloadRef = item.spec.workloadRef
            if workloadRef is not None:
                ret = await loop.run_in_executor(
                    self.executor,
                    lambda: self.apps.read_namespaced_deployment(
                        namespace=item.metadata.namespace, name=workloadRef.name

View source record →