Merge branch 'refs/heads/main' into release/0.9.0

Author: danyi1212

horizon/config.py

@@ -265,6 +265,16 @@ def parse_plugins(value: Any) -> dict[str, dict[str, int | bool | str]]:
         description="The path to the file that contains the PDP version",
     )
 
+    HORIZON_NICENESS = confi.int(
+        "HORIZON_NICENESS",
+        10,
+        description=(
+            "The niceness value for the PDP Horizon process (Python process). "
+            "Niceness values range from -20 (highest priority) to 19 (lowest priority) with 0 is neutral. "
+            "Adjusting this can help manage CPU resource allocation. "
+        ),
+    )
+
     @staticmethod
     def parse_callbacks(value: Any) -> list[CallbackEntry]:
         if isinstance(value, str):

horizon/facts/opal_forwarder.py

@@ -1,6 +1,6 @@
 from functools import cache
 from urllib.parse import urljoin
-from uuid import uuid4
+from uuid import UUID
 
 from opal_common.fetcher.providers.http_fetch_provider import HttpFetcherConfig
 from opal_common.schemas.data import DataSourceEntry, DataUpdate
@@ -34,6 +34,8 @@ def create_data_source_entry(
     obj_id: str,
     obj_key: str,
     authorization_header: str,
+    *,
+    update_id: UUID,
 ) -> DataSourceEntry:
     obj_id = obj_id.replace("-", "")  # convert UUID to Hex
     url = urljoin(
@@ -45,6 +47,7 @@ def create_data_source_entry(
 
     headers = {
         "Authorization": authorization_header,
+        "X-Permit-Update-Id": update_id.hex,
     }
     if sidecar_config.SHARD_ID:
         headers["X-Shard-Id"] = sidecar_config.SHARD_ID
@@ -59,10 +62,15 @@ def create_data_source_entry(
     )
 
 
-def create_data_update_entry(entries: list[DataSourceEntry]) -> DataUpdate:
+def create_data_update_entry(
+    entries: list[DataSourceEntry],
+    *,
+    update_id: UUID,
+) -> DataUpdate:
     entries_text = ", ".join(entry.dst_path for entry in entries)
+
     return DataUpdate(
-        id=uuid4().hex,
+        id=update_id.hex,
         entries=entries,
         reason=f"Local facts upload for {entries_text}",
     )

horizon/facts/router.py

@@ -1,5 +1,6 @@
 from collections.abc import Callable, Iterable
 from typing import Any
+from uuid import UUID, uuid4
 
 from fastapi import (
     APIRouter,
@@ -45,12 +46,13 @@ async def create_user(
         update_subscriber,
         wait_timeout,
         path="/users",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="users",
                 obj_id=body["id"],
                 obj_key=body["key"],
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             )
         ],
         timeout_policy=timeout_policy,
@@ -71,12 +73,13 @@ async def create_tenant(
         update_subscriber,
         wait_timeout,
         path="/tenants",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="tenants",
                 obj_id=body["id"],
                 obj_key=body["key"],
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             )
         ],
         timeout_policy=timeout_policy,
@@ -98,12 +101,13 @@ async def sync_user(
         update_subscriber,
         wait_timeout,
         path=f"/users/{user_id}",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="users",
                 obj_id=body["id"],
                 obj_key=body["key"],
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             )
         ],
         timeout_policy=timeout_policy,
@@ -125,31 +129,36 @@ async def update_user(
         update_subscriber,
         wait_timeout,
         path=f"/users/{user_id}",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="users",
                 obj_id=body["id"],
                 obj_key=body["key"],
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             )
         ],
         timeout_policy=timeout_policy,
     )
 
 
-def create_role_assignment_data_entries(request: FastApiRequest, body: dict[str, Any]) -> Iterable[DataSourceEntry]:
+def create_role_assignment_data_entries(
+    request: FastApiRequest, body: dict[str, Any], update_id: UUID | None
+) -> Iterable[DataSourceEntry]:
     if not body.get("resource_instance"):
         yield create_data_source_entry(
             obj_type="role_assignments",
             obj_id=body["user_id"],
             obj_key=f"user:{body['user']}",
             authorization_header=request.headers.get("Authorization"),
+            update_id=update_id,
         )
         yield create_data_source_entry(
             obj_type="users",
             obj_id=body["user_id"],
             obj_key=body["user"],
             authorization_header=request.headers.get("Authorization"),
+            update_id=update_id,
         )
     else:
         # note that user_id == subject_id,
