feat(#223): map logical usageMetrics names to catalog metrics so models price from the catalog

infra_cost_model/engine/engine.py

@@ -450,6 +450,16 @@ def _compute_flat_cost(self, address: str, node: dict,
                 result = self.catalog.query(
                     provider, service, region, metric_name, total_quantity
                 )
+                if result is None:
+                    # The node used a logical metric name (e.g. "natHours"); map it
+                    # to the catalog usage_metric ("NAT-Gateway-Hour") via the
+                    # owning handler and retry, so catalog pricing (live/seed) is
+                    # reached instead of falling back to embedded pricingRates.
+                    mapped = self._resolve_catalog_metric(address, node, metric_name)
+                    if mapped is not None:
+                        result = self.catalog.query(
+                            provider, service, region, mapped, total_quantity
+                        )
                 if result is not None:
                     metric_cost = result.total_cost
             if metric_cost is None and metric_name in pricing_rates:
@@ -523,6 +533,16 @@ def _compute_tiered_cost(self, address: str, node: dict, invocations: float) ->
                 result = self.catalog.query(
                     provider, service, region, metric_name, total_quantity
                 )
+                if result is None:
+                    # The node used a logical metric name (e.g. "natHours"); map it
+                    # to the catalog usage_metric ("NAT-Gateway-Hour") via the
+                    # owning handler and retry, so catalog pricing (live/seed) is
+                    # reached instead of falling back to embedded pricingRates.
+                    mapped = self._resolve_catalog_metric(address, node, metric_name)
+                    if mapped is not None:
+                        result = self.catalog.query(
+                            provider, service, region, mapped, total_quantity
+                        )
                 if result is not None:
                     metric_cost = result.total_cost
             # Fallback: flat pricingRates
@@ -538,6 +558,21 @@ def _compute_tiered_cost(self, address: str, node: dict, invocations: float) ->
 
         return (variable_cost, fixed_cost)
 
+    def _resolve_catalog_metric(self, address: str, node: dict, logical_metric: str):
+        """Translate a node's logical usageMetrics key to a catalog usage_metric
+        name via the handler that owns the node's resource address.
+
+        Returns the catalog name, or ``None`` when no handler matches the address
+        or the handler declares no mapping for that logical name (in which case
+        the caller falls back to embedded ``pricingRates``).
+        """
+        from infra_cost_model.resources.registry import ResourceRegistry
+
+        resource_address = node.get("resourceAddress") or address
+        if not resource_address:
+            return None
+        return ResourceRegistry.resolve_catalog_metric(resource_address, logical_metric)
+
     def _resolve_param(self, value) -> float:
         """Resolve a value that may be a parameter name or a numeric literal.
 

infra_cost_model/resources/cloudwatch.py

@@ -20,6 +20,11 @@ class CloudWatchLogGroup(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["ingestedGb", "storedGb"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"ingestedGb": "CloudWatch-Log-Ingestion",
+                "storedGb": "CloudWatch-Log-Storage"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["CloudWatchLogGroup"]:
         if (resource_address.startswith("aws_cloudwatch_log_group.") or
@@ -82,6 +87,12 @@ class CloudWatchMetricAlarm(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["alarmsCount", "customMetricsCount", "getMetricDataRequests"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"alarmsCount": "CloudWatch-Alarm-Month",
+                "customMetricsCount": "CloudWatch-Metric-Month",
+                "getMetricDataRequests": "CloudWatch-GetMetricData"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["CloudWatchMetricAlarm"]:
         if (resource_address.startswith("aws_cloudwatch_metric_alarm.") or

infra_cost_model/resources/data_transfer.py

@@ -36,6 +36,12 @@ class DataTransferNode(ExternalResource):
     def valid_metrics(self) -> list[str]:
         return ["interRegionGb", "internetOutGb", "interAzGb"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"interRegionGb": "DataTransfer-InterRegion-GB",
+                "internetOutGb": "DataTransfer-Internet-Out-GB",
+                "interAzGb": "DataTransfer-InterAZ-GB"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["DataTransferNode"]:
         if resource_address.startswith(_ADDRESS_PREFIXES):

infra_cost_model/resources/kms.py

@@ -17,6 +17,10 @@ class KMSKey(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["keysCount", "apiRequests"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"keysCount": "KMS-Key-Month", "apiRequests": "KMS-API-Request"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["KMSKey"]:
         if (resource_address.startswith("aws_kms_key.") or

infra_cost_model/resources/misc_services.py

@@ -18,6 +18,11 @@ class SecretsManagerSecret(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["secretsCount", "apiCalls"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"secretsCount": "SecretsManager-Secret",
+                "apiCalls": "SecretsManager-API-Call"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["SecretsManagerSecret"]:
         if (resource_address.startswith("aws_secretsmanager_secret.") or
@@ -82,6 +87,10 @@ class ECRRepository(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["storedGb"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"storedGb": "ECR-Storage"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["ECRRepository"]:
         if (resource_address.startswith("aws_ecr_repository.") or
@@ -137,6 +146,10 @@ class Route53Zone(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["hostedZones", "queries"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"hostedZones": "Route53-HostedZone", "queries": "Route53-Query"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["Route53Zone"]:
         if (resource_address.startswith("aws_route53_zone.") or

infra_cost_model/resources/networking.py

@@ -22,6 +22,11 @@ class NATGateway(RoutingResource):
     def valid_metrics(self) -> list[str]:
         return ["natHours", "dataProcessedGb"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"natHours": "NAT-Gateway-Hour",
+                "dataProcessedGb": "NAT-Gateway-DataProcessed"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["NATGateway"]:
         if (resource_address.startswith("aws_nat_gateway.") or
@@ -82,6 +87,11 @@ class VpcEndpoint(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["endpointHours", "dataProcessedGb"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"endpointHours": "VPC-Endpoint-Hour",
+                "dataProcessedGb": "VPC-Endpoint-DataProcessed"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["VpcEndpoint"]:
         if (resource_address.startswith("aws_vpc_endpoint.") or
@@ -146,6 +156,10 @@ class ElasticIP(StorageResource):
     def valid_metrics(self) -> list[str]:
         return ["inUseHours", "idleHours"]
 
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        return {"inUseHours": "IPv4-InUse-Hours", "idleHours": "IPv4-Idle-Hours"}
+
     @classmethod
     def from_address(cls, resource_address: str) -> Optional["ElasticIP"]:
         if (resource_address.startswith("aws_eip.") or

infra_cost_model/resources/registry.py

@@ -114,6 +114,26 @@ def from_address(cls, resource_address: str,
                 return handler
         return None
 
+    @classmethod
+    def resolve_catalog_metric(cls, resource_address: str,
+                               logical_metric: str) -> Optional[str]:
+        """Map a node's logical usageMetrics key to a catalog usage_metric name.
+
+        Finds the handler that owns ``resource_address`` and looks up
+        ``logical_metric`` in its ``catalog_metrics`` map. Resolution is
+        per-handler (not per-service) so resources sharing a service can reuse a
+        logical name for different catalog metrics (e.g. ``dataProcessedGb`` maps
+        to ``NAT-Gateway-DataProcessed`` for NAT Gateway but
+        ``VPC-Endpoint-DataProcessed`` for a VPC endpoint).
+
+        Returns ``None`` when no handler matches or the handler has no mapping for
+        that logical name.
+        """
+        handler = cls.from_address(resource_address)
+        if handler is None:
+            return None
+        return handler().catalog_metrics.get(logical_metric)
+
     @classmethod
     def known_prefixes(cls) -> set[str]:
         """Return set of handler class names registered."""

infra_cost_model/resources/types.py

@@ -39,7 +39,21 @@ def node_type(self) -> str:
     def valid_metrics(self) -> list[str]:
         """Return list of valid usage metric names for this type."""
         pass
-
+
+    @property
+    def catalog_metrics(self) -> dict[str, str]:
+        """Map logical usageMetrics names to pricing-catalog usage_metric names.
+
+        Empty by default. Handlers whose logical metrics correspond to catalog
+        pricing rows override this so the engine can price nodes from the catalog
+        (Principle 13) — using the live/seed pricing — instead of falling back to
+        embedded per-node ``pricingRates``. Keyed per handler (not per service),
+        because different resources of the same service can reuse a logical name
+        for a different catalog metric (e.g. ``dataProcessedGb`` on NAT Gateway
+        vs VPC Endpoint).
+        """
+        return {}
+
     @classmethod
     @abstractmethod
     def from_address(cls, resource_address: str) -> Optional["ResourceType"]:

tests/test_engine.py

@@ -2696,3 +2696,85 @@ def test_fixed_metric_uses_catalog(self):
             costs = aggregator.aggregate()
 
             assert costs["nat"] == pytest.approx(730 * 0.045)
+
+
+class TestCatalogMetricMapping:
+    """Issue #223: a node authored with a handler's logical usageMetrics name is
+    priced from the catalog (live/seed), not just embedded pricingRates."""
+
+    @staticmethod
+    def _catalog(tmpdir, usage_metric, price):
+        from infra_cost_model.pricing.cache import PricingCache, Price
+        from infra_cost_model.pricing.catalog import PricingCatalog
+        from pathlib import Path
+        cache = PricingCache(db_path=Path(tmpdir) / "m.db")
+        cache.upsert(Price(
+            vendor="aws", service="AmazonVPC", region="us-east-1",
+            product_family="", attributes={}, usage_metric=usage_metric,
+            unit="Hours", price_usd=price, start_usage_amount=None,
+            end_usage_amount=None, source="test", effective_date="2024-01-01",
+            fetched_at="2024-01-01T00:00:00"))
+        return PricingCatalog(db_path=Path(tmpdir) / "m.db")
+
+    def _nat_node(self, address):
+        return {"nat": {
+            "nodeType": "routing", "resourceAddress": address,
+            "provider": "aws", "service": "AmazonVPC", "region": "us-east-1",
+            "usageMetrics": {"natHours": {"unit": "hours", "value": 730, "fixed": True}},
+        }}
+
+    def test_logical_metric_prices_from_catalog(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as tmp:
+            catalog = self._catalog(tmp, "NAT-Gateway-Hour", 0.045)  # catalog name only
+            nodes = self._nat_node("aws_nat_gateway.main")
+            derived = {"nat": DerivedUsage("nat", 1.0)}
+            costs = CostAggregator(nodes, derived, [], catalog).aggregate()
+            # 730 * $0.045, reached only if natHours -> NAT-Gateway-Hour resolved.
+            assert costs["nat"] == pytest.approx(32.85)
+
+    def test_unmapped_address_without_rates_is_zero(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as tmp:
+            catalog = self._catalog(tmp, "NAT-Gateway-Hour", 0.045)
+            nodes = self._nat_node("not.a.known.resource")  # no handler → no mapping
+            derived = {"nat": DerivedUsage("nat", 1.0)}
+            costs = CostAggregator(nodes, derived, [], catalog).aggregate()
+            assert costs["nat"] == 0.0
+
+    def test_pricing_rates_still_win_for_unmapped_metric(self):
+        """Backward compat: an unmapped logical metric with no catalog/seed row
+        falls back to embedded pricingRates."""
+        import tempfile
+        with tempfile.TemporaryDirectory() as tmp:
+            catalog = self._catalog(tmp, "NAT-Gateway-Hour", 0.045)
+            nodes = self._nat_node("aws_nat_gateway.main")
+            # A logical name the handler does NOT map and the catalog/seed lack.
+            nodes["nat"]["usageMetrics"] = {"customThing": {"unit": "x", "value": 100, "fixed": True}}
+            nodes["nat"]["pricingRates"] = {"customThing": 0.02}
+            derived = {"nat": DerivedUsage("nat", 1.0)}
+            costs = CostAggregator(nodes, derived, [], catalog).aggregate()
+            assert costs["nat"] == pytest.approx(100 * 0.02)
+
+
+class TestCatalogMetricMappingTiered:
+    """Issue #223: the tiered cost path also resolves logical -> catalog metrics."""
+
+    def test_tiered_path_prices_from_catalog(self):
+        import tempfile
+        from infra_cost_model.pricing.catalog import PricingCatalog
+        from pathlib import Path
+        with tempfile.TemporaryDirectory() as tmp:
+            # Fresh catalog; the seed (which carries the tiered CloudWatch-GetMetricData
+            # rows) auto-loads on the first query miss.
+            catalog = PricingCatalog(db_path=Path(tmp) / "t.db")
+            nodes = {"cw": {
+                "nodeType": "storage", "resourceAddress": "aws_cloudwatch_metric_alarm.x",
+                "provider": "aws", "service": "AmazonCloudWatch", "region": "us-east-1",
+                "pricingModel": "tiered",
+                "usageMetrics": {"getMetricDataRequests": {"unit": "metrics", "value": 2_000_000, "fixed": True}},
+            }}
+            derived = {"cw": DerivedUsage("cw", 1.0)}
+            costs = CostAggregator(nodes, derived, [], catalog).aggregate()
+            # 1M free + 1M x $0.00001 = $10.00, via getMetricDataRequests -> CloudWatch-GetMetricData
+            assert costs["cw"] == pytest.approx(10.0)

tests/test_registry.py

@@ -357,3 +357,26 @@ def test_extract_cdk_empty_resources(self):
         cdk_json = {"Resources": {}}
         nodes = extract_resources_from_cdk(cdk_json)
         assert nodes == {}
+
+
+def test_resolve_catalog_metric_per_handler():
+    """Issue #223: logical usageMetrics names resolve to catalog usage_metric
+    names per-handler, disambiguating shared logical names across services."""
+    from infra_cost_model.resources.registry import ResourceRegistry as R
+    assert R.resolve_catalog_metric("aws_nat_gateway.main", "natHours") == "NAT-Gateway-Hour"
+    assert R.resolve_catalog_metric("aws_kms_key.k", "keysCount") == "KMS-Key-Month"
+    assert R.resolve_catalog_metric("aws_eip.nat", "inUseHours") == "IPv4-InUse-Hours"
+    # Same logical name, different catalog metric per handler:
+    assert R.resolve_catalog_metric("aws_nat_gateway.main", "dataProcessedGb") == "NAT-Gateway-DataProcessed"
+    assert R.resolve_catalog_metric("aws_vpc_endpoint.s3", "dataProcessedGb") == "VPC-Endpoint-DataProcessed"
+    # Unknown address or unmapped logical name → None (caller uses pricingRates).
+    assert R.resolve_catalog_metric("aws_lambda_function.f", "natHours") is None
+    assert R.resolve_catalog_metric("aws_nat_gateway.main", "bogusMetric") is None
+    assert R.resolve_catalog_metric("not.a.resource", "natHours") is None
+
+
+def test_resolve_catalog_metric_storedgb_disambiguation():
+    """`storedGb` maps to different catalog metrics per handler (ECR vs CW logs)."""
+    from infra_cost_model.resources.registry import ResourceRegistry as R
+    assert R.resolve_catalog_metric("aws_ecr_repository.r", "storedGb") == "ECR-Storage"
+    assert R.resolve_catalog_metric("aws_cloudwatch_log_group.g", "storedGb") == "CloudWatch-Log-Storage"