Merge pull request #170 from permitio/omer/per-10669-adapt-pdp-ci-to-…

…autouse-the-go-service

omer/per-10669-adapt-pdp-ci-to-autouse-the-go-service

.github/workflows/pdp_cicd.yml

@@ -149,6 +149,14 @@ jobs:
         cd ./permit-opa
         find * -name '*go*' -print0 | xargs -0 tar -czf "$build_root"/custom/custom_opa.tar.gz --exclude '.*'
 
+    - uses: robinraju/release-downloader@v1
+      with:
+        repository: permitio/datasync
+        latest: true
+        fileName: factstore_server*
+        token: ${{ secrets.CLONE_REPO_TOKEN }}
+        out-file-path: "factstore_server"
+
     - name: Build and push PDP image - (pre-release)
       if: "github.event.release.prerelease"
       uses: docker/build-push-action@v5

.github/workflows/pdp_cicd_manual.yml

@@ -0,0 +1,68 @@
+name: Build and Push PDP Docker Image Manually
+
+on:
+  workflow_dispatch:
+    inputs:
+      tag:
+        required: true
+        description: 'The tag for the image'
+        type: string
+
+jobs:
+  build-and-push-pdp:
+    runs-on: ubuntu-latest
+    env:
+      IMAGE_TAG: ${{ inputs.tag }}
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up QEMU
+      uses: docker/setup-qemu-action@v3
+
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@v3
+
+    - name: Login to Docker Hub
+      uses: docker/login-action@v3
+      with:
+        username: ${{ secrets.DOCKERHUB_USERNAME }}
+        password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+    - uses: actions/checkout@v3
+      with:
+        repository: permitio/permit-opa
+        ref: main
+        path: './permit-opa'
+        token: ${{ secrets.CLONE_REPO_TOKEN }}
+
+    - name: Pre build PDP
+      run: |
+        echo "${IMAGE_TAG}" | cut -d '-' -f 1 > permit_pdp_version
+        rm -rf custom
+        mkdir custom
+        build_root="$PWD"
+        cd ./permit-opa
+        find * -name '*go*' -print0 | xargs -0 tar -czf "$build_root"/custom/custom_opa.tar.gz --exclude '.*'
+        if [[ "${IMAGE_TAG}" != *alpha* ]]; then
+          echo "Alpha version not detected, exiting..."
+          exit 1
+        fi
+
+    - uses: robinraju/release-downloader@v1
+      with:
+        repository: permitio/datasync
+        latest: true
+        fileName: factstore_server*
+        token: ${{ secrets.CLONE_REPO_TOKEN }}
+        out-file-path: "factstore_server"
+
+    - name: Build and push PDP image - (pre-release)
+      uses: docker/build-push-action@v5
+      with:
+        push: true
+        context: .
+        platforms: linux/amd64,linux/arm64
+        tags: permitio/pdp-v2:${{ inputs.tag }}
+        cache-from: type=gha
+        cache-to: type=gha,mode=max

Dockerfile

@@ -67,6 +67,10 @@ RUN mkdir -p /config
 RUN chown -R permit:permit /opa
 RUN chown -R permit:permit /config
 
+COPY factstore_server/factstore_server-linux-amd64 horizon/data_manager/data_manager-amd
+COPY factstore_server/factstore_server-linux-arm64 horizon/data_manager/data_manager-arm
+RUN chmod +x horizon/data_manager/data_manager-amd
+RUN chmod +x horizon/data_manager/data_manager-arm
 # copy wait-for-it (use only for development! e.g: docker compose)
 COPY scripts/wait-for-it.sh /usr/wait-for-it.sh
 RUN chmod +x /usr/wait-for-it.sh

horizon/config.py

@@ -2,6 +2,7 @@
 
 MOCK_API_KEY = "MUST BE DEFINED"
 
+
 # scopes enum
 class ApiKeyLevel(str):
     ORGANIZATION = "organization"
@@ -45,12 +46,6 @@ def __new__(cls, prefix=None, is_model=True):
         "DATA_MANAGER_ENABLE_REMOTE_BACKUP is true",
     )
 
-    DATA_MANAGER_TOKEN = confi.str(
-        "DATA_MANAGER_TOKEN",
-        "secret",
-        description="The Data Manager service token for internal data manager communication",
-    )
-
     SHARD_ID = confi.str(
         "SHARD_ID",
         None,

horizon/data_manager/client.py

@@ -5,7 +5,7 @@
 from loguru import logger
 from opal_client import OpalClient
 from opal_client.callbacks.api import init_callbacks_api
-from opal_client.config import opal_client_config
+from opal_client.config import opal_client_config, EngineLogFormat
 from opal_client.data.api import init_data_router
 from opal_client.data.updater import DataUpdater
 from opal_client.engine.options import OpaServerOptions, CedarServerOptions
@@ -110,13 +110,14 @@ def __init__(
     ):
         self._data_manager_runner = DataManagerRunner(
             data_manager_url=sidecar_config.DATA_MANAGER_SERVICE_URL,
-            data_manager_token=sidecar_config.DATA_MANAGER_TOKEN,
+            data_manager_token=opal_client_config.CLIENT_TOKEN,
             data_manager_remote_backup_enabled=sidecar_config.DATA_MANAGER_ENABLE_REMOTE_BACKUP,
             data_manager_remote_backup_url=sidecar_config.DATA_MANAGER_REMOTE_BACKUP_URL,
             engine_token=sidecar_config.API_KEY,
+            piped_logs_format=EngineLogFormat.FULL,
         )
         policy_store = policy_store or DataManagerPolicyStoreClient(
-            data_manager_client=self._data_manager_runner.client,
+            data_manager_client=lambda: self._data_manager_runner.client,
             opa_server_url=opal_client_config.POLICY_STORE_URL,
             opa_auth_token=opal_client_config.POLICY_STORE_AUTH_TOKEN,
             auth_type=opal_client_config.POLICY_STORE_AUTH_TYPE,

horizon/data_manager/data_update.py

@@ -1,4 +1,5 @@
-from typing import Any, Iterator, Self
+from typing import Any, Iterator
+from typing_extensions import Self
 
 from pydantic import BaseModel
 

horizon/data_manager/policy_store.py

@@ -1,5 +1,5 @@
 import time
-from typing import Optional, Iterator
+from typing import Optional, Iterator, Callable
 
 import aiohttp
 from aiohttp import ClientSession
@@ -20,7 +20,7 @@
 class DataManagerPolicyStoreClient(OpaClient):
     def __init__(
         self,
-        data_manager_client: ClientSession,
+        data_manager_client: ClientSession | Callable[[], ClientSession],
         opa_server_url=None,
         opa_auth_token: Optional[str] = None,
         auth_type: PolicyStoreAuth = PolicyStoreAuth.NONE,
@@ -50,6 +50,12 @@ def __init__(
         )
         self._client = data_manager_client
 
+    @property
+    def client(self):
+        if isinstance(self._client, ClientSession):
+            return self._client
+        return self._client()
+
     async def set_policy_data(
         self,
         policy_data: JsonableValue,
@@ -94,7 +100,7 @@ async def _apply_data_update(
         self, data_update: DataUpdate
     ) -> aiohttp.ClientResponse:
         start_time = time.perf_counter_ns()
-        res = await self._client.post(
+        res = await self.client.post(
             "/v1/facts/applyUpdate",
             json=data_update.dict(),
         )

horizon/data_manager/runner.py

@@ -1,10 +1,14 @@
+import json
+import logging
 import os
 import platform
 from pathlib import Path
 
 import aiohttp
 from opal_client.config import EngineLogFormat
+from opal_client.engine.logger import logging_level_from_string, log_entire_dict
 from opal_client.engine.runner import PolicyEngineRunner
+from opal_client.logger import logger
 
 
 class DataManagerRunner(PolicyEngineRunner):
@@ -16,13 +20,15 @@ def __init__(
         data_manager_remote_backup_enabled: bool,
         data_manager_remote_backup_url: str | None,
         piped_logs_format: EngineLogFormat = EngineLogFormat.NONE,
+        binary_file_name: str = "data_manager",
     ):
         super().__init__(piped_logs_format=piped_logs_format)
         self._engine_token = engine_token
         self._data_manager_url = data_manager_url
         self._data_manager_token = data_manager_token
         self._data_manager_remote_backup_enabled = data_manager_remote_backup_enabled
         self._data_manager_remote_backup_url = data_manager_remote_backup_url
+        self._binary_file_name = binary_file_name
         self._client = None
 
     @property
@@ -34,6 +40,17 @@ def client(self) -> aiohttp.ClientSession:
             )
         return self._client
 
+    async def handle_log_line(self, line: bytes) -> None:
+        try:
+            log_line = json.loads(line)
+            level = logging.getLevelName(
+                logging_level_from_string(log_line.pop("level", "info"))
+            )
+            msg = log_line.pop("msg", None)
+            log_entire_dict(level, msg, log_line)
+        except json.JSONDecodeError:
+            logger.info(line.decode("utf-8"))
+
     async def __aenter__(self):
         self.set_envs()
         await super().__aenter__()
@@ -78,9 +95,9 @@ def command(self) -> str:
 
         arch = platform.machine()
         if arch == "x86_64":
-            binary_path = "data_manager-amd"
+            binary_path = f"{self._binary_file_name}-amd"
         elif arch == "arm64" or arch == "aarch64":
-            binary_path = "data_manager-arm"
+            binary_path = f"{self._binary_file_name}-arm"
         else:
             raise ValueError(f"Unsupported architecture: {arch}")
         return os.path.join(current_dir, binary_path)

horizon/data_manager/update_operations.py

@@ -57,7 +57,7 @@ def _get_operations_for_update_role_assigment(
                         type="role_assignments",
                         attributes={
                             "id": str(uuid4()),
-                            "actor": user_key,
+                            "actor": f"user:{user_key}",
                             "tenant": subject_key,
                             "role": role_key,
                             "resource": "",
@@ -70,7 +70,7 @@ def _get_operations_for_update_role_assigment(
                         type="role_assignments",
                         attributes={
                             "id": str(uuid4()),
-                            "actor": user_key,
+                            "actor": f"user:{user_key}",
                             "tenant": "",
                             "role": role_key,
                             "resource": subject,
@@ -88,7 +88,7 @@ def _get_operations_for_update_user(
             fact=Fact(
                 type="users",
                 attributes={
-                    "id": user_key,
+                    "id": f"user:{user_key}",
                     "attributes": json.dumps(attributes),
                     # TODO remove the json.dumps after fixing the map[string]string issue in Go
                 },
@@ -100,13 +100,13 @@ def _get_operations_for_update_user(
         yield DeleteOperation(
             fact=Fact(
                 type="users",
-                attributes={"id": user_key},
+                attributes={"id": f"user:{user_key}"},
             ),
         )
         yield DeleteOperation(
             fact=Fact(
                 type="role_assignments",
-                attributes={"actor": user_key},
+                attributes={"actor": f"user:{user_key}"},
             ),
         )