feat(telemetry): licensing resource-allocation telemetry [INFP-589]

backend/infrahub/services/component.py

@@ -3,14 +3,15 @@
 import re
 from typing import TYPE_CHECKING, Any
 
-from attr import dataclass
+from attr import Factory, dataclass
 
 from infrahub.components import ComponentType
 from infrahub.core.constants import GLOBAL_BRANCH_NAME
 from infrahub.core.registry import registry
 from infrahub.core.timestamp import Timestamp
 from infrahub.log import get_logger
 from infrahub.message_bus.types import KVTTL
+from infrahub.telemetry.resources import ProcessResources, WorkerResourceReading
 from infrahub.worker import WORKER_IDENTITY
 
 if TYPE_CHECKING:
@@ -20,6 +21,17 @@
 
 PRIMARY_API_SERVER = "workers:primary:api_server"
 WORKER_MATCH = re.compile(r":worker:([^:]+)")
+RESOURCE_COMPONENT_MATCH = re.compile(r"workers:resources:([^:]+):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()
 
@@ -30,6 +42,7 @@ class InfrahubComponent:
     db: InfrahubDatabase
     message_bus: InfrahubMessageBus
     component_type: ComponentType
+    process_resources: ProcessResources = Factory(ProcessResources)
 
     @classmethod
     async def new(
@@ -106,12 +119,64 @@ async def refresh_heartbeat(self) -> None:
                 value=Timestamp().to_string(),
                 expires=KVTTL.FIFTEEN,
             )
+            await self.cache.set(
+                key=f"workers:resources:{component}:worker:{WORKER_IDENTITY}",
+                value=self._read_own_resources().model_dump_json(),
+                expires=KVTTL.FIFTEEN,
+            )
         if self.component_type == ComponentType.API_SERVER:
             await self._set_primary_api_server()
         await self.cache.set(
             key=f"workers:worker:{WORKER_IDENTITY}", value=Timestamp().to_string(), expires=KVTTL.TWO_HOURS
         )
 
+    def _read_own_resources(self) -> WorkerResourceReading:
+        """Read this process's resource allocation, retrying a transient failure.
+
+        A read that still fails after its retries is logged with the component and
+        the failing source, then reported as a null-valued reading so a worker that
+        silently stops reporting resources leaves a trace rather than only an
+        aggregate undercount.
+        """
+        last_error: Exception | None = None
+        for _ in range(RESOURCE_READ_MAX_ATTEMPTS):
+            try:
+                return self.process_resources.read()
+            except Exception as exc:
+                last_error = exc
+
+        log.warning(
+            "Unable to read process resource allocation for telemetry; reporting null",
+            component_type=self.component_type.name,
+            worker_id=WORKER_IDENTITY,
+            error=str(last_error),
+        )
+        return WorkerResourceReading(host=_UNKNOWN_HOST)
+
+    async def read_worker_resources(self) -> dict[str, dict[str, WorkerResourceReading]]:
+        """Return the latest worker resource readings grouped by component and host.
+
+        Readings that fail to parse are skipped; the several processes of one host
+        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:*")
+        values = await self.cache.get_values(keys=keys)
+
+        grouped: dict[str, dict[str, WorkerResourceReading]] = {}
+        for key, value in zip(keys, values, strict=False):
+            if value is None:
+                continue
+            match = RESOURCE_COMPONENT_MATCH.search(key)
+            if not match:
+                continue
+            try:
+                reading = WorkerResourceReading.model_validate_json(value)
+            except ValueError:
+                continue
+            grouped.setdefault(match.group(1), {})[reading.host] = reading
+        return grouped
+
     async def _set_primary_api_server(self) -> None:
         result = await self.cache.set(
             key=PRIMARY_API_SERVER, value=WORKER_IDENTITY, expires=KVTTL.FIFTEEN, not_exists=True

backend/infrahub/telemetry/database.py

@@ -13,6 +13,11 @@
 from .models import TelemetryDatabaseData, TelemetryDatabaseServerData, TelemetryDatabaseSystemInfoData
 from .utils import safe_metric
 
+# Neo4j setting capping Cypher query parallelism. It defaults to 0 (auto = use
+# every available core), which is not an enforced limit and is reported as an
+# absent assignment; a positive value is the configured cap.
+DB_WORKER_LIMIT_SETTING = "server.cypher.parallel.worker_limit"
+
 
 async def get_server_info(db: InfrahubDatabase) -> list[TelemetryDatabaseServerData]:
     data: list[TelemetryDatabaseServerData] = []
@@ -33,6 +38,46 @@ async def get_server_info(db: InfrahubDatabase) -> list[TelemetryDatabaseServerD
     return data
 
 
+def _worker_limit_from_value(value: object) -> int | None:
+    """Interpret a raw ``worker_limit`` setting value as a configured core cap.
+
+    ``0`` (auto) is not an enforced limit and maps to ``None``; a positive integer
+    is the configured cap. An absent, non-numeric, or non-positive value is also
+    reported as no configured limit.
+    """
+    if not isinstance(value, (str, int)):
+        return None
+    try:
+        limit = int(value)
+    except ValueError:
+        return None
+    return limit if limit > 0 else None
+
+
+async def get_processor_assigned(db: InfrahubDatabase) -> int | None:
+    """Read the configured Cypher-parallelism core cap, or ``None`` when unbounded.
+
+    A missing setting or a non-positive/unparseable value maps to ``None`` — the
+    same reading a deployment with no configured limit yields. A failure to run the
+    query is left to raise so the caller's degradation boundary logs it, rather than
+    being swallowed silently here.
+    """
+    query = """
+    SHOW SETTINGS YIELD name, value
+    WHERE name = $setting_name
+    RETURN value AS value
+    """
+    results = await db.execute_query(
+        query=query,
+        params={"setting_name": DB_WORKER_LIMIT_SETTING},
+        name="get_processor_assigned",
+        type=QueryType.READ,
+    )
+    if not results:
+        return None
+    return _worker_limit_from_value(results[0]["value"])
+
+
 async def get_system_info(db: InfrahubDatabase) -> TelemetryDatabaseSystemInfoData:
     query = """
     CALL dbms.queryJmx("java.lang:type=OperatingSystem")
@@ -48,6 +93,10 @@ async def get_system_info(db: InfrahubDatabase) -> TelemetryDatabaseSystemInfoDa
         memory_total=results[0]["memory_total"]["value"],
         memory_available=results[0]["memory_available"]["value"],
         processor_available=results[0]["processor_available"]["value"],
+        # The assigned read is a separate source from the JMX figures above; a failure
+        # to reach it must null only this field rather than the whole system-info block,
+        # so it degrades independently even when it raises outside its own catch.
+        processor_assigned=await safe_metric(get_processor_assigned(db=db)),
     )
 
 

