Remove no org (#2368)

Co-authored-by: Andrew Svetlov <andrew.svetlov@gmail.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: neuro-auto-bot <68340147+neuro-auto-bot@users.noreply.github.com>

Author: 4 people

PLATFORMSECRETS_IMAGE

@@ -1 +1 @@
-ghcr.io/neuro-inc/platformsecrets:25.8.1
+ghcr.io/neuro-inc/platformsecrets:latest

platform_api/api.py

@@ -82,12 +82,6 @@ async def handle_config(self, request: aiohttp.web.Request) -> aiohttp.web.Respo
 
         If the requesting user is authorized, the response will contain the details
         about the user's orgs, clusters, and projects.
-
-        In case the user has direct access to a cluster outside of any org,
-        the list of orgs will not have a None entry, but the cluster will have a None
-        entry in its orgs list.
-
-        Similarly, a project in the response can have a None org.
         """
         data: dict[str, Any] = {}
 

platform_api/config.py

@@ -11,9 +11,6 @@
 
 STORAGE_URI_SCHEME = "storage"
 
-NO_ORG = "NO_ORG"
-NO_ORG_NORMALIZED = "no-org"
-
 
 @dataclass(frozen=True)
 class RegistryConfig:

platform_api/handlers/jobs_handler.py

@@ -21,7 +21,7 @@
 from neuro_config_client import Cluster, ResourcePreset, TPUResource
 from yarl import URL
 
-from platform_api.config import NO_ORG, STORAGE_URI_SCHEME, Config
+from platform_api.config import STORAGE_URI_SCHEME, Config
 from platform_api.log import log_debug_time
 from platform_api.orchestrator.job import (
     JOB_NAME_SEPARATOR,
@@ -250,16 +250,15 @@ def _set_preset_resources(payload: dict[str, Any]) -> dict[str, Any]:
 def create_job_cluster_org_name_validator(
     *,
     default_cluster_name: str,
-    default_org_name: str | None,
+    default_org_name: str,
     default_project_name: str,
 ) -> t.Trafaret:
     return t.Dict(
         {
             t.Key(
                 "cluster_name", default=default_cluster_name
             ): create_cluster_name_validator(),
-            t.Key("org_name", default=default_org_name): create_org_name_validator()
-            | t.Null,
+            t.Key("org_name", default=default_org_name): create_org_name_validator(),
             t.Key(
                 "project_name", default=default_project_name
             ): create_project_name_validator(),
@@ -550,7 +549,7 @@ def infer_permissions_from_container(
     container: Container,
     registry_host: str,
     cluster_name: str,
-    org_name: str | None,
+    org_name: str,
     *,
     project_name: str,
 ) -> list[Permission]:
@@ -563,7 +562,7 @@ def infer_permissions_from_container(
     if container.belongs_to_registry(registry_host):
         permissions.append(
             Permission(
-                uri=str(container.to_image_uri(registry_host, cluster_name)),
+                uri=str(container.to_image_uri(registry_host, cluster_name, org_name)),
                 action="read",
             )
         )
@@ -695,26 +694,51 @@ async def create_job(self, request: aiohttp.web.Request) -> aiohttp.web.Response
         cluster_configs = await self._jobs_service.get_user_cluster_configs(user)
         self._check_user_can_submit_jobs(cluster_configs)
         default_cluster_name = cluster_configs[0].config.name
-        cluster_for_default_org = (
-            orig_payload.get("cluster_name") or default_cluster_name
-        )
-        cluster_config_for_default_org = next(
-            (
-                cluster_config
-                for cluster_config in cluster_configs
-                if cluster_config.config.name == cluster_for_default_org
-            ),
-            None,
-        )
-        # always use NO_ORG as default if a user has direct access to cluster.
-        # if cluster_config_for_default_org is None,
-        # the validator below will raise an error
-        default_org_name = None
-        if (
-            cluster_config_for_default_org is not None
-            and None not in cluster_config_for_default_org.orgs
-        ):
-            default_org_name = cluster_config_for_default_org.orgs[0]
+
+        # Check if user explicitly specified a cluster they don't have access to
+        requested_cluster = orig_payload.get("cluster_name")
+        if requested_cluster is not None:
+            cluster_config_for_default_org = next(
+                (
+                    cluster_config
+                    for cluster_config in cluster_configs
+                    if cluster_config.config.name == requested_cluster
+                ),
+                None,
+            )
+            if cluster_config_for_default_org is None:
+                raise aiohttp.web.HTTPForbidden(
+                    text=json.dumps(
+                        {
+                            "error": (
+                                "User is not allowed to submit jobs to the "
+                                "specified cluster"
+                            )
+                        }
+                    ),
+                    content_type="application/json",
+                )
+            if not cluster_config_for_default_org.orgs:
+                raise aiohttp.web.HTTPForbidden(
+                    text=json.dumps(
+                        {
+                            "error": (
+                                "User is not allowed to submit jobs to the "
+                                "specified cluster as a member of given organization"
+                            )
+                        }
+                    ),
+                    content_type="application/json",
+                )
+        else:
+            cluster_config_for_default_org = cluster_configs[0]
+            if not cluster_config_for_default_org.orgs:
+                raise aiohttp.web.HTTPForbidden(
+                    text=json.dumps({"error": "User must have at least one org"}),
+                    content_type="application/json",
+                )
+
+        default_org_name = cluster_config_for_default_org.orgs[0]
 
         job_cluster_org_name_validator = create_job_cluster_org_name_validator(
             default_cluster_name=default_cluster_name,
@@ -1108,7 +1132,7 @@ def create_from_query(self, query: MultiDictProxy) -> JobFilter:  # type: ignore
                 for cluster_name in query.getall("cluster_name", [])
             }
             orgs = {
-                self._parse_org_name(org_name)
+                self._org_name_validator.check(org_name)
                 for org_name in query.getall("org_name", [])
             }
             projects = {
@@ -1154,13 +1178,6 @@ def create_from_query(self, query: MultiDictProxy) -> JobFilter:  # type: ignore
             **bool_filters,  # type: ignore
         )
 
-    def _parse_org_name(self, org_name: str) -> str | None:
-        return (
-            None
-            if org_name.upper() == NO_ORG
-            else self._org_name_validator.check(org_name)
-        )
-
 
 @dataclass(frozen=True)
 class BulkJobFilter:
@@ -1180,11 +1197,11 @@ def __init__(
         self._has_access_to_all: bool = False
         self._has_clusters_shared_all: bool = False
         self._has_orgs_shared_all: bool = False
-        self._clusters_shared_any: dict[str, dict[str | None, dict[str, set[str]]]] = (
+        self._clusters_shared_any: dict[str, dict[str, dict[str, set[str]]]] = (
             defaultdict(lambda: defaultdict(lambda: defaultdict(set)))
         )
         self._projects_shared_any: set[str] = set()
-        self._orgs_shared_any: set[str | None] = set()
+        self._orgs_shared_any: set[str] = set()
         self._shared_ids: set[str] = set()
 
     def build(self) -> BulkJobFilter:
@@ -1239,10 +1256,6 @@ def _traverse_clusters(self, tree: ClientAccessSubTreeView) -> None:
                 self._clusters_shared_any[cluster_name] = {}
             else:
                 self._traverse_orgs(sub_tree, cluster_name)
-                if self._query_filter.orgs and None not in self._query_filter.orgs:
-                    # skipping None org
-                    continue
-                self._traverse_projects(sub_tree, cluster_name, None)
 
     def _traverse_orgs(self, tree: ClientAccessSubTreeView, cluster_name: str) -> None:
         for org_name, sub_tree in tree.children.items():
@@ -1267,7 +1280,7 @@ def _traverse_projects(
         self,
         tree: ClientAccessSubTreeView,
         cluster_name: str,
-        org_name: str | None,
+        org_name: str,
     ) -> None:
         for project, sub_tree in tree.children.items():
             if not sub_tree.can_list():
@@ -1294,7 +1307,7 @@ def _traverse_jobs(
         self,
         tree: ClientAccessSubTreeView,
         cluster_name: str,
-        org_name: str | None,
+        org_name: str,
         project_name: str,
     ) -> None:
         for name, sub_tree in tree.children.items():
@@ -1365,7 +1378,7 @@ def _create_bulk_filter(self) -> JobFilter | None:
 
     def _optimize_clusters_projects(
         self,
-        orgs: AbstractSet[str | None],
+        orgs: AbstractSet[str],
         projects: AbstractSet[str],
         name: str | None,
     ) -> None:

platform_api/orchestrator/job.py

@@ -15,7 +15,6 @@
 from neuro_config_client.entities import DEFAULT_ENERGY_SCHEDULE_NAME
 from yarl import URL
 
-from platform_api.config import NO_ORG
 from platform_api.old_kube_client.apolo import generate_namespace_name
 
 from .job_request import (
@@ -290,7 +289,7 @@ class JobRecord:
     project_name: str
     org_project_hash: bytes
     namespace: str
-    org_name: str | None = None
+    org_name: str
     name: str | None = None
     preset_name: str | None = None
     tags: Sequence[str] = ()
@@ -335,18 +334,18 @@ def create(
         if not kwargs.get("project_name"):
             kwargs["project_name"] = get_base_owner(kwargs["owner"])
 
-        org_name = kwargs.get("org_name")
+        org_name = kwargs["org_name"]
         project_name = kwargs["project_name"]
 
         kwargs["org_project_hash"] = cls._create_org_project_hash(
             org_name, project_name
         )
-        kwargs["namespace"] = generate_namespace_name(org_name or NO_ORG, project_name)
+        kwargs["namespace"] = generate_namespace_name(org_name, project_name)
         return cls(**kwargs)
 
     @classmethod
-    def _create_org_project_hash(cls, org_name: str | None, project_name: str) -> bytes:
-        return cls._create_hash(org_name or NO_ORG, project_name)[:5]
+    def _create_org_project_hash(cls, org_name: str, project_name: str) -> bytes:
+        return cls._create_hash(org_name, project_name)[:5]
 
     @classmethod
     def _create_hash(cls, *args: str) -> bytes:
@@ -540,7 +539,7 @@ def from_primitive(
             request.job_id, payload
         )
         owner = payload.get("owner") or orphaned_job_owner
-        org_name = payload.get("org_name", None)
+        org_name = payload["org_name"]
         project_name = payload["project_name"]
         org_project_hash = payload.get("org_project_hash")
         if org_project_hash and isinstance(org_project_hash, str):
@@ -949,7 +948,7 @@ def get_total_price_credits(self) -> Decimal:
         return runtime_hours * self.price_credits_per_hour
 
     @property
-    def org_name(self) -> str | None:
+    def org_name(self) -> str:
         return self._record.org_name
 
     @property

platform_api/orchestrator/job_request.py

@@ -10,8 +10,6 @@
 from neuro_config_client import ResourcePoolType, ResourcePreset
 from yarl import URL
 
-from platform_api.config import NO_ORG_NORMALIZED
-
 
 class JobException(Exception):
     pass
@@ -83,23 +81,8 @@ def to_uri(self) -> URL:
         )
 
     def to_permission_uri(self) -> str:
-        """
-        Permission URI must not include a `no-org`.
-        This method can be removed whenever we'll get rid of `no-org` concept.
-        """
-        joined_path = self.path
-        if "/" in self.path:
-            path = []
-            org_name, project_name, *parts = self.path.split("/")
-            if org_name == NO_ORG_NORMALIZED:
-                path.extend([project_name, *parts])
-            else:
-                path.extend([org_name, project_name, *parts])
-            joined_path = "/".join(path)
         return str(
-            URL.build(scheme="disk", host=self.cluster_name)
-            / joined_path
-            / self.disk_id
+            URL.build(scheme="disk", host=self.cluster_name) / self.path / self.disk_id
         )
 
     @classmethod
@@ -175,8 +158,6 @@ def create(cls, secret_uri: str | URL) -> "Secret":
 
     @staticmethod
     def path_with_org(path: str) -> str:
-        if "/" not in path:
-            path = f"{NO_ORG_NORMALIZED}/{path}"
         return path
 
 
@@ -448,13 +429,14 @@ def belongs_to_registry(self, registry_host: str) -> bool:
         prefix = f"{registry_host}/"
         return self.image.startswith(prefix)
 
-    def to_image_uri(self, registry_host: str, cluster_name: str) -> URL:
+    def to_image_uri(self, registry_host: str, cluster_name: str, org_name: str) -> URL:
         assert self.belongs_to_registry(registry_host), "Unknown registry"
         prefix = f"{registry_host}/"
         repo = self.image[len(prefix) :]
         path, *_ = repo.split(":", 1)
         assert cluster_name
-        return URL.build(scheme="image", host=cluster_name) / path
+        assert org_name
+        return URL.build(scheme="image", host=cluster_name) / org_name / path
 
     def get_secrets(self) -> list[Secret]:
         return list(

platform_api/orchestrator/jobs_poller.py

@@ -131,7 +131,6 @@ def _parse_container(data: Mapping[str, Any]) -> Container:
             http_server=http_server,
         )
 
-    project_name = payload["project_name"]
     return JobRecord(
         request=JobRequest(
             job_id=payload["id"],
@@ -143,8 +142,8 @@ def _parse_container(data: Mapping[str, Any]) -> Container:
             [_parse_status_item(item) for item in payload["statuses"]]
         ),
         cluster_name=payload["cluster_name"],
-        org_name=payload.get("org_name"),
-        project_name=project_name,
+        org_name=payload["org_name"],
+        project_name=payload["project_name"],
         org_project_hash=bytes.fromhex(payload["org_project_hash"]),
         namespace=payload["namespace"],
         name=payload.get("name"),

platform_api/orchestrator/jobs_service.py

@@ -91,8 +91,7 @@ def create_for_org(cls, org: str) -> "NoCreditsError":
 @dataclass(frozen=True)
 class UserClusterConfig:
     config: Cluster
-    # None value means the direct access to cluster without any or:
-    orgs: list[str | None]
+    orgs: list[str]
 
 
 @dataclass(frozen=True)
@@ -503,7 +502,8 @@ async def _get_user_cluster_configs(
         cluster_configs = await self._get_user_cluster_configs_by_name(response)
         cluster_to_orgs = defaultdict(list)
         for user_cluster in response.clusters:
-            cluster_to_orgs[user_cluster.cluster_name].append(user_cluster.org_name)
+            if user_cluster.org_name is not None:
+                cluster_to_orgs[user_cluster.cluster_name].append(user_cluster.org_name)
         for cluster_name, orgs in cluster_to_orgs.items():
             if cluster_config := cluster_configs.get(cluster_name):
                 configs.append(UserClusterConfig(config=cluster_config, orgs=orgs))

platform_api/orchestrator/jobs_storage/base.py

@@ -28,7 +28,7 @@ def __init__(self, job_name: str, project_name: str, found_job_id: str):
         )
 
 
-ClusterOrgProjectNameSet = dict[str, dict[str | None, dict[str, AbstractSet[str]]]]
+ClusterOrgProjectNameSet = dict[str, dict[str, dict[str, AbstractSet[str]]]]
 
 
 @dataclass(frozen=True)
@@ -39,9 +39,7 @@ class JobFilter:
     clusters: ClusterOrgProjectNameSet = field(
         default_factory=cast(type[ClusterOrgProjectNameSet], dict)
     )
-    orgs: AbstractSet[str | None] = field(
-        default_factory=cast(type[AbstractSet[str | None]], set)
-    )
+    orgs: AbstractSet[str] = field(default_factory=cast(type[AbstractSet[str]], set))
     owners: AbstractSet[str] = field(default_factory=cast(type[AbstractSet[str]], set))
     projects: AbstractSet[str] = field(
         default_factory=cast(type[AbstractSet[str]], set)