Onprem Compatibility Change

Author: charlesahn-scale

charts/model-engine/templates/_helpers.tpl

@@ -256,6 +256,10 @@ env:
   - name: ABS_CONTAINER_NAME
     value: {{ .Values.azure.abs_container_name }}
   {{- end }}
+  {{- if .Values.s3EndpointUrl }}
+  - name: S3_ENDPOINT_URL
+    value: {{ .Values.s3EndpointUrl | quote }}
+  {{- end }}
 {{- end }}
 
 {{- define "modelEngine.syncForwarderTemplateEnv" -}}
@@ -342,9 +346,27 @@ env:
     value: "/workspace/model-engine/model_engine_server/core/configs/config.yaml"
   {{- end }}
   - name: CELERY_ELASTICACHE_ENABLED
-    value: "true"
+    value: {{ .Values.celeryElasticacheEnabled | default true | quote }}
   - name: LAUNCH_SERVICE_TEMPLATE_FOLDER
     value: "/workspace/model-engine/model_engine_server/infra/gateways/resources/templates"
+  {{- if .Values.s3EndpointUrl }}
+  - name: S3_ENDPOINT_URL
+    value: {{ .Values.s3EndpointUrl | quote }}
+  {{- end }}
+  {{- if .Values.redisHost }}
+  - name: REDIS_HOST
+    value: {{ .Values.redisHost | quote }}
+  - name: REDIS_PORT
+    value: {{ .Values.redisPort | default "6379" | quote }}
+  {{- end }}
+  {{- if .Values.celeryBrokerUrl }}
+  - name: CELERY_BROKER_URL
+    value: {{ .Values.celeryBrokerUrl | quote }}
+  {{- end }}
+  {{- if .Values.celeryResultBackend }}
+  - name: CELERY_RESULT_BACKEND
+    value: {{ .Values.celeryResultBackend | quote }}
+  {{- end }}
   {{- if .Values.redis.auth}}
   - name: REDIS_AUTH_TOKEN
     value: {{ .Values.redis.auth }}

model-engine/model_engine_server/api/dependencies.py