backend/infrahub/telemetry/models.py

@@ -1,5 +1,3 @@
-from typing import Self
-
 from pydantic import BaseModel, Field
 
 from .constants import InfrahubType
@@ -8,6 +6,10 @@
 class TelemetryWorkerData(BaseModel):
     total: int
     active: int
+    processor_available: int | None = None
+    processor_assigned: int | None = None
+    memory_total: int | None = None
+    memory_available: int | None = None
 
 
 class TelemetryBranchData(BaseModel):
@@ -19,10 +21,6 @@ class TelemetryAccountData(BaseModel):
     active: int | None = Field(default=None)
     groups: int | None = Field(default=None)
 
-    @classmethod
-    def default(cls) -> Self:
-        return cls()
-
 
 class TelemetryActivity24hData(BaseModel):
     logins: int | None = Field(default=None)
@@ -38,10 +36,6 @@ class TelemetryActivity24hData(BaseModel):
     webhooks_fired_success: int | None = Field(default=None)
     webhooks_fired_failure: int | None = Field(default=None)
 
-    @classmethod
-    def default(cls) -> Self:
-        return cls()
-
 
 class TelemetrySchemaData(BaseModel):
     node_count: int
@@ -58,6 +52,14 @@ class TelemetryDatabaseSystemInfoData(BaseModel):
     memory_total: int
     memory_available: int
     processor_available: int
+    processor_assigned: int | None = None
+
+
+class TelemetryServerData(BaseModel):
+    processor_available: int | None = None
+    processor_assigned: int | None = None
+    memory_total: int | None = None
+    memory_available: int | None = None
 
 
 class TelemetryDatabaseData(BaseModel):
@@ -89,9 +91,10 @@ class TelemetryData(BaseModel):
     python_version: str
     platform: str
     workers: TelemetryWorkerData
+    server: TelemetryServerData = Field(default_factory=TelemetryServerData)
     branches: TelemetryBranchData
-    accounts: TelemetryAccountData = Field(default_factory=TelemetryAccountData.default)
-    activity_24h: TelemetryActivity24hData = Field(default_factory=TelemetryActivity24hData.default)
+    accounts: TelemetryAccountData = Field(default_factory=TelemetryAccountData)
+    activity_24h: TelemetryActivity24hData = Field(default_factory=TelemetryActivity24hData)
     features: dict[str, int]
     schema_info: TelemetrySchemaData
     database: TelemetryDatabaseData