Align sandbox preview APIs with dotnet

Author: YunchuWang

durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker_profiles.py

@@ -19,25 +19,43 @@
 DEFAULT_CPU = "1000m"
 DEFAULT_MEMORY = "2048Mi"
 DEFAULT_MAX_CONCURRENT_ACTIVITIES = 100
+MIN_CPU_MILLICORES = 250
+MAX_CPU_MILLICORES = 16000
+CPU_STEP_MILLICORES = 250
+MEMORY_MIB_PER_CORE = 2 * 1024
 
 
 @dataclass
-class SandboxWorkerProfileOptions:
-    """Options for a decorated sandbox worker profile."""
+class SandboxWorkerProfileImageOptions:
+    """Options for the sandbox worker image DTS should start."""
 
-    worker_profile_id: str
     # Full OCI image reference for the sandbox worker container, for example
     # "myregistry.azurecr.io/workers/hello:1.0" or
     # "myregistry.azurecr.io/workers/hello@sha256:0123456789abcdef...".
-    container_image: Optional[str] = None
-    image_pull_managed_identity_client_id: Optional[str] = None
+    image_ref: Optional[str] = None
+    managed_identity_client_id: Optional[str] = None
+    entrypoint: list[str] = field(default_factory=list[str])
+    cmd: list[str] = field(default_factory=list[str])
+
+
+@dataclass
+class SandboxWorkerProfileOptions:
+    """Options for a decorated sandbox worker profile."""
+
+    @dataclass(frozen=True)
+    class Activity:
+        """Activity name and optional version for a sandbox worker profile."""
+
+        name: str
+        version: Optional[str]
+
+    worker_profile_id: str
+    image: SandboxWorkerProfileImageOptions = field(default_factory=SandboxWorkerProfileImageOptions)
     scheduler_managed_identity_client_id: Optional[str] = None
     cpu: str = DEFAULT_CPU
     memory: str = DEFAULT_MEMORY
     environment_variables: dict[str, str] = field(default_factory=dict[str, str])
     max_concurrent_activities: int = DEFAULT_MAX_CONCURRENT_ACTIVITIES
-    entrypoint: list[str] = field(default_factory=list[str])
-    cmd: list[str] = field(default_factory=list[str])
     activities: list[SandboxActivity] = field(default_factory=list[SandboxActivity])
 
     def add_activity(
@@ -50,6 +68,11 @@ def add_activity(
             name=normalize_required(activity_name, "Sandbox activity name is required."),
             version=(version.strip() if version and version.strip() else None)))
 
+    def add_activities(self, activities: Iterable[Activity]) -> None:
+        """Add activity names and versions to the sandbox worker profile."""
+        for activity in activities:
+            self.add_activity(activity.name, activity.version)
+
 
 class SandboxWorkerProfile:
     """Base class for configuring a decorated sandbox worker profile."""
@@ -94,21 +117,19 @@ def _build_sandbox_worker_profile(
         activities: Iterable[SandboxActivity],
         scheduler_managed_identity_client_id: Optional[str],
         worker_profile_id: str,
-        container_image: Optional[str] = None,
-        image_pull_managed_identity_client_id: Optional[str] = None,
+        image: Optional[SandboxWorkerProfileImageOptions] = None,
         cpu: str = DEFAULT_CPU,
         memory: str = DEFAULT_MEMORY,
         environment_variables: Optional[dict[str, str]] = None,
-        max_concurrent_activities: int = DEFAULT_MAX_CONCURRENT_ACTIVITIES,
-        entrypoint: Optional[Iterable[str]] = None,
-        cmd: Optional[Iterable[str]] = None) -> pb.SandboxWorkerProfile:
+        max_concurrent_activities: int = DEFAULT_MAX_CONCURRENT_ACTIVITIES) -> pb.SandboxWorkerProfile:
     """Build a sandbox activity worker_profile.
 
     Args:
-        container_image: Full OCI image reference for the sandbox worker container,
+        image: Sandbox worker image options with the full OCI image reference,
             such as "myregistry.azurecr.io/workers/hello:1.0" or
             "myregistry.azurecr.io/workers/hello@sha256:0123456789abcdef...".
     """
+    image_options = image or SandboxWorkerProfileImageOptions()
     resolved_activities = resolve_activities(activities)
     if not resolved_activities:
         raise ValueError("Sandbox activity worker_profile requires at least one activity.")
@@ -120,7 +141,7 @@ def _build_sandbox_worker_profile(
         raise ValueError("Sandbox activity max concurrent activities must be greater than zero.")
 
     image_ref = normalize_required(
-        container_image,
+        image_options.image_ref,
         "Sandbox activity image metadata requires a container image reference like "
         "'myregistry.azurecr.io/workers/hello:1.0' or "
         "'myregistry.azurecr.io/workers/hello@sha256:...'.")
@@ -129,11 +150,11 @@ def _build_sandbox_worker_profile(
         scheduler_managed_identity_client_id,
         "Sandbox activity worker_profile requires the managed identity client ID workers use to connect to Durable Task Scheduler.")
     resolved_image_pull_managed_identity_client_id = normalize_required(
-        image_pull_managed_identity_client_id,
-        "Sandbox activity worker_profile requires the managed identity client ID ADC uses to pull the worker image.")
+        image_options.managed_identity_client_id,
+        "Sandbox activity worker_profile requires the managed identity client ID used to pull the worker image.")
 
-    resolved_cpu = _normalize_cpu(cpu)
-    resolved_memory = _normalize_memory(memory)
+    resolved_cpu, cpu_millicores = _normalize_cpu(cpu)
+    resolved_memory = _normalize_memory(memory, cpu_millicores)
 
     worker_profile = pb.SandboxWorkerProfile(
         worker_profile_id=worker_profile_id.strip(),
@@ -150,8 +171,8 @@ def _build_sandbox_worker_profile(
         for activity in resolved_activities
     ])
     worker_profile.environment_variables.update(environment_variables or {})
-    worker_profile.image.entrypoint.extend(_normalize_optional_strings(entrypoint or []))
-    worker_profile.image.cmd.extend(_normalize_optional_strings(cmd or []))
+    worker_profile.image.entrypoint.extend(_normalize_optional_strings(image_options.entrypoint))
+    worker_profile.image.cmd.extend(_normalize_optional_strings(image_options.cmd))
     return worker_profile
 
 
@@ -175,15 +196,12 @@ def build_sandbox_worker_profiles() -> list[pb.SandboxWorkerProfile]:
         worker_profiles.append(_build_sandbox_worker_profile(
             activities=activities,
             worker_profile_id=profile.worker_profile_id,
-            container_image=profile.container_image,
-            image_pull_managed_identity_client_id=profile.image_pull_managed_identity_client_id,
+            image=profile.image,
             scheduler_managed_identity_client_id=profile.scheduler_managed_identity_client_id,
             cpu=profile.cpu,
             memory=profile.memory,
             environment_variables=profile.environment_variables,
-            max_concurrent_activities=profile.max_concurrent_activities,
-            entrypoint=profile.entrypoint,
-            cmd=profile.cmd))
+            max_concurrent_activities=profile.max_concurrent_activities))
 
     return worker_profiles
 
@@ -236,46 +254,61 @@ def _normalize_optional_strings(values: Iterable[str]) -> list[str]:
     return [value.strip() for value in values if value and value.strip()]
 
 
-def _normalize_cpu(value: str) -> str:
+def _normalize_cpu(value: str) -> tuple[str, int]:
     normalized = normalize_required(value, "Sandbox activity worker_profile requires CPU resources.")
     milli_cpu = _try_parse_cpu_millicores(normalized)
-    if milli_cpu is None or milli_cpu <= 0:
+    if (milli_cpu is None
+            or milli_cpu < MIN_CPU_MILLICORES
+            or milli_cpu > MAX_CPU_MILLICORES
+            or milli_cpu % CPU_STEP_MILLICORES != 0):
         raise ValueError(
-            "Sandbox activity CPU resources must be a positive Kubernetes-style CPU quantity. "
+            "Sandbox activity CPU resources must match an ADC sandbox CPU tier: "
+            "250m through 16000m, in 250m increments. "
             "Use formats like '500m', '2', or '0.5'.")
-    return normalized
+    return normalized, milli_cpu
 
 
-def _normalize_memory(value: str) -> str:
+def _normalize_memory(value: str, cpu_millicores: int) -> str:
     normalized = normalize_required(value, "Sandbox activity worker_profile requires memory resources.")
+    max_memory_mib = cpu_millicores * MEMORY_MIB_PER_CORE // 1000
     memory_mib = _try_parse_memory_mib(normalized)
     if memory_mib is None or memory_mib <= 0:
         raise ValueError(
             "Sandbox activity memory resources must be a positive Kubernetes-style memory quantity. "
             "Use formats like '256Mi', '1Gi', or '2048'.")
+    if memory_mib > max_memory_mib:
+        raise ValueError(
+            "Sandbox activity memory resources exceed the ADC sandbox tier maximum for the configured CPU. "
+            f"Maximum memory for CPU '{cpu_millicores}m' is {max_memory_mib}Mi.")
     return normalized
 
 
 def _try_parse_cpu_millicores(value: str) -> Optional[int]:
     try:
         if value[-1:].lower() == "m":
             return int(value[:-1])
-        return int(Decimal(value) * 1000)
+        millicores = Decimal(value) * 1000
+        return int(millicores) if millicores == millicores.to_integral_value() else None
     except (InvalidOperation, ValueError):
         return None
 
 
 def _try_parse_memory_mib(value: str) -> Optional[int]:
     try:
         if value[-2:].lower() == "gi":
-            return int(Decimal(value[:-2]) * 1024)
+            return _try_convert_memory_to_mib(Decimal(value[:-2]), 1024)
         if value[-2:].lower() == "mi":
-            return int(Decimal(value[:-2]))
-        return int(Decimal(value))
+            return _try_convert_memory_to_mib(Decimal(value[:-2]), 1)
+        return _try_convert_memory_to_mib(Decimal(value), 1)
     except (InvalidOperation, ValueError):
         return None
 
 
+def _try_convert_memory_to_mib(value: Decimal, multiplier: int) -> Optional[int]:
+    memory_mib = value * multiplier
+    return int(memory_mib) if memory_mib == memory_mib.to_integral_value() else None
+
+
 def _parse_sandbox_provider(sandbox_provider: Optional[str]) -> "pb.SandboxProviderKind":
     if not sandbox_provider:
         return pb.SANDBOX_PROVIDER_KIND_UNSPECIFIED

examples/sandboxes/README.md

@@ -21,9 +21,7 @@ Set these before running the declarer app:
 Bash:
 
 ~~~bash
-export DTS_ENDPOINT="<scheduler endpoint>"
-export DTS_TASK_HUB="<task hub name>"
-export DTS_WORKER_PROFILE_ID="python-sandbox-worker"
+export DURABLE_TASK_SCHEDULER_CONNECTION_STRING="Endpoint=https://<scheduler-endpoint>;TaskHub=<task-hub-name>;Authentication=DefaultAzure"
 export DTS_SANDBOX_CONTAINER_IMAGE="<container image reference>"
 export DTS_SANDBOX_IMAGE_PULL_UMI_CLIENT_ID="<image-pull UMI client ID>"
 export DTS_SANDBOX_SCHEDULER_UMI_CLIENT_ID="<scheduler UMI client ID>"
@@ -32,21 +30,25 @@ export DTS_SANDBOX_SCHEDULER_UMI_CLIENT_ID="<scheduler UMI client ID>"
 PowerShell:
 
 ~~~powershell
-$env:DTS_ENDPOINT = "<scheduler endpoint>"
-$env:DTS_TASK_HUB = "<task hub name>"
-$env:DTS_WORKER_PROFILE_ID = "python-sandbox-worker"
+$env:DURABLE_TASK_SCHEDULER_CONNECTION_STRING = "Endpoint=https://<scheduler-endpoint>;TaskHub=<task-hub-name>;Authentication=DefaultAzure"
 $env:DTS_SANDBOX_CONTAINER_IMAGE = "<container image reference>"
 $env:DTS_SANDBOX_IMAGE_PULL_UMI_CLIENT_ID = "<image-pull UMI client ID>"
 $env:DTS_SANDBOX_SCHEDULER_UMI_CLIENT_ID = "<scheduler UMI client ID>"
 ~~~
 
+For `Authentication=DefaultAzure`, sign in with Azure CLI or configure another
+supported Azure identity before running the declarer app. For a local emulator,
+use `Authentication=None`.
+
 After pushing the remote worker image, set `DTS_SANDBOX_CONTAINER_IMAGE` to
 the pushed image reference. `RemoteWorkerProfile.configure()` declares CPU,
 memory, max concurrency, customer environment variables, and sandbox activity
-identities with `options.add_activity(name, version)`. This sample uses an
-unversioned activity identity (`version=None`) because Python orchestrations
-currently schedule activities by name. The declarer and remote worker both use
-`activities.py` so they stay in sync.
+identities with `options.image.image_ref`,
+`options.image.managed_identity_client_id`, and
+`options.add_activity(name, version)`. This sample uses an unversioned activity
+identity (`version=None`) because Python orchestrations currently schedule
+activities by name. The declarer and remote worker both use `activities.py` so
+they stay in sync.
 
 The remote worker code cannot pass Durable Task Scheduler runtime settings to the SDK. In a
 sandbox, `SandboxWorker()` reads `DTS_ENDPOINT`,
@@ -81,7 +83,8 @@ docker build `
 docker push <public container image reference>
 ~~~
 
-Private preview requires the image to be publicly pullable by the sandbox platform.
+The sandbox platform pulls the image with `DTS_SANDBOX_IMAGE_PULL_UMI_CLIENT_ID`,
+so grant that identity pull access to the pushed image.
 
 ## Run the declarer app
 

examples/sandboxes/main_app.py

@@ -2,6 +2,7 @@
 
 import os
 
+from azure.core.credentials import TokenCredential
 from azure.identity import DefaultAzureCredential
 
 from durabletask import client, task
@@ -13,6 +14,8 @@
 
 from activities import REMOTE_HELLO
 
+WORKER_PROFILE_ID = "remote-hello-profile"
+
 
 def hello_orchestrator(ctx: task.OrchestrationContext, name: str):
     """Orchestrator that calls an activity executed by the remote worker image."""
@@ -26,25 +29,58 @@ def _get_required_env(name: str) -> str:
     raise RuntimeError(f"Set {name} before running the sandbox sample.")
 
 
-taskhub_name = os.getenv("DTS_TASK_HUB") or "SandboxPocHub"
-endpoint = os.getenv("DTS_ENDPOINT", "http://localhost:8080")
-worker_profile_id = _get_required_env("DTS_WORKER_PROFILE_ID")
-container_image = (
-    os.getenv("DTS_SANDBOX_CONTAINER_IMAGE")
-    or "sandboxes-remote-worker:local")
+def _parse_scheduler_connection_string(connection_string: str) -> dict[str, str]:
+    settings: dict[str, str] = {}
+    for segment in connection_string.split(";"):
+        if not segment.strip():
+            continue
+        key, separator, value = segment.partition("=")
+        if not separator or not key.strip():
+            raise RuntimeError(
+                "DURABLE_TASK_SCHEDULER_CONNECTION_STRING must use key=value segments.")
+        settings[key.strip().lower()] = value.strip()
+    return settings
+
+
+def _resolve_scheduler_connection() -> tuple[str, str, bool, TokenCredential | None]:
+    settings = _parse_scheduler_connection_string(
+        _get_required_env("DURABLE_TASK_SCHEDULER_CONNECTION_STRING"))
+    endpoint = settings.get("endpoint")
+    taskhub = settings.get("taskhub")
+    if not endpoint:
+        raise RuntimeError("DURABLE_TASK_SCHEDULER_CONNECTION_STRING must include Endpoint.")
+    if not taskhub:
+        raise RuntimeError("DURABLE_TASK_SCHEDULER_CONNECTION_STRING must include TaskHub.")
+
+    authentication = settings.get("authentication", "").lower()
+    if authentication in ("", "none"):
+        credential = None
+    elif authentication == "defaultazure":
+        credential = DefaultAzureCredential()
+    else:
+        raise RuntimeError(
+            "DURABLE_TASK_SCHEDULER_CONNECTION_STRING Authentication must be DefaultAzure or None.")
+
+    endpoint = endpoint.strip()
+    secure_channel = endpoint.lower().startswith(("https://", "grpcs://"))
+    return endpoint, taskhub.strip(), secure_channel, credential
+
+
+endpoint, taskhub_name, secure_channel, credential = _resolve_scheduler_connection()
+container_image = _get_required_env("DTS_SANDBOX_CONTAINER_IMAGE")
 image_pull_managed_identity_client_id = _get_required_env("DTS_SANDBOX_IMAGE_PULL_UMI_CLIENT_ID")
 scheduler_managed_identity_client_id = _get_required_env("DTS_SANDBOX_SCHEDULER_UMI_CLIENT_ID")
 sample_input = os.getenv("DTS_SAMPLE_HELLO_INPUT", "sandbox Python")
 sample_timeout_seconds = int(os.getenv("DTS_SAMPLE_TIMEOUT_SECONDS", "300"))
 
 
-@sandbox_worker_profile(worker_profile_id)
+@sandbox_worker_profile(WORKER_PROFILE_ID)
 class RemoteWorkerProfile(SandboxWorkerProfile):
     """Sandbox worker profile used by the sample remote activity."""
 
     def configure(self, options) -> None:
-        options.container_image = container_image
-        options.image_pull_managed_identity_client_id = image_pull_managed_identity_client_id
+        options.image.image_ref = container_image
+        options.image.managed_identity_client_id = image_pull_managed_identity_client_id
         options.scheduler_managed_identity_client_id = scheduler_managed_identity_client_id
         options.cpu = "1000m"
         options.memory = "2048Mi"
@@ -57,9 +93,6 @@ def configure(self, options) -> None:
 print(f"Using endpoint: {endpoint}")
 print(f"Declaring sandbox activity image: {container_image}")
 
-secure_channel = endpoint.startswith("https://") or endpoint.startswith("grpcs://")
-credential = DefaultAzureCredential() if secure_channel else None
-
 sandboxes_client = SandboxActivitiesClient(
     host_address=endpoint,
     secure_channel=secure_channel,