@@ -159,6 +168,7 @@ def create_role_assignment_data_entries(request: FastApiRequest, body: dict[str,
             obj_id=body["user_id"],
             obj_key=f"user:{body['user']}",
             authorization_header=request.headers.get("Authorization"),
+            update_id=update_id,
         )
 
 
@@ -215,12 +225,13 @@ async def create_resource_instance(
         update_subscriber,
         wait_timeout,
         path="/resource_instances",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="resource_instances",
                 obj_id=body["id"],
                 obj_key=f"{body['resource']}:{body['key']}",
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             ),
         ],
         timeout_policy=timeout_policy,
@@ -242,12 +253,13 @@ async def update_resource_instance(
         update_subscriber,
         wait_timeout,
         path=f"/resource_instances/{instance_id}",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="resource_instances",
                 obj_id=body["id"],
                 obj_key=f"{body['resource']}:{body['key']}",
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             ),
         ],
         timeout_policy=timeout_policy,
@@ -268,12 +280,13 @@ async def create_relationship_tuple(
         update_subscriber,
         wait_timeout,
         path="/relationship_tuples",
-        entries_callback=lambda r, body: [
+        entries_callback=lambda r, body, update_id: [
             create_data_source_entry(
                 obj_type="relationships",
                 obj_id=body["object_id"],
                 obj_key=body["object"],
                 authorization_header=r.headers.get("Authorization"),
+                update_id=update_id,
             ),
         ],
         timeout_policy=timeout_policy,
@@ -287,16 +300,20 @@ async def forward_request_then_wait_for_update(
     wait_timeout: float | None,
     *,
     path: str,
-    entries_callback: Callable[[FastApiRequest, dict[str, Any]], Iterable[DataSourceEntry]],
+    update_id: UUID | None = None,
+    entries_callback: Callable[[FastApiRequest, dict[str, Any], UUID | None], Iterable[DataSourceEntry]],
     timeout_policy: TimeoutPolicy = TimeoutPolicy.IGNORE,
 ) -> Response:
+    _update_id = update_id or uuid4()
     response = await client.send_forward_request(request, path)
     body = client.extract_body(response)
     if body is None:
         return client.convert_response(response)
 
     try:
-        data_update_entry = create_data_update_entry(list(entries_callback(request, body)))
+        data_update_entry = create_data_update_entry(
+            list(entries_callback(request, body, _update_id)), update_id=_update_id
+        )
     except KeyError as e:
         logger.warning(f"Missing required field {e.args[0]} in the response body, skipping wait for update.")
         return client.convert_response(response)
@@ -315,6 +332,7 @@ async def forward_request_then_wait_for_update(
         )
     else:
         logger.warning("Timeout waiting for update and policy is set to ignore")
+        return client.convert_response(response)
 
 
 @facts_router.api_route(

horizon/pdp.py

@@ -1,4 +1,5 @@
 import logging
+import os
 import sys
 from pathlib import Path
 from uuid import UUID, uuid4
@@ -45,6 +46,37 @@
 OPA_LOGGER_MODULE = "opal_client.opa.logger"
 
 
+def set_process_niceness(target_nice: int) -> None:
+    """
+    Attempts to set the current process's niceness value to `target_nice`.
+    This operation is performed only once during the call.
+
+    This function is idempotent if the current niceness already equals `target_nice`.
+    Setting a lower niceness value (increasing priority) may require CAP_SYS_NICE
+    capabilities and could fail if the process lacks sufficient privileges.
+    """
+    if target_nice < -20 or target_nice > 19:
+        raise ValueError(f"Target niceness must be between -20 and 19, got {target_nice}")
+
+    try:
+        current_niceness = os.nice(0)  # Read current niceness without changing it
+        delta = target_nice - current_niceness
+        if delta != 0:
+            os.nice(delta)  # Apply the change
+            new_niceness = os.nice(0)  # Read the new niceness to confirm
+            logging.info(
+                "Changed the process niceness by %d from %d to %d (target was %d).",
+                delta,
+                current_niceness,
+                new_niceness,
+                target_nice,
+            )
+        else:
+            logging.debug("Process niceness is already %d, which matches the target; no change made.", current_niceness)
+    except OSError as exc:
+        logging.warning("Failed to change process niceness to %d: %s", target_nice, exc)
+
+
 def apply_config(overrides_dict: dict, config_object: Confi):
     """
     apply config values from dict into a confi object
@@ -131,6 +163,9 @@ def __init__(self):
         if sidecar_config.ENABLE_MONITORING:
             self._configure_monitoring()
 
+        if sidecar_config.HORIZON_NICENESS:
+            set_process_niceness(sidecar_config.HORIZON_NICENESS)
+
         self._opal = OpalClient(shard_id=sidecar_config.SHARD_ID, data_topics=self._fix_data_topics())
         self._inject_extra_callbacks()
         # remove default data update callbacks that are not needed and might be managed by the control plane