csit/libraries/MdsalLowlevelPy.py

   1 """
   2 Python invocation of several parallel publish-notifications RPCs.
   3 """
   4 from robot.api import logger
   5 import time
   6 import Queue
   7 import requests
   8 import string
   9 import threading
  10
  11
  12 _globals = {}
  13
  14
  15 def _send_http_request_thread_impl(rqueue, prefix_id, url, data, http_timeout):
  16     """Start either publish or write transactions rpc based on input.
  17
  18     :param rqueue: result queue
  19     :type rqueue: Queue.Queue
  20     :param prefix_id: identifier for prefix, should imply cluster member index
  21     :type prefix_id: str
  22     :param url: rpc url
  23     :type url: str
  24     :param data: http request content
  25     :type data: str
  26     :param http_timeout: http response timeout
  27     :type http_timeout: int
  28     """
  29     logger.info('rpc invoked with details: {}'.format(data))
  30     try:
  31         resp = requests.post(url=url, headers={'Content-Type': 'application/xml'},
  32                              data=data, auth=('admin', 'admin'), timeout=http_timeout)
  33     except Exception as exc:
  34         resp = exc
  35         logger.debug(exc)
  36     rqueue.put((time.ctime(), prefix_id, resp))
  37
  38
  39 def _initiate_rpcs(host_list, index_list, url_templ, data_templ, subst_dict):
  40     """Initiate rpc on given hosts.
  41
  42     :param host_list: IP addresses of odl nodes
  43     :type host_list: list[str]
  44     :param index_list: node indices which correspond to the ip addresses
  45     :type index_list: list[int]
  46     :param url_templ: url template
  47     :type url_templ: string.Template object
  48     :param data_templ: http request data
  49     :type data_templ: string.Template object
  50     :param subst_dict: dictionary with key value pairs to be used with template
  51     :type subst_dict: dict
  52     """
  53     resqueue = _globals.pop('result_queue', Queue.Queue())
  54     lthreads = _globals.pop('threads', [])
  55     for i, host in enumerate(host_list):
  56         url = url_templ.substitute({'HOST': host})
  57         timeout = int(subst_dict['DURATION']) + 3 * 125 + 10
  58         prefix_id = subst_dict['ID_PREFIX'] + str(index_list[i])
  59         subst_dict['ID'] = prefix_id
  60         data = data_templ.substitute(subst_dict)
  61         logger.info('url: {}, data: {}, timeout: {}'.format(url, data, timeout))
  62         t = threading.Thread(target=_send_http_request_thread_impl,
  63                              args=(resqueue, prefix_id, url, data, timeout))
  64         t.daemon = True
  65         t.start()
  66         lthreads.append(t)
  67
  68     _globals.update({'threads': lthreads, 'result_queue': resqueue})
  69
  70
  71 def start_write_transactions_on_nodes(host_list, index_list, id_prefix, duration, rate, chained_flag=False,
  72                                       reset_globals=True):
  73     """Invoke write-transactions rpc on given nodes.
  74
  75     :param host_list: IP addresses of odl nodes
  76     :type host_list: list[str]
  77     :param index_list: node indices which correspond to the ip addresses
  78     :type index_list: list[int]
  79     :param id_prefix: identifier prefix
  80     :type id_prefix: str
  81     :param duration: time in seconds
  82     :type duration: int
  83     :param rate: writing transactions rate in transactions per second
  84     :type rate: int
  85     :param chained_flag: specify chained vs. simple transactions
  86     :type chained_flag: bool
  87     :param reset_globals: reset global variable dict
  88     :type reset_globals: bool
  89     """
  90     if reset_globals:
  91         _globals.clear()
  92
  93     logger.info(
  94         "Input parameters: host_list:{}, index_list:{}, id_prefix:{}, duration:{}, rate:{}, chained_flag:{}".format(
  95             host_list, index_list, id_prefix, duration, rate, chained_flag))
  96     datat = string.Template('''<input xmlns="tag:opendaylight.org,2017:controller:yang:lowlevel:control">
  97   <id>$ID</id>
  98   <seconds>$DURATION</seconds>
  99   <transactions-per-second>$RATE</transactions-per-second>
 100   <chained-transactions>$CHAINED_FLAG</chained-transactions>
 101 </input>''')
 102     subst_dict = {'ID_PREFIX': id_prefix, 'DURATION': duration,
 103                   'RATE': rate, 'CHAINED_FLAG': 'true' if chained_flag else 'false'}
 104     urlt = string.Template('''http://$HOST:8181/restconf/operations/odl-mdsal-lowlevel-control:write-transactions''')
 105     _initiate_rpcs(host_list, index_list, urlt, datat, subst_dict)
 106
 107
 108 def start_produce_transactions_on_nodes(host_list, index_list, id_prefix,
 109                                         duration, rate, isolated_transactions_flag=False, reset_globals=True):
 110     """Invoke produce-transactions rpcs on given nodes.
 111
 112     :param host_list: IP addresses of odl nodes
 113     :type host_list: list[str]
 114     :param index_list: node indices which correspond to the ip addresses
 115     :type index_list: list[int]
 116     :param id_prefix: identifier prefix
 117     :type id_prefix: str
 118     :param duration: time in seconds
 119     :type duration: int
 120     :param rate: produce transactions rate in transactions per second
 121     :type rate: int
 122     :param isolated_transactions_flag: isolated transactions flag
 123     :type isolated_transactions_flag: bool
 124     :param reset_globals: reset global variable dict
 125     :type reset_globals: bool
 126     """
 127     if reset_globals:
 128         _globals.clear()
 129
 130     msg = "host_list:{}, index_list:{} ,id_prefix:{}, duration:{}, rate:{}, isolated_transactions:{}".format(
 131         host_list, index_list, id_prefix, duration, rate, isolated_transactions_flag)
 132     msg = "Input parameters: " + msg
 133     logger.info(msg)
 134     datat = string.Template('''<input xmlns="tag:opendaylight.org,2017:controller:yang:lowlevel:control">
 135   <id>$ID</id>
 136   <seconds>$DURATION</seconds>
 137   <transactions-per-second>$RATE</transactions-per-second>
 138   <isolated-transactions>$ISOLATED_TRANSACTIONS</isolated-transactions>
 139 </input>''')
 140     subst_dict = {'ID_PREFIX': id_prefix, 'DURATION': duration, 'RATE': rate,
 141                   'ISOLATED_TRANSACTIONS': 'true' if isolated_transactions_flag else 'false'}
 142     urlt = string.Template('''http://$HOST:8181/restconf/operations/odl-mdsal-lowlevel-control:produce-transactions''')
 143     _initiate_rpcs(host_list, index_list, urlt, datat, subst_dict)
 144
 145
 146 def wait_for_transactions():
 147     """Blocking call, waitig for responses from all threads.
 148
 149     :return: list of triples; triple consists of response time, prefix identifier and response object
 150     :rtype: list[(str, str, requests.Response)]
 151     """
 152     lthreads = _globals.pop('threads')
 153     resqueue = _globals.pop('result_queue')
 154
 155     for t in lthreads:
 156         t.join()
 157
 158     results = []
 159     while not resqueue.empty():
 160         results.append(resqueue.get())
 161     for rsp in results:
 162         if isinstance(rsp[2], requests.Response):
 163             logger.info(rsp[2].text)
 164         else:
 165             logger.info(rsp[2])
 166     return results
 167
 168
 169 def get_next_transactions_response():
 170     """Get http response from write-transactions rpc if available.
 171
 172     :return: None or a triple consisting of response time, prefix identifier and response object
 173     :rtype: (str, str, requests.Response)
 174     """
 175     resqueue = _globals.get('result_queue')
 176
 177     if not resqueue.empty():
 178         rsp = resqueue.get()
 179         if isinstance(rsp[2], requests.Response):
 180             logger.info(rsp[2].text)
 181         else:
 182             logger.info(rsp[2])
 183         return rsp
 184     return None