Align serverless declarations with dotnet

Author: YunchuWang

durabletask-azuremanaged/CHANGELOG.md

@@ -9,6 +9,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 - Added preview `durabletask.azuremanaged.extensions.serverless` APIs for
   declaring DTS serverless activities and running a sandbox activity worker.
+- Added profile-based DTS serverless declarations with
+  `@serverless_worker_profile`, `ServerlessWorkerProfile.configure()`,
+  `ServerlessWorkerProfileOptions.add_activity(...)`, and
+  `ServerlessActivitiesClient.enable_serverless_activities()`.
+- Changed serverless worker profile declarations so the decorator names the
+  profile and the profile class configures image, resources, concurrency, and
+  customer environment variables explicitly.
 - Changed serverless activity worker configuration to require explicit SDK
   parameters instead of reading DTS environment variables inside the SDK.
 - Removed the serverless worker wakeup HTTP listener because ADC sandbox

durabletask-azuremanaged/durabletask/azuremanaged/extensions/serverless/__init__.py

@@ -16,9 +16,15 @@
 """
 
 from durabletask.azuremanaged.extensions.serverless.client import ServerlessActivitiesClient
+from durabletask.azuremanaged.extensions.serverless.client import ServerlessWorkerProfile
+from durabletask.azuremanaged.extensions.serverless.client import ServerlessWorkerProfileOptions
+from durabletask.azuremanaged.extensions.serverless.client import serverless_worker_profile
 from durabletask.azuremanaged.extensions.serverless.worker import ServerlessWorker
 
 __all__ = [
     "ServerlessWorker",
+    "ServerlessWorkerProfile",
+    "ServerlessWorkerProfileOptions",
     "ServerlessActivitiesClient",
+    "serverless_worker_profile",
 ]

durabletask-azuremanaged/durabletask/azuremanaged/extensions/serverless/client.py

@@ -1,11 +1,13 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from typing import Iterable, Optional, Sequence
+from dataclasses import dataclass, field
+from typing import Callable, Iterable, Optional, Sequence
 
 import grpc
 from azure.core.credentials import TokenCredential
 
+from durabletask import task
 from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import (
     DTSDefaultClientInterceptorImpl,
 )
@@ -21,6 +23,65 @@
 DEFAULT_MAX_CONCURRENT_ACTIVITIES = 100
 
 
+@dataclass
+class ServerlessWorkerProfileOptions:
+    """Options for a decorated serverless worker profile."""
+
+    worker_profile_id: str
+    container_image: Optional[str] = None
+    registry_server: Optional[str] = None
+    repository: Optional[str] = None
+    tag: Optional[str] = None
+    image_digest: Optional[str] = None
+    cpu: str = DEFAULT_CPU
+    memory: str = DEFAULT_MEMORY
+    environment_variables: dict[str, str] = field(default_factory=dict)
+    max_concurrent_activities: int = DEFAULT_MAX_CONCURRENT_ACTIVITIES
+    entrypoint: list[str] = field(default_factory=list)
+    cmd: list[str] = field(default_factory=list)
+    activity_names: list[str] = field(default_factory=list)
+
+    def add_activity(self, activity: str | Callable) -> None:
+        """Add an activity to the serverless worker profile declaration."""
+        activity_name = task.get_name(activity) if callable(activity) else activity
+        self.activity_names.append(
+            _normalize_required(activity_name, "Serverless activity name is required."))
+
+
+class ServerlessWorkerProfile:
+    """Base class for configuring a decorated serverless worker profile."""
+
+    def configure(self, options: ServerlessWorkerProfileOptions) -> None:
+        """Configure the serverless worker profile declaration options."""
+
+
+_worker_profiles: dict[str, ServerlessWorkerProfileOptions] = {}
+
+
+def serverless_worker_profile(worker_profile_id: str) -> Callable[[type], type]:
+    """Declare a serverless worker profile using a decorated marker class."""
+    normalized_profile = _normalize_required(worker_profile_id, "Serverless worker profile ID is required.")
+
+    def decorator(cls: type) -> type:
+        if normalized_profile in _worker_profiles:
+            raise ValueError(f"Serverless worker profile '{normalized_profile}' is declared more than once.")
+
+        options = ServerlessWorkerProfileOptions(worker_profile_id=normalized_profile)
+        try:
+            profile = cls()
+        except TypeError as ex:
+            raise TypeError("Serverless worker profile classes must have a parameterless constructor.") from ex
+
+        configure = getattr(profile, "configure", None)
+        if callable(configure):
+            configure(options)
+
+        _worker_profiles[normalized_profile] = options
+        return cls
+
+    return decorator
+
+
 def resolve_activity_names(activity_names: str | Iterable[str]) -> list[str]:
     resolved: list[str] = []
     seen: set[str] = set()
@@ -114,6 +175,41 @@ def build_serverless_activity_declaration(
     return declaration
 
 
+def build_profile_serverless_activity_declarations() -> list[pb.ServerlessActivityDeclaration]:
+    """Build serverless declarations from worker profile configuration."""
+    declarations: list[pb.ServerlessActivityDeclaration] = []
+    activity_owners: dict[str, str] = {}
+    for profile in _worker_profiles.values():
+        activity_names = resolve_activity_names(profile.activity_names)
+        if not activity_names:
+            continue
+
+        for activity_name in activity_names:
+            existing_profile = activity_owners.get(activity_name)
+            if existing_profile and existing_profile != profile.worker_profile_id:
+                raise ValueError(
+                    f"Serverless activity '{activity_name}' is assigned to both worker profile "
+                    f"'{existing_profile}' and '{profile.worker_profile_id}'.")
+            activity_owners[activity_name] = profile.worker_profile_id
+
+        declarations.append(build_serverless_activity_declaration(
+            activity_names=activity_names,
+            worker_profile_id=profile.worker_profile_id,
+            container_image=profile.container_image,
+            registry_server=profile.registry_server,
+            repository=profile.repository,
+            tag=profile.tag,
+            image_digest=profile.image_digest,
+            cpu=profile.cpu,
+            memory=profile.memory,
+            environment_variables=profile.environment_variables,
+            max_concurrent_activities=profile.max_concurrent_activities,
+            entrypoint=profile.entrypoint,
+            cmd=profile.cmd))
+
+    return declarations
+
+
 def build_serverless_worker_start(
         *,
         taskhub: str,
@@ -188,37 +284,14 @@ def close(self) -> None:
         if self._owns_channel:
             self._channel.close()
 
-    def declare_serverless_activities(
-            self,
-            *,
-            activity_names: str | Iterable[str],
-            worker_profile_id: str = DEFAULT_WORKER_PROFILE_ID,
-            container_image: Optional[str] = None,
-            registry_server: Optional[str] = None,
-            repository: Optional[str] = None,
-            tag: Optional[str] = None,
-            image_digest: Optional[str] = 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) -> None:
-        declaration = build_serverless_activity_declaration(
-            activity_names=activity_names,
-            worker_profile_id=worker_profile_id,
-            container_image=container_image,
-            registry_server=registry_server,
-            repository=repository,
-            tag=tag,
-            image_digest=image_digest,
-            cpu=cpu,
-            memory=memory,
-            environment_variables=environment_variables,
-            max_concurrent_activities=max_concurrent_activities,
-            entrypoint=entrypoint,
-            cmd=cmd)
-        self._stub.DeclareServerlessActivities(declaration)
+    def enable_serverless_activities(self) -> None:
+        """Declare all configured serverless worker profiles with DTS."""
+        declarations = build_profile_serverless_activity_declarations()
+        if not declarations:
+            raise ValueError("No configured serverless activities were found.")
+
+        for declaration in declarations:
+            self._stub.DeclareServerlessActivities(declaration)
 
     def remove_serverless_activity_declaration(self, worker_profile_id: str) -> None:
         worker_profile_id = _normalize_required(worker_profile_id, "Worker profile ID is required.")

examples/serverless/README.md

@@ -6,6 +6,8 @@ This sample mirrors the .NET serverless sample with two customer-owned pieces:
    serverlessly, starts the orchestration, and waits for the result.
 2. A **remote worker image** (`remote_worker.py` plus `Containerfile`) that
    DTS starts in a sandbox to execute the declared activity.
+3. A tiny shared module (`activity_names.py`) that keeps the declarer and remote
+   worker on the same activity name constants.
 
 Reference .NET template:
 <https://github.com/microsoft/durabletask-dotnet/compare/wangbill/serverless-private-preview>
@@ -24,12 +26,14 @@ Set these before running the declarer app:
 $env:DTS_ENDPOINT = "<scheduler endpoint>"
 $env:DTS_TASK_HUB = "<task hub name>"
 $env:DTS_WORKER_PROFILE_ID = "default"
-$env:DTS_SERVERLESS_ACTIVITY_IMAGE = "<public container image reference>"
-$env:DTS_SERVERLESS_CPU = "1000m"
-$env:DTS_SERVERLESS_MEMORY = "2048Mi"
-$env:DTS_SERVERLESS_MAX_ACTIVITIES = "1"
 ```
 
+After pushing the remote worker image, set `options.container_image` in
+`RemoteWorkerProfile.configure()` to the pushed image reference. That method is
+also where the sample declares CPU, memory, max concurrency, customer
+environment variables, and serverless activity names with `options.add_activity(...)`.
+The declarer and remote worker both use `activity_names.py` so they stay in sync.
+
 The remote worker code cannot pass DTS runtime settings to the SDK. In a
 sandbox, `ServerlessWorker()` reads `DTS_ENDPOINT`,
 `DTS_TASK_HUB`, `DTS_WORKER_PROFILE_ID`, `DTS_SERVERLESS_MAX_ACTIVITIES`,
@@ -67,3 +71,6 @@ python examples\serverless\main_app.py
 
 The declarer app registers the serverless activity metadata, starts
 `hello_orchestrator`, and the remote worker sandbox executes `remote_hello`.
+The result includes `SERVERLESS_SAMPLE_MARKER=serverless-python-sample-marker`,
+proving the customer environment variable declared on the worker profile reached
+the sandbox.

examples/serverless/activity_names.py

@@ -0,0 +1,4 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+REMOTE_HELLO = "remote_hello"

examples/serverless/main_app.py

@@ -7,29 +7,40 @@
 from durabletask import client, task
 from durabletask.azuremanaged.client import DurableTaskSchedulerClient
 from durabletask.azuremanaged.extensions.serverless import ServerlessActivitiesClient
+from durabletask.azuremanaged.extensions.serverless import ServerlessWorkerProfile
+from durabletask.azuremanaged.extensions.serverless import serverless_worker_profile
 from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker
 
-
-REMOTE_ACTIVITY_NAME = "remote_hello"
+from activity_names import REMOTE_HELLO
 
 
 def hello_orchestrator(ctx: task.OrchestrationContext, name: str):
     """Orchestrator that calls an activity executed by the remote worker image."""
-    return (yield ctx.call_activity(REMOTE_ACTIVITY_NAME, input=name))
+    return (yield ctx.call_activity(REMOTE_HELLO, input=name))
 
 
 taskhub_name = os.getenv("DTS_TASK_HUB") or "ServerlessPocHub"
 endpoint = os.getenv("DTS_ENDPOINT", "http://localhost:8080")
 worker_profile_id = os.getenv("DTS_WORKER_PROFILE_ID", "default")
-serverless_image = os.getenv("DTS_SERVERLESS_ACTIVITY_IMAGE", "serverless-remote-worker:local")
-serverless_cpu = os.getenv("DTS_SERVERLESS_CPU", "1000m")
-serverless_memory = os.getenv("DTS_SERVERLESS_MEMORY", "2048Mi")
-serverless_max_activities = int(os.getenv("DTS_SERVERLESS_MAX_ACTIVITIES", "1"))
 sample_input = os.getenv("DTS_SAMPLE_HELLO_INPUT", "serverless Python")
 
+
+@serverless_worker_profile(worker_profile_id)
+class RemoteWorkerProfile(ServerlessWorkerProfile):
+    """Serverless worker profile used by the sample remote activity."""
+
+    def configure(self, options) -> None:
+        options.container_image = "serverless-remote-worker:local"
+        options.cpu = "1000m"
+        options.memory = "2048Mi"
+        options.max_concurrent_activities = 1
+        options.environment_variables["SERVERLESS_SAMPLE_MARKER"] = "serverless-python-sample-marker"
+        options.add_activity(REMOTE_HELLO)
+
+
 print(f"Using taskhub: {taskhub_name}")
 print(f"Using endpoint: {endpoint}")
-print(f"Declaring serverless activity image: {serverless_image}")
+print("Declaring serverless activity image: serverless-remote-worker:local")
 
 secure_channel = endpoint.startswith("https://") or endpoint.startswith("grpcs://")
 credential = DefaultAzureCredential() if secure_channel else None
@@ -39,13 +50,7 @@ def hello_orchestrator(ctx: task.OrchestrationContext, name: str):
     secure_channel=secure_channel,
     taskhub=taskhub_name,
     token_credential=credential)
-serverless_client.declare_serverless_activities(
-    activity_names=REMOTE_ACTIVITY_NAME,
-    worker_profile_id=worker_profile_id,
-    container_image=serverless_image,
-    cpu=serverless_cpu,
-    memory=serverless_memory,
-    max_concurrent_activities=serverless_max_activities)
+serverless_client.enable_serverless_activities()
 
 with DurableTaskSchedulerWorker(
         host_address=endpoint,

examples/serverless/remote_worker.py

@@ -6,15 +6,21 @@
 from durabletask import task
 from durabletask.azuremanaged.extensions.serverless import ServerlessWorker
 
+from activity_names import REMOTE_HELLO
 
-def remote_hello(ctx: task.ActivityContext, name: str) -> str:
+
+def _remote_hello(ctx: task.ActivityContext, name: str) -> str:
     """Activity function that runs inside the serverless worker container."""
     sandbox_id = os.getenv("DTS_SANDBOX_ID", "unknown-sandbox")
-    return f"Hello {name} from Python serverless worker {sandbox_id}!"
+    marker = os.getenv("SERVERLESS_SAMPLE_MARKER", "<missing>")
+    return f"Hello {name} from Python serverless worker {sandbox_id}! SERVERLESS_SAMPLE_MARKER={marker}"
+
+
+_remote_hello.__name__ = REMOTE_HELLO
 
 
 with ServerlessWorker() as worker:
-    worker.add_activity(remote_hello)
+    worker.add_activity(_remote_hello)
     worker.start()
     print("Python serverless remote worker is running. Press Ctrl+C to stop.")
     try: