feat: populate default image name

deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller.go

@@ -82,6 +82,7 @@ const (
 	KubeAnnotationLWSSize        = "nvidia.com/lws-size"
 	DeploymentTypeStandard       = "standard"
 	DeploymentTypeLeaderWorker   = "leader-worker"
+	ComponentTypePlanner         = "Planner"
 )
 
 // DynamoComponentDeploymentReconciler reconciles a DynamoComponentDeployment object
@@ -1454,7 +1455,9 @@ func (r *DynamoComponentDeploymentReconciler) generatePodTemplateSpec(ctx contex
 		if opt.dynamoComponentDeployment.Spec.DynamoNamespace != nil && *opt.dynamoComponentDeployment.Spec.DynamoNamespace != "" {
 			args = append(args, fmt.Sprintf("--%s.ServiceArgs.dynamo.namespace=%s", opt.dynamoComponentDeployment.Spec.ServiceName, *opt.dynamoComponentDeployment.Spec.DynamoNamespace))
 		}
-		args = append(args, fmt.Sprintf("--%s.environment=%s", opt.dynamoComponentDeployment.Spec.ServiceName, KubernetesDeploymentStrategy))
+		if componentType, exists := opt.dynamoComponentDeployment.Labels[commonconsts.KubeLabelDynamoComponent]; exists && componentType == ComponentTypePlanner {
+			args = append(args, fmt.Sprintf("--%s.environment=%s", opt.dynamoComponentDeployment.Spec.ServiceName, KubernetesDeploymentStrategy))
+		}
 	}
 
 	if len(opt.dynamoComponentDeployment.Spec.Envs) > 0 {

deploy/cloud/operator/internal/controller/dynamocomponentdeployment_controller_test.go

@@ -934,7 +934,7 @@ func TestDynamoComponentDeploymentReconciler_generateLeaderWorkerSet(t *testing.
 										Name:    "main",
 										Image:   "test-image:latest",
 										Command: []string{"sh", "-c"},
-										Args:    []string{"ray start --head --port=6379 && cd src && uv run dynamo serve --system-app-port 5000 --enable-system-app --use-default-health-checks --service-name test-lws-deploy-service test-tag --test-lws-deploy-service.ServiceArgs.dynamo.namespace=default --test-lws-deploy-service.environment=kubernetes"},
+										Args:    []string{"ray start --head --port=6379 && cd src && uv run dynamo serve --system-app-port 5000 --enable-system-app --use-default-health-checks --service-name test-lws-deploy-service test-tag --test-lws-deploy-service.ServiceArgs.dynamo.namespace=default"},
 										Env:     []corev1.EnvVar{{Name: "DYNAMO_PORT", Value: "3000"}},
 										VolumeMounts: []corev1.VolumeMount{
 											{

deploy/sdk/src/dynamo/sdk/cli/build.py

@@ -129,13 +129,18 @@ def from_service(cls, service: ServiceInterface[T]) -> ServiceInfo:
             if DynamoTransport.HTTP in endpoint.transports:
                 api_endpoints.append(f"/{ep_name}")
 
+        image = service.config.image or DYNAMO_IMAGE
+        assert (
+            image is not None
+        ), "Please set DYNAMO_IMAGE environment variable or image field in service config"
+
         # Create config
         config = ServiceConfig(
             name=name,
             service="",
-            resource=service.config.resource.model_dump(),
+            resource=service.config.resources.model_dump(),
             workers=service.config.workers,
-            image=service.config.image,
+            image=image,
             dynamo=service.config.dynamo.model_dump(),
             http_exposed=len(api_endpoints) > 0,
             api_endpoints=api_endpoints,
@@ -423,7 +428,6 @@ def to_package_name(name: str) -> str:
         s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name)
         s2 = re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1)
         ret = s2.replace(":", "_")
-        print(f"Converting {name} to snake_case: {ret}")
         return ret
 
     @staticmethod

deploy/sdk/src/dynamo/sdk/core/lib.py

@@ -60,7 +60,6 @@ def service(
 ) -> Any:
     """Service decorator that's adapter-agnostic"""
     config = ServiceConfig(**kwargs)
-    logger.info(f"inner: {inner} config: {config}")
 
     def decorator(inner: Type[G]) -> ServiceInterface[G]:
         provider = get_target()

deploy/sdk/src/dynamo/sdk/core/protocol/interface.py

@@ -17,10 +17,10 @@
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from enum import Enum, auto
-from typing import Any, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar
+from typing import Any, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union
 
 from fastapi import FastAPI
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 
 from dynamo.sdk.core.protocol.deployment import Env
 
@@ -59,16 +59,22 @@ class DynamoTransport(Enum):
 class ResourceConfig(BaseModel):
     """Configuration for Dynamo resources"""
 
-    cpu: int = 1
-    memory: str = "100Mi"
-    gpu: str = "0"
+    cpu: str = Field(default="1")
+    memory: str = Field(default="500Mi")
+    gpu: str = Field(default="0")
+
+    @field_validator("gpu", mode="before")
+    @classmethod
+    def convert_gpu_to_string(cls, v: Union[str, int]) -> str:
+        """Convert gpu value to string if it's an integer"""
+        return str(v)
 
 
 class ServiceConfig(BaseModel):
     """Base service configuration that can be extended by adapters"""
 
     dynamo: DynamoConfig
-    resource: ResourceConfig = ResourceConfig()
+    resources: ResourceConfig = ResourceConfig()
     workers: int = 1
     image: str | None = None
     envs: List[Env] | None = None

deploy/sdk/src/dynamo/sdk/tests/test_resources.py

@@ -16,6 +16,7 @@
 import pytest
 
 from dynamo.sdk.cli.utils import configure_target_environment
+from dynamo.sdk.core.protocol.interface import ServiceInterface
 from dynamo.sdk.core.runner import TargetEnum
 
 pytestmark = pytest.mark.pre_merge
@@ -40,4 +41,8 @@ class MyService:
         def __init__(self) -> None:
             pass
 
-    assert MyService.config is not None  # type: ignore
+    dyn_svc: ServiceInterface = MyService
+    assert dyn_svc.config is not None  # type: ignore
+    assert dyn_svc.config.resources.cpu == "2"
+    assert dyn_svc.config.resources.gpu == "1"
+    assert dyn_svc.config.resources.memory == "4Gi"

examples/llm/configs/agg_router.yaml

@@ -40,6 +40,8 @@ VllmWorker:
     workers: 1
     resources:
       gpu: '1'
+      cpu: '10'
+      memory: '20Gi'
   common-configs: [model, block-size, max-model-len, router, kv-transfer-config]
 
 Planner:

examples/tensorrt_llm/common/utils.py

@@ -21,7 +21,7 @@
 import weakref
 from enum import Enum
 from queue import Queue
-from typing import Callable, Optional, TypedDict, Union
+from typing import Any, Callable, Coroutine, Optional, TypedDict, Union
 
 logger = logging.getLogger(__name__)
 
@@ -54,7 +54,9 @@ class ConversationMessage(TypedDict):
 class ManagedThread(threading.Thread):
     def __init__(
         self,
-        task: Optional[Union[Callable[..., bool], weakref.WeakMethod]],
+        task: Optional[
+            Union[Callable[..., Coroutine[Any, Any, bool]], weakref.WeakMethod]
+        ],
         error_queue: Optional[Queue] = None,
         name: Optional[str] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
@@ -74,7 +76,9 @@ def set_loop(self, loop: asyncio.AbstractEventLoop):
 
     def run(self):
         while not self.stop_event.is_set():
-            task: Optional[Union[Callable[..., bool], weakref.WeakMethod]] = self.task
+            task: Optional[
+                Union[Callable[..., Coroutine[Any, Any, bool]], weakref.WeakMethod]
+            ] = self.task
             if isinstance(task, weakref.WeakMethod):
                 task = task()
                 if task is None: