Metrics and Monitoring V0.0.2: Model Container, Latencies, and Example (#357)

* Allow port specification for Prometheus

* Add monitoring example with grafana

This commit adds an example in the example folder. It helps
the user to visualize clipper metric.

The init_grafana.py script launches a grafana/grafana docker
container. It adds prometheus as a data source. I attempted
to add the dashboard via Grafana API but failed after trying
it out for hours. This seems like a persistent issue for at
least two years: grafana/grafana#2816

<Simon Mo>

* Fix dict unpacking lint error

* Format Code

* Update yapf

* Update yapf and reformat code

* Check yapf version in shell script

* Change `echo` to `cat`

* Add python version check; almost ready

* Format code with python version 2.7.12

* Allow port specification for Prometheus

* Add monitoring example with grafana

This commit adds an example in the example folder. It helps
the user to visualize clipper metric.

The init_grafana.py script launches a grafana/grafana docker
container. It adds prometheus as a data source. I attempted
to add the dashboard via Grafana API but failed after trying
it out for hours. This seems like a persistent issue for at
least two years: grafana/grafana#2816

<Simon Mo>

* Update Metric Config; Finish Implement Model Container

* Add Integration Test

* Fix Integration Tests

* Format Code and Rebase from Develop

* Format Code Again with python2.7.12

* Update monitoring readme; trigger Jenkins

* Address Comments

1. Fix naming and typos
2. Update configuration buckets

* Format Code

* Address comments

- use 1000.0 instead 1000 rpc arithmatic.
- change docs

* Address RPC Test issue

Add a block of code make sure config file is found.

* Update tensowflowcifar

Author: simon-mo

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

@@ -80,15 +80,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",
@@ -101,7 +101,7 @@ def run_metric_image(docker_client, common_labels, extra_container_kwargs):
         "prom/prometheus",
         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',

containers/python/rpc.py

@@ -7,10 +7,12 @@
 from datetime import datetime
 import socket
 import sys
+import os
+import yaml
 from collections import deque
 from multiprocessing import Pipe, Process
 from prometheus_client import start_http_server
-from prometheus_client.core import GaugeMetricFamily, REGISTRY
+from prometheus_client.core import Counter, Gauge, Histogram, Summary
 
 INPUT_TYPE_BYTES = 0
 INPUT_TYPE_INTS = 1
@@ -193,8 +195,6 @@ def run(self, metric_conn):
         sys.stdout.flush()
         sys.stderr.flush()
 
-        pred_metric = dict(model_pred_count=0)
-
         while True:
             socket = self.context.socket(zmq.DEALER)
             poller.register(socket, zmq.POLLIN)
@@ -312,13 +312,25 @@ def run(self, metric_conn):
 
                         response.send(socket, self.event_history)
 
-                        pred_metric['model_pred_count'] += 1
-
-                        metric_conn.send(pred_metric)
+                        recv_time = (t2 - t1).microseconds
+                        parse_time = (t3 - t2).microseconds
+                        handle_time = (t4 - t3).microseconds
+
+                        model_container_metric = {}
+                        model_container_metric['pred_total'] = 1
+                        model_container_metric[
+                            'recv_time_ms'] = recv_time / 1000.0
+                        model_container_metric[
+                            'parse_time_ms'] = parse_time / 1000.0
+                        model_container_metric[
+                            'handle_time_ms'] = handle_time / 1000.0
+                        model_container_metric['end_to_end_latency_ms'] = (
+                            recv_time + parse_time + handle_time) / 1000.0
+                        metric_conn.send(model_container_metric)
 
                         print("recv: %f us, parse: %f us, handle: %f us" %
-                              ((t2 - t1).microseconds, (t3 - t2).microseconds,
-                               (t4 - t3).microseconds))
+                              (recv_time, parse_time, handle_time))
+
                         sys.stdout.flush()
                         sys.stderr.flush()
 
@@ -499,39 +511,95 @@ def start(self, model, host, port, model_name, model_version, input_type):
         self.server.model_input_type = model_input_type
         self.server.model = model
 
-        parent_conn, child_conn = Pipe(duplex=True)
+        child_conn, parent_conn = Pipe(duplex=False)
         metrics_proc = Process(target=run_metric, args=(child_conn, ))
         metrics_proc.start()
         self.server.run(parent_conn)
 
 
 class MetricCollector:
+    """
+    Note this is no longer a Prometheus Collector.
+    Instead, this is simply a class to encapsulate metric recording.
+    """
+
     def __init__(self, pipe_child_conn):
         self.pipe_conn = pipe_child_conn
+        self.metrics = {}
+        self.name_to_type = {}
+
+        self._load_config()
+
+    def _load_config(self):
+        config_file_path = 'metrics_config.yaml'
+
+        # Make sure we are inside /container, where the config file lives.
+        cwd = os.path.split(os.getcwd())[1]
+        if cwd != 'container':
+            config_file_path = os.path.join(os.getcwd(), 'container',
+                                            config_file_path)
+
+        with open(config_file_path, 'r') as f:
+            config = yaml.load(f)
+        config = config['Model Container']
+
+        prefix = 'clipper_{}_'.format(config.pop('prefix'))
+
+        for name, spec in config.items():
+            metric_type = spec.get('type')
+            metric_description = spec.get('description')
+
+            if not metric_type and not metric_description:
+                raise Exception(
+                    "{}: Metric Type and Metric Description are Required in Config File.".
+                    format(name))
+
+            if metric_type == 'Counter':
+                self.metrics[name] = Counter(prefix + name, metric_description)
+            elif metric_type == 'Gauge':
+                self.metrics[name] = Gauge(prefix + name, metric_description)
+            elif metric_type == 'Histogram':
+                if 'bucket' in spec.keys():
+                    buckets = spec['bucket'] + [float("inf")]
+                    self.metrics[name] = Histogram(
+                        prefix + name, metric_description, buckets=buckets)
+                else:
+                    self.metrics[name] = Histogram(prefix + name,
+                                                   metric_description)
+            elif metric_type == 'Summary':
+                self.metrics[name] = Summary(prefix + name, metric_description)
+            else:
+                raise Exception(
+                    "Unknown Metric Type: {}. See config file.".format(
+                        metric_type))
+
+            self.name_to_type[name] = metric_type
 
     def collect(self):
-        latest_metric_dict = None
-        while self.pipe_conn.poll():
-            latest_metric_dict = self.pipe_conn.recv()
-        if latest_metric_dict:
-            for name, val in latest_metric_dict.items():
-                try:
-                    yield GaugeMetricFamily(
-                        name=name,
-                        documentation=name,  # Required Argument
-                        value=val)
-                except ValueError:
-                    pass
+        while True:
+            latest_metric_dict = self.pipe_conn.recv(
+            )  # This call is blocking.
+            for name, value in latest_metric_dict.items():
+                metric = self.metrics[name]
+                if self.name_to_type[name] == 'Counter':
+                    metric.inc(value)
+                elif self.name_to_type[name] == 'Gauge':
+                    metric.set(value)
+                elif self.name_to_type[name] == 'Histogram' or self.name_to_type[name] == 'Summary':
+                    metric.observe(value)
+                else:
+                    raise Exception(
+                        "Unknown Metric Type for {}. See config file.".format(
+                            name))
 
 
 def run_metric(child_conn):
     """
     This function takes a child_conn at the end of the pipe and
     receive object to update prometheus metric.
 
-    It is recommended to be ran in a separate process. 
+    It is recommended to be ran in a separate process.
     """
-    REGISTRY.register(MetricCollector(child_conn))
+    collector = MetricCollector(child_conn)
     start_http_server(1390)
-    while True:
-        time.sleep(1)
+    collector.collect()

dockerfiles/RPCDockerfile

@@ -9,12 +9,14 @@ RUN mkdir -p /model \
       && apt-get update \
       && apt-get install -y libzmq3 libzmq3-dev \
       && conda install -y pyzmq \
-      && pip install prometheus_client
+      && pip install prometheus_client pyyaml
 
 WORKDIR /container
 
 COPY containers/python/__init__.py containers/python/rpc.py /container/
 
+COPY monitoring/metrics_config.yaml /container/
+
 ENV CLIPPER_MODEL_PATH=/model
 
 # vim: set filetype=dockerfile:

dockerfiles/TensorFlowCifarDockerfile

@@ -3,7 +3,7 @@ FROM clipper/py-rpc:${CODE_VERSION}
 
 MAINTAINER Dan Crankshaw <dscrankshaw@gmail.com>
 
-RUN conda install tensorflow
+RUN conda install -c conda-forge tensorflow
 
 COPY containers/python/tf_cifar_container.py /container/