[azure][fix] Warn and ignore HttpResponseError (#2034)

* [azure][fix] Warn and ignore HttpResponseError

* - process pool
- allow expected errors

* fix aws imports

* name the variable better

Author: aquamatthias

fixlib/fixlib/core/actions.py

@@ -1,6 +1,7 @@
 import threading
 import time
 from contextlib import suppress, AbstractContextManager
+from itertools import islice
 from logging import Logger
 from queue import Queue
 
@@ -18,7 +19,7 @@
 from fixlib.args import ArgumentParser
 from fixlib.jwt import encode_jwt_to_headers
 from fixlib.core.ca import TLSData
-from typing import Callable, Dict, Optional, List, Any
+from typing import Callable, Dict, Optional, List, Any, Set
 
 from fixlib.types import Json
 from fixlib.utils import utc_str
@@ -84,6 +85,61 @@ def child_context(self, *context: str) -> "CoreFeedback":
         return self.with_context(*(self.context + list(context)))
 
 
+@define
+class ErrorSummary:
+    error: str
+    message: str
+    info: bool
+    region: Optional[str] = None
+    service_actions: Dict[str, Set[str]] = field(factory=dict)
+
+
+class ErrorAccumulator:
+    def __init__(self) -> None:
+        self.regional_errors: Dict[Optional[str], Dict[str, ErrorSummary]] = {}
+
+    def add_error(
+        self, as_info: bool, error_kind: str, service: str, action: str, message: str, region: Optional[str] = None
+    ) -> None:
+        if region not in self.regional_errors:
+            self.regional_errors[region] = {}
+        regional_errors = self.regional_errors[region]
+
+        key = f"{error_kind}:{message}:{as_info}"
+        if key not in regional_errors:
+            regional_errors[key] = ErrorSummary(error_kind, message, as_info, region, {service: {action}})
+        else:
+            summary = regional_errors[key]
+            if service not in summary.service_actions:
+                summary.service_actions[service] = {action}
+            else:
+                summary.service_actions[service].add(action)
+
+    def report_region(self, core_feedback: CoreFeedback, region: Optional[str]) -> None:
+        if regional_errors := self.regional_errors.get(region):
+            # reset errors for this region
+            self.regional_errors[region] = {}
+            # add region as context
+            feedback = core_feedback.child_context(region) if region else core_feedback
+            # send to core
+            for err in regional_errors.values():
+                srv_acts = []
+                for service, actions in islice(err.service_actions.items(), 10):
+                    suffix = " and more" if len(actions) > 3 else ""
+                    srv_acts.append(service + ": " + ", ".join(islice(actions, 3)) + suffix)
+                message = f"[{err.error}] {err.message} Services and actions affected: {', '.join(srv_acts)}"
+                if len(err.service_actions) > 10:
+                    message += " and more..."
+                if err.info:
+                    feedback.info(message)
+                else:
+                    feedback.error(message)
+
+    def report_all(self, core_feedback: CoreFeedback) -> None:
+        for region in self.regional_errors.keys():
+            self.report_region(core_feedback, region)
+
+
 class SuppressWithFeedback(AbstractContextManager[None]):
     def __init__(self, message: str, feedback: CoreFeedback, logger: Optional[Logger] = None) -> None:
         self.message = message

plugins/aws/fix_plugin_aws/aws_client.py

@@ -3,21 +3,18 @@
 import logging
 from datetime import datetime
 from functools import cached_property
-from itertools import islice
-from typing import Any, Callable, Dict, List, Optional, Set, TypeVar
+from typing import Any, Callable, List, Optional, TypeVar
 
-from attr import define, field
 from botocore.config import Config
 from botocore.exceptions import ClientError, EndpointConnectionError
 from botocore.model import ServiceModel
-from fix_plugin_aws.configuration import AwsConfig
 from retrying import retry
 
-from fixlib.core.actions import CoreFeedback
+from fix_plugin_aws.configuration import AwsConfig
+from fixlib.core.actions import ErrorAccumulator
 from fixlib.json import value_in_path
 from fixlib.types import Json, JsonElement
 from fixlib.utils import log_runtime, utc_str
-
 from .utils import global_region_by_partition
 
 log = logging.getLogger("fix.plugins.aws")
@@ -49,67 +46,6 @@ def is_retryable_exception(e: Exception) -> bool:
     return False
 
 
-@define
-class ErrorSummary:
-    error: str
-    message: str
-    info: bool
-    region: Optional[str] = None
-    service_actions: Dict[str, Set[str]] = field(factory=dict)
-
-
-class ErrorAccumulator:
-    def __init__(self) -> None:
-        self.regional_errors: Dict[Optional[str], Dict[str, ErrorSummary]] = {}
-
-    def add_error(
-        self,
-        as_info: bool,
-        error_kind: str,
-        service: str,
-        action: str,
-        message: str,
-        region: Optional[str],
-    ) -> None:
-        if region not in self.regional_errors:
-            self.regional_errors[region] = {}
-        regional_errors = self.regional_errors[region]
-
-        key = f"{error_kind}:{message}:{as_info}"
-        if key not in regional_errors:
-            regional_errors[key] = ErrorSummary(error_kind, message, as_info, region, {service: {action}})
-        else:
-            summary = regional_errors[key]
-            if service not in summary.service_actions:
-                summary.service_actions[service] = {action}
-            else:
-                summary.service_actions[service].add(action)
-
-    def report_region(self, core_feedback: CoreFeedback, region: Optional[str]) -> None:
-        if regional_errors := self.regional_errors.get(region):
-            # reset errors for this region
-            self.regional_errors[region] = {}
-            # add region as context
-            feedback = core_feedback.child_context(region) if region else core_feedback
-            # send to core
-            for err in regional_errors.values():
-                srv_acts = []
-                for aws_service, actions in islice(err.service_actions.items(), 10):
-                    suffix = " and more" if len(actions) > 3 else ""
-                    srv_acts.append(aws_service + ": " + ", ".join(islice(actions, 3)) + suffix)
-                message = f"[{err.error}] {err.message} Services and actions affected: {', '.join(srv_acts)}"
-                if len(err.service_actions) > 10:
-                    message += " and more..."
-                if err.info:
-                    feedback.info(message)
-                else:
-                    feedback.error(message)
-
-    def report_all(self, core_feedback: CoreFeedback) -> None:
-        for region in self.regional_errors.keys():
-            self.report_region(core_feedback, region)
-
-
 class AwsClient:
     def __init__(
         self,

plugins/aws/fix_plugin_aws/collector.py

@@ -4,7 +4,7 @@
 from typing import List, Type, Optional, ClassVar, Union
 from datetime import datetime, timezone
 
-from fix_plugin_aws.aws_client import AwsClient, ErrorAccumulator
+from fix_plugin_aws.aws_client import AwsClient
 from fix_plugin_aws.configuration import AwsConfig
 from fix_plugin_aws.resource import (
     apigateway,
@@ -48,7 +48,7 @@
 from fix_plugin_aws.resource.base import AwsAccount, AwsApiSpec, AwsRegion, AwsResource, GraphBuilder
 
 from fixlib.baseresources import Cloud, EdgeType, BaseOrganizationalRoot, BaseOrganizationalUnit
-from fixlib.core.actions import CoreFeedback
+from fixlib.core.actions import CoreFeedback, ErrorAccumulator
 from fixlib.core.progress import ProgressDone, ProgressTree
 from fixlib.graph import Graph, BySearchCriteria, ByNodeId
 from fixlib.proc import set_thread_name

plugins/aws/test/aws_client_test.py

@@ -2,8 +2,9 @@
 
 from botocore.exceptions import ClientError
 
-from fix_plugin_aws.aws_client import AwsClient, ErrorAccumulator, is_retryable_exception
+from fix_plugin_aws.aws_client import AwsClient, is_retryable_exception
 from fix_plugin_aws.configuration import AwsConfig
+from fixlib.core.actions import ErrorAccumulator
 from test.resources import BotoFileBasedSession, BotoErrorSession
 
 

plugins/azure/fix_plugin_azure/__init__.py

@@ -1,7 +1,7 @@
 import logging
 import multiprocessing
 from collections import namedtuple
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import as_completed, ProcessPoolExecutor
 from typing import Optional, Tuple, Any
 
 from attr import evolve
@@ -61,6 +61,7 @@ def collect(self) -> None:
             )
             for name, ac in account_configs.items()
             for subscription in AzureSubscription.list_subscriptions(ac.credentials())
+            if ac.allowed(subscription.subscription_id)
         }
         args = list(args_by_subscription_id.values())
 
@@ -72,7 +73,7 @@ def collect(self) -> None:
         self.core_feedback.progress(progress)
 
         # Collect all subscriptions
-        with ThreadPoolExecutor(max_workers=config.subscription_pool_size) as executor:
+        with ProcessPoolExecutor(max_workers=config.subscription_pool_size) as executor:
             wait_for = [executor.submit(collect_in_process, sub, self.task_data) for sub in args]
             for future in as_completed(wait_for):
                 subscription, graph = future.result()

plugins/azure/fix_plugin_azure/azure_client.py

@@ -4,7 +4,7 @@
 from abc import ABC, abstractmethod
 from typing import List, MutableMapping, Optional, Any, Union, Dict
 
-from attr import define
+from attr import define, field
 from retrying import retry
 from azure.core.exceptions import (
     ClientAuthenticationError,
@@ -21,6 +21,7 @@
 from azure.mgmt.resource.resources.models import GenericResource
 
 from fix_plugin_azure.config import AzureCredentials
+from fixlib.core.actions import CoreFeedback, ErrorAccumulator
 from fixlib.types import Json
 
 log = logging.getLogger("fix.plugins.azure")
@@ -55,6 +56,7 @@ class AzureApiSpec:
     query_parameters: List[str] = []
     access_path: Optional[str] = None
     expect_array: bool = False
+    expected_error_codes: List[str] = field(factory=list)
 
 
 class AzureClient(ABC):
@@ -82,9 +84,13 @@ def delete_resource_tag(self, tag_name: str, resource_id: str) -> bool:
     def __create_management_client(
         credential: AzureCredentials,
         subscription_id: str,
+        core_feedback: Optional[CoreFeedback] = None,
+        error_accumulator: Optional[ErrorAccumulator] = None,
         resource_group: Optional[str] = None,
     ) -> AzureClient:
-        return AzureResourceManagementClient(credential, subscription_id, resource_group)
+        return AzureResourceManagementClient(
+            credential, subscription_id, resource_group, core_feedback, error_accumulator
+        )
 
     create = __create_management_client
 
@@ -95,10 +101,14 @@ def __init__(
         credential: AzureCredentials,
         subscription_id: str,
         location: Optional[str] = None,
+        core_feedback: Optional[CoreFeedback] = None,
+        accumulator: Optional[ErrorAccumulator] = None,
     ) -> None:
         self.credential = credential
         self.subscription_id = subscription_id
         self.location = location
+        self.core_feedback = core_feedback
+        self.accumulator = accumulator or ErrorAccumulator()
         self.client = ResourceManagementClient(self.credential, self.subscription_id)
 
     def list(self, spec: AzureApiSpec, **kwargs: Any) -> List[Json]:
@@ -175,10 +185,14 @@ def _list_with_retry(self, spec: AzureApiSpec, **kwargs: Any) -> Optional[List[J
             if error := e.error:
                 if error.code == "NoRegisteredProviderFound":
                     return None  # API not available in this region
+                elif error.code in spec.expected_error_codes:
+                    return None
                 elif error.code == "BadRequest" and spec.service == "metric":
                     raise MetricRequestError from e
-            log.warning(f"[Azure] Error encountered while requesting resource: {e}")
-            raise e
+                code = error.code or "Unknown"
+                self.accumulator.add_error(False, code, spec.service, spec.path, str(e), self.location)
+            log.warning(f"[Azure] Client Error: status={e.status_code}, error={e.error}, message={e}")
+            return None
         except Exception as e:
             log.warning(f"[Azure] called service={spec.service}: hit unexpected error: {e}", exc_info=e)
             return None
@@ -271,4 +285,4 @@ def _make_request(self, url: str, params: MutableMapping[str, Any], headers: Mut
         return response
 
     def for_location(self, location: str) -> AzureClient:
-        return AzureClient.create(self.credential, self.subscription_id, location)
+        return AzureClient.create(self.credential, self.subscription_id, self.core_feedback, self.accumulator, location)

plugins/azure/fix_plugin_azure/collector.py

@@ -26,7 +26,7 @@
 )
 from fix_plugin_azure.resource.containerservice import resources as aks_resources
 from fixlib.baseresources import Cloud, GraphRoot
-from fixlib.core.actions import CoreFeedback
+from fixlib.core.actions import CoreFeedback, ErrorAccumulator
 from fixlib.graph import Graph
 from fixlib.json import value_in_path
 from fixlib.threading import ExecutorQueue, GatherFutures
@@ -72,7 +72,13 @@ def collect(self) -> None:
         ) as executor:
             self.core_feedback.progress_done(self.subscription.subscription_id, 0, 1, context=[self.cloud.id])
             queue = ExecutorQueue(executor, "azure_collector")
-            client = AzureClient.create(self.credentials, self.subscription.subscription_id)
+            error_accumulator = ErrorAccumulator()
+            client = AzureClient.create(
+                self.credentials,
+                self.subscription.subscription_id,
+                core_feedback=self.core_feedback,
+                error_accumulator=error_accumulator,
+            )
 
             def get_last_run() -> Optional[datetime]:
                 td = self.task_data
@@ -123,7 +129,8 @@ def get_last_run() -> Optional[datetime]:
 
             # delete unnecessary nodes after all work is completed
             self.after_collect_filter()
-
+            # report all accumulated errors
+            error_accumulator.report_all(self.core_feedback)
             self.core_feedback.progress_done(self.subscription.subscription_id, 1, 1, context=[self.cloud.id])
             log.info(f"[Azure:{self.subscription.safe_name}] Collecting resources done.")
 

plugins/azure/fix_plugin_azure/config.py

@@ -42,6 +42,13 @@ def credentials(self) -> AzureCredentials:
 
         return DefaultAzureCredential()
 
+    def allowed(self, subscription_id: str) -> bool:
+        if self.subscriptions is not None:
+            return subscription_id in self.subscriptions
+        if self.exclude_subscriptions is not None:
+            return subscription_id not in self.exclude_subscriptions
+        return True
+
 
 @define
 class AzureConfig:

plugins/azure/fix_plugin_azure/resource/network.py

@@ -4701,6 +4701,7 @@ class AzureUsage(AzureResource, BaseNetworkQuota):
         query_parameters=["api-version"],
         access_path="value",
         expect_array=True,
+        expected_error_codes=["SubscriptionHasNoUsages"],
     )
     mapping: ClassVar[Dict[str, Bender]] = {
         "id": S("id"),