improved configuration for workflows calling for GPUs

Author: ksatzke

ManagementService/python/deployWorkflow.py

@@ -132,8 +132,9 @@ def compile_resource_info_map(resource_names, uploaded_resources, email, sapi, d
                 resource_metadata = json.loads(resource_metadata)
                 if "runtime" in resource_metadata:
                     resource_info["runtime"] = resource_metadata["runtime"]
-                #if "on_gpu" in resource_metadata:
-                #    resource_info["on_gpu"] = True
+                print("RESOURCE_INFO_ALL: " +str(resource_info))
+                #if "num_gpu" in resource_metadata:
+                #    print("RESOURCE_INFO: " + str(resource_info["num_gpu"]))
 
             num_chunks_str = dlc.get("grain_source_zip_num_chunks_" + resource_id)
             try:
@@ -296,12 +297,23 @@ def create_k8s_deployment(email, workflow_info, runtime, management=False):
     env.append({'name': 'WORKFLOWID', 'value': workflow_info["workflowId"]})
     env.append({'name': 'WORKFLOWNAME', 'value': workflow_info["workflowName"]})
 
+    """
+    if "num_gpu" in workflow_info.keys():
+        print("INSIDE K8S Deploy, num_gpu: " + str(workflow_info['num_gpu']))
+        num_gpu = int(workflow_info['num_gpu'])
+        # overwrite values from values.yaml for new workflows
+        kservice['spec']['template']['spec']['containers'][0]['resources']['limits']['nvidia.com/gpu'] = str(num_gpu)
+        kservice['spec']['template']['spec']['containers'][0]['resources']['requests']['nvidia.com/gpu'] = str(num_gpu)
+        kservice['spec']['template']['spec']['containers'][0]['image'] = "localhost:5000/microfn/sandbox" 
+        if num_gpu > 0:
+            kservice['spec']['template']['spec']['containers'][0]['image'] = "localhost:5000/microfn/sandbox" 
+    """ 
     # Special handling for the management container
     if management:
         kservice['spec']['template']['spec']['volumes'] = [{ 'name': 'new-workflow-conf', 'configMap': {'name': new_workflow_conf['configmap']}}]
         kservice['spec']['template']['spec']['containers'][0]['volumeMounts'] = [{'name': 'new-workflow-conf', 'mountPath': '/opt/mfn/SandboxAgent/conf'}]
         kservice['spec']['template']['spec']['serviceAccountName'] = new_workflow_conf['mgmtserviceaccount']
-
+        
         # management container should not consume a CPU and use standard sandbox image
         if (labels['workflowid'] == "Management"):
             kservice['spec']['template']['spec']['containers'][0]['resources']['limits']['nvidia.com/gpu'] = "0"
@@ -464,6 +476,13 @@ def handle(value, sapi):
                 runtime = "Java"
             else:
                 runtime = "Python"
+
+            if "num_gpu" in resource_info_map.keys():
+                print ("RESOURCE_INFO_MAP: " + str(resource_info_map))
+                workflow_info['num_gpu'] = resource_info_map['num_gpu']
+            else:
+                workflow_info['num_gpu'] = 0
+
             url, endpoint_key = create_k8s_deployment(email, workflow_info, runtime)
             if url is not None and len(url) > 0:
                 status = "deploying"

mfn_sdk/mfn_sdk/workflow.py

@@ -127,7 +127,7 @@ def json(self,json):
             self.client.action('uploadWorkflowJSON',{'workflow':{'id':self.id,'json':base64.b64encode(self._json.encode()).decode()}})
 
 
-    def deploy(self, timeout=None):
+    def deploy(self, timeout=None): #, num_gpu=None):
         """ deploy a workflow and optionally wait in linearly increasing multiples of 1000ms
         :timeout: By default returns after calling deploy on the workflow without waiting for it to be actually deployed.
             If timeout is set to a numeric <= 0, it waits indefinitely in intervals of 1000ms, 2000ms, 3000ms, ...
@@ -145,6 +145,10 @@ def deploy(self, timeout=None):
         else:
             self.client.action('deployWorkflow',{'workflow':{'id':self.id}})
 
+        #if num_gpu is not None:
+        #    print("NUM_GPU:" + str(num_gpu))
+
+
         # if timeout is None, do not wait but return immediately even if it's not yet deployed
         if timeout is None:
             return

tests/asl_DLIB/test.py

@@ -34,6 +34,6 @@ def test_dlib(self):
 
         testtuplelist =[(inp1, res1)]
 
-        test = MFNTest(test_name = "Dlib__Test")
+        test = MFNTest(test_name = "Dlib_Test", num_gpu = 1)
         test.exec_tests(testtuplelist)
 

tests/mfn_test_utils.py

@@ -43,7 +43,7 @@ class MfnAppTextFormat():
 mfntestfailed = MfnAppTextFormat.STYLE_BOLD + MfnAppTextFormat.COLOR_RED + 'FAILED' + MfnAppTextFormat.END + MfnAppTextFormat.END
 
 class MFNTest():
-    def __init__(self, test_name=None, timeout=None, workflow_filename=None, new_user=False, delete_user=False):
+    def __init__(self, test_name=None, timeout=None, workflow_filename=None, new_user=False, delete_user=False, num_gpu=None):
 
         self._settings = self._get_settings()
 
@@ -84,6 +84,9 @@ def __init__(self, test_name=None, timeout=None, workflow_filename=None, new_use
         if timeout is not None:
             self._settings["timeout"] = timeout
 
+        if num_gpu is not None:
+            self._settings["num_gpu"] = num_gpu
+
         self._log_clear_timestamp = int(time.time() * 1000.0 * 1000.0)
 
         # will be the deployed workflow object in self._client
@@ -190,6 +193,9 @@ def _get_resource_info_map(self, workflow_description=None, resource_info_map=No
                         resource_info["resource_req_filename"] = "requirements/" + resource_ref + "_requirements.txt"
                         resource_info["resource_env_filename"] = "environment_variables/" + resource_ref + "_environment_variables.txt"
                         resource_info_map[resource_ref] = resource_info
+                        resource_info_map[resource_ref]['num_gpu'] = self._settings['num_gpu']
+                        print("resource_info_map: " + json.dumps(resource_info_map))
+
 
         elif "States" in workflow_description:
             states = workflow_description["States"]
@@ -203,6 +209,9 @@ def _get_resource_info_map(self, workflow_description=None, resource_info_map=No
                         resource_info["resource_req_filename"] = "requirements/" + resource_name + "_requirements.txt"
                         resource_info["resource_env_filename"] = "environment_variables/" + resource_name + "_environment_variables.txt"
                         resource_info_map[resource_name] = resource_info
+                        resource_info_map[resource_name]['num_gpu'] = self._settings['num_gpu']
+                        print("resource_info_map: " + json.dumps(resource_info_map))
+
 
                 if "Type" in state and state["Type"] == "Parallel":
                     branches = state['Branches']
@@ -219,10 +228,6 @@ def _get_resource_info_map(self, workflow_description=None, resource_info_map=No
             print("ERROR: invalid workflow description.")
             assert False
 
-        #resource_info_map[resource_name]['on_gpu'] = True
-        
-        #print("resource_info_map: " + str(resource_info_map))
-
         return resource_info_map
 
     def _delete_resource_if_existing(self, existing_resources, resource_name):
@@ -299,7 +304,7 @@ def deploy_workflow(self):
         try:
             wf = self._client.add_workflow(self._workflow_name)
             wf.json = json.dumps(self._workflow_description)
-            wf.deploy(self._settings["timeout"])
+            wf.deploy(self._settings["timeout"]) #, num_gpu=self._settings['num_gpu'])
             self._workflow = wf
             if self._workflow.status != "failed":
                 print("MFN workflow " + self._workflow_name + " deployed.")