You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@marvin.apache.org by we...@apache.org on 2020/10/09 14:27:41 UTC
[incubator-marvin] 02/05: Benchmark support added
This is an automated email from the ASF dual-hosted git repository.
weichen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/incubator-marvin.git
commit 95335cc3d81d22838c72cbde927a4ba4b99519c6
Author: cardosolucas <ca...@gmail.com>
AuthorDate: Wed Sep 23 14:19:49 2020 -0300
Benchmark support added
---
marvin-cli/bin/marvin | 2 +-
marvin-cli/marvin_cli/management/engine.py | 84 ++++++++++++-
marvin-cli/marvin_cli/utils/benchmark.py | 196 +++++++++++++++++++++++++++++
marvin-cli/tests/utils/test_benchmark.py | 137 ++++++++++++++++++++
4 files changed, 417 insertions(+), 2 deletions(-)
diff --git a/marvin-cli/bin/marvin b/marvin-cli/bin/marvin
index f639271..b7fa19b 100644
--- a/marvin-cli/bin/marvin
+++ b/marvin-cli/bin/marvin
@@ -33,7 +33,7 @@ from marvin_cli.management.test import cli as cli_test
EXCLUDE_BY_TYPE = {
'toolbox': ['engine-dryrun', 'engine-grpcserver', 'engine-httpserver', 'notebook',
'lab', 'project-export', 'engine-logs', 'edit-metadata', 'test', 'test-tdd',
- 'test-tox', 'engine-bumpversion'],
+ 'test-tox', 'engine-bumpversion', 'benchmark', 'benchmark-plot'],
'engine': ['project-import']
}
diff --git a/marvin-cli/marvin_cli/management/engine.py b/marvin-cli/marvin_cli/management/engine.py
index 8b51135..fea564b 100644
--- a/marvin-cli/marvin_cli/management/engine.py
+++ b/marvin-cli/marvin_cli/management/engine.py
@@ -32,6 +32,7 @@ from ..utils.misc import call_logs, package_to_name, get_executor_path_or_downlo
from ..utils.misc import generate_timestamp, persist_process
from ..utils.git import git_init, bump_version
from ..utils.log import get_logger
+from ..utils.benchmark import benchmark_thread, create_poi, make_graph
logger = get_logger('engine')
@@ -220,4 +221,85 @@ def http(ctx, grpchost, grpcport, host, port, protocol, action, max_workers,
type=click.Choice(['major', 'minor', 'patch']),
help='The part of the version to increase.')
def bumpversion(verbose, dry_run, part):
- bump_version(part, verbose, dry_run)
\ No newline at end of file
+ bump_version(part, verbose, dry_run)
+
+POI_LABELS = {
+ 'acquisitor': 'ac',
+ 'tpreparator': 'tp',
+ 'trainer': 't',
+ 'evaluator': 'e'
+}
+
+def _sleep(sec):
+ logger.info("Sleeping for {0} seconds...".format(sec))
+ time.sleep(5)
+
+@cli.command("benchmark", help="Collect engine benchmark stats.")
+@click.option('--host', '-gh', help='gRPC Host Address', default='localhost')
+@click.option('--port', '-gp', help='gRPC Port', default='50057')
+@click.option('--profiling', '-p', default=False, is_flag=True, help='Deterministic profiling of user code.')
+@click.option('--delay', '-d', default=False, is_flag=True, help='Delay the benchmark for 5 seconds.')
+@click.pass_context
+def btest(ctx, host, port, profiling, delay):
+ timestamp = generate_timestamp()
+ b_thread = benchmark_thread(ctx.obj['package_name'], timestamp)
+ actions = ['acquisitor', 'tpreparator', 'trainer',
+ 'evaluator']
+ rc = RemoteCalls(host, port)
+ initial_time = time.time()
+ b_thread.start()
+ if delay:
+ _sleep(5)
+ for action in actions:
+ logger.info("Executing {0} action...".format(action))
+ create_poi('{0}-i'.format(POI_LABELS[action]),
+ time.time() - initial_time,
+ timestamp)
+ rc.run_dryrun(action, profiling)
+ create_poi('{0}-e'.format(POI_LABELS[action]),
+ time.time() - initial_time,
+ timestamp)
+ if delay:
+ _sleep(5)
+ b_thread.terminate()
+ sys.exit(0)
+
+PLOTS = {
+ 'cpu': 'CPU Usage',
+ 'memory': 'Memory Usage',
+ 'r_disk': 'Disk read',
+ 'w_disk': 'Disk write',
+ 'r_net': 'Network received',
+ 't_net': 'Network transfered'
+}
+
+@cli.command("benchmark-plot", help="Plot engine benchmark stats.")
+@click.option('--protocol', '-p', prompt='Protocol',
+ help='Unique protocol from poi and benchmark files.')
+def plot(protocol):
+ options = (
+ 'cpu',
+ 'memory',
+ 'r_disk',
+ 'w_disk',
+ 'r_net',
+ 't_net'
+ )
+ try:
+ while(True):
+ print('1 - CPU Usage')
+ print('2 - Memory Usage')
+ print('3 - Disk read')
+ print('3 - Disk write')
+ print('4 - Network received')
+ print('5 - Network transfered')
+ print('Press Ctrl-c to end.')
+ option = int(input('Option:'))
+ if option < 1 or option > 5:
+ logger.error('Option not available.')
+ sys.exit(1)
+ op_name = options[option - 1]
+ label = PLOTS[op_name]
+ make_graph(op_name, label, protocol)
+ except KeyboardInterrupt:
+ sys.exit(0)
\ No newline at end of file
diff --git a/marvin-cli/marvin_cli/utils/benchmark.py b/marvin-cli/marvin_cli/utils/benchmark.py
new file mode 100644
index 0000000..cff69d3
--- /dev/null
+++ b/marvin-cli/marvin_cli/utils/benchmark.py
@@ -0,0 +1,196 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+# Copyright [2020] [Apache Software Foundation]
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import sys
+import csv
+import json
+import time
+import multiprocessing
+from json.decoder import JSONDecodeError
+import matplotlib.pyplot as plt
+from .docker import get_stats
+from .log import get_logger
+from .misc import generate_timestamp
+
+logger = get_logger("benchmark")
+
+#to-do
+def create_or_get_benchmark_folder():
+ _folder_path = os.path.join(os.getcwd(), 'benchmarks')
+ if not os.path.exists(_folder_path):
+ os.makedirs(_folder_path)
+ return _folder_path
+
+def create_or_return_poi(timestamp):
+ _folder_path = create_or_get_benchmark_folder()
+ _file_path = os.path.join(_folder_path, 'poi_{0}.json'.format(timestamp))
+ if not os.path.exists(_file_path):
+ with open(_file_path, 'w') as _:
+ pass
+ return _file_path
+
+def get_internal_keys(input_data, *args, default=None):
+ _data = input_data
+ for arg in args:
+ try:
+ _data = _data[arg]
+ except (KeyError, ValueError, TypeError, AttributeError):
+ logger.error("Failed getting {0} key...".format(arg))
+ return default
+ return _data
+
+def calculate_disk_io(data_input):
+ _stats = get_internal_keys(data_input, "blkio_stats", "io_service_bytes_recursive")
+ if not _stats:
+ return 0, 0
+ r = 0
+ w = 0
+ for stat in _stats:
+ if stat["op"] == "Read":
+ r += stat["value"]
+ elif stat["op"] == "Write":
+ w += stat["value"]
+ r = r / 1000 if (r > 0) else 0
+ w = w / 1000 if (w > 0) else 0
+ return r, w
+
+def calculate_network_bytes(data_input):
+ networks = get_internal_keys(data_input, "networks")
+ if not networks:
+ return 0, 0
+ r = 0
+ t = 0
+ for if_name, data in networks.items():
+ logger.debug("getting stats for interface %r", if_name)
+ r += data["rx_bytes"]
+ t += data["tx_bytes"]
+ r = r / 1000 if (r > 0) else 0
+ t = t / 1000 if (t > 0) else 0
+ return r, t
+
+def calculate_cpu_percent(input_data):
+ cpu_count = len(get_internal_keys(input_data, "cpu_stats", "cpu_usage", "percpu_usage"))
+ cpu_percent = 0.0
+ cpu_delta = float(get_internal_keys(input_data, "cpu_stats", "cpu_usage", "total_usage")) - \
+ float(get_internal_keys(input_data, "precpu_stats", "cpu_usage", "total_usage"))
+ system_delta = float(get_internal_keys(input_data, "cpu_stats", "system_cpu_usage")) - \
+ float(get_internal_keys(input_data, "precpu_stats", "system_cpu_usage"))
+ if system_delta > 0.0:
+ cpu_percent = cpu_delta / system_delta * 100.0 * cpu_count
+ return cpu_percent
+
+def filter_data(input_data, initial_time):
+ _data = input_data
+ _r_net, _t_net = calculate_network_bytes(_data)
+ _r_disk, _w_disk = calculate_disk_io(_data)
+ _time = time.time() - initial_time
+ return (_time,
+ calculate_cpu_percent(_data),
+ int(float(_data["memory_stats"]["usage"]) * (10 >> 6)),
+ int(_r_disk),
+ int(_w_disk),
+ int(_r_net),
+ int(_t_net))
+
+def get_and_persist_stats(engine_package, initial_time, timestamp):
+ _stats = get_stats(engine_package)
+ _colleted_stats = filter_data(_stats, initial_time)
+ _filename = 'benchmark_{0}.csv'.format(timestamp)
+ _path = os.path.join(create_or_get_benchmark_folder(), _filename)
+ with open(_path, 'a') as f:
+ writer = csv.writer(f)
+ writer.writerow(_colleted_stats)
+
+def repeat_stats_call(engine_package, timestamp, initial_time):
+ while True:
+ getattr(sys.modules[__name__],
+ 'get_and_persist_stats')(engine_package, initial_time, timestamp)
+
+def benchmark_thread(engine_package, timestamp, initial_time=time.time()):
+ return multiprocessing.Process(target=repeat_stats_call,
+ args=(engine_package, timestamp, initial_time,))
+
+def create_poi(key, value, timestamp):
+ _file_path = create_or_return_poi(timestamp)
+ pois = None
+ with open(_file_path, 'r') as f:
+ try:
+ pois = json.load(f)
+ except JSONDecodeError:
+ pois = {}
+ pois[key] = value
+ with open(_file_path, 'w') as f:
+ json.dump(pois, f)
+
+def read_poi(timestamp):
+ _file_path = create_or_return_poi(timestamp)
+ pois = None
+ with open(_file_path, 'r') as f:
+ try:
+ pois = json.load(f)
+ except:
+ pois = {}
+
+ return pois
+
+def read_csv(timestamp):
+ _file_name = 'benchmark_{0}.csv'.format(timestamp)
+ _file_path = os.path.join(create_or_get_benchmark_folder(), _file_name)
+ print(_file_path)
+ _data_dict = {
+ 'time': [],
+ 'cpu': [],
+ 'memory': [],
+ 'r_disk': [],
+ 'w_disk': [],
+ 'r_net': [],
+ 't_net': []
+ }
+ if os.path.exists(_file_path):
+ with open(_file_path, 'r') as f:
+ try:
+ csv_buffer = csv.reader(f)
+ for row in csv_buffer:
+ _data_dict['time'].append(float(row[0]))
+ _data_dict['cpu'].append(float(row[1]))
+ _data_dict['memory'].append(float(row[2]))
+ _data_dict['r_disk'].append(float(row[3]))
+ _data_dict['w_disk'].append(float(row[4]))
+ _data_dict['r_net'].append(float(row[5]))
+ _data_dict['t_net'].append(float(row[6]))
+ except:
+ logger.error('Something went wrong when writing csv file.')
+ return _data_dict
+
+def make_graph(name, label, timestamp):
+ info_dict = read_csv(timestamp)
+ time_dict = read_poi(timestamp)
+
+ plt.plot(info_dict['time'], info_dict[name], color = "r")
+
+ for key, value in time_dict.items():
+ x_line_annotation = value
+ x_text_annotation = value
+ plt.axvline(x=x_line_annotation, linestyle='dashed', alpha=0.5, color='black')
+ t = plt.text(x=x_text_annotation, y=max(info_dict[name])/2, s=key, alpha=0.5, color='black')
+ t.set_bbox(dict(facecolor='white', alpha=0.5, edgecolor='white'))
+
+ plt.xlabel('Time (s)')
+ plt.ylabel(label)
+
+ plt.show()
\ No newline at end of file
diff --git a/marvin-cli/tests/utils/test_benchmark.py b/marvin-cli/tests/utils/test_benchmark.py
new file mode 100644
index 0000000..2162779
--- /dev/null
+++ b/marvin-cli/tests/utils/test_benchmark.py
@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+# Copyright [2020] [Apache Software Foundation]
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import os
+
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
+
+from marvin_cli.utils.benchmark import create_or_get_benchmark_folder
+from marvin_cli.utils.benchmark import calculate_cpu_percent, calculate_disk_io, calculate_network_bytes
+from marvin_cli.utils.benchmark import create_or_return_poi, create_poi
+from marvin_cli.utils.benchmark import make_graph, read_csv, read_poi, filter_data
+
+mocked_data = {
+ "memory_stats": {
+ "usage": 0
+ },
+ "cpu_stats": {
+ "cpu_usage": {
+ "total_usage": 0,
+ "percpu_usage": []
+ },
+ "system_cpu_usage": 0,
+ },
+ "precpu_stats": {
+ "cpu_usage": {
+ "total_usage": 0,
+ "percpu_usage": []
+ },
+ "system_cpu_usage": 0
+ },
+ "networks": {
+ "interface": {
+ "rx_bytes": 0,
+ "tx_bytes": 0
+ }
+ },
+ "blkio_stats": {
+ "io_service_bytes_recursive": {
+ "stat": {
+ "op": {
+ "Read": {
+ "value": 0
+ },
+ "Write": {
+ "value": 0
+ }
+ }
+ }
+ }
+ }
+}
+
+def test_create_or_get_benchmark_folder():
+ path = create_or_get_benchmark_folder()
+ assert path == os.path.join(os.getcwd(), 'benchmarks')
+
+def test_create_or_return_poi():
+ timestamp = 'mocked_timestamp'
+ path = create_or_return_poi(timestamp)
+ assert path == os.path.join(create_or_get_benchmark_folder(),
+ 'poi_{0}.json'.format(timestamp))
+
+@mock.patch("marvin_cli.utils.benchmark.os.path.exists")
+@mock.patch("marvin_cli.utils.benchmark.open")
+def test_read_csv(open_mocked, exists_mocked):
+ path = os.path.join(create_or_get_benchmark_folder(), 'benchmark_mocked.csv')
+ read_csv('mocked')
+ exists_mocked.assert_called_with(path)
+ open_mocked.assert_called_with(path, 'r')
+
+@mock.patch("marvin_cli.utils.benchmark.open")
+def test_read_poi(open_mocked):
+ timestamp = 'mocked'
+ path = create_or_return_poi(timestamp)
+ read_poi('mocked')
+ open_mocked.assert_called_with(path, 'r')
+
+@mock.patch("marvin_cli.utils.benchmark.read_csv")
+@mock.patch("marvin_cli.utils.benchmark.read_poi")
+@mock.patch("marvin_cli.utils.benchmark.plt.show")
+def test_make_graph(show_mocked, poi_mocked, csv_mocked):
+ timestamp = 'mocked'
+ make_graph('mocked_name', 'mocked_label', timestamp)
+ poi_mocked.assert_called_with(timestamp)
+ csv_mocked.assert_called_with(timestamp)
+ show_mocked.assert_called_once()
+
+@mock.patch("marvin_cli.utils.benchmark.open")
+@mock.patch("marvin_cli.utils.benchmark.json.load")
+@mock.patch("marvin_cli.utils.benchmark.json.dump")
+def test_create_poi(dump_mocked, load_mocked, open_mocked):
+ timestamp = 'mocked'
+ create_poi('mock', 'ed', timestamp)
+ load_mocked.assert_called_once()
+ dump_mocked.assert_called_once()
+ open_mocked.assert_called()
+
+
+@mock.patch("marvin_cli.utils.benchmark.calculate_network_bytes")
+@mock.patch("marvin_cli.utils.benchmark.calculate_disk_io")
+def test_filter_data(disk_io_mocked, net_bytes_mocked):
+ disk_io_mocked.return_value = (0,0)
+ net_bytes_mocked.return_value = (0,0)
+ filter_data(mocked_data, 0)
+ disk_io_mocked.assert_called_once()
+ net_bytes_mocked.assert_called_once()
+
+@mock.patch("marvin_cli.utils.benchmark.get_internal_keys")
+def test_calculate_disk_io(int_keys_mocked):
+ calculate_disk_io(mocked_data)
+ int_keys_mocked.assert_called_with(mocked_data, "blkio_stats", "io_service_bytes_recursive")
+
+@mock.patch("marvin_cli.utils.benchmark.get_internal_keys")
+def test_calculate_network_bytes(int_keys_mocked):
+ calculate_network_bytes(mocked_data)
+ int_keys_mocked.assert_called_with(mocked_data, "networks")
+
+@mock.patch("marvin_cli.utils.benchmark.get_internal_keys")
+def test_calculate_cpu_percent(int_keys_mocked):
+ calculate_cpu_percent(mocked_data)
+ int_keys_mocked.assert_called()