Skip to content

Commit 27d2745

Browse files
YunchuWangYunchuWang
committed
Split sandbox profile builders
1 parent 1e9ed95 commit 27d2745

6 files changed

Lines changed: 260 additions & 221 deletions

File tree

‎durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
from azure.core.credentials import TokenCredential
88

99
from durabletask.azuremanaged.preview.sandboxes.helpers import normalize_required
10-
from durabletask.azuremanaged.preview.sandboxes.worker_profiles import (
10+
from durabletask.azuremanaged.preview.sandboxes.profile_builder import (
1111
build_sandbox_worker_profiles,
1212
)
1313
from durabletask.azuremanaged.preview.sandboxes.transport import (

‎durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/profile_builder.py‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
# Copyright (c) Microsoft Corporation.
2+
# Licensed under the MIT License.
3+
4+
from decimal import Decimal, InvalidOperation
5+
from typing import Iterable, Optional
6+
7+
from durabletask.azuremanaged.internal import sandbox_service_pb2 as pb
8+
from durabletask.azuremanaged.preview.sandboxes.helpers import (
9+
SandboxActivity,
10+
activities_overlap,
11+
format_activity,
12+
normalize_required,
13+
resolve_activities,
14+
)
15+
from durabletask.azuremanaged.preview.sandboxes.worker_profiles import (
16+
DEFAULT_CPU,
17+
DEFAULT_MAX_CONCURRENT_ACTIVITIES,
18+
DEFAULT_MEMORY,
19+
SandboxWorkerProfileImageOptions,
20+
registered_sandbox_worker_profiles,
21+
)
22+
23+
24+
MIN_CPU_MILLICORES = 250
25+
MAX_CPU_MILLICORES = 16000
26+
CPU_STEP_MILLICORES = 250
27+
MEMORY_MIB_PER_CORE = 2 * 1024
28+
29+
30+
def _build_sandbox_worker_profile(
31+
*,
32+
activities: Iterable[SandboxActivity],
33+
scheduler_managed_identity_client_id: str = "",
34+
worker_profile_id: str,
35+
image: Optional[SandboxWorkerProfileImageOptions] = None,
36+
cpu: str = DEFAULT_CPU,
37+
memory: str = DEFAULT_MEMORY,
38+
environment_variables: Optional[dict[str, str]] = None,
39+
max_concurrent_activities: int = DEFAULT_MAX_CONCURRENT_ACTIVITIES) -> pb.SandboxWorkerProfile:
40+
"""Build a sandbox activity worker_profile.
41+
42+
Args:
43+
image: Sandbox worker image options with the full OCI image reference,
44+
such as "myregistry.azurecr.io/workers/hello:1.0" or
45+
"myregistry.azurecr.io/workers/hello@sha256:0123456789abcdef...".
46+
"""
47+
image_options = image or SandboxWorkerProfileImageOptions()
48+
resolved_activities = resolve_activities(activities)
49+
if not resolved_activities:
50+
raise ValueError("Sandbox activity worker_profile requires at least one activity.")
51+
52+
if not worker_profile_id or not worker_profile_id.strip():
53+
raise ValueError("Sandbox activity worker_profile requires a worker profile ID.")
54+
55+
if max_concurrent_activities <= 0:
56+
raise ValueError("Sandbox activity max concurrent activities must be greater than zero.")
57+
58+
image_ref = normalize_required(
59+
image_options.image_ref,
60+
"Sandbox activity image metadata requires a container image reference like "
61+
"'myregistry.azurecr.io/workers/hello:1.0' or "
62+
"'myregistry.azurecr.io/workers/hello@sha256:...'.")
63+
64+
resolved_scheduler_managed_identity_client_id = normalize_required(
65+
scheduler_managed_identity_client_id,
66+
"Sandbox activity worker_profile requires the managed identity client ID workers use to connect to Durable Task Scheduler.")
67+
resolved_image_pull_managed_identity_client_id = normalize_required(
68+
image_options.managed_identity_client_id,
69+
"Sandbox activity worker_profile requires the managed identity client ID used to pull the worker image.")
70+
71+
resolved_cpu, cpu_millicores = _normalize_cpu(cpu)
72+
resolved_memory = _normalize_memory(memory, cpu_millicores)
73+
74+
worker_profile = pb.SandboxWorkerProfile(
75+
worker_profile_id=worker_profile_id.strip(),
76+
image=pb.SandboxActivityImage(
77+
image_ref=image_ref,
78+
managed_identity_client_id=resolved_image_pull_managed_identity_client_id),
79+
resources=pb.SandboxActivityResources(
80+
cpu=resolved_cpu,
81+
memory=resolved_memory),
82+
scheduler_managed_identity_client_id=resolved_scheduler_managed_identity_client_id,
83+
max_concurrent_activities=max_concurrent_activities)
84+
worker_profile.activities.extend([
85+
pb.SandboxActivity(name=activity.name, version=activity.version or "")
86+
for activity in resolved_activities
87+
])
88+
worker_profile.environment_variables.update(environment_variables or {})
89+
worker_profile.image.entrypoint.extend(_normalize_optional_strings(image_options.entrypoint))
90+
worker_profile.image.cmd.extend(_normalize_optional_strings(image_options.cmd))
91+
return worker_profile
92+
93+
94+
def build_sandbox_worker_profiles() -> list[pb.SandboxWorkerProfile]:
95+
"""Build sandbox worker_profiles from worker profile configuration."""
96+
worker_profiles: list[pb.SandboxWorkerProfile] = []
97+
activity_owners: list[tuple[SandboxActivity, str]] = []
98+
for profile in registered_sandbox_worker_profiles():
99+
activities = resolve_activities(profile.activities)
100+
101+
for activity in activities:
102+
existing_profile = next((owner_profile for owner_activity, owner_profile in activity_owners
103+
if activities_overlap(owner_activity, activity)
104+
and owner_profile != profile.worker_profile_id), None)
105+
if existing_profile:
106+
raise ValueError(
107+
f"Sandbox activity '{format_activity(activity)}' is assigned to both worker profile "
108+
f"'{existing_profile}' and '{profile.worker_profile_id}'.")
109+
activity_owners.append((activity, profile.worker_profile_id))
110+
111+
worker_profiles.append(_build_sandbox_worker_profile(
112+
activities=activities,
113+
worker_profile_id=profile.worker_profile_id,
114+
image=profile.image,
115+
scheduler_managed_identity_client_id=profile.scheduler_managed_identity_client_id,
116+
cpu=profile.cpu,
117+
memory=profile.memory,
118+
environment_variables=profile.environment_variables,
119+
max_concurrent_activities=profile.max_concurrent_activities))
120+
121+
return worker_profiles
122+
123+
124+
def _normalize_optional_strings(values: Iterable[str]) -> list[str]:
125+
return [value.strip() for value in values if value and value.strip()]
126+
127+
128+
def _normalize_cpu(value: str) -> tuple[str, int]:
129+
normalized = normalize_required(value, "Sandbox activity worker_profile requires CPU resources.")
130+
milli_cpu = _try_parse_cpu_millicores(normalized)
131+
if (milli_cpu is None
132+
or milli_cpu < MIN_CPU_MILLICORES
133+
or milli_cpu > MAX_CPU_MILLICORES
134+
or milli_cpu % CPU_STEP_MILLICORES != 0):
135+
raise ValueError(
136+
"Sandbox activity CPU resources must match an ADC sandbox CPU tier: "
137+
"250m through 16000m, in 250m increments. "
138+
"Use formats like '500m', '2', or '0.5'.")
139+
return normalized, milli_cpu
140+
141+
142+
def _normalize_memory(value: str, cpu_millicores: int) -> str:
143+
normalized = normalize_required(value, "Sandbox activity worker_profile requires memory resources.")
144+
max_memory_mib = cpu_millicores * MEMORY_MIB_PER_CORE // 1000
145+
memory_mib = _try_parse_memory_mib(normalized)
146+
if memory_mib is None or memory_mib <= 0:
147+
raise ValueError(
148+
"Sandbox activity memory resources must be a positive Kubernetes-style memory quantity. "
149+
"Use formats like '256Mi', '1Gi', or '2048'.")
150+
if memory_mib > max_memory_mib:
151+
raise ValueError(
152+
"Sandbox activity memory resources exceed the ADC sandbox tier maximum for the configured CPU. "
153+
f"Maximum memory for CPU '{cpu_millicores}m' is {max_memory_mib}Mi.")
154+
return normalized
155+
156+
157+
def _try_parse_cpu_millicores(value: str) -> Optional[int]:
158+
try:
159+
if value[-1:].lower() == "m":
160+
return int(value[:-1])
161+
millicores = Decimal(value) * 1000
162+
return int(millicores) if millicores == millicores.to_integral_value() else None
163+
except (InvalidOperation, ValueError):
164+
return None
165+
166+
167+
def _try_parse_memory_mib(value: str) -> Optional[int]:
168+
try:
169+
if value[-2:].lower() == "gi":
170+
return _try_convert_memory_to_mib(Decimal(value[:-2]), 1024)
171+
if value[-2:].lower() == "mi":
172+
return _try_convert_memory_to_mib(Decimal(value[:-2]), 1)
173+
return _try_convert_memory_to_mib(Decimal(value), 1)
174+
except (InvalidOperation, ValueError):
175+
return None
176+
177+
178+
def _try_convert_memory_to_mib(value: Decimal, multiplier: int) -> Optional[int]:
179+
memory_mib = value * multiplier
180+
return int(memory_mib) if memory_mib == memory_mib.to_integral_value() else None

‎durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@
1515
from durabletask.azuremanaged.preview.sandboxes.helpers import resolve_activities
1616
from durabletask.azuremanaged.preview.sandboxes.worker_profiles import (
1717
DEFAULT_MAX_CONCURRENT_ACTIVITIES,
18+
)
19+
from durabletask.azuremanaged.preview.sandboxes.worker_messages import (
1820
build_sandbox_worker_heartbeat,
1921
build_sandbox_worker_start,
2022
)

‎durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/worker_messages.py‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
# Copyright (c) Microsoft Corporation.
2+
# Licensed under the MIT License.
3+
4+
from typing import Iterable, Optional
5+
6+
from durabletask.azuremanaged.internal import sandbox_service_pb2 as pb
7+
from durabletask.azuremanaged.preview.sandboxes.helpers import SandboxActivity
8+
from durabletask.azuremanaged.preview.sandboxes.helpers import resolve_activities
9+
10+
11+
def build_sandbox_worker_start(
12+
*,
13+
taskhub: str,
14+
worker_profile_id: str,
15+
max_activities_count: int,
16+
activities: Iterable[SandboxActivity],
17+
sandbox_provider: Optional[str] = None,
18+
dts_sandbox_identifier: Optional[str] = None) -> pb.SandboxActivityWorkerMessage:
19+
if not taskhub or not taskhub.strip():
20+
raise ValueError("Sandbox activity worker registration requires a task hub name.")
21+
22+
if not worker_profile_id or not worker_profile_id.strip():
23+
raise ValueError("Sandbox activity worker registration requires a worker profile ID.")
24+
25+
if max_activities_count <= 0:
26+
raise ValueError("Sandbox activity worker max activity count must be greater than zero.")
27+
28+
resolved_activities = resolve_activities(activities)
29+
if not resolved_activities:
30+
raise ValueError("Sandbox activity worker registration requires at least one registered activity.")
31+
32+
message = pb.SandboxActivityWorkerMessage(
33+
start=pb.SandboxActivityWorkerStart(
34+
task_hub=taskhub.strip(),
35+
worker_profile_id=worker_profile_id.strip(),
36+
max_activities_count=max_activities_count,
37+
sandbox_provider=_parse_sandbox_provider(sandbox_provider),
38+
dts_sandbox_identifier=(dts_sandbox_identifier or "").strip()))
39+
message.start.activities.extend([
40+
pb.SandboxActivity(name=activity.name, version=activity.version or "")
41+
for activity in resolved_activities
42+
])
43+
return message
44+
45+
46+
def build_sandbox_worker_heartbeat(active_activities_count: int) -> pb.SandboxActivityWorkerMessage:
47+
if active_activities_count < 0:
48+
raise ValueError("Sandbox activity worker active activity count cannot be negative.")
49+
50+
return pb.SandboxActivityWorkerMessage(
51+
heartbeat=pb.SandboxActivityWorkerHeartbeat(
52+
active_activities_count=active_activities_count))
53+
54+
55+
def _parse_sandbox_provider(sandbox_provider: Optional[str]) -> "pb.SandboxProviderKind":
56+
if not sandbox_provider:
57+
return pb.SANDBOX_PROVIDER_KIND_UNSPECIFIED
58+
if sandbox_provider.lower() == "sandbox":
59+
return pb.SANDBOX_PROVIDER_KIND_SANDBOX
60+
if sandbox_provider.lower() == "acasessionpool":
61+
return pb.SANDBOX_PROVIDER_KIND_ACA_SESSION_POOL
62+
return pb.SANDBOX_PROVIDER_KIND_UNSPECIFIED

0 commit comments

Comments
 (0)