Dev

CHANGELOG.md

@@ -1,3 +1,14 @@
+# v0.8.4
+
+### Features
+- [x] Add basic config in helm chart and CLI for Istio.
+- [x] Add worker route to get pieces repository.
+
+
+### Fixes
+- [x] Add classic authorization to default route for get pieces repositories.
+
+
 # v0.8.3
 
 ### Features

helm/domino/Chart.yaml

@@ -2,8 +2,8 @@ apiVersion: v2
 name: domino
 description: A Helm chart for Domino
 type: application
-version: 0.1.9
-appVersion: 0.1.9
+version: 0.1.10
+appVersion: 0.1.10
 home: https://github.com/Tauffer-Consulting/domino
 sources:
   - https://github.com/Tauffer-Consulting/domino

helm/domino/templates/jobs/domino-migrations.yml

@@ -8,6 +8,9 @@ spec:
   parallelism: 1
   backoffLimit: 4
   template:
+    metadata:
+      annotations:
+        sidecar.istio.io/inject: "false"
     spec:
       restartPolicy: OnFailure
       containers:

rest/routers/piece_repository_router.py

@@ -120,8 +120,7 @@ def get_piece_repository_release_data(
         status.HTTP_500_INTERNAL_SERVER_ERROR: {'model': SomethingWrongError},
         status.HTTP_403_FORBIDDEN: {'model': ForbiddenError},
     },
-    # TODO - I commented this to make it easier to test, but we should solve the auth service
-    # dependencies=[Depends(auth_service.workspace_access_authorizer)]
+    dependencies=[Depends(auth_service.workspace_access_authorizer)]
 )
 def get_pieces_repositories(
     workspace_id: int,
@@ -142,6 +141,39 @@ def get_pieces_repositories(
         raise HTTPException(status_code=e.status_code, detail=e.message)
 
 
+@router.get(
+    path="/worker",
+    status_code=status.HTTP_200_OK,
+    responses={
+        status.HTTP_200_OK: {'model': GetWorkspaceRepositoriesResponse},
+        status.HTTP_500_INTERNAL_SERVER_ERROR: {'model': SomethingWrongError},
+        status.HTTP_403_FORBIDDEN: {'model': ForbiddenError},
+    },
+)
+def get_pieces_repositories_worker(
+    workspace_id: int,
+    page: Optional[int] = 0,
+    page_size: Optional[int] = 100,
+    filters: ListRepositoryFilters = Depends(),
+) -> GetWorkspaceRepositoriesResponse:
+    """
+    Get pieces repositories for workspace.
+    This endpoint is used by the worker to get the repositories to be processed.
+    Is the same endpoint as the one above, but without the auth service.
+    The authorization is done by our service mesh Authorization Policy.
+    """
+    try:
+        response = piece_repository_service.get_pieces_repositories(
+            workspace_id=workspace_id,
+            page=page,
+            page_size=page_size,
+            filters=filters
+        )
+        return response
+    except (BaseException, ForbiddenException) as e:
+        raise HTTPException(status_code=e.status_code, detail=e.message)
+
+
 @router.delete(
     path="/{piece_repository_id}",
     status_code=status.HTTP_204_NO_CONTENT,

rest/routers/secret_router.py

@@ -70,7 +70,7 @@ def update_repository_secret(
         raise HTTPException(status_code=e.status_code, detail=e.message)
 
 @router.get(
-    path='/{piece_name}',
+    path='/{piece_name}/secrets-values', # using sufix /secrets-values only because istio does not support wildcards in paths
     status_code=200,
     responses={
         status.HTTP_200_OK: {'model': List[GetSecretsByPieceResponse]},
@@ -79,13 +79,15 @@ def update_repository_secret(
     },
     include_in_schema=False
 )
-#@auth_service.authorize_repository_workspace_access # TODO authorize only worker
 def get_piece_secrets(
     piece_repository_id: int,
-    piece_name: str, # TODO check what is better to use. query or path ?
-    #auth_context: AuthorizationContextData = Depends(auth_service.auth_wrapper)
+    piece_name: str,
 ) -> List[GetSecretsByPieceResponse]:
-    """Get secrets for a specific Piece from an piece repository, in a specific workspace"""
+    """
+    Get secrets values for a specific Piece from an piece repository, in a specific workspace
+    This endpoint is not using authorization service because it is used by airflow to get secrets values
+    In production this endpoint should be blocked from external access using security strategies like authorization policies.
+    """
     try:
         response = secret_service.get_piece_secrets(
             piece_repository_id=piece_repository_id,

rest/tests/secret/fixtures.py

@@ -33,7 +33,7 @@ def patch_piece_secret(client: ApiTestClient, authorization_token: Dict, piece_r
 @pytest.fixture(scope="function")
 def get_secrets_by_piece_name(request, client: ApiTestClient, authorization_token: Dict, piece_repository: PieceRepository):
     return client.get(
-        f"/pieces-repositories/{piece_repository.id}/secrets/{request.param['piece_name']}",
+        f"/pieces-repositories/{piece_repository.id}/secrets/{request.param['piece_name']}/secrets-values",
         headers={"Authorization": authorization_token["header"]}
     )
 
@@ -72,5 +72,5 @@ def get_repository_secrets_mock_response():
         # )
     ]
     return mock_response
-    
+
 

rest/tests/secret/test_secret_router.py

@@ -9,7 +9,7 @@
     "tests.secret.fixtures"
 ]
 @pytest.mark.usefixtures("register", "login", "add_piece_repository", "teardown_piece_repository")
-class TestSecretRouter:    
+class TestSecretRouter:
     @staticmethod
     def test_get_repository_secrets(get_repository_secrets: Response, get_repository_secrets_mock_response: List):
         mock_response = get_repository_secrets_mock_response
@@ -26,18 +26,18 @@ def test_get_repository_secrets(get_repository_secrets: Response, get_repository
             secret_content = json.loads(secret.model_dump_json())
             mock_response_secrets_names.append(secret_content.get("name"))
         mock_response_secrets_names.sort()
-        
+
         assert response.status_code == 200
 
         mock_response_content = json.loads(mock_response[0].model_dump_json())
         assert content[0].keys() == mock_response_content.keys()
         assert response_secrets_names == mock_response_secrets_names
-    
+
     @staticmethod
     def test_patch_piece_secret(patch_piece_secret: Response):
         response = patch_piece_secret
         assert response.status_code == 204
-    
+
     @staticmethod
     @pytest.mark.parametrize("get_secrets_by_piece_name", [{"piece_name": "SimpleLogPiece"}], indirect=True)
     def test_get_secrets_by_piece_name(get_secrets_by_piece_name: Response):
@@ -55,4 +55,3 @@ def test_get_secrets_by_piece_name(get_secrets_by_piece_name: Response):
         assert response.status_code == 200
         assert content[0] == mock_response_content
 
-

src/domino/VERSION

@@ -1 +1 @@
-0.8.3
+0.8.4

src/domino/cli/utils/platform.py

@@ -397,6 +397,22 @@ def create_platform(install_airflow: bool = True, use_gpu: bool = False) -> None
                 "sshKeySecret": "airflow-ssh-secret"
             },
         },
+        "migrateDatabaseJob": {
+            "jobAnnotations": {
+                "sidecar.istio.io/inject": "false"
+            },
+            "annotations": {
+                "sidecar.istio.io/inject": "false"
+            },
+        },
+        "createUserJob": {
+            "jobAnnotations": {
+                "sidecar.istio.io/inject": "false"
+            },
+            "annotations": {
+                "sidecar.istio.io/inject": "false"
+            },
+        },
         **workers,
         **scheduler,
     }
@@ -417,7 +433,7 @@ def create_platform(install_airflow: bool = True, use_gpu: bool = False) -> None
                 "-f", str(fp.name),
                 "airflow",
                 "apache-airflow/airflow",
-                "--version", " 1.9.0",
+                "--version", " 1.11.0",
             ]
             subprocess.run(commands)
 

src/domino/client/domino_backend_client.py

@@ -31,30 +31,22 @@ def health_check(self) -> requests.Response:
         return response
 
     def get_piece_secrets(self, piece_repository_id: int, piece_name: str) -> requests.Response:
-        resource = f"/pieces-repositories/{piece_repository_id}/secrets/{piece_name}"
-        response = self.request(
-            method='get',
-            resource=resource
-        )
-        return response
-
-    def get_piece_repository(self, piece_repository_id: int) -> requests.Response:
-        resource = f"/pieces-repositories/{piece_repository_id}"
+        resource = f"/pieces-repositories/{piece_repository_id}/secrets/{piece_name}/secrets-values"
         response = self.request(
             method='get',
             resource=resource
         )
         return response
 
     def get_piece_repositories_from_workspace_id(self, params: dict) -> requests.Response:
-        resource = "/pieces-repositories"
+        resource = "/pieces-repositories/worker"
         response = self.request(
             method='get',
             resource=resource,
             params=params
         )
         return response
-    
+
     def check_create_airflow_connection(self, conn_id: str, conn_type: str):
         """
         This should check if a specific Airflow connection exists and create it if it doesn't.

src/domino/task.py

@@ -103,6 +103,7 @@ def _set_operator(self) -> BaseOperator:
             # - https://www.astronomer.io/guides/templating/
             # - good example: https://github.com/apache/airflow/blob/main/tests/system/providers/cncf/kubernetes/example_kubernetes.py
             # - commands HAVE to go in a list object: https://stackoverflow.com/a/55149915/11483674
+
             return DominoKubernetesPodOperator(
                 dag_id=self.dag_id,
                 task_id=self.task_id,
@@ -115,11 +116,12 @@ def _set_operator(self) -> BaseOperator:
                 workflow_shared_storage=self.workflow_shared_storage,
                 container_resources=self.container_resources,
                 # ----------------- Kubernetes -----------------
-                namespace='default',  # TODO - separate namespace by User or Workspace?
+                namespace='default',
                 image=self.piece.get("source_image"),
                 image_pull_policy='IfNotPresent',
                 name=f"airflow-worker-pod-{self.task_id}",
                 startup_timeout_seconds=600,
+                annotations={"sidecar.istio.io/inject": "false"}, # TODO - remove this when istio is working with airflow k8s pod
                 # cmds=["/bin/bash"],
                 # arguments=["-c", "sleep 120;"],
                 cmds=["domino"],