@@ -227,26 +227,24 @@ def _get_external_interfaces(
     queue_delegate: QueueEndpointResourceDelegate
     if CIRCLECI:
         queue_delegate = FakeQueueEndpointResourceDelegate()
-    elif infra_config().cloud_provider == "azure":
-        queue_delegate = ASBQueueEndpointResourceDelegate()
     elif infra_config().cloud_provider == "onprem":
         queue_delegate = OnPremQueueEndpointResourceDelegate()
+    elif infra_config().cloud_provider == "azure":
+        queue_delegate = ASBQueueEndpointResourceDelegate()
     else:
         queue_delegate = SQSQueueEndpointResourceDelegate(
             sqs_profile=os.getenv("SQS_PROFILE", hmi_config.sqs_profile)
         )
 
     inference_task_queue_gateway: TaskQueueGateway
     infra_task_queue_gateway: TaskQueueGateway
-    if CIRCLECI:
+    if CIRCLECI or infra_config().cloud_provider == "onprem":
+        # On-prem uses Redis-based task queues
         inference_task_queue_gateway = redis_24h_task_queue_gateway
         infra_task_queue_gateway = redis_task_queue_gateway
     elif infra_config().cloud_provider == "azure":
         inference_task_queue_gateway = servicebus_task_queue_gateway
         infra_task_queue_gateway = servicebus_task_queue_gateway
-    elif infra_config().cloud_provider == "onprem":
-        inference_task_queue_gateway = redis_task_queue_gateway
-        infra_task_queue_gateway = redis_task_queue_gateway
     elif infra_config().celery_broker_type_redis:
         inference_task_queue_gateway = redis_task_queue_gateway
         infra_task_queue_gateway = redis_task_queue_gateway
@@ -288,10 +286,8 @@ def _get_external_interfaces(
     if infra_config().cloud_provider == "azure":
         filesystem_gateway = ABSFilesystemGateway()
         llm_artifact_gateway = ABSLLMArtifactGateway()
-    elif infra_config().cloud_provider == "onprem":
-        filesystem_gateway = S3FilesystemGateway()  # Uses MinIO via s3_utils
-        llm_artifact_gateway = S3LLMArtifactGateway()  # Uses MinIO via s3_utils
     else:
+        # AWS uses S3, on-prem uses MinIO (S3-compatible)
         filesystem_gateway = S3FilesystemGateway()
         llm_artifact_gateway = S3LLMArtifactGateway()
     model_endpoints_schema_gateway = LiveModelEndpointsSchemaGateway(
@@ -341,10 +337,8 @@ def _get_external_interfaces(
     if infra_config().cloud_provider == "azure":
         llm_fine_tune_repository = ABSFileLLMFineTuneRepository(file_path=file_path)
         llm_fine_tune_events_repository = ABSFileLLMFineTuneEventsRepository()
-    elif infra_config().cloud_provider == "onprem":
-        llm_fine_tune_repository = S3FileLLMFineTuneRepository(file_path=file_path)  # Uses MinIO
-        llm_fine_tune_events_repository = S3FileLLMFineTuneEventsRepository()  # Uses MinIO
     else:
+        # AWS uses S3, on-prem uses MinIO (S3-compatible)
         llm_fine_tune_repository = S3FileLLMFineTuneRepository(file_path=file_path)
         llm_fine_tune_events_repository = S3FileLLMFineTuneEventsRepository()
     llm_fine_tuning_service = DockerImageBatchJobLLMFineTuningService(
@@ -360,18 +354,17 @@ def _get_external_interfaces(
     file_storage_gateway: FileStorageGateway
     if infra_config().cloud_provider == "azure":
         file_storage_gateway = ABSFileStorageGateway()
-    elif infra_config().cloud_provider == "onprem":
-        file_storage_gateway = S3FileStorageGateway()  # Uses MinIO via s3_utils
     else:
+        # AWS uses S3, on-prem uses MinIO (S3-compatible)
         file_storage_gateway = S3FileStorageGateway()
 
     docker_repository: DockerRepository
     if CIRCLECI:
         docker_repository = FakeDockerRepository()
-    elif infra_config().cloud_provider == "azure":
-        docker_repository = ACRDockerRepository()
     elif infra_config().cloud_provider == "onprem":
         docker_repository = OnPremDockerRepository()
+    elif infra_config().cloud_provider == "azure":
+        docker_repository = ACRDockerRepository()
     else:
         docker_repository = ECRDockerRepository()
 

model-engine/model_engine_server/common/config.py

@@ -70,12 +70,13 @@ class HostedModelInferenceServiceConfig:
     user_inference_tensorflow_repository: str
     docker_image_layer_cache_repository: str
     sensitive_log_mode: bool
-    # Exactly one of the following three must be specified
+    # Exactly one of the following must be specified for Redis cache
     cache_redis_aws_url: Optional[str] = None  # also using this to store sync autoscaling metrics
     cache_redis_azure_host: Optional[str] = None
     cache_redis_aws_secret_name: Optional[str] = (
         None  # Not an env var because the redis cache info is already here
     )
+    cache_redis_onprem_url: Optional[str] = None  # For on-prem Redis (e.g., redis://redis:6379/0)
     sglang_repository: Optional[str] = None
 
     @classmethod
@@ -90,8 +91,13 @@ def from_yaml(cls, yaml_path):
 
     @property
     def cache_redis_url(self) -> str:
+        # On-prem Redis support (explicit URL, no cloud provider dependency)
+        if self.cache_redis_onprem_url:
+            return self.cache_redis_onprem_url
+
         cloud_provider = infra_config().cloud_provider
 
+        # On-prem: support REDIS_HOST env var fallback
         if cloud_provider == "onprem":
             if self.cache_redis_aws_url:
                 logger.info("On-prem deployment using cache_redis_aws_url")

model-engine/model_engine_server/common/io.py

@@ -3,17 +3,19 @@
 import os
 from typing import Any
 
+import boto3
 import smart_open
 from model_engine_server.core.config import infra_config
 
 
 def open_wrapper(uri: str, mode: str = "rt", **kwargs):
     client: Any
+    cloud_provider: str
+    # This follows the 5.1.0 smart_open API
     try:
         cloud_provider = infra_config().cloud_provider
     except Exception:
         cloud_provider = "aws"
-
     if cloud_provider == "azure":
         from azure.identity import DefaultAzureCredential
         from azure.storage.blob import BlobServiceClient
@@ -23,9 +25,9 @@ def open_wrapper(uri: str, mode: str = "rt", **kwargs):
             DefaultAzureCredential(),
         )
     else:
-        from model_engine_server.infra.gateways.s3_utils import get_s3_client
-
-        client = get_s3_client(kwargs)
+        profile_name = kwargs.get("aws_profile", os.getenv("AWS_PROFILE"))
+        session = boto3.Session(profile_name=profile_name)
+        client = session.client("s3")
 
     transport_params = {"client": client}
     return smart_open.open(uri, mode, transport_params=transport_params)

model-engine/model_engine_server/core/aws/roles.py

@@ -119,12 +119,21 @@ def session(role: Optional[str], session_type: SessionT = Session) -> SessionT:
 
     :param:`session_type` defines the type of session to return. Most users will use
     the default boto3 type. Some users required a special type (e.g aioboto3 session).
+
+    For on-prem deployments without AWS profiles, pass role=None or role=""
+    to use default credentials from environment variables (AWS_ACCESS_KEY_ID, etc).
     """
     # Do not assume roles in CIRCLECI
     if os.getenv("CIRCLECI"):
         logger.warning(f"In circleci, not assuming role (ignoring: {role})")
         role = None
-    sesh: SessionT = session_type(profile_name=role)
+
+    # Use profile-based auth only if role is specified
+    # For on-prem with MinIO, role will be None or empty - use env var credentials
+    if role:
+        sesh: SessionT = session_type(profile_name=role)
+    else:
+        sesh: SessionT = session_type()  # Uses default credential chain (env vars)
     return sesh
 
 

model-engine/model_engine_server/core/aws/storage_client.py

@@ -1,3 +1,4 @@
+import os
 import time
 from typing import IO, Callable, Iterable, Optional, Sequence
 
@@ -20,6 +21,10 @@
 
 
 def sync_storage_client(**kwargs) -> BaseClient:
+    # Support for MinIO/on-prem S3-compatible storage
+    endpoint_url = os.getenv("S3_ENDPOINT_URL")
+    if endpoint_url and "endpoint_url" not in kwargs:
+        kwargs["endpoint_url"] = endpoint_url
     return session(infra_config().profile_ml_worker).client("s3", **kwargs)  # type: ignore