Merge pull request #216 from JosepSampe/pywren-dev

Take into account pywren version in knative runtimes

Author: gilv

docs/data-processing.md

@@ -6,8 +6,7 @@ Additionally, the built-in data-processing logic integrates a **data partitioner
 
 
 ## Processing data from IBM Cloud Object Storage
-This mode is activated when you write the parameter **obj** into the function arguments. The input to the partitioner may be either a list of buckets, a list of buckets with object prefix, or a list of data objects. If you set the *size of the chunk* or the *number of chunks*, the partitioner is activated inside PyWren and it is responsible to split the objects into smaller chunks, eventually running one function activation for each generated chunk. If *size of the chunk* and *number of chunks* are not set, chunk is an entire object, so one function activation is executed for each individual object. For example consider the following function:
-
+This mode is activated when you write the parameter **obj** into the function arguments. The input to the partitioner may be either a list of buckets, a list of buckets with object prefix, or a list of data objects. If you set the *size of the chunk* or the *number of chunks*, the partitioner is activated inside PyWren and it is responsible to split the objects into smaller chunks, eventually running one function activation for each generated chunk. If *size of the chunk* and *number of chunks* are not set, chunk is an entire object, so one function activation is executed for each individual object.
 
 The *obj* parameter is a python class from where you can access all the information related to the object (or chunk) that the function is processing. For example, consider the following function that shows all the available attributes in *obj*:
 
@@ -24,26 +23,26 @@ def my_map_function(obj):
 
 As stated above, the allowed inputs of the function can be:
 
-- Input data is a bucket or a list of buckets. See a complete example in [map_reduce_cos_bucket.py](../examples/map_reduce_cos_bucket.py):
+- Input data is a bucket or a list of buckets. See an example in [map_reduce_cos_bucket.py](../examples/map_reduce_cos_bucket.py):
     ```python
     iterdata = 'cos://bucket1'
     ```
 
--  Input data is a bucket(s) with object prefix. See a complete example in [map_cos_prefix.py](../examples/map_cos_prefix.py):
+-  Input data is a bucket(s) with object prefix. See an example in [map_cos_prefix.py](../examples/map_cos_prefix.py):
     ```python
     iterdata = ['cos://bucket1/images/', 'cos://bucket1/videos/']
     ```
     Notice that you must write the end slash (/) to inform partitioner you are providing an object prefix.
 
-- Input data is a list of object keys. See a complete example in [map_reduce_cos_key.py](../examples/map_reduce_cos_key.py):
+- Input data is a list of object keys. See an example in [map_reduce_cos_key.py](../examples/map_reduce_cos_key.py):
     ```python
     iterdata = ['cos://bucket1/object1', 'cos://bucket1/object2', 'cos://bucket1/object3'] 
     ```
 
-Notice that *iterdata* must be only one of the previous 3 types. Intermingled types are not allowed. For example, you cannot set in the same *iterdata* list a bucket and some object keys:
+Notice that *iterdata* must be only one of the previous 3 types. Intermingled types are not allowed. For example, you cannot set in the same *iterdata* a bucket and some object keys:
 
 ```python
-iterdata = ['cos://bucket1', 'cos://bucket1/object2', 'cos://bucket1/object3'] 
+iterdata = ['cos://bucket1', 'cos://bucket1/object2', 'cos://bucket1/object3']  # Not allowed
 ```
 
 Once iterdata is defined, you can execute PyWren as usual, either using *map()* or **map_reduce()* calls. If you need to split the files in smaller chunks, you can set (optionally) the *chunk_size* or *chunk_n* parameters.
@@ -92,8 +91,7 @@ See a complete example in [map_reduce_url.py](../examples/map_reduce_url.py).
 
 
 ## Reducer granularity            
-By default there will be one reducer for all the object chunks. If you need one reducer for each object, you must set the parameter
-`reducer_one_per_object=True` into the *map()* or *map_reduce()* methods.
+When using the `map_reduce()` API call with `chunk_size` or `chunk_n`, by default there will be only one reducer for all the object chunks from all the objects. Alternatively, you can spawn one reducer for each object by setting the parameter `reducer_one_per_object=True`.
 
 ```python
 pw.map_reduce(my_map_function, bucket_name, my_reduce_function, 

pywren_ibm_cloud/compute/backends/knative/config.py

@@ -99,8 +99,6 @@
         value: .
       - name: pathToDockerFile
         value: runtime/knative/Dockerfile
-      - name: imageTag
-        value: latest
   serviceAccount: pywren-build-pipeline
 """
 

pywren_ibm_cloud/compute/backends/knative/knative.py

@@ -163,10 +163,18 @@ def _create_build_resources(self):
         task_def = yaml.safe_load(kconfig.task_def)
         task_name = task_def['metadata']['name']
 
-        git_url_param = {'name': 'url', 'value': kconfig.GIT_URL_DEFAULT}
-        git_rev_param = {'name': 'revision', 'value': kconfig.GIT_REV_DEFAULT}
-        params = [git_url_param, git_rev_param]
+        if 'git_url' in self.knative_config:
+            git_url_param = {'name': 'url', 'value': self.knative_config['git_url']}
+        else:
+            git_url_param = {'name': 'url', 'value': kconfig.GIT_URL_DEFAULT}
+
+        if 'git_rev' in self.knative_config:
+            git_rev_param = {'name': 'revision', 'value': self.knative_config['git_rev']}
+        else:
+            revision = 'master' if 'SNAPSHOT' in __version__ else __version__
+            git_rev_param = {'name': 'revision', 'value': revision}
 
+        params = [git_url_param, git_rev_param]
         git_res['spec']['params'] = params
 
         try:
@@ -216,16 +224,22 @@ def _build_docker_image_from_git(self, docker_image_name):
         Builds the docker image and pushes it to the docker container registry
         """
         logger.debug("Building default docker image from git")
+
+        revision = 'latest' if 'SNAPSHOT' in __version__ else __version__
+
+        if self.knative_config['docker_repo'] == 'docker.io' and revision != 'latest':
+            resp = requests.get('https://index.docker.io/v1/repositories/{}/tags/{}'
+                                .format(docker_image_name, revision))
+            if resp.status_code == 200:
+                logger.debug('Docker image docker.io/{}:{} already created in Dockerhub. '
+                             'Skipping build process.'.format(docker_image_name, revision))
+                return
+
         task_run = yaml.safe_load(kconfig.task_run)
         image_url = {'name': 'imageUrl', 'value': '/'.join([self.knative_config['docker_repo'], docker_image_name])}
         task_run['spec']['inputs']['params'].append(image_url)
-        #image_tag = {'name': 'imageTag', 'value':  __version__}
-        #task_run['spec']['inputs']['params'].append(image_tag)
-
-        resp = requests.get('https://index.docker.io/v1/repositories/{}/tags/latest'.format(docker_image_name))
-        if resp.status_code == 200:
-            logger.debug('Docker image already created in Dockerhub. Skipping build process.')
-            return
+        image_tag = {'name': 'imageTag', 'value':  revision}
+        task_run['spec']['inputs']['params'].append(image_tag)
 
         self._create_account_resources()
         self._create_build_resources()
@@ -251,6 +265,7 @@ def _build_docker_image_from_git(self, docker_image_name):
                     body=task_run
                 )
 
+        logger.debug("Building image...")
         pod_name = None
         w = watch.Watch()
         for event in w.stream(self.api.list_namespaced_custom_object, namespace=self.namespace,
@@ -290,15 +305,15 @@ def _create_service(self, docker_image_name, runtime_memory, timeout):
         logger.debug("Creating PyWren runtime service resource in k8s")
         svc_res = yaml.safe_load(kconfig.service_res)
 
+        revision = 'latest' if 'SNAPSHOT' in __version__ else __version__
+        # TODO: Take into account revision in service name
         service_name = self._format_service_name(docker_image_name, runtime_memory)
         svc_res['metadata']['name'] = service_name
         svc_res['metadata']['namespace'] = self.namespace
 
         svc_res['spec']['template']['spec']['timeoutSeconds'] = timeout
-
         docker_image = '/'.join([self.knative_config['docker_repo'], docker_image_name])
-        svc_res['spec']['template']['spec']['container']['image'] = docker_image
-
+        svc_res['spec']['template']['spec']['container']['image'] = '{}:{}'.format(docker_image, revision)
         svc_res['spec']['template']['spec']['container']['resources']['limits']['memory'] = '{}Mi'.format(runtime_memory)
 
         try:
@@ -353,8 +368,9 @@ def create_runtime(self, docker_image_name, memory, timeout=kconfig.RUNTIME_TIME
         """
         default_runtime_img_name = self._get_default_runtime_image_name()
         if docker_image_name in ['default', default_runtime_img_name]:
-            # We only build default image. rest of images must already exist
+            # We only build the default image. rest of images must already exist
             # in the docker registry.
+            docker_image_name = default_runtime_img_name
             self._build_docker_image_from_git(default_runtime_img_name)
 
         service_url = self._create_service(docker_image_name, memory, timeout)
@@ -399,28 +415,16 @@ def delete_runtime(self, docker_image_name, memory):
                     plural="services",
                     body=client.V1DeleteOptions()
                 )
-        except Exception as e:
-            if json.loads(e.body)['code'] == 404:
-                log_msg = 'Knative service: resource "{}" Not Found'.format(service_name)
-                logger.debug(log_msg)
+        except Exception:
+            pass
 
     def delete_all_runtimes(self):
         """
         Deletes all runtimes deployed in knative
         """
         runtimes = self.list_runtimes()
-        for image_name, memory in runtimes:
-            service_name = self._format_service_name(image_name, memory)
-            log_msg = 'Deleting runtime: {}'.format(service_name)
-            logger.debug(log_msg)
-            self.api.delete_namespaced_custom_object(
-                    group="serving.knative.dev",
-                    version="v1alpha1",
-                    name=service_name,
-                    namespace=self.namespace,
-                    plural="services",
-                    body=client.V1DeleteOptions()
-                )
+        for docker_image_name, memory in runtimes:
+            self.delete_runtime(docker_image_name, memory)
 
     def list_runtimes(self, docker_image_name='all'):
         """

pywren_ibm_cloud/executor.py

@@ -275,7 +275,7 @@ def map_reduce(self, map_function, map_iterdata, reduce_function, extra_params=N
         self._state = ExecutorState.running
 
         if reducer_wait_local:
-            self.monitor(futures=map_futures)
+            self.wait(fs=map_futures)
 
         reduce_job_id = 'R{}'.format(job_id)
 
@@ -548,7 +548,7 @@ def clean(self, local_execution=True, delete_all=False):
                          'STORE_RESULT': False}
             old_stdout = sys.stdout
             sys.stdout = open(os.devnull, 'w')
-            self.executor.call_async(clean_os_bucket, [storage_bucket, storage_prerix], extra_env=extra_env)
+            self.call_async(clean_os_bucket, [storage_bucket, storage_prerix], extra_env=extra_env)
             sys.stdout = old_stdout
 
         self._state = ExecutorState.finished

pywren_ibm_cloud/runtime/function_handler/handler.py

@@ -23,7 +23,7 @@
 import logging
 import traceback
 import subprocess
-from multiprocessing import Value
+from multiprocessing import Pipe
 from distutils.util import strtobool
 from pywren_ibm_cloud import version
 from pywren_ibm_cloud.utils import sizeof_fmt
@@ -136,8 +136,8 @@ def function_handler(event):
         setup_time = time.time()
         response_status['setup_time'] = round(setup_time - start_time, 8)
 
-        jr_success_flag = Value('i', 0)
-        tr = JobRunner(jobrunner_config, jr_success_flag)
+        handler_conn, jobrunner_conn = Pipe()
+        tr = JobRunner(jobrunner_config, jobrunner_conn)
         tr.daemon = True
         logger.debug('Starting JobRunner process')
         tr.start()
@@ -152,7 +152,9 @@ def function_handler(event):
                    'seconds and was killed'.format(execution_timeout))
             raise Exception('OUTATIME',  msg)
 
-        if jr_success_flag.value == 0:
+        try:
+            handler_conn.recv()
+        except EOFError:
             logger.error('No completion message received from JobRunner process')
             logger.debug('Assuming memory overflow...')
             # Only 1 message is returned by jobrunner when it finishes.

pywren_ibm_cloud/runtime/function_handler/jobrunner.py

@@ -55,11 +55,11 @@ def __del__(self):
 
 class JobRunner(Process):
 
-    def __init__(self, jr_config, jr_success_flag):
+    def __init__(self, jr_config, jobrunner_conn):
         super().__init__()
         start_time = time.time()
         self.jr_config = jr_config
-        self.success_flag = jr_success_flag
+        self.jobrunner_conn = jobrunner_conn
 
         log_level = self.jr_config['log_level']
         cloud_logging_config(log_level)
@@ -312,5 +312,5 @@ def run(self):
                 self.internal_storage.put_data(self.output_key, pickled_output)
                 output_upload_timestamp_t2 = time.time()
                 self.stats.write("output_upload_time", round(output_upload_timestamp_t2 - output_upload_timestamp_t1, 8))
-            self.success_flag.value = 1
+            self.jobrunner_conn.send("Finished")
             logger.info("Finished")

pywren_ibm_cloud/storage/storage.py

@@ -217,7 +217,7 @@ def put_runtime_meta(self, key, runtime_meta):
 
         if not is_remote_cluster():
             filename_local_path = os.path.join(CACHE_DIR, *path)
-            logger.debug("Saving runtime metadata into local cache: {}".format(filename_local_path))
+            logger.debug("Storing runtime metadata into local cache: {}".format(filename_local_path))
 
             if not os.path.exists(os.path.dirname(filename_local_path)):
                 os.makedirs(os.path.dirname(filename_local_path))