INNER CODE UNIT · Python

_list_scannable_objects

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

    async def _list_scannable_objects(
        self,
        kind: KindLiteral,
        all_namespaces_request: Callable,
        namespaced_request: Callable,
        extract_containers: Callable[[Any], Union[Iterable[V1Container], Awaitable[Iterable[V1Container]]]],
        filter_workflows: Optional[Callable[[Any], bool]] = None,
    ) -> list[K8sObjectData]:
        if not self._should_list_resource(kind):
            logger.debug(f"Skipping {kind}s in {self.cluster}")
            return []

        if not self.__kind_available[kind]:
            return []

        result = []
        try:
            for item in await self._list_namespaced_or_global_objects(kind, all_namespaces_request, namespaced_request):

View source record →