Merge branch 'develop' into develop

Author: dcrankshaw

clipper_admin/clipper_admin/clipper_admin.py

@@ -281,7 +281,9 @@ def build_and_deploy_model(self,
             for a model can be changed at any time with
             :py:meth:`clipper.ClipperConnection.set_num_replicas`.
         batch_size : int, optional
-            The user-defined batch size.
+            The user-defined query batch size for the model. Replicas of the model will attempt
+            to process at most `batch_size` queries simultaneously. They may process smaller 
+            batches if `batch_size` queries are not immediately available.
             If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
             replicas of this model.
         Raises
@@ -443,7 +445,9 @@ def deploy_model(self,
             for a model can be changed at any time with
             :py:meth:`clipper.ClipperConnection.set_num_replicas`.
         batch_size : int, optional
-            The user-defined batch size.
+            The user-defined query batch size for the model. Replicas of the model will attempt
+            to process at most `batch_size` queries simultaneously. They may process smaller 
+            batches if `batch_size` queries are not immediately available.
             If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
             replicas of this model.
 
@@ -516,7 +520,9 @@ def register_model(self,
             A list of strings annotating the model. These are ignored by Clipper
             and used purely for user annotations.
         batch_size : int, optional
-            The user-defined batch size.
+            The user-defined query batch size for the model. Replicas of the model will attempt
+            to process at most `batch_size` queries simultaneously. They may process smaller 
+            batches if `batch_size` queries are not immediately available.
             If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
             replicas of this model.
 

clipper_admin/clipper_admin/deployers/check_and_write_deps.py

@@ -2,6 +2,7 @@
 from conda.api import get_index
 from conda.base.context import context
 from conda.exceptions import UnsatisfiableError, NoPackagesFoundError
+from requests.exceptions import ChunkedEncodingError
 import conda.resolve
 import conda_env.specs as specs
 import sys
@@ -66,7 +67,15 @@ def check_solvability_write_deps(env_path, directory, platform,
         on the container os. Otherwise returns False.
     """
 
-    index = get_index(platform=platform)
+    def get_packages_index(tries=5):
+        for n in range(tries):
+            try:
+                return get_index(platform=platform)
+            except ChunkedEncodingError as e:
+                if n == tries - 1:
+                    raise e
+
+    index = get_packages_index()
     r = conda.resolve.Resolve(index)
     spec = specs.detect(filename=env_path)
     env = spec.environment

clipper_admin/clipper_admin/deployers/pyspark.py

@@ -26,7 +26,8 @@ def create_endpoint(
         labels=None,
         registry=None,
         base_image="clipper/pyspark-container:{}".format(__version__),
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Registers an app and deploys the provided predict function with PySpark model as
     a Clipper model.
 
@@ -79,13 +80,19 @@ def create_endpoint(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
     """
 
     clipper_conn.register_application(name, input_type, default_output,
                                       slo_micros)
     deploy_pyspark_model(clipper_conn, name, version, input_type, func,
                          pyspark_model, sc, base_image, labels, registry,
-                         num_replicas)
+                         num_replicas, batch_size)
 
     clipper_conn.link_model_to_app(name, name)
 
@@ -101,7 +108,8 @@ def deploy_pyspark_model(
         base_image="clipper/pyspark-container:{}".format(__version__),
         labels=None,
         registry=None,
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Deploy a Python function with a PySpark model.
 
     The function must take 3 arguments (in order): a SparkSession, the PySpark model, and a list of
@@ -141,7 +149,12 @@ def deploy_pyspark_model(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
-
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
 
     Example
     -------
@@ -214,7 +227,7 @@ def predict(spark, model, inputs):
     # Deploy model
     clipper_conn.build_and_deploy_model(name, version, input_type,
                                         serialization_dir, base_image, labels,
-                                        registry, num_replicas)
+                                        registry, num_replicas, batch_size)
 
     # Remove temp files
     shutil.rmtree(serialization_dir)

clipper_admin/clipper_admin/deployers/python.py

@@ -22,7 +22,8 @@ def create_endpoint(
         labels=None,
         registry=None,
         base_image="clipper/python-closure-container:{}".format(__version__),
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Registers an application and deploys the provided predict function as a model.
 
     Parameters
@@ -70,12 +71,19 @@ def create_endpoint(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
     """
 
     clipper_conn.register_application(name, input_type, default_output,
                                       slo_micros)
     deploy_python_closure(clipper_conn, name, version, input_type, func,
-                          base_image, labels, registry, num_replicas)
+                          base_image, labels, registry, num_replicas,
+                          batch_size)
 
     clipper_conn.link_model_to_app(name, name)
 
@@ -89,7 +97,8 @@ def deploy_python_closure(
         base_image="clipper/python-closure-container:{}".format(__version__),
         labels=None,
         registry=None,
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Deploy an arbitrary Python function to Clipper.
 
     The function should take a list of inputs of the type specified by `input_type` and
@@ -125,7 +134,12 @@ def deploy_python_closure(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
-
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
 
     Example
     -------
@@ -171,6 +185,6 @@ def centered_predict(inputs):
     # Deploy function
     clipper_conn.build_and_deploy_model(name, version, input_type,
                                         serialization_dir, base_image, labels,
-                                        registry, num_replicas)
+                                        registry, num_replicas, batch_size)
     # Remove temp files
     shutil.rmtree(serialization_dir)

clipper_admin/clipper_admin/deployers/pytorch.py

@@ -28,7 +28,8 @@ def create_endpoint(
         labels=None,
         registry=None,
         base_image="clipper/pytorch-container:{}".format(__version__),
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Registers an app and deploys the provided predict function with PyTorch model as
     a Clipper model.
     Parameters
@@ -78,13 +79,19 @@ def create_endpoint(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
     """
 
     clipper_conn.register_application(name, input_type, default_output,
                                       slo_micros)
     deploy_pytorch_model(clipper_conn, name, version, input_type, func,
                          pytorch_model, base_image, labels, registry,
-                         num_replicas)
+                         num_replicas, batch_size)
 
     clipper_conn.link_model_to_app(name, name)
 
@@ -99,7 +106,8 @@ def deploy_pytorch_model(
         base_image="clipper/pytorch-container:{}".format(__version__),
         labels=None,
         registry=None,
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Deploy a Python function with a PyTorch model.
     Parameters
     ----------
@@ -133,6 +141,13 @@ def deploy_pytorch_model(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
+        
     Example
     -------
     
@@ -185,7 +200,7 @@ def predict(model, inputs):
     # Deploy model
     clipper_conn.build_and_deploy_model(name, version, input_type,
                                         serialization_dir, base_image, labels,
-                                        registry, num_replicas)
+                                        registry, num_replicas, batch_size)
 
     # Remove temp files
     shutil.rmtree(serialization_dir)

clipper_admin/clipper_admin/deployers/tensorflow.py

@@ -26,7 +26,8 @@ def create_endpoint(clipper_conn,
                     labels=None,
                     registry=None,
                     base_image="clipper/tf-container:{}".format(__version__),
-                    num_replicas=1):
+                    num_replicas=1,
+                    batch_size=-1):
     """Registers an app and deploys the provided predict function with TensorFlow model as
     a Clipper model.
 
@@ -76,13 +77,19 @@ def create_endpoint(clipper_conn,
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
     """
 
     clipper_conn.register_application(name, input_type, default_output,
                                       slo_micros)
     deploy_tensorflow_model(clipper_conn, name, version, input_type, func,
                             tf_sess_or_saved_model_path, base_image, labels,
-                            registry, num_replicas)
+                            registry, num_replicas, batch_size)
 
     clipper_conn.link_model_to_app(name, name)
 
@@ -97,7 +104,8 @@ def deploy_tensorflow_model(
         base_image="clipper/tf-container:{}".format(__version__),
         labels=None,
         registry=None,
-        num_replicas=1):
+        num_replicas=1,
+        batch_size=-1):
     """Deploy a Python prediction function with a Tensorflow session or saved Tensorflow model.
     Parameters
     ----------
@@ -131,7 +139,12 @@ def deploy_tensorflow_model(
         The number of replicas of the model to create. The number of replicas
         for a model can be changed at any time with
         :py:meth:`clipper.ClipperConnection.set_num_replicas`.
-
+    batch_size : int, optional
+        The user-defined query batch size for the model. Replicas of the model will attempt
+        to process at most `batch_size` queries simultaneously. They may process smaller 
+        batches if `batch_size` queries are not immediately available.
+        If the default value of -1 is used, Clipper will adaptively calculate the batch size for individual
+        replicas of this model.
 
     Example
     -------
@@ -239,7 +252,7 @@ def predict(sess, inputs):
     # Deploy model
     clipper_conn.build_and_deploy_model(name, version, input_type,
                                         serialization_dir, base_image, labels,
-                                        registry, num_replicas)
+                                        registry, num_replicas, batch_size)
 
     # Remove temp files
     shutil.rmtree(serialization_dir)

clipper_admin/clipper_admin/docker/docker_container_manager.py

@@ -25,6 +25,7 @@ def __init__(self,
                  clipper_rpc_port=7000,
                  redis_ip=None,
                  redis_port=6379,
+                 prometheus_port=9090,
                  docker_network="clipper_network",
                  extra_container_kwargs={}):
         """
@@ -63,6 +64,7 @@ def __init__(self,
         else:
             self.external_redis = True
         self.redis_port = redis_port
+        self.prometheus_port = prometheus_port
         if docker_network is "host":
             raise ClipperException(
                 "DockerContainerManager does not support running Clipper on the "
@@ -156,7 +158,7 @@ def start_clipper(self, query_frontend_image, mgmt_frontend_image,
         setup_metric_config(query_frontend_metric_name,
                             CLIPPER_INTERNAL_METRIC_PORT)
         run_metric_image(self.docker_client, self.common_labels,
-                         self.extra_container_kwargs)
+                         self.prometheus_port, self.extra_container_kwargs)
 
         self.connect()
 

clipper_admin/clipper_admin/docker/docker_metric_utils.py

@@ -4,6 +4,8 @@
 import os
 from ..version import __version__
 
+PROM_VERSION = "v2.1.0"
+
 
 def ensure_clipper_tmp():
     """
@@ -80,15 +82,15 @@ def setup_metric_config(query_frontend_metric_name,
         yaml.dump(prom_config, f)
 
 
-def run_metric_image(docker_client, common_labels, extra_container_kwargs):
+def run_metric_image(docker_client, common_labels, prometheus_port,
+                     extra_container_kwargs):
     """
     Run the prometheus image.
     :param docker_client: The docker client object
     :param common_labels: Labels to pass in
     :param extra_container_kwargs: Kwargs to pass in.
     :return: None
     """
-
     metric_cmd = [
         "--config.file=/etc/prometheus/prometheus.yml",
         "--storage.tsdb.path=/prometheus",
@@ -98,10 +100,10 @@ def run_metric_image(docker_client, common_labels, extra_container_kwargs):
     ]
     metric_labels = common_labels.copy()
     docker_client.containers.run(
-        "prom/prometheus",
+        "prom/prometheus:{}".format(PROM_VERSION),
         metric_cmd,
         name="metric_frontend-{}".format(random.randint(0, 100000)),
-        ports={'9090/tcp': 9090},
+        ports={'9090/tcp': prometheus_port},
         volumes={
             '/tmp/clipper/prometheus.yml': {
                 'bind': '/etc/prometheus/prometheus.yml',