clean up on-prem artificats

Author: TarunRavikumar

model-engine/model_engine_server/domain/entities/model_bundle_entity.py

@@ -75,6 +75,13 @@ def validate_fields_present_for_framework_type(cls, field_values):
                 "Expected `image_tag` to be non-null because the custom framework "
                 "type was selected."
             )
+            if not field_values.get("ecr_repo"):
+                from model_engine_server.core.config import infra_config
+                if infra_config().cloud_provider != "onprem":
+                    raise ValueError(
+                        "Expected `ecr_repo` to be non-null for custom framework. "
+                        "For on-prem deployments, ecr_repo can be omitted to use direct image references."
+                    )
         return field_values
 
     model_config = ConfigDict(from_attributes=True)

model-engine/model_engine_server/infra/gateways/resources/onprem_queue_endpoint_resource_delegate.py

@@ -26,7 +26,7 @@ async def create_queue_if_not_exists(
             f"(Redis queues don't require explicit creation)"
         )
 
-        return QueueInfo(queue_name=queue_name, queue_url=None)
+        return QueueInfo(queue_name=queue_name, queue_url=queue_name)
 
     async def delete_queue(self, endpoint_id: str) -> None:
         queue_name = QueueEndpointResourceDelegate.endpoint_id_to_queue_name(endpoint_id)
@@ -37,7 +37,11 @@ async def delete_queue(self, endpoint_id: str) -> None:
     async def get_queue_attributes(self, endpoint_id: str) -> Dict[str, Any]:
         queue_name = QueueEndpointResourceDelegate.endpoint_id_to_queue_name(endpoint_id)
 
-        logger.debug(f"Getting attributes for queue {queue_name}")
+        logger.warning(
+            f"Getting queue attributes for {queue_name} - returning hardcoded values. "
+            f"On-prem Redis queues do not support real-time message counts. "
+            f"Do not rely on ApproximateNumberOfMessages for autoscaling decisions."
+        )
 
         return {
             "Attributes": {

model-engine/model_engine_server/infra/gateways/s3_file_storage_gateway.py

@@ -92,11 +92,12 @@ async def list_files(self, owner: str) -> List[FileMetadata]:
             files = []
             for obj in objects.get("Contents", []):
                 key = obj["Key"]
-                file_id = key[len(owner) :].lstrip("/")
-                if file_id:
-                    file_metadata = await self.get_file(owner, file_id)
-                    if file_metadata:
-                        files.append(file_metadata)
+                if key.startswith(owner):
+                    file_id = key[len(owner):].lstrip("/")
+                    if file_id:
+                        file_metadata = await self.get_file(owner, file_id)
+                        if file_metadata:
+                            files.append(file_metadata)
             logger.debug(f"Listed {len(files)} files for owner {owner}")
             return files
         except Exception as e:

model-engine/model_engine_server/infra/gateways/s3_utils.py

@@ -8,16 +8,12 @@
 logger = make_logger(logger_name())
 
 
-def is_onprem_mode() -> bool:
-    return os.getenv("DEPLOYMENT_MODE") == "onprem"
-
-
 def get_s3_client(kwargs: Optional[Dict[str, Any]] = None):
     kwargs = kwargs or {}
     session = boto3.Session()
     client_kwargs = {}
 
-    if is_onprem_mode():
+    if infra_config().cloud_provider == "onprem":
         logger.debug("Using on-prem/MinIO S3-compatible configuration")
 
         s3_endpoint = getattr(infra_config(), "s3_endpoint_url", None) or os.getenv(
@@ -45,7 +41,7 @@ def get_s3_resource(kwargs: Optional[Dict[str, Any]] = None):
     session = boto3.Session()
     resource_kwargs = {}
 
-    if is_onprem_mode():
+    if infra_config().cloud_provider == "onprem":
         logger.debug("Using on-prem/MinIO S3-compatible configuration")
 
         s3_endpoint = getattr(infra_config(), "s3_endpoint_url", None) or os.getenv(

model-engine/model_engine_server/infra/repositories/onprem_docker_repository.py

@@ -13,10 +13,17 @@ def image_exists(
         self, image_tag: str, repository_name: str, aws_profile: Optional[str] = None
     ) -> bool:
         if not repository_name:
-            logger.debug(f"Direct image reference: {image_tag}, assuming exists")
+            logger.warning(
+                f"Direct image reference: {image_tag}, assuming exists. "
+                f"Image validation skipped for on-prem deployments."
+            )
             return True
 
-        logger.debug(f"Registry image: {repository_name}:{image_tag}, assuming exists")
+        logger.warning(
+            f"Registry image: {repository_name}:{image_tag}, assuming exists. "
+            f"Image validation skipped for on-prem deployments. "
+            f"Deployment will fail if image does not exist in registry."
+        )
         return True
 
     def get_image_url(self, image_tag: str, repository_name: str) -> str: