test-scripts/stress_test.py

   1 import unittest
   2 import os
   3 import re
   4 import sys
   5 import logging
   6 import time
   7 import argparse
   8 import requests
   9
  10 import xml.dom.minidom as md
  11 from xml.etree import ElementTree as ET
  12
  13 from openvswitch.mininet_tools import MininetTools
  14 from openvswitch.flow_tools import  FlowAdderThread, FlowRemoverThread, loglevels, TO_GET, WAIT_TIME, OPERATIONAL_DELAY, FLOWS_PER_SECOND
  15 from openvswitch.testclass_templates import TestClassAdd, TestClassRemove
  16 from openvswitch.testclass_components import CheckSwitchDump, CheckOperFlowsComponent, CheckConfigFlowsComponent
  17
  18 REMOVE_FLOWS_PER_THREAD = 250
  19
  20 class MultiTest(unittest.TestCase, TestClassAdd, TestClassRemove, CheckSwitchDump, CheckOperFlowsComponent, CheckConfigFlowsComponent):
  21
  22     log = logging.getLogger('MultiTest')
  23     total_errors = 0
  24
  25     def setUp(self):
  26         MultiTest.log.info('test setup...')
  27         self.threads_count = 50
  28         self.thread_pool = list()
  29
  30         self.active_map = dict()
  31
  32         self.__start_MN()
  33         self.__setup_threads()
  34         self.__run_threads()
  35
  36
  37     def tearDown(self):
  38         MultiTest.log.info('test cleanup...')
  39         MultiTest.total_deleted = 0
  40         self.__delete_flows()
  41
  42     def inc_error(self, value=1):
  43         MultiTest.total_errors += value
  44
  45     def inc_flow(self, flow_id= None, cookie_id=None):
  46         if flow_id is not None and cookie_id is not None:
  47             self.active_map[cookie_id] = flow_id
  48
  49     def delete_flow_from_map(self, flow_id, cookie_id):
  50         del self.active_map[cookie_id]
  51
  52     def __start_MN(self):
  53         self.net = MininetTools.create_network(self.host, self.mn_port)
  54         self.net.start()
  55         MultiTest.log.info('mininet stared')
  56         MultiTest.log.info('waiting {0} seconds...'.format(WAIT_TIME))
  57         time.sleep(WAIT_TIME)
  58
  59
  60     def __setup_threads(self):
  61         if args.threads is not None:
  62             self.threads_count = int(args.threads)
  63
  64         for i in range(0, self.threads_count):
  65             t = FlowAdderThread(self, i, self.host, self.port, self.net,\
  66                                 flows_ids_from=i*MultiTest.flows + 1, flows_ids_to=(i+1)*MultiTest.flows + 1)
  67
  68             self.thread_pool.append(t)
  69
  70     def __run_threads(self):
  71         for t in self.thread_pool:
  72             t.start()
  73
  74         for t in self.thread_pool:
  75             t.join()
  76
  77     def test(self):
  78         cookie_regexp = re.compile("cookie=0x[0-9,a-f,A-F]+")
  79         switch_flows = 0
  80         oper_flows = 0
  81         flows_on_controller_operational = None
  82         flows_on_controller = None
  83
  84         total_flows = len(self.active_map.keys())
  85
  86         assert total_flows > 0, ('Stored flows should be greater than 0, actual is {0}'.format(total_flows))
  87
  88         MultiTest.log.info('\n\n---------- preparation finished, running test ----------')
  89
  90         # check config
  91         flows_on_controller = self.check_config_flows(self.host, self.port, self.active_map)
  92
  93         #check operational
  94         flows_on_controller_operational = self.check_oper_flows_loop(self.host, self.port, self.active_map)
  95
  96         # check switch
  97         switch_flows_list = MininetTools.get_flows_string(self.net)
  98         MultiTest.log.info('flow dump has {0} entries (including informational)'.format(len(switch_flows_list)))
  99         for item in switch_flows_list:
 100             if self.get_id_by_entry(item, self.active_map) is not None:
 101                 MultiTest.log.debug('flow_id:{0} = {1}'.format(self.get_id_by_entry(item, self.active_map), item))
 102                 switch_flows += 1
 103
 104         # print info
 105         MultiTest.log.info('{0} flows are stored by results from threads, {1} errors'.format(total_flows, MultiTest.total_errors))
 106         MultiTest.log.info('{0} flows are stored in controller config'.format(flows_on_controller))
 107         MultiTest.log.info('{0} flows are stored in controller operational'.format(flows_on_controller_operational))
 108         MultiTest.log.info('{0} flows are stored on switch'.format(switch_flows))
 109
 110         assert total_flows == switch_flows, 'Added amount of flows to switch should be equal to successfully added flows to controller {0} <> {1}'.format(switch_flows,total_flows)
 111
 112     def __delete_flows(self):
 113         flows_deleted = 0
 114         flows_on_controller = 0
 115         MultiTest.log.info('deleting flows added during test')
 116
 117         # using threads to delete to speed up cleanup
 118         items_to_delete = list(self.active_map.items())
 119         self.thread_pool = []
 120         thread_id = 0
 121         slice_from = REMOVE_FLOWS_PER_THREAD * thread_id
 122         slice_to = REMOVE_FLOWS_PER_THREAD * (thread_id + 1)
 123
 124         total_flows = len(self.active_map.keys())
 125         total_deleted = 0
 126
 127         while(slice_from < len(items_to_delete)):
 128             self.thread_pool.append(FlowRemoverThread(self, thread_id, self.host, self.port, self.net, items_to_delete[slice_from:slice_to]))
 129             thread_id += 1
 130             slice_from = REMOVE_FLOWS_PER_THREAD * thread_id
 131             slice_to = REMOVE_FLOWS_PER_THREAD * (thread_id + 1)
 132
 133         for t in self.thread_pool:
 134             t.start()
 135
 136         for t in self.thread_pool:
 137             t.join()
 138
 139         for t in self.thread_pool:
 140             total_deleted += t.removed
 141
 142         MultiTest.log.info('deleted {0} flows'.format(total_deleted))
 143         if total_flows <> total_deleted:
 144             raise StandardError('Not all flows have been deleted, flows added'\
 145                 ' during test: {0} <> deleted flows: {1},\nflows ids left on controller: {2}'.format(\
 146                 total_flows, total_deleted, sorted(self.active_map.values())))
 147
 148 if __name__ == '__main__':
 149
 150     requests_log = logging.getLogger("requests")
 151     requests_log.setLevel(logging.WARNING)
 152
 153     # parse cmdline arguments
 154     parser = argparse.ArgumentParser(description='End to end stress tests of flows '
 155                         'addition from multiple connections')
 156     parser.add_argument('--odlhost', default='127.0.0.1', help='host where '
 157                         'odl controller is running  (default = 127.0.0.1) ')
 158     parser.add_argument('--odlport', type=int, default=8080, help='port on '
 159                         'which odl\'s RESTCONF is listening  (default = 8080) ')
 160     parser.add_argument('--mnport', type=int, default=6653, help='port on '
 161                         'which odl\'s controller is listening  (default = 6653)')
 162     parser.add_argument('--threads', default=50, help='how many threads '
 163                         'should be used  (default = 50)')
 164     parser.add_argument('--flows', default=20, help='how many flows will add'
 165                         ' one thread  (default = 20)')
 166     parser.add_argument('--log', default='info', help='log level, permitted values are'
 167                         ' debug/info/warning/error  (default = info)')
 168     args = parser.parse_args()
 169
 170     #logging.basicConfig(level=logging.DEBUG)
 171     logging.basicConfig(level=loglevels.get(args.log, logging.INFO))
 172
 173     # set host and port of ODL controller for test cases
 174     MultiTest.port = args.odlport
 175     MultiTest.host = args.odlhost
 176     MultiTest.mn_port = args.mnport
 177     MultiTest.threads = int(args.threads)
 178     MultiTest.flows = int(args.flows)
 179
 180     del sys.argv[1:]
 181
 182     suite = unittest.TestSuite()
 183     test = MultiTest('test')
 184     suite.addTest(test)
 185
 186     try:
 187         unittest.TextTestRunner(verbosity=2).run(suite)
 188         #unittest.main()
 189     finally:
 190         test.net.stop()
 191         print 'end'