refactor(telemetry): give the failed-reading contract and key layout one owner each [INFP-589]

The producer half of "what a failed reading looks like" (the unknown-host
sentinel) lived in the component service while the consumer half (the
all-None predicate) lived in the resources module, tied together only by
prose. Both now live on WorkerResourceReading itself — a failed()
constructor and an is_failed property derived from the field set, so a new
figure is covered automatically.

The resource cache-key layout was restated three times in component.py
(writer f-string, list-keys glob, parse regex); all three now derive from a
single RESOURCE_KEY_PREFIX so they cannot drift apart silently.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: saltas888

backend/infrahub/services/component.py

@@ -21,18 +21,17 @@
 
 PRIMARY_API_SERVER = "workers:primary:api_server"
 WORKER_MATCH = re.compile(r":worker:([^:]+)")
-RESOURCE_COMPONENT_MATCH = re.compile(r"workers:resources:([^:]+):worker:")
+
+# Single owner of the resource-key layout: the writer f-string, the list-keys glob
+# and the parse regex are all derived from this prefix so they cannot drift apart.
+RESOURCE_KEY_PREFIX = "workers:resources:"
+RESOURCE_COMPONENT_MATCH = re.compile(re.escape(RESOURCE_KEY_PREFIX) + r"([^:]+):worker:")
 
 # The per-process resource read can transiently fail (a psutil hiccup, a momentary
 # hostname-lookup failure); a few immediate retries cover that before the reading
 # is written as null and the failure logged for traceability.
 RESOURCE_READ_MAX_ATTEMPTS = 3
 
-# Host stand-in written when the resource read fails outright; such a reading
-# carries no figures and is dropped from the aggregate, so the value is never
-# summed and only needs to be non-raising.
-_UNKNOWN_HOST = "unknown"
-
 log = get_logger()
 
 
@@ -129,7 +128,7 @@ async def refresh_heartbeat(self) -> None:
                 expires=KVTTL.FIFTEEN,
             )
             await self.cache.set(
-                key=f"workers:resources:{component}:worker:{WORKER_IDENTITY}",
+                key=f"{RESOURCE_KEY_PREFIX}{component}:worker:{WORKER_IDENTITY}",
                 value=self._read_own_resources().model_dump_json(),
                 expires=KVTTL.FIFTEEN,
             )
@@ -160,7 +159,7 @@ def _read_own_resources(self) -> WorkerResourceReading:
             worker_id=WORKER_IDENTITY,
             error=str(last_error),
         )
-        return WorkerResourceReading(host=_UNKNOWN_HOST)
+        return WorkerResourceReading.failed()
 
     async def read_worker_resources(self) -> dict[str, dict[str, WorkerResourceReading]]:
         """Return the latest worker resource readings grouped by component and host.
@@ -169,7 +168,7 @@ async def read_worker_resources(self) -> dict[str, dict[str, WorkerResourceReadi
         report identical values, so a later reading for a host simply overwrites
         the earlier one.
         """
-        keys = await self.cache.list_keys(filter_pattern="workers:resources:*")
+        keys = await self.cache.list_keys(filter_pattern=f"{RESOURCE_KEY_PREFIX}*")
         values = await self.cache.get_values(keys=keys)
 
         grouped: dict[str, dict[str, WorkerResourceReading]] = {}

backend/infrahub/telemetry/resources.py

@@ -60,6 +60,27 @@ class WorkerResourceReading(BaseModel):
     memory_total: int | None = None
     memory_available: int | None = None
 
+    @classmethod
+    def failed(cls) -> WorkerResourceReading:
+        """The reading written when a process's self-read failed outright.
+
+        It carries no figures and a host stand-in, so it is recognisable as
+        failed (see ``is_failed``) and is dropped from aggregation rather than
+        summed or used for host dedup.
+        """
+        return cls(host="unknown")
+
+    @property
+    def is_failed(self) -> bool:
+        """Whether this reading is a failed self-read rather than a contribution.
+
+        A healthy host always reports at least its logical CPU count and memory
+        capacity; only a read that failed after its retries carries every figure
+        as ``None``. Derived from the field set so a new figure is covered
+        automatically.
+        """
+        return all(value is None for value in self.model_dump(exclude={"host"}).values())
+
 
 @dataclass(frozen=True)
 class ResourceAggregate:
@@ -227,21 +248,6 @@ def _sum_over_hosts(values: list[int | None]) -> int | None:
     return sum(value for value in values if value is not None)
 
 
-def _is_failed_reading(reading: WorkerResourceReading) -> bool:
-    """A reading carrying a host but no figure at all is a failed self-read.
-
-    A healthy host always reports its logical CPU count and memory capacity; only a
-    read that failed after its retries carries every figure as ``None``. Such a
-    reading is not a real contribution and must not null the fleet.
-    """
-    return (
-        reading.processor_available is None
-        and reading.processor_assigned is None
-        and reading.memory_total is None
-        and reading.memory_available is None
-    )
-
-
 def aggregate(readings: Iterable[WorkerResourceReading]) -> ResourceAggregate:
     """Collapse per-process readings into one figure per field for a component.
 
@@ -253,7 +259,7 @@ def aggregate(readings: Iterable[WorkerResourceReading]) -> ResourceAggregate:
     """
     by_host: dict[str, WorkerResourceReading] = {}
     for reading in readings:
-        if _is_failed_reading(reading):
+        if reading.is_failed:
             continue
         by_host.setdefault(reading.host, reading)