test-scripts/openvswitch/flow_tools.py

   1 import logging
   2 import requests
   3 import threading
   4
   5 from openvswitch.parser_tools import ParseTools
   6 import xml.dom.minidom as md
   7
   8
   9 TO_GET = 30
  10 TO_PUT = 10
  11 TO_DEL = 30
  12 WAIT_TIME = 15
  13 OPERATIONAL_DELAY = 11
  14 FLOWS_PER_SECOND = 5
  15
  16 FLOW_ID_TEMPLATE = 'FLOW_ID_TEMPLATE'
  17 COOKIE_TEMPLATE = 'COOKIE_TEMPLATE'
  18 HARD_TO_TEMPLATE = 'HARD_TO_TEMPLATE'
  19 FLOW_NAME_TEMPLATE = 'FLOW_NAME_TEMPLATE'
  20 IPV4DST_TEMPLATE = 'IPV4DST_TEMPLATE'
  21 PRIORITY_TEMPLATE = 'PRIORITY_TEMPLATE'
  22
  23 xml_template = '<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"no\"?>' \
  24 '<flow xmlns=\"urn:opendaylight:flow:inventory\">' \
  25     '<strict>false</strict>' \
  26     '<instructions><instruction><order>0</order><apply-actions><action><order>0</order><dec-nw-ttl/>' \
  27     '</action></apply-actions></instruction></instructions><table_id>2</table_id>' \
  28     '<id>'+FLOW_ID_TEMPLATE+'</id><cookie_mask>255</cookie_mask><installHw>false</installHw>' \
  29     '<match><ethernet-match><ethernet-type><type>2048</type></ethernet-type></ethernet-match>' \
  30     '<ipv4-destination>'+IPV4DST_TEMPLATE+'</ipv4-destination></match><hard-timeout>'+HARD_TO_TEMPLATE+'</hard-timeout>' \
  31     '<flags>FlowModFlags [_cHECKOVERLAP=false, _rESETCOUNTS=false, _nOPKTCOUNTS=false, _nOBYTCOUNTS=false, _sENDFLOWREM=false]</flags>' \
  32     '<cookie>'+COOKIE_TEMPLATE+'</cookie><idle-timeout>34000</idle-timeout><flow-name>'+FLOW_NAME_TEMPLATE+'</flow-name><priority>'+PRIORITY_TEMPLATE+'</priority>' \
  33     '<barrier>false</barrier></flow>'
  34
  35 class MapNames():
  36     TEST = 'TEST'
  37     DUMMY = 'DUMMY'
  38
  39 loglevels = {
  40     'debug' : logging.DEBUG,
  41     'info' : logging.INFO,
  42     'warning' : logging.WARNING,
  43     'error' : logging.ERROR
  44 }
  45
  46
  47 class FlowAdderThread(threading.Thread):
  48     """
  49     Thread to remove flows from TestClassAdd
  50
  51     TestClassAdd should implement methods inc_flow(value, key) and inc_error()
  52     """
  53
  54     def __init__(self, test_class, thread_id, host, port, net, flows_ids_from=0, flows_ids_to=1):
  55         """
  56         test_class: should be type of TestClassAdd
  57         thread_id: id of thread
  58         host: controller's ip address
  59         port: controller's port
  60         net: mininet instance
  61         flows_ids_from: minimum id of flow to be added (including)
  62         flows_ids_to: maximum id of flow to be added (excluding)
  63         """
  64
  65         threading.Thread.__init__(self)
  66         self.thread_id = thread_id
  67         self.flows = 0
  68         self.errors = 0
  69         self.flows_ids_from = flows_ids_from
  70         self.flows_ids_to = flows_ids_to
  71
  72         self.test_class = test_class
  73         self.net = net
  74         self.host = host
  75         self.port = port
  76
  77         self.flows = 0
  78         self.errors = 0
  79
  80         self.log = logging.getLogger('FlowAdderThread: ' + str(thread_id))
  81         self.log.propagate = False
  82         self.channel = logging.StreamHandler()
  83         self.log.addHandler(self.channel)
  84         formatter = logging.Formatter('THREAD A{0}: %(levelname)s: %(message)s'.format(self.thread_id))
  85         self.channel.setFormatter(formatter)
  86
  87         self.log.info('created new FlowAdderThread->id:{0}, flows id: {1} -> {2}'.format(self.thread_id, self.flows_ids_from, self.flows_ids_to))
  88
  89     def make_cookie_marker(self, val_input, number=0):
  90         return '0x' + "{0:x}".format(int(''.join("{0:x}".format(ord(c)) for c in (val_input)), 16) + number)
  91
  92     def make_ipv4_address(self, number, octet_count=4, octet_size=255):
  93         mask = 24
  94         ip = ['10', '0', '0', '0']
  95
  96         if number < (255**3):
  97             for o in range(1, octet_count):
  98                 ip[octet_count - 1 - o] = str(number % octet_size)
  99                 number = number / octet_size
 100                 #mask -= 8
 101                 if number == 0:
 102                     break
 103
 104         return '.'.join(ip) + '/{0}'.format(mask)
 105
 106     def __add_flows(self, act_flow_id):
 107         cookie_id = None
 108
 109         try:
 110             self.log.debug('adding flow id: {0}'.format(act_flow_id))
 111
 112             cookie_id = self.make_cookie_marker('stress', act_flow_id)
 113
 114             xml_string = str(xml_template).replace(FLOW_ID_TEMPLATE, str(act_flow_id))\
 115             .replace(COOKIE_TEMPLATE, str(int(cookie_id, 16)))\
 116             .replace(HARD_TO_TEMPLATE, '1200').replace(FLOW_NAME_TEMPLATE,'FooXf{0}'.format(act_flow_id))\
 117             .replace(IPV4DST_TEMPLATE,self.make_ipv4_address(act_flow_id)).replace(PRIORITY_TEMPLATE,str(act_flow_id))
 118
 119             #TestRestartMininet.log.info('loaded xml: {0}'.format(''.join(xml_string.split())))
 120             tree = md.parseString(xml_string)
 121             ids = ParseTools.get_values(tree.documentElement, 'table_id', 'id')
 122             data = (self.host, self.port, ids['table_id'], ids['id'])
 123
 124             url = 'http://%s:%d/restconf/config/opendaylight-inventory:nodes' \
 125                   '/node/openflow:1/table/%s/flow/%s' % data
 126             # send request via RESTCONF
 127             headers = {
 128                 'Content-Type': 'application/xml',
 129                 'Accept': 'application/xml',
 130             }
 131             self.log.debug('sending request to url: {0}'.format(url))
 132             rsp = requests.put(url, auth=('admin', 'admin'), data=xml_string,
 133                                headers=headers, timeout=TO_PUT)
 134             self.log.debug('received status code: {0}'.format(rsp.status_code))
 135             self.log.debug('received content: {0}'.format(rsp.text))
 136             assert rsp.status_code == 204 or rsp.status_code == 200, 'Status' \
 137                             ' code returned %d' % rsp.status_code
 138
 139             # we expect that controller doesn't fail to store flow on switch
 140             self.test_class.inc_flow(flow_id=act_flow_id, cookie_id=cookie_id)
 141             self.flows += 1
 142         except Exception as e:
 143             self.log.error('Error storing flow id:{0}, cookie-id:{1}, reason: {2}'.format(act_flow_id, cookie_id, str(e)))
 144             self.test_class.inc_error()
 145             self.errors += 1
 146
 147     def run(self):
 148         self.flows, self.errors = 0, 0
 149         self.log.info('adding flows {0} to {1}'.format(self.flows_ids_from, self.flows_ids_to))
 150         for i in range(self.flows_ids_from, self.flows_ids_to + 1):
 151             self.__add_flows(i)
 152
 153         self.log.info('finished, added {0} flows, {1} errors'.format(self.flows,self.errors))
 154
 155
 156
 157 class FlowRemoverThread(threading.Thread):
 158     """
 159     Thread to remove flows from TestClassDelete
 160
 161     TestClassDelete should implement method delete_flows_from_map(value, key)
 162     """
 163
 164
 165     def __init__(self, test_class, thread_id, host, port, net, flows_to_delete=[]):
 166         """
 167         test_class: should be type of TestClassDelete
 168         thread_id: id of thread
 169         host: controller's ip address
 170         port: controller's port
 171         net: mininet instance
 172         flows_to_delete: dictionary of flows to delete with items to match method delete_flows_from_map(value, key)
 173         """
 174
 175         threading.Thread.__init__(self)
 176         self.thread_id = thread_id
 177         self.flows_to_delete = flows_to_delete
 178         self.test_class = test_class
 179
 180
 181         self.removed = 0
 182         self.errors = 0
 183
 184         self.net = net
 185         self.host = host
 186         self.port = port
 187
 188         self.log = logging.getLogger('FlowRemoverThread: ' + str(thread_id))
 189         self.log.propagate = False
 190         self.channel = logging.StreamHandler()
 191         self.log.addHandler(self.channel)
 192         formatter = logging.Formatter('THREAD R{0}: %(levelname)s: %(message)s'.format(self.thread_id))
 193         self.channel.setFormatter(formatter)
 194
 195         self.log.info('created new FlowRemoverThread->id:{0}'.format(self.thread_id))
 196
 197     def __remove_flows(self, act_flow_id, cookie_id):
 198         headers = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
 199         url = 'http://%s:%d/restconf/config/opendaylight-inventory:nodes' \
 200                 '/node/openflow:1/table/2/flow/%s' % (self.host, self.port, act_flow_id)
 201
 202         try:
 203             response = requests.delete(url, auth=('admin','admin'), headers=headers, timeout=TO_DEL)
 204             self.log.debug('deletion flow: {0} from controller: response: {1}'.format(act_flow_id, response.status_code))
 205
 206             assert response.status_code == 200 or response.status_code == 204, 'Delete response should be 200 or 204 is {0}'.format(response.status_code)
 207             self.test_class.delete_flow_from_map(act_flow_id, cookie_id)
 208             self.removed += 1
 209
 210         except Exception as e:
 211             self.log.error('Error deleting flow:{0}, reason: {1}'.format(act_flow_id, str(e)))
 212             self.errors += 1
 213         except requests.exceptions.Timeout as te:
 214             self.log.error('Error deleting flow: {0}, timeout reached: {1}'.format(act_flow_id, str(te)))
 215             self.errors += 1
 216
 217
 218     def run(self):
 219         self.log.debug('started removing flows')
 220         for flow_ids in set(self.flows_to_delete):
 221             self.__remove_flows(flow_ids[1], flow_ids[0])
 222
 223         self.log.info('finished removing {0} flows, {1} errors'.format(self.removed, self.errors))
 224