✨ Use S3 links in internal computational backend cluster, prepare for temporary tokens (⚠️ devops) (ITISFoundation#3006)

Author: sanderegg

api/specs/storage/openapi.yaml

@@ -383,6 +383,26 @@ paths:
         default:
           $ref: "#/components/responses/DefaultErrorResponse"
 
+  /simcore-s3:access:
+    post:
+      summary: Returns the temporary access credentials for the user storage space
+      operationId: get_or_create_temporary_s3_access
+      parameters:
+        - name: user_id
+          in: query
+          required: true
+          schema:
+            type: integer
+      responses:
+        "200":
+          description: the S3 access credentials
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/S3AccessCredentialsEnveloped"
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
+
   /simcore-s3/files/metadata:search:
     post:
       summary: Returns metadata for all files matching a pattern
@@ -880,6 +900,35 @@ components:
       example:
         link: "example_link"
 
+    S3AccessCredentialsEnveloped:
+      type: object
+      required:
+        - data
+        - error
+      properties:
+        data:
+          $ref: "#/components/schemas/S3AccessCredentials"
+        error:
+          nullable: true
+          default: null
+
+    S3AccessCredentials:
+      type: object
+      required:
+        - access
+        - secret
+        - token
+        - endpoint
+      properties:
+        access:
+          type: string
+        secret:
+          type: string
+        token:
+          type: string
+        endpoint:
+          type: string
+
     Project:
       $ref: "../common/schemas/project.yaml#/components/schemas/ProjectIn"
 

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/io.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Union, cast
 
+from models_library.basic_regex import MIME_TYPE_RE
 from models_library.generics import DictModel
 from models_library.services import PROPERTY_KEY_RE
 from pydantic import (
@@ -51,7 +52,7 @@ class Config(PortSchema.Config):
                 },
                 {
                     "required": False,
-                    "url": "ftp://some_file_url",
+                    "url": "s3://another_file_url",
                 },
             ]
         }
@@ -63,15 +64,20 @@ class FileUrl(BaseModel):
         None,
         description="Local file relpath name (if given), otherwise it takes the url filename",
     )
+    file_mime_type: Optional[str] = Field(
+        None, description="the file MIME type", regex=MIME_TYPE_RE
+    )
 
     class Config:
         extra = Extra.forbid
         schema_extra = {
             "examples": [
+                {"url": "https://some_file_url", "file_mime_type": "application/json"},
                 {
                     "url": "https://some_file_url",
+                    "file_mapping": "some_file_name.txt",
+                    "file_mime_type": "application/json",
                 },
-                {"url": "s3://some_file_url", "file_mapping": "some_file_name.txt"},
             ]
         }
 
@@ -98,7 +104,7 @@ class Config(DictModel.Config):
                     "int_input": -45,
                     "float_input": 4564.45,
                     "string_input": "nobody thinks like a string",
-                    "file_input": {"url": "s3://some_file_url"},
+                    "file_input": {"url": "s3://thatis_file_url"},
                 },
             ]
         }

packages/models-library/src/models_library/basic_regex.py

@@ -25,3 +25,7 @@
 #  - cannot start with spaces, _ (we only want public) or numbers
 # https://docs.python.org/3/reference/lexical_analysis.html#identifiers
 PUBLIC_VARIABLE_NAME_RE = r"^[^_\W0-9]\w*$"
+
+MIME_TYPE_RE = (
+    r"([\w\*]*)\/(([\w\-\*]+\.)+)?([\w\-\*]+)(\+([\w\-\.]+))?(; ([\w+-\.=]+))?"
+)

packages/models-library/src/models_library/projects_nodes_io.py

@@ -88,12 +88,12 @@ class BaseFileLink(BaseModel):
     )
 
     label: Optional[str] = Field(
-        None,
+        default=None,
         description="The real file name",
     )
 
     e_tag: Optional[str] = Field(
-        None,
+        default=None,
         description="Entity tag that uniquely represents the file. The method to generate the tag is not specified (black box).",
         alias="eTag",
     )
@@ -103,7 +103,7 @@ class SimCoreFileLink(BaseFileLink):
     """I/O port type to hold a link to a file in simcore S3 storage"""
 
     dataset: Optional[str] = Field(
-        None,
+        default=None,
         deprecated=True
         # TODO: Remove with storage refactoring
     )

packages/pytest-simcore/src/pytest_simcore/services_api_mocks_for_aiohttp_clients.py

@@ -334,10 +334,13 @@ async def storage_v0_service_mock(
 
     def get_download_link_cb(url: URL, **kwargs) -> CallbackResult:
         file_id = url.path.rsplit("/files/")[1]
-
+        assert "params" in kwargs
+        assert "link_type" in kwargs["params"]
+        link_type = kwargs["params"]["link_type"]
+        scheme = {"presigned": "http", "s3": "s3"}
         return CallbackResult(
             status=web.HTTPOk.status_code,
-            payload={"data": {"link": f"file://{file_id}"}},
+            payload={"data": {"link": f"{scheme[link_type]}://{file_id}"}},
         )
 
     get_file_metadata_pattern = re.compile(

packages/simcore-sdk/src/simcore_sdk/node_ports_common/filemanager.py

@@ -37,29 +37,31 @@ async def _get_download_link(
     store_id: str,
     file_id: str,
     session: ClientSession,
+    link_type: storage_client.LinkType,
 ) -> URL:
-    presigned_link: AnyUrl = await storage_client.get_download_file_presigned_link(
-        session, file_id, store_id, user_id
+    link: AnyUrl = await storage_client.get_download_file_link(
+        session, file_id, store_id, user_id, link_type
     )
-    if not presigned_link:
+    if not link:
         raise exceptions.S3InvalidPathError(file_id)
 
-    return URL(presigned_link)
+    return URL(link)
 
 
 async def _get_upload_link(
     user_id: int,
     store_id: str,
     file_id: str,
     session: ClientSession,
+    link_type: storage_client.LinkType,
 ) -> URL:
-    presigned_link: AnyUrl = await storage_client.get_upload_file_presigned_link(
-        session, file_id, store_id, user_id
+    link: AnyUrl = await storage_client.get_upload_file_link(
+        session, file_id, store_id, user_id, link_type
     )
-    if not presigned_link:
+    if not link:
         raise exceptions.S3InvalidPathError(file_id)
 
-    return URL(presigned_link)
+    return URL(link)
 
 
 async def _download_link_to_file(session: ClientSession, url: URL, file_path: Path):
@@ -140,6 +142,7 @@ async def get_download_link_from_s3(
     store_name: Optional[str],
     store_id: Optional[str],
     s3_object: str,
+    link_type: storage_client.LinkType,
     client_session: Optional[ClientSession] = None,
 ) -> Optional[URL]:
     if store_name is None and store_id is None:
@@ -151,7 +154,9 @@ async def get_download_link_from_s3(
                 user_id, store_name, session
             )
         assert store_id is not None  # nosec
-        return await _get_download_link(user_id, store_id, s3_object, session)
+        return await _get_download_link(
+            user_id, store_id, s3_object, session, link_type
+        )
 
 
 async def get_upload_link_from_s3(
@@ -160,6 +165,7 @@ async def get_upload_link_from_s3(
     store_name: Optional[str],
     store_id: Optional[str],
     s3_object: str,
+    link_type: storage_client.LinkType,
     client_session: Optional[ClientSession] = None,
 ) -> Tuple[str, URL]:
     if store_name is None and store_id is None:
@@ -173,7 +179,7 @@ async def get_upload_link_from_s3(
         assert store_id is not None  # nosec
         return (
             store_id,
-            await _get_upload_link(user_id, store_id, s3_object, session),
+            await _get_upload_link(user_id, store_id, s3_object, session, link_type),
         )
 
 
@@ -210,6 +216,7 @@ async def download_file_from_s3(
             store_id=store_id,
             s3_object=s3_object,
             client_session=session,
+            link_type=storage_client.LinkType.PRESIGNED,
         )
 
         # the link contains the file name
@@ -274,6 +281,7 @@ async def upload_file(
             store_id=store_id,
             s3_object=s3_object,
             client_session=session,
+            link_type=storage_client.LinkType.PRESIGNED,
         )
 
         if not upload_link:

packages/simcore-sdk/src/simcore_sdk/node_ports_common/storage_client.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from functools import wraps
 from json import JSONDecodeError
 from typing import Any, Callable, Dict
@@ -61,9 +62,18 @@ async def get_storage_locations(
         return locations_enveloped.data
 
 
+class LinkType(str, Enum):
+    PRESIGNED = "presigned"
+    S3 = "s3"
+
+
 @handle_client_exception
-async def get_download_file_presigned_link(
-    session: ClientSession, file_id: str, location_id: str, user_id: UserID
+async def get_download_file_link(
+    session: ClientSession,
+    file_id: str,
+    location_id: str,
+    user_id: UserID,
+    link_type: LinkType,
 ) -> AnyUrl:
     if (
         not isinstance(file_id, str)
@@ -80,7 +90,7 @@ async def get_download_file_presigned_link(
 
     async with session.get(
         f"{_base_url()}/locations/{location_id}/files/{quote(file_id, safe='')}",
-        params={"user_id": f"{user_id}"},
+        params={"user_id": f"{user_id}", "link_type": link_type.value},
     ) as response:
         response.raise_for_status()
 
@@ -93,8 +103,12 @@ async def get_download_file_presigned_link(
 
 
 @handle_client_exception
-async def get_upload_file_presigned_link(
-    session: ClientSession, file_id: str, location_id: str, user_id: UserID
+async def get_upload_file_link(
+    session: ClientSession,
+    file_id: str,
+    location_id: str,
+    user_id: UserID,
+    link_type: LinkType,
 ) -> AnyUrl:
     if (
         not isinstance(file_id, str)
@@ -110,7 +124,7 @@ async def get_upload_file_presigned_link(
         )
     async with session.put(
         f"{_base_url()}/locations/{location_id}/files/{quote(file_id, safe='')}",
-        params={"user_id": f"{user_id}"},
+        params={"user_id": f"{user_id}", "link_type": link_type.value},
     ) as response:
         response.raise_for_status()
 

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/__init__.py

@@ -4,6 +4,7 @@
 from ..node_ports_common import config as node_config
 from ..node_ports_common import exceptions
 from ..node_ports_common.dbmanager import DBManager
+from ..node_ports_common.storage_client import LinkType as FileLinkType
 from .nodeports_v2 import Nodeports
 from .port import Port
 from .serialization_v2 import load
@@ -33,4 +34,4 @@ async def ports(
     )
 
 
-__all__ = ["ports", "node_config", "exceptions", "Port"]
+__all__ = ("ports", "node_config", "exceptions", "Port", "FileLinkType")

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/nodeports_v2.py

@@ -5,6 +5,7 @@
 
 from pydantic import BaseModel, Field
 from servicelib.utils import logged_gather
+from simcore_sdk.node_ports_common.storage_client import LinkType
 
 from ..node_ports_common.dbmanager import DBManager
 from ..node_ports_common.exceptions import PortNotFound, UnboundPortError
@@ -55,14 +56,20 @@ async def outputs(self) -> OutputsList:
             await self._auto_update_from_db()
         return self.internal_outputs
 
-    async def get_value_link(self, item_key: str) -> Optional[ItemValue]:
+    async def get_value_link(
+        self, item_key: str, *, file_link_type: LinkType
+    ) -> Optional[ItemValue]:
         try:
-            return await (await self.inputs)[item_key].get_value()
+            return await (await self.inputs)[item_key].get_value(
+                file_link_type=file_link_type
+            )
         except UnboundPortError:
             # not available try outputs
             pass
         # if this fails it will raise an exception
-        return await (await self.outputs)[item_key].get_value()
+        return await (await self.outputs)[item_key].get_value(
+            file_link_type=file_link_type
+        )
 
     async def get(self, item_key: str) -> Optional[ItemConcreteValue]:
         try:

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/port.py

@@ -7,6 +7,7 @@
 import jsonschema
 from models_library.services import PROPERTY_KEY_RE, BaseServiceIOModel
 from pydantic import AnyUrl, Field, PrivateAttr, validator
+from simcore_sdk.node_ports_common.storage_client import LinkType
 
 from ..node_ports_common.exceptions import (
     AbsoluteSymlinkIsNotUploadableException,
@@ -125,13 +126,18 @@ def __init__(self, **data: Any):
         assert self._py_value_type  # nosec
         assert self._py_value_converter  # nosec
 
-    async def get_value(self) -> Optional[ItemValue]:
+    async def get_value(
+        self, *, file_link_type: Optional[LinkType] = None
+    ) -> Optional[ItemValue]:
         """returns the value of the link after resolving the port links"""
+        if not file_link_type:
+            file_link_type = LinkType.PRESIGNED
         log.debug(
-            "getting value of %s[%s] containing '%s'",
+            "getting value of %s[%s] containing '%s' using %s",
             self.key,
             self.property_type,
             pformat(self.value),
+            file_link_type,
         )
 
         if isinstance(self.value, PortLink):
@@ -140,12 +146,14 @@ async def get_value(self) -> Optional[ItemValue]:
                 # pylint: disable=protected-access
                 self.value,
                 self._node_ports._node_ports_creator_cb,
+                file_link_type=file_link_type,
             )
         if isinstance(self.value, FileLink):
             # let's get the download/upload link from storage
             return await port_utils.get_download_link_from_storage(
                 user_id=self._node_ports.user_id,
                 value=self.value,
+                link_type=file_link_type,
             )
 
         if isinstance(self.value, DownloadLink):