feat: add rss router serving cached resource status (#940)

* feat: add rss router serving cached resource status

Adds the concrete resource status sources (storage, compute, FTS, site)
backed by ResourceStatusDB and serves them through new
/api/rss/{storage,compute,site,fts} endpoints. Responses are read from
an all-VO snapshot cached per source, filtered to the caller's VO, and
carry ETag/Last-Modified headers with 304 support; any authenticated
user may read. Includes the regenerated diracx and gubbins clients.

Closes #839

Co-authored-by: Loxeris <30194187+Loxeris@users.noreply.github.com>

* fix(rss): filter revision query by element_type

Compute and FTS sources share the all status type, so latest_revision()
computed identical revisions for both and counted rows from the other
element type. Pass element_type to get_resource_status_date() so the
revision tracks exactly the rows read_raw returns.

---------

Co-authored-by: Loxeris <30194187+Loxeris@users.noreply.github.com>

Author: chrisburr

diracx-client/src/diracx/client/_generated/_client.py

@@ -15,7 +15,7 @@
 from . import models as _models
 from ._configuration import DiracConfiguration
 from ._utils.serialization import Deserializer, Serializer
-from .operations import AuthOperations, ConfigOperations, JobsOperations, WellKnownOperations
+from .operations import AuthOperations, ConfigOperations, JobsOperations, RssOperations, WellKnownOperations
 
 
 class Dirac:  # pylint: disable=client-accepts-api-version-keyword
@@ -29,6 +29,8 @@ class Dirac:  # pylint: disable=client-accepts-api-version-keyword
     :vartype config: _generated.operations.ConfigOperations
     :ivar jobs: JobsOperations operations
     :vartype jobs: _generated.operations.JobsOperations
+    :ivar rss: RssOperations operations
+    :vartype rss: _generated.operations.RssOperations
     :keyword endpoint: Service URL. Required. Default value is "".
     :paramtype endpoint: str
     """
@@ -65,6 +67,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self.auth = AuthOperations(self._client, self._config, self._serialize, self._deserialize)
         self.config = ConfigOperations(self._client, self._config, self._serialize, self._deserialize)
         self.jobs = JobsOperations(self._client, self._config, self._serialize, self._deserialize)
+        self.rss = RssOperations(self._client, self._config, self._serialize, self._deserialize)
 
     def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs: Any) -> HttpResponse:
         """Runs the network request through the client's chained policies.

diracx-client/src/diracx/client/_generated/aio/_client.py

@@ -15,7 +15,7 @@
 from .. import models as _models
 from .._utils.serialization import Deserializer, Serializer
 from ._configuration import DiracConfiguration
-from .operations import AuthOperations, ConfigOperations, JobsOperations, WellKnownOperations
+from .operations import AuthOperations, ConfigOperations, JobsOperations, RssOperations, WellKnownOperations
 
 
 class Dirac:  # pylint: disable=client-accepts-api-version-keyword
@@ -29,6 +29,8 @@ class Dirac:  # pylint: disable=client-accepts-api-version-keyword
     :vartype config: _generated.aio.operations.ConfigOperations
     :ivar jobs: JobsOperations operations
     :vartype jobs: _generated.aio.operations.JobsOperations
+    :ivar rss: RssOperations operations
+    :vartype rss: _generated.aio.operations.RssOperations
     :keyword endpoint: Service URL. Required. Default value is "".
     :paramtype endpoint: str
     """
@@ -65,6 +67,7 @@ def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
         self.auth = AuthOperations(self._client, self._config, self._serialize, self._deserialize)
         self.config = ConfigOperations(self._client, self._config, self._serialize, self._deserialize)
         self.jobs = JobsOperations(self._client, self._config, self._serialize, self._deserialize)
+        self.rss = RssOperations(self._client, self._config, self._serialize, self._deserialize)
 
     def send_request(
         self, request: HttpRequest, *, stream: bool = False, **kwargs: Any

diracx-client/src/diracx/client/_generated/aio/operations/__init__.py

@@ -14,6 +14,7 @@
 from ._operations import AuthOperations  # type: ignore
 from ._operations import ConfigOperations  # type: ignore
 from ._operations import JobsOperations  # type: ignore
+from ._operations import RssOperations  # type: ignore
 
 from ._patch import __all__ as _patch_all
 from ._patch import *
@@ -24,6 +25,7 @@
     "AuthOperations",
     "ConfigOperations",
     "JobsOperations",
+    "RssOperations",
 ]
 __all__.extend([p for p in _patch_all if p not in __all__])  # pyright: ignore
 _patch_sdk()

diracx-client/src/diracx/client/_generated/aio/operations/_operations.py

@@ -51,6 +51,10 @@
     build_jobs_summary_request,
     build_jobs_unassign_bulk_jobs_sandboxes_request,
     build_jobs_unassign_job_sandboxes_request,
+    build_rss_get_compute_status_request,
+    build_rss_get_fts_status_request,
+    build_rss_get_site_status_request,
+    build_rss_get_storage_status_request,
     build_well_known_get_installation_metadata_request,
     build_well_known_get_jwks_request,
     build_well_known_get_openid_configuration_request,
@@ -2319,3 +2323,303 @@ async def submit_jdl_jobs(self, body: Union[list[str], IO[bytes]], **kwargs: Any
             return cls(pipeline_response, deserialized, {})  # type: ignore
 
         return deserialized  # type: ignore
+
+
+class RssOperations:
+    """
+    .. warning::
+        **DO NOT** instantiate this class directly.
+
+        Instead, you should access the following operations through
+        :class:`~_generated.aio.Dirac`'s
+        :attr:`rss` attribute.
+    """
+
+    models = _models
+
+    def __init__(self, *args, **kwargs) -> None:
+        input_args = list(args)
+        self._client: AsyncPipelineClient = input_args.pop(0) if input_args else kwargs.pop("client")
+        self._config: DiracConfiguration = input_args.pop(0) if input_args else kwargs.pop("config")
+        self._serialize: Serializer = input_args.pop(0) if input_args else kwargs.pop("serializer")
+        self._deserialize: Deserializer = input_args.pop(0) if input_args else kwargs.pop("deserializer")
+
+    @distributed_trace_async
+    async def get_storage_status(
+        self,
+        *,
+        if_modified_since: Optional[str] = None,
+        etag: Optional[str] = None,
+        match_condition: Optional[MatchConditions] = None,
+        **kwargs: Any
+    ) -> dict[str, _models.StorageElementStatus]:
+        """Get Storage Status.
+
+        Get the latest status of storage elements, scoped to the caller's VO.
+
+        :keyword if_modified_since: Default value is None.
+        :paramtype if_modified_since: str
+        :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is
+         None.
+        :paramtype etag: str
+        :keyword match_condition: The match condition to use upon the etag. Default value is None.
+        :paramtype match_condition: ~azure.core.MatchConditions
+        :return: dict mapping str to StorageElementStatus
+        :rtype: dict[str, ~_generated.models.StorageElementStatus]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        if match_condition == MatchConditions.IfNotModified:
+            error_map[412] = ResourceModifiedError
+        elif match_condition == MatchConditions.IfPresent:
+            error_map[412] = ResourceNotFoundError
+        elif match_condition == MatchConditions.IfMissing:
+            error_map[412] = ResourceExistsError
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[dict[str, _models.StorageElementStatus]] = kwargs.pop("cls", None)
+
+        _request = build_rss_get_storage_status_request(
+            if_modified_since=if_modified_since,
+            etag=etag,
+            match_condition=match_condition,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        _stream = False
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            raise HttpResponseError(response=response)
+
+        deserialized = self._deserialize("{StorageElementStatus}", pipeline_response.http_response)
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    async def get_compute_status(
+        self,
+        *,
+        if_modified_since: Optional[str] = None,
+        etag: Optional[str] = None,
+        match_condition: Optional[MatchConditions] = None,
+        **kwargs: Any
+    ) -> dict[str, _models.ComputeElementStatus]:
+        """Get Compute Status.
+
+        Get the latest status of compute elements, scoped to the caller's VO.
+
+        :keyword if_modified_since: Default value is None.
+        :paramtype if_modified_since: str
+        :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is
+         None.
+        :paramtype etag: str
+        :keyword match_condition: The match condition to use upon the etag. Default value is None.
+        :paramtype match_condition: ~azure.core.MatchConditions
+        :return: dict mapping str to ComputeElementStatus
+        :rtype: dict[str, ~_generated.models.ComputeElementStatus]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        if match_condition == MatchConditions.IfNotModified:
+            error_map[412] = ResourceModifiedError
+        elif match_condition == MatchConditions.IfPresent:
+            error_map[412] = ResourceNotFoundError
+        elif match_condition == MatchConditions.IfMissing:
+            error_map[412] = ResourceExistsError
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[dict[str, _models.ComputeElementStatus]] = kwargs.pop("cls", None)
+
+        _request = build_rss_get_compute_status_request(
+            if_modified_since=if_modified_since,
+            etag=etag,
+            match_condition=match_condition,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        _stream = False
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            raise HttpResponseError(response=response)
+
+        deserialized = self._deserialize("{ComputeElementStatus}", pipeline_response.http_response)
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    async def get_site_status(
+        self,
+        *,
+        if_modified_since: Optional[str] = None,
+        etag: Optional[str] = None,
+        match_condition: Optional[MatchConditions] = None,
+        **kwargs: Any
+    ) -> dict[str, _models.SiteStatus]:
+        """Get Site Status.
+
+        Get the latest status of sites, scoped to the caller's VO.
+
+        :keyword if_modified_since: Default value is None.
+        :paramtype if_modified_since: str
+        :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is
+         None.
+        :paramtype etag: str
+        :keyword match_condition: The match condition to use upon the etag. Default value is None.
+        :paramtype match_condition: ~azure.core.MatchConditions
+        :return: dict mapping str to SiteStatus
+        :rtype: dict[str, ~_generated.models.SiteStatus]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        if match_condition == MatchConditions.IfNotModified:
+            error_map[412] = ResourceModifiedError
+        elif match_condition == MatchConditions.IfPresent:
+            error_map[412] = ResourceNotFoundError
+        elif match_condition == MatchConditions.IfMissing:
+            error_map[412] = ResourceExistsError
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[dict[str, _models.SiteStatus]] = kwargs.pop("cls", None)
+
+        _request = build_rss_get_site_status_request(
+            if_modified_since=if_modified_since,
+            etag=etag,
+            match_condition=match_condition,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        _stream = False
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            raise HttpResponseError(response=response)
+
+        deserialized = self._deserialize("{SiteStatus}", pipeline_response.http_response)
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore
+
+    @distributed_trace_async
+    async def get_fts_status(
+        self,
+        *,
+        if_modified_since: Optional[str] = None,
+        etag: Optional[str] = None,
+        match_condition: Optional[MatchConditions] = None,
+        **kwargs: Any
+    ) -> dict[str, _models.FTSStatus]:
+        """Get Fts Status.
+
+        Get the latest status of FTS servers, scoped to the caller's VO.
+
+        :keyword if_modified_since: Default value is None.
+        :paramtype if_modified_since: str
+        :keyword etag: check if resource is changed. Set None to skip checking etag. Default value is
+         None.
+        :paramtype etag: str
+        :keyword match_condition: The match condition to use upon the etag. Default value is None.
+        :paramtype match_condition: ~azure.core.MatchConditions
+        :return: dict mapping str to FTSStatus
+        :rtype: dict[str, ~_generated.models.FTSStatus]
+        :raises ~azure.core.exceptions.HttpResponseError:
+        """
+        error_map: MutableMapping = {
+            401: ClientAuthenticationError,
+            404: ResourceNotFoundError,
+            409: ResourceExistsError,
+            304: ResourceNotModifiedError,
+        }
+        if match_condition == MatchConditions.IfNotModified:
+            error_map[412] = ResourceModifiedError
+        elif match_condition == MatchConditions.IfPresent:
+            error_map[412] = ResourceNotFoundError
+        elif match_condition == MatchConditions.IfMissing:
+            error_map[412] = ResourceExistsError
+        error_map.update(kwargs.pop("error_map", {}) or {})
+
+        _headers = kwargs.pop("headers", {}) or {}
+        _params = kwargs.pop("params", {}) or {}
+
+        cls: ClsType[dict[str, _models.FTSStatus]] = kwargs.pop("cls", None)
+
+        _request = build_rss_get_fts_status_request(
+            if_modified_since=if_modified_since,
+            etag=etag,
+            match_condition=match_condition,
+            headers=_headers,
+            params=_params,
+        )
+        _request.url = self._client.format_url(_request.url)
+
+        _stream = False
+        pipeline_response: PipelineResponse = await self._client._pipeline.run(  # pylint: disable=protected-access
+            _request, stream=_stream, **kwargs
+        )
+
+        response = pipeline_response.http_response
+
+        if response.status_code not in [200]:
+            map_error(status_code=response.status_code, response=response, error_map=error_map)
+            raise HttpResponseError(response=response)
+
+        deserialized = self._deserialize("{FTSStatus}", pipeline_response.http_response)
+
+        if cls:
+            return cls(pipeline_response, deserialized, {})  # type: ignore
+
+        return deserialized  # type: ignore