perf: optimize licensing calculator, resource counter, and consolidate _is_managed_service

- Module-level frozensets for DDI/asset type lookups (eliminate per-call set reconstruction)
- Single-pass provider grouping in _get_provider_breakdown (eliminate double _determine_provider scan)
- Module-level _IP_KEY_MAP constant in resource_counter (eliminate per-resource dict rebuild)
- Consolidate _is_managed_service into data-driven base class (3 copies -> 1)
- Proof manifest single write (serialize -> hash -> write, not write -> read -> rewrite)
- Single _scan_timestamp per discovery run (cached in __init__)
- Remove duplicate timestamp creation in all 3 provider discover.py files

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

aws_discovery/aws_discovery.py

@@ -35,6 +35,18 @@
 class AWSDiscovery(BaseDiscovery):
     """AWS Cloud Discovery implementation."""
 
+    _managed_key_prefixes = (
+        "aws:ecs:",
+        "aws:eks:",
+        "eks.amazonaws.com/",
+        "lambda:",
+        "aws:lambda:",
+        "elasticmapreduce:",
+        "aws:elasticmapreduce:",
+    )
+    _managed_key_exact = frozenset({"managed-by", "managed_by", "aws-managed"})
+    _managed_value_exact = frozenset({"aws-managed", "ecs", "lambda", "eks"})
+
     def __init__(self, config: AWSConfig):
         """
         Initialize AWS discovery.
@@ -752,41 +764,6 @@ def _discover_route53_zones_and_records(self) -> List[Dict]:
 
         return resources
 
-    def _is_managed_service(self, tags: Dict[str, str]) -> bool:
-        """Check if a resource is a managed service (Management Token-free).
-
-        Detects resources created/managed by AWS platform services (ECS tasks,
-        Lambda ENIs, EKS system pods, etc.). Avoids false positives from generic
-        aws:cloudformation:* or aws:autoscaling:* auto-tags.
-        """
-        if not tags:
-            return False
-
-        # Specific tag key prefixes that indicate AWS-managed resources
-        managed_key_prefixes = (
-            "aws:ecs:",
-            "aws:eks:",
-            "eks.amazonaws.com/",
-            "lambda:",
-            "aws:lambda:",
-            "elasticmapreduce:",
-            "aws:elasticmapreduce:",
-        )
-        managed_key_exact = {"managed-by", "managed_by", "aws-managed"}
-
-        for key, value in tags.items():
-            key_lower = key.lower()
-            value_lower = value.lower()
-
-            if key_lower in managed_key_exact:
-                return True
-            if any(key_lower.startswith(prefix) for prefix in managed_key_prefixes):
-                return True
-            if value_lower in ("aws-managed", "ecs", "lambda", "eks"):
-                return True
-
-        return False
-
     def get_management_token_free_assets(self) -> List[Dict]:
         """
         Get list of Management Token-free assets.

aws_discovery/discover.py

@@ -157,8 +157,6 @@ def main(args=None):
         calculator = UniversalDDILicensingCalculator()
         licensing_results = calculator.calculate_from_discovery_results(native_objects, provider="aws")
 
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-
         # Export CSV for Sales Engineers
         csv_file = os.path.join("output", f"aws_universal_ddi_licensing_{timestamp}.csv")
         calculator.export_csv(csv_file, provider="aws")

azure_discovery/azure_discovery.py

@@ -79,6 +79,10 @@ def make_retry_policy(sub_name: str, print_lock: threading.Lock) -> VisibleRetry
 class AzureDiscovery(BaseDiscovery):
     """Azure Cloud Discovery implementation."""
 
+    _managed_key_prefixes = ("aks-managed-", "k8s-azure-", "ms-resource-usage:")
+    _managed_key_exact = frozenset({"managed-by", "managed_by", "azure-managed"})
+    _managed_value_exact = frozenset({"azure-managed", "aks", "appservice", "azure-functions"})
+
     def __init__(
         self,
         config: AzureConfig,
@@ -999,37 +1003,6 @@ def _discover_azure_dns_zones_and_records(self) -> List[Dict]:
 
         return resources
 
-    def _is_managed_service(self, tags: Dict[str, str]) -> bool:
-        """Check if a resource is a managed service (Management Token-free).
-
-        Detects resources created/managed by Azure platform services (AKS system
-        pools, App Service infra, etc.). Avoids false positives from generic tags
-        that happen to contain 'azure' or 'service'.
-        """
-        if not tags:
-            return False
-
-        # Specific tag key prefixes that indicate Azure-managed resources
-        managed_key_prefixes = (
-            "aks-managed-",
-            "k8s-azure-",
-            "ms-resource-usage:",
-        )
-        managed_key_exact = {"managed-by", "managed_by", "azure-managed"}
-
-        for key, value in tags.items():
-            key_lower = key.lower()
-            value_lower = value.lower()
-
-            if key_lower in managed_key_exact:
-                return True
-            if any(key_lower.startswith(prefix) for prefix in managed_key_prefixes):
-                return True
-            if value_lower in ("azure-managed", "aks", "appservice", "azure-functions"):
-                return True
-
-        return False
-
     def get_scanned_subscription_ids(self) -> list:
         """Return the Azure Subscription ID(s) scanned."""
         return [self.subscription_id] if self.subscription_id else []

azure_discovery/discover.py

@@ -405,8 +405,6 @@ def discover_subscription(sub_id):
         calculator = UniversalDDILicensingCalculator()
         calculator.calculate_from_discovery_results(all_native_objects, provider="azure")
 
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-
         # Export CSV for Sales Engineers
         csv_file = os.path.join("output", f"azure_universal_ddi_licensing_{timestamp}.csv")
         calculator.export_csv(csv_file, provider="azure")

gcp_discovery/discover.py

@@ -253,8 +253,6 @@ def discover_project(project_info):
         calculator = UniversalDDILicensingCalculator()
         calculator.calculate_from_discovery_results(all_native_objects, provider="gcp")
 
-        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-
         # Export CSV for Sales Engineers
         csv_file = os.path.join("output", f"gcp_universal_ddi_licensing_{timestamp}.csv")
         calculator.export_csv(csv_file, provider="gcp")

gcp_discovery/gcp_discovery.py

@@ -33,6 +33,10 @@
 class GCPDiscovery(BaseDiscovery):
     """GCP Cloud Discovery implementation."""
 
+    _managed_key_prefixes = ("goog-managed-by", "gke-managed", "cloud-run", "cloud-functions")
+    _managed_key_exact = frozenset({"managed-by", "managed_by", "google-managed"})
+    _managed_value_exact = frozenset({"google-managed", "gke", "cloud-run", "cloud-functions"})
+
     def __init__(self, config: GCPConfig, shared_compute_clients: Optional[dict] = None):
         """
         Initialize GCP discovery.
@@ -794,38 +798,6 @@ def _discover_dns_records(self, zone) -> List[Dict]:
 
         return resources
 
-    def _is_managed_service(self, labels: Dict[str, str]) -> bool:
-        """Check if a resource is a managed service (Management Token-free).
-
-        Detects resources created/managed by GCP platform services (GKE system
-        pods, Cloud Run infra, Cloud Functions, etc.). Uses specific key prefixes
-        and exact value matches to avoid false positives.
-        """
-        if not labels:
-            return False
-
-        # Specific label key prefixes that indicate GCP-managed resources
-        managed_key_prefixes = (
-            "goog-managed-by",
-            "gke-managed",
-            "cloud-run",
-            "cloud-functions",
-        )
-        managed_key_exact = {"managed-by", "managed_by", "google-managed"}
-
-        for key, value in labels.items():
-            key_lower = key.lower()
-            value_lower = value.lower()
-
-            if key_lower in managed_key_exact:
-                return True
-            if any(key_lower.startswith(prefix) for prefix in managed_key_prefixes):
-                return True
-            if value_lower in ("google-managed", "gke", "cloud-run", "cloud-functions"):
-                return True
-
-        return False
-
     def get_scanned_project_ids(self) -> list:
         """Return the GCP Project ID(s) scanned."""
         return [self.project_id] if self.project_id else []

shared/base_discovery.py

@@ -21,6 +21,12 @@ class DiscoveryConfig:
 class BaseDiscovery(ABC):
     """Base class for cloud discovery implementations."""
 
+    # Subclasses override these to customize managed-service detection.
+    # The base _is_managed_service checks key exact, key prefix, and value exact.
+    _managed_key_prefixes: tuple = ()
+    _managed_key_exact: frozenset = frozenset({"managed-by", "managed_by"})
+    _managed_value_exact: frozenset = frozenset()
+
     def __init__(self, config: DiscoveryConfig):
         """
         Initialize the base discovery class.
@@ -31,6 +37,7 @@ def __init__(self, config: DiscoveryConfig):
         self.config = config
         self._discovered_resources: Optional[List[Dict]] = None
         self.resource_counter = ResourceCounter(config.provider)
+        self._scan_timestamp: str = datetime.now().isoformat()
 
         logging.basicConfig(level=logging.WARNING)
         self.logger = logging.getLogger(self.__class__.__name__)
@@ -147,18 +154,27 @@ def _format_resource(
             "requires_management_token": requires_management_token,
             "tags": tags or {},
             "details": resource_data,
-            "discovered_at": datetime.now().isoformat(),
+            "discovered_at": self._scan_timestamp,
         }
 
     def _is_managed_service(self, tags: Dict[str, str]) -> bool:
-        """Base managed-service check. Provider subclasses override with specific indicators."""
+        """Data-driven managed-service check using class-level indicator sets.
+
+        Subclasses customize detection by setting _managed_key_prefixes,
+        _managed_key_exact, and _managed_value_exact class attributes.
+        """
         if not tags:
             return False
 
-        managed_key_exact = {"managed-by", "managed_by"}
         for key, value in tags.items():
             key_lower = key.lower()
-            if key_lower in managed_key_exact:
+            value_lower = value.lower() if isinstance(value, str) else ""
+
+            if key_lower in self._managed_key_exact:
+                return True
+            if any(key_lower.startswith(prefix) for prefix in self._managed_key_prefixes):
+                return True
+            if value_lower in self._managed_value_exact:
                 return True
 
         return False