1515from transformers import pipeline
1616import torch
1717
18- app = FastAPI (title = "ML Platform API" )
18+ app = FastAPI (
19+ title = "ML Platform API" ,
20+ root_path = os .getenv ("ROOT_PATH" , "/api" ) # For ingress path-based routing
21+ )
1922
2023app .add_middleware (
2124 CORSMiddleware ,
2831UPLOAD_DIR = Path (os .getenv ("UPLOAD_DIR" , "/mnt/uploads" ))
2932UPLOAD_DIR .mkdir (parents = True , exist_ok = True )
3033
34+ # Azure VM auto-start config
35+ AZURE_RESOURCE_GROUP = os .getenv ("AZURE_RESOURCE_GROUP" , "rg-mlplatform" )
36+ AZURE_GPU_VM_NAME = os .getenv ("AZURE_GPU_VM_NAME" , "vm-mlplatform-gpu" )
37+
38+ # Azure Service Principal credentials
39+ AZURE_CLIENT_ID = os .getenv ("AZURE_CLIENT_ID" )
40+ AZURE_CLIENT_SECRET = os .getenv ("AZURE_CLIENT_SECRET" )
41+ AZURE_TENANT_ID = os .getenv ("AZURE_TENANT_ID" )
42+
43+ _azure_logged_in = False
44+
45+ def ensure_azure_login ():
46+ """Ensure Azure CLI is logged in using Service Principal."""
47+ global _azure_logged_in
48+ if _azure_logged_in :
49+ return True
50+
51+ if not all ([AZURE_CLIENT_ID , AZURE_CLIENT_SECRET , AZURE_TENANT_ID ]):
52+ print ("Azure credentials not configured" )
53+ return False
54+
55+ try :
56+ result = subprocess .run (
57+ ["az" , "login" , "--service-principal" ,
58+ "-u" , AZURE_CLIENT_ID ,
59+ "-p" , AZURE_CLIENT_SECRET ,
60+ "--tenant" , AZURE_TENANT_ID ],
61+ capture_output = True , text = True , timeout = 30
62+ )
63+ if result .returncode == 0 :
64+ _azure_logged_in = True
65+ print ("Azure CLI login successful" )
66+ return True
67+ else :
68+ print (f"Azure login failed: { result .stderr } " )
69+ return False
70+ except Exception as e :
71+ print (f"Azure login error: { e } " )
72+ return False
73+
3174_model_cache = {}
3275
3376BASE_MODELS = {
@@ -72,16 +115,54 @@ def load_model(model_id: str):
72115 return None
73116
74117def check_gpu_worker () -> dict :
118+ """Check if GPU worker node is ready in K8s."""
75119 try :
76120 result = subprocess .run (
77- ["kubectl" , "get" , "nodes" , "-l" , "nvidia.com/gpu.present=true" , "-o" , "name" ],
121+ ["kubectl" , "get" , "nodes" , "-l" , "nvidia.com/gpu.present=true" ,
122+ "-o" , "jsonpath={.items[0].status.conditions[?(@.type=='Ready')].status}" ],
78123 capture_output = True , text = True , timeout = 5
79124 )
80- has_gpu = bool ( result .stdout .strip ())
81- return {"available" : has_gpu , "method" : "k8s" }
125+ is_ready = result .stdout .strip () == "True"
126+ return {"available" : is_ready , "method" : "k8s" }
82127 except :
83128 return {"available" : False , "method" : "none" }
84129
130+ def start_gpu_vm () -> dict :
131+ """Start GPU VM using Azure CLI with Service Principal auth."""
132+ # Ensure we're logged in to Azure
133+ if not ensure_azure_login ():
134+ return {"status" : "error" , "message" : "Azure authentication failed - credentials not configured" }
135+
136+ try :
137+ # Check if VM is already running
138+ status_result = subprocess .run (
139+ ["az" , "vm" , "get-instance-view" ,
140+ "--resource-group" , AZURE_RESOURCE_GROUP ,
141+ "--name" , AZURE_GPU_VM_NAME ,
142+ "--query" , "instanceView.statuses[1].displayStatus" , "-o" , "tsv" ],
143+ capture_output = True , text = True , timeout = 30
144+ )
145+ current_status = status_result .stdout .strip ()
146+
147+ if current_status == "VM running" :
148+ return {"status" : "already_running" , "message" : "GPU VM is already running" }
149+
150+ # Start the VM
151+ start_result = subprocess .run (
152+ ["az" , "vm" , "start" ,
153+ "--resource-group" , AZURE_RESOURCE_GROUP ,
154+ "--name" , AZURE_GPU_VM_NAME ,
155+ "--no-wait" ],
156+ capture_output = True , text = True , timeout = 60
157+ )
158+
159+ if start_result .returncode == 0 :
160+ return {"status" : "starting" , "message" : "GPU VM start initiated" }
161+ else :
162+ return {"status" : "error" , "message" : start_result .stderr }
163+ except Exception as e :
164+ return {"status" : "error" , "message" : str (e )}
165+
85166def scan_models_dir () -> dict :
86167 finetuned = {}
87168 if MODELS_DIR .exists ():
@@ -116,7 +197,7 @@ def create_training_job(job_id: str, config: dict) -> dict:
116197 "kind" : "Job" ,
117198 "metadata" : {
118199 "name" : job_id ,
119- "namespace" : "default "
200+ "namespace" : "ml-platform "
120201 },
121202 "spec" : {
122203 "ttlSecondsAfterFinished" : 3600 ,
@@ -125,7 +206,7 @@ def create_training_job(job_id: str, config: dict) -> dict:
125206 "restartPolicy" : "Never" ,
126207 "containers" : [{
127208 "name" : "trainer" ,
128- "image" : "mlplatformacr2024 .azurecr.io/ml-jupyterlab:v1 " ,
209+ "image" : "mlplatformacrwqckxk .azurecr.io/ml-jupyter:latest " ,
129210 "command" : ["python3" , "-c" , f"""
130211import os
131212from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer, TextDataset, DataCollatorForLanguageModeling
@@ -190,7 +271,10 @@ def create_training_job(job_id: str, config: dict) -> dict:
190271 ],
191272 "nodeSelector" : {
192273 "nvidia.com/gpu.present" : "true"
193- }
274+ },
275+ "imagePullSecrets" : [
276+ {"name" : "acr-secret" }
277+ ]
194278 }
195279 }
196280 }
@@ -229,6 +313,32 @@ def health():
229313def gpu_status ():
230314 return check_gpu_worker ()
231315
316+ @app .post ("/gpu/start" )
317+ def gpu_start ():
318+ """Manually start the GPU VM."""
319+ return start_gpu_vm ()
320+
321+ @app .post ("/gpu/stop" )
322+ def gpu_stop ():
323+ """Stop the GPU VM to save costs."""
324+ if not ensure_azure_login ():
325+ return {"status" : "error" , "message" : "Azure authentication failed" }
326+
327+ try :
328+ result = subprocess .run (
329+ ["az" , "vm" , "deallocate" ,
330+ "--resource-group" , AZURE_RESOURCE_GROUP ,
331+ "--name" , AZURE_GPU_VM_NAME ,
332+ "--no-wait" ],
333+ capture_output = True , text = True , timeout = 60
334+ )
335+ if result .returncode == 0 :
336+ return {"status" : "stopping" , "message" : "GPU VM stop initiated" }
337+ else :
338+ return {"status" : "error" , "message" : result .stderr }
339+ except Exception as e :
340+ return {"status" : "error" , "message" : str (e )}
341+
232342@app .get ("/models" )
233343def list_models ():
234344 models = []
@@ -395,22 +505,21 @@ async def submit_training_job(
395505 "learning_rate" : learning_rate
396506 }
397507
398- if gpu ["available" ]:
399- k8s_result = create_training_job (job_id , config )
400- return {
401- "job_id" : job_id ,
402- "status" : "submitted" if k8s_result ["success" ] else "failed" ,
403- "gpu_available" : True ,
404- "config" : config ,
405- "output_path" : f"/mnt/models/{ model_name .lower ()} " ,
406- "k8s_result" : k8s_result
407- }
408- else :
409- return {
410- "job_id" : job_id ,
411- "status" : "pending" ,
412- "gpu_available" : False ,
413- "config" : config ,
414- "output_path" : f"/mnt/models/{ model_name .lower ()} " ,
415- "message" : "GPU not available - job queued"
416- }
508+ # Auto-start GPU VM if not available
509+ vm_start_result = None
510+ if not gpu ["available" ]:
511+ vm_start_result = start_gpu_vm ()
512+
513+ # Create job - K8s will schedule it when GPU node is ready
514+ k8s_result = create_training_job (job_id , config )
515+
516+ return {
517+ "job_id" : job_id ,
518+ "status" : "submitted" if k8s_result ["success" ] else "failed" ,
519+ "gpu_available" : gpu ["available" ],
520+ "gpu_vm_start" : vm_start_result ,
521+ "config" : config ,
522+ "output_path" : f"/mnt/models/{ model_name .lower ()} " ,
523+ "k8s_result" : k8s_result ,
524+ "message" : "Job submitted. GPU VM starting..." if vm_start_result else None
525+ }
0 commit comments