Modify simulator.py tool

Introduce mechanism of generating real life messages to the tool
using the information gathered during Rally testing. This change
allows to generate messages of the specfic length due to the
distribution observed on real environment.

messages_length.txt file contains lengths of string JSON objects
that were later sent through the MQ layer during deployment and
deletion of 50 VMs.

simulator.py was modified to use this data as a baseline to generate
random string messages of the required length with the needed
probability.

Change-Id: Iae21f90b5ca202bf0e83f1149baef8b42c64eb55

tools/simulator.py

@@ -13,14 +13,19 @@
 import eventlet
 eventlet.monkey_patch()
 
-import os
-
 import argparse
+import collections
 import datetime
 import logging
+import os
+import random
+import string
 import sys
 import threading
 import time
+import yaml
+
+from scipy.stats import rv_discrete
 
 from oslo_config import cfg
 import oslo_messaging as messaging
@@ -28,6 +33,8 @@ from oslo_messaging import notify  # noqa
 from oslo_messaging import rpc  # noqa
 
 LOG = logging.getLogger()
+RANDOM_VARIABLE = None
+CURRENT_PID = None
 
 USAGE = """ Usage: ./simulator.py [-h] [--url URL] [-d DEBUG]\
  {notify-server,notify-client,rpc-server,rpc-client} ...
@@ -40,6 +47,29 @@ Usage example:
  --exit-wait 15000 -p 64 -m 64"""
 
 
+def init_random_generator():
+    data = []
+    with open('./messages_length.yaml') as m_file:
+        content = yaml.load(m_file)
+        data += [int(n) for n in content[
+            'test_data']['string_lengths'].split(', ')]
+
+    ranges = collections.defaultdict(int)
+    for msg_length in data:
+        range_start = (msg_length / 500) * 500 + 1
+        ranges[range_start] += 1
+
+    ranges_start = sorted(ranges.keys())
+    total_count = len(data)
+    ranges_dist = []
+    for r in ranges_start:
+        r_dist = float(ranges[r]) / total_count
+        ranges_dist.append(r_dist)
+
+    random_var = rv_discrete(values=(ranges_start, ranges_dist))
+    return random_var
+
+
 class LoggingNoParsingFilter(logging.Filter):
     def filter(self, record):
         msg = record.getMessage()
@@ -136,9 +166,19 @@ def send_msg(_id, transport, target, messages, wait_after_msg, timeout,
     client = client.prepare(timeout=timeout)
     rpc_method = _rpc_cast if is_cast else _rpc_call
 
-    for i in range(0, messages):
-        msg = "test message %d" % i
-        LOG.info("SEND: %s" % msg)
+    ranges = RANDOM_VARIABLE.rvs(size=messages)
+    i = 0
+    for range_start in ranges:
+        length = random.randint(range_start, range_start + 497)
+        msg = ''.join(random.choice(string.lowercase) for x in range(length)) \
+              + ' ' + str(i)
+        i += 1
+        # temporary file to log approximate bytes size of messages
+        with open('./oslo_%s_%s.log' % (target.topic, CURRENT_PID), 'a+') as f:
+            # 37 additional bytes for Python String object size canculation.
+            # In fact we may ignore these bytes, and estimate the data flow
+            # via number of symbols
+            f.write(str(length + 37) + '\n')
         rpc_method(client, msg)
         if wait_after_msg > 0:
             time.sleep(wait_after_msg)
@@ -197,6 +237,9 @@ def main():
     parser.add_argument('-d', '--debug', dest='debug', type=bool,
                         default=False,
                         help="Turn on DEBUG logging level instead of WARN")
+    parser.add_argument('-tp', '--topic', dest='topic',
+                        default="profiler_topic",
+                        help="Topic to publish/receive messages to/from.")
     subparsers = parser.add_subparsers(dest='mode',
                                        help='notify/rpc server/client mode')
 
@@ -246,7 +289,7 @@ def main():
     cfg.CONF.project = 'oslo.messaging'
 
     transport = messaging.get_transport(cfg.CONF, url=args.url)
-    target = messaging.Target(topic='profiler_topic', server='profiler_server')
+    target = messaging.Target(topic=args.topic, server='profiler_server')
 
     if args.mode == 'rpc-server':
         if args.url.startswith('zmq'):
@@ -266,11 +309,29 @@ def main():
                         args.is_cast)
         time_ellapsed = (datetime.datetime.now() - start).total_seconds()
         msg_count = args.messages * args.threads
-        print ('%d messages was sent for %s seconds. Bandwight is %s msg/sec'
-               % (msg_count, time_ellapsed, (msg_count / time_ellapsed)))
+        log_msg = '%d messages was sent for %s seconds. ' \
+                  'Bandwidth is %s msg/sec' % (msg_count, time_ellapsed,
+                                               (msg_count / time_ellapsed))
+        print (log_msg)
+        with open('./oslo_res_%s.txt' % args.topic, 'a+') as f:
+            f.write(log_msg + '\n')
+
+        with open('./oslo_%s_%s.log' % (args.topic, CURRENT_PID), 'a+') as f:
+            data = f.read()
+        data = [int(i) for i in data.split()]
+        data_sum = sum(data)
+        log_msg = '%s bytes were sent for %s seconds. Bandwidth is %s b/s' % (
+            data_sum, time_ellapsed, (data_sum / time_ellapsed))
+        print(log_msg)
+        with open('./oslo_res_%s.txt' % args.topic, 'a+') as f:
+            f.write(log_msg + '\n')
+        os.remove('./oslo_%s_%s.log' % (args.topic, CURRENT_PID))
+
         LOG.info("calls finished, wait %d seconds" % args.exit_wait)
         time.sleep(args.exit_wait)
 
 
 if __name__ == '__main__':
+    RANDOM_VARIABLE = init_random_generator()
+    CURRENT_PID = os.getpid()
     main()