Merge branch 'master' into introduce-comp-runs-tasks-snapshot

Author: matusdrobuliak66

packages/service-library/src/servicelib/fastapi/tracing.py

@@ -101,7 +101,6 @@ def _startup(tracing_settings: TracingSettings, service_name: str) -> None:
     span_processor = BatchSpanProcessor(otlp_exporter)
     global_tracer_provider.add_span_processor(span_processor)
 
-    FastAPIInstrumentor().instrument()
     if HAS_AIOPG:
         with log_context(
             _logger,
@@ -181,14 +180,18 @@ def _shutdown() -> None:
             _logger.exception("Failed to uninstrument RequestsInstrumentor")
 
 
+def initialize_fastapi_app_tracing(app: FastAPI):
+    FastAPIInstrumentor.instrument_app(app)
+
+
 def setup_httpx_client_tracing(client: AsyncClient | Client):
     HTTPXClientInstrumentor.instrument_client(client)
 
 
 def setup_tracing(
     app: FastAPI, tracing_settings: TracingSettings, service_name: str
 ) -> None:
-
+    # NOTE: This does not instrument the app itself. Call setup_fastapi_app_tracing to do that.
     _startup(tracing_settings=tracing_settings, service_name=service_name)
 
     def _on_shutdown() -> None:
@@ -200,7 +203,7 @@ def _on_shutdown() -> None:
 def get_tracing_instrumentation_lifespan(
     tracing_settings: TracingSettings, service_name: str
 ):
-
+    # NOTE: This lifespan does not instrument the app itself. Call setup_fastapi_app_tracing to do that.
     _startup(tracing_settings=tracing_settings, service_name=service_name)
 
     async def tracing_instrumentation_lifespan(

packages/service-library/src/servicelib/file_utils.py

@@ -1,10 +1,11 @@
 import asyncio
 import hashlib
 import shutil
+from collections.abc import Iterator
 from contextlib import contextmanager
 from logging import Logger
 from pathlib import Path
-from typing import Final, Iterator, Protocol
+from typing import Final, Protocol
 
 # https://docs.python.org/3/library/shutil.html#shutil.rmtree
 # https://docs.python.org/3/library/os.html#os.remove
@@ -13,11 +14,13 @@
 from pydantic import ByteSize, TypeAdapter
 
 CHUNK_4KB: Final[ByteSize] = TypeAdapter(ByteSize).validate_python("4kb")  # 4K blocks
+CHUNK_8MB: Final[ByteSize] = TypeAdapter(ByteSize).validate_python(
+    "8MiB"
+)  # 8mIB blocks
 
 
 class AsyncStream(Protocol):
-    async def read(self, size: int = -1) -> bytes:
-        ...
+    async def read(self, size: int = -1) -> bytes: ...
 
 
 _shutil_rmtree = sync_to_async(shutil.rmtree)
@@ -45,7 +48,7 @@ async def remove_directory(
 
 
 async def create_sha256_checksum(
-    async_stream: AsyncStream, *, chunk_size: ByteSize = CHUNK_4KB
+    async_stream: AsyncStream, *, chunk_size: ByteSize = CHUNK_8MB
 ) -> str:
     """
     Usage:

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

@@ -291,7 +291,7 @@ async def _generate_checksum(
         return checksum
     if isinstance(path_to_upload, Path):
         async with aiofiles.open(path_to_upload, mode="rb") as f:
-            checksum = SHA256Str(await create_sha256_checksum(f))
+            checksum = await create_sha256_checksum(f)
     elif isinstance(path_to_upload, UploadableFileObject):
         checksum = path_to_upload.sha256_checksum
     return checksum

packages/simcore-sdk/tests/integration/conftest.py

@@ -337,7 +337,7 @@ def _assign_config(
 @pytest.fixture
 async def r_clone_settings_factory(
     minio_s3_settings: S3Settings, storage_service: URL
-) -> Awaitable[RCloneSettings]:
+) -> Callable[[], Awaitable[RCloneSettings]]:
     async def _factory() -> RCloneSettings:
         settings = RCloneSettings(
             R_CLONE_S3=minio_s3_settings, R_CLONE_PROVIDER=S3Provider.MINIO
@@ -347,35 +347,35 @@ async def _factory() -> RCloneSettings:
 
         return settings
 
-    return _factory()
+    return _factory
 
 
 @pytest.fixture
 async def aws_s3_cli_settings_factory(
     minio_s3_settings: S3Settings, storage_service: URL
-) -> Awaitable[AwsS3CliSettings]:
+) -> Callable[[], Awaitable[AwsS3CliSettings]]:
     async def _factory() -> AwsS3CliSettings:
         settings = AwsS3CliSettings(AWS_S3_CLI_S3=minio_s3_settings)
         if not await is_aws_s3_cli_available(settings):
             pytest.skip("aws cli not installed")
 
         return settings
 
-    return _factory()
+    return _factory
 
 
 @pytest.fixture
 async def r_clone_settings(
-    r_clone_settings_factory: Awaitable[RCloneSettings],
+    r_clone_settings_factory: Callable[[], Awaitable[RCloneSettings]],
 ) -> RCloneSettings:
-    return await r_clone_settings_factory
+    return await r_clone_settings_factory()
 
 
 @pytest.fixture
 async def aws_s3_cli_settings(
-    aws_s3_cli_settings_factory: Awaitable[AwsS3CliSettings],
+    aws_s3_cli_settings_factory: Callable[[], Awaitable[AwsS3CliSettings]],
 ) -> AwsS3CliSettings:
-    return await aws_s3_cli_settings_factory
+    return await aws_s3_cli_settings_factory()
 
 
 @pytest.fixture

packages/simcore-sdk/tests/integration/test_node_ports_common_filemanager.py

@@ -56,15 +56,17 @@ class _SyncSettings(BaseModel):
         "Both RClone and AwsS3Cli disabled",
     ],
 )
-def optional_sync_settings(
-    r_clone_settings: RCloneSettings,
-    aws_s3_cli_settings: AwsS3CliSettings,
+async def optional_sync_settings(
+    r_clone_settings_factory: Callable[[], Awaitable[RCloneSettings]],
+    aws_s3_cli_settings_factory: Callable[[], Awaitable[AwsS3CliSettings]],
     request: pytest.FixtureRequest,
 ) -> _SyncSettings:
     _rclone_enabled, _aws_s3_cli_enabled = request.param
 
-    _r_clone_settings = r_clone_settings if _rclone_enabled else None
-    _aws_s3_cli_settings = aws_s3_cli_settings if _aws_s3_cli_enabled else None
+    _r_clone_settings = await r_clone_settings_factory() if _rclone_enabled else None
+    _aws_s3_cli_settings = (
+        await aws_s3_cli_settings_factory() if _aws_s3_cli_enabled else None
+    )
 
     return _SyncSettings(
         r_clone_settings=_r_clone_settings, aws_s3_cli_settings=_aws_s3_cli_settings

packages/simcore-sdk/tests/integration/test_node_ports_v2_nodeports2.py

@@ -21,7 +21,6 @@
 import pytest
 import sqlalchemy as sa
 from faker import Faker
-from models_library.projects import ProjectIDStr
 from models_library.projects_nodes_io import (
     BaseFileLink,
     DownloadLink,
@@ -156,10 +155,10 @@ def config_value_symlink_path(symlink_path: Path) -> dict[str, Any]:
 
 @pytest.fixture(params=[True, False])
 async def option_r_clone_settings(
-    request, r_clone_settings_factory: Awaitable[RCloneSettings]
+    request, r_clone_settings_factory: Callable[[], Awaitable[RCloneSettings]]
 ) -> RCloneSettings | None:
     if request.param:
-        return await r_clone_settings_factory
+        return await r_clone_settings_factory()
     return None
 
 
@@ -174,7 +173,7 @@ async def test_default_configuration(
     await check_config_valid(
         await node_ports_v2.ports(
             user_id=user_id,
-            project_id=ProjectIDStr(project_id),
+            project_id=project_id,
             node_uuid=node_uuid,
             r_clone_settings=option_r_clone_settings,
         ),
@@ -192,7 +191,7 @@ async def test_invalid_ports(
     config_dict, _, _ = create_special_configuration()
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -238,7 +237,7 @@ async def test_port_value_accessors(
 
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -298,7 +297,7 @@ async def test_port_file_accessors(
 
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -375,7 +374,7 @@ async def test_adding_new_ports(
     config_dict, project_id, node_uuid = create_special_configuration()
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -429,7 +428,7 @@ async def test_removing_ports(
     )  # pylint: disable=W0612
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -489,7 +488,7 @@ async def test_get_value_from_previous_node(
 
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -541,7 +540,7 @@ async def test_get_file_from_previous_node(
     )
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -598,7 +597,7 @@ async def test_get_file_from_previous_node_with_mapping_of_same_key_name(
     )
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -659,7 +658,7 @@ async def test_file_mapping(
     )
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -752,7 +751,7 @@ async def test_regression_concurrent_port_update_fails(
 
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )
@@ -841,7 +840,7 @@ async def test_batch_update_inputs_outputs(
 
     PORTS = await node_ports_v2.ports(
         user_id=user_id,
-        project_id=ProjectIDStr(project_id),
+        project_id=project_id,
         node_uuid=node_uuid,
         r_clone_settings=option_r_clone_settings,
     )

services/agent/src/simcore_service_agent/core/application.py

@@ -5,7 +5,10 @@
     get_common_oas_options,
     override_fastapi_openapi_method,
 )
-from servicelib.fastapi.tracing import setup_tracing
+from servicelib.fastapi.tracing import (
+    initialize_fastapi_app_tracing,
+    setup_tracing,
+)
 from servicelib.logging_utils import config_all_loggers
 
 from .._meta import (
@@ -66,6 +69,9 @@ def create_app() -> FastAPI:
     setup_rest_api(app)
     setup_rpc_api_routes(app)
 
+    if settings.AGENT_TRACING:
+        initialize_fastapi_app_tracing(app)
+
     async def _on_startup() -> None:
         print(APP_STARTED_BANNER_MSG, flush=True)  # noqa: T201
 

services/api-server/src/simcore_service_api_server/core/application.py

@@ -5,7 +5,10 @@
 from models_library.basic_types import BootModeEnum
 from packaging.version import Version
 from servicelib.fastapi.profiler import initialize_profiler
-from servicelib.fastapi.tracing import setup_tracing
+from servicelib.fastapi.tracing import (
+    initialize_fastapi_app_tracing,
+    setup_tracing,
+)
 from servicelib.logging_utils import config_all_loggers
 
 from .. import exceptions
@@ -93,6 +96,9 @@ def init_app(settings: ApplicationSettings | None = None) -> FastAPI:
     if app.state.settings.API_SERVER_PROMETHEUS_INSTRUMENTATION_ENABLED:
         setup_prometheus_instrumentation(app)
 
+    if settings.API_SERVER_TRACING:
+        initialize_fastapi_app_tracing(app)
+
     if settings.API_SERVER_WEBSERVER:
         webserver.setup(
             app,

services/autoscaling/src/simcore_service_autoscaling/core/application.py

@@ -1,7 +1,10 @@
 import logging
 
 from fastapi import FastAPI
-from servicelib.fastapi.tracing import setup_tracing
+from servicelib.fastapi.tracing import (
+    initialize_fastapi_app_tracing,
+    setup_tracing,
+)
 
 from .._meta import (
     API_VERSION,
@@ -71,6 +74,9 @@ def create_app(settings: ApplicationSettings) -> FastAPI:
     setup_ssm(app)
     setup_redis(app)
 
+    if app.state.settings.AUTOSCALING_TRACING:
+        initialize_fastapi_app_tracing(app)
+
     setup_auto_scaler_background_task(app)
     setup_buffer_machines_pool_task(app)
 

services/catalog/src/simcore_service_catalog/core/application.py

@@ -8,7 +8,10 @@
     setup_prometheus_instrumentation,
 )
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
-from servicelib.fastapi.tracing import setup_tracing
+from servicelib.fastapi.tracing import (
+    initialize_fastapi_app_tracing,
+    setup_tracing,
+)
 from starlette.middleware.base import BaseHTTPMiddleware
 
 from .._meta import (
@@ -68,6 +71,9 @@ def create_app() -> FastAPI:
     if settings.CATALOG_PROMETHEUS_INSTRUMENTATION_ENABLED:
         setup_prometheus_instrumentation(app)
 
+    if settings.CATALOG_TRACING:
+        initialize_fastapi_app_tracing(app)
+
     if settings.SC_BOOT_MODE != BootModeEnum.PRODUCTION:
         # middleware to time requests (ONLY for development)
         app.add_middleware(