OpenStack Networking (Neutron)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

327 lines
13 KiB

  1. # Copyright (C) 2014,2015 VA Linux Systems Japan K.K.
  2. # Copyright (C) 2014,2015 YAMAMOTO Takashi <yamamoto at valinux co jp>
  3. # All Rights Reserved.
  4. #
  5. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  6. # not use this file except in compliance with the License. You may obtain
  7. # a copy of the License at
  8. #
  9. # http://www.apache.org/licenses/LICENSE-2.0
  10. #
  11. # Unless required by applicable law or agreed to in writing, software
  12. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  13. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  14. # License for the specific language governing permissions and limitations
  15. # under the License.
  16. import functools
  17. import random
  18. import eventlet
  19. import netaddr
  20. from neutron_lib import exceptions
  21. import os_ken.app.ofctl.api as ofctl_api
  22. import os_ken.exception as os_ken_exc
  23. from os_ken.lib import ofctl_string
  24. from os_ken.ofproto import ofproto_parser
  25. from oslo_config import cfg
  26. from oslo_log import log as logging
  27. from oslo_utils import excutils
  28. from oslo_utils import timeutils
  29. import six
  30. from neutron._i18n import _
  31. from neutron.agent.common import ovs_lib
  32. LOG = logging.getLogger(__name__)
  33. BUNDLE_ID_WIDTH = 1 << 32
  34. COOKIE_DEFAULT = object()
  35. class ActiveBundleRunning(exceptions.NeutronException):
  36. message = _("Another active bundle 0x%(bundle_id)x is running")
  37. class OpenFlowSwitchMixin(object):
  38. """Mixin to provide common convenient routines for an openflow switch.
  39. NOTE(yamamoto): super() points to ovs_lib.OVSBridge.
  40. See ovs_bridge.py how this class is actually used.
  41. """
  42. @staticmethod
  43. def _cidr_to_os_ken(ip):
  44. n = netaddr.IPNetwork(ip)
  45. if n.hostmask:
  46. return (str(n.ip), str(n.netmask))
  47. return str(n.ip)
  48. def __init__(self, *args, **kwargs):
  49. self._app = kwargs.pop('os_ken_app')
  50. self.active_bundles = set()
  51. super(OpenFlowSwitchMixin, self).__init__(*args, **kwargs)
  52. def _get_dp_by_dpid(self, dpid_int):
  53. """Get os-ken datapath object for the switch."""
  54. timeout_sec = cfg.CONF.OVS.of_connect_timeout
  55. start_time = timeutils.now()
  56. while True:
  57. dp = ofctl_api.get_datapath(self._app, dpid_int)
  58. if dp is not None:
  59. break
  60. # The switch has not established a connection to us; retry again
  61. # until timeout.
  62. if timeutils.now() > start_time + timeout_sec:
  63. m = _("Switch connection timeout")
  64. LOG.error(m)
  65. # NOTE(yamamoto): use RuntimeError for compat with ovs_lib
  66. raise RuntimeError(m)
  67. return dp
  68. def _send_msg(self, msg, reply_cls=None, reply_multi=False,
  69. active_bundle=None):
  70. timeout_sec = cfg.CONF.OVS.of_request_timeout
  71. timeout = eventlet.Timeout(seconds=timeout_sec)
  72. if active_bundle is not None:
  73. (dp, ofp, ofpp) = self._get_dp()
  74. msg = ofpp.ONFBundleAddMsg(dp, active_bundle['id'],
  75. active_bundle['bundle_flags'], msg, [])
  76. try:
  77. result = ofctl_api.send_msg(self._app, msg, reply_cls, reply_multi)
  78. except os_ken_exc.OSKenException as e:
  79. m = _("ofctl request %(request)s error %(error)s") % {
  80. "request": msg,
  81. "error": e,
  82. }
  83. LOG.error(m)
  84. # NOTE(yamamoto): use RuntimeError for compat with ovs_lib
  85. raise RuntimeError(m)
  86. except eventlet.Timeout as e:
  87. with excutils.save_and_reraise_exception() as ctx:
  88. if e is timeout:
  89. ctx.reraise = False
  90. m = _("ofctl request %(request)s timed out") % {
  91. "request": msg,
  92. }
  93. LOG.error(m)
  94. # NOTE(yamamoto): use RuntimeError for compat with ovs_lib
  95. raise RuntimeError(m)
  96. finally:
  97. timeout.cancel()
  98. LOG.debug("ofctl request %(request)s result %(result)s",
  99. {"request": msg, "result": result})
  100. return result
  101. @staticmethod
  102. def _match(_ofp, ofpp, match, **match_kwargs):
  103. if match is not None:
  104. return match
  105. return ofpp.OFPMatch(**match_kwargs)
  106. def uninstall_flows(self, table_id=None, strict=False, priority=0,
  107. cookie=COOKIE_DEFAULT, cookie_mask=0,
  108. match=None, active_bundle=None, **match_kwargs):
  109. (dp, ofp, ofpp) = self._get_dp()
  110. if table_id is None:
  111. table_id = ofp.OFPTT_ALL
  112. if cookie == ovs_lib.COOKIE_ANY:
  113. cookie = 0
  114. if cookie_mask != 0:
  115. raise Exception(_("cookie=COOKIE_ANY but cookie_mask set to "
  116. "%s") %
  117. cookie_mask)
  118. elif cookie == COOKIE_DEFAULT:
  119. cookie = self._default_cookie
  120. cookie_mask = ovs_lib.UINT64_BITMASK
  121. match = self._match(ofp, ofpp, match, **match_kwargs)
  122. if strict:
  123. cmd = ofp.OFPFC_DELETE_STRICT
  124. else:
  125. cmd = ofp.OFPFC_DELETE
  126. msg = ofpp.OFPFlowMod(dp,
  127. command=cmd,
  128. cookie=cookie,
  129. cookie_mask=cookie_mask,
  130. table_id=table_id,
  131. match=match,
  132. priority=priority,
  133. out_group=ofp.OFPG_ANY,
  134. out_port=ofp.OFPP_ANY)
  135. self._send_msg(msg, active_bundle=active_bundle)
  136. def dump_flows(self, table_id=None):
  137. (dp, ofp, ofpp) = self._get_dp()
  138. if table_id is None:
  139. table_id = ofp.OFPTT_ALL
  140. msg = ofpp.OFPFlowStatsRequest(dp, table_id=table_id)
  141. replies = self._send_msg(msg,
  142. reply_cls=ofpp.OFPFlowStatsReply,
  143. reply_multi=True)
  144. flows = []
  145. for rep in replies:
  146. flows += rep.body
  147. return flows
  148. def _dump_and_clean(self, table_id=None):
  149. cookies = set([f.cookie for f in self.dump_flows(table_id)]) - \
  150. self.reserved_cookies
  151. for c in cookies:
  152. LOG.warning("Deleting flow with cookie 0x%(cookie)x",
  153. {'cookie': c})
  154. self.uninstall_flows(cookie=c, cookie_mask=ovs_lib.UINT64_BITMASK)
  155. def cleanup_flows(self):
  156. LOG.info("Reserved cookies for %s: %s", self.br_name,
  157. self.reserved_cookies)
  158. for table_id in self.of_tables:
  159. self._dump_and_clean(table_id)
  160. def install_goto_next(self, table_id, active_bundle=None):
  161. self.install_goto(table_id=table_id, dest_table_id=table_id + 1,
  162. active_bundle=active_bundle)
  163. def install_output(self, port, table_id=0, priority=0,
  164. match=None, **match_kwargs):
  165. (_dp, ofp, ofpp) = self._get_dp()
  166. actions = [ofpp.OFPActionOutput(port, 0)]
  167. instructions = [ofpp.OFPInstructionActions(
  168. ofp.OFPIT_APPLY_ACTIONS, actions)]
  169. self.install_instructions(table_id=table_id, priority=priority,
  170. instructions=instructions,
  171. match=match, **match_kwargs)
  172. def install_normal(self, table_id=0, priority=0,
  173. match=None, **match_kwargs):
  174. (_dp, ofp, _ofpp) = self._get_dp()
  175. self.install_output(port=ofp.OFPP_NORMAL,
  176. table_id=table_id, priority=priority,
  177. match=match, **match_kwargs)
  178. def install_goto(self, dest_table_id, table_id=0, priority=0,
  179. match=None, **match_kwargs):
  180. (_dp, _ofp, ofpp) = self._get_dp()
  181. instructions = [ofpp.OFPInstructionGotoTable(table_id=dest_table_id)]
  182. self.install_instructions(table_id=table_id, priority=priority,
  183. instructions=instructions,
  184. match=match, **match_kwargs)
  185. def install_drop(self, table_id=0, priority=0, match=None, **match_kwargs):
  186. self.install_instructions(table_id=table_id, priority=priority,
  187. instructions=[], match=match, **match_kwargs)
  188. def install_instructions(self, instructions,
  189. table_id=0, priority=0,
  190. match=None, active_bundle=None, **match_kwargs):
  191. (dp, ofp, ofpp) = self._get_dp()
  192. match = self._match(ofp, ofpp, match, **match_kwargs)
  193. if isinstance(instructions, six.string_types):
  194. # NOTE: instructions must be str for the ofctl of_interface.
  195. # After the ofctl driver is removed, a deprecation warning
  196. # could be added here.
  197. jsonlist = ofctl_string.ofp_instruction_from_str(
  198. ofp, instructions)
  199. instructions = ofproto_parser.ofp_instruction_from_jsondict(
  200. dp, jsonlist)
  201. msg = ofpp.OFPFlowMod(dp,
  202. table_id=table_id,
  203. cookie=self.default_cookie,
  204. match=match,
  205. priority=priority,
  206. instructions=instructions)
  207. self._send_msg(msg, active_bundle=active_bundle)
  208. def install_apply_actions(self, actions,
  209. table_id=0, priority=0,
  210. match=None, **match_kwargs):
  211. (dp, ofp, ofpp) = self._get_dp()
  212. instructions = [
  213. ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, actions),
  214. ]
  215. self.install_instructions(table_id=table_id,
  216. priority=priority,
  217. match=match,
  218. instructions=instructions,
  219. **match_kwargs)
  220. def bundled(self, atomic=False, ordered=False):
  221. return BundledOpenFlowBridge(self, atomic, ordered)
  222. class BundledOpenFlowBridge(object):
  223. def __init__(self, br, atomic, ordered):
  224. self.br = br
  225. self.active_bundle = None
  226. self.bundle_flags = 0
  227. if not atomic and not ordered:
  228. return
  229. (dp, ofp, ofpp) = self.br._get_dp()
  230. if atomic:
  231. self.bundle_flags |= ofp.ONF_BF_ATOMIC
  232. if ordered:
  233. self.bundle_flags |= ofp.ONF_BF_ORDERED
  234. def __getattr__(self, name):
  235. if name.startswith('install') or name.startswith('uninstall'):
  236. under = getattr(self.br, name)
  237. if self.active_bundle is None:
  238. return under
  239. return functools.partial(under, active_bundle=dict(
  240. id=self.active_bundle, bundle_flags=self.bundle_flags))
  241. raise AttributeError(_("Only install_* or uninstall_* methods "
  242. "can be used"))
  243. def __enter__(self):
  244. if self.active_bundle is not None:
  245. raise ActiveBundleRunning(bundle_id=self.active_bundle)
  246. while True:
  247. self.active_bundle = random.randrange(BUNDLE_ID_WIDTH)
  248. if self.active_bundle not in self.br.active_bundles:
  249. self.br.active_bundles.add(self.active_bundle)
  250. break
  251. try:
  252. (dp, ofp, ofpp) = self.br._get_dp()
  253. msg = ofpp.ONFBundleCtrlMsg(dp, self.active_bundle,
  254. ofp.ONF_BCT_OPEN_REQUEST,
  255. self.bundle_flags, [])
  256. reply = self.br._send_msg(msg, reply_cls=ofpp.ONFBundleCtrlMsg)
  257. if reply.type != ofp.ONF_BCT_OPEN_REPLY:
  258. raise RuntimeError(
  259. _("Unexpected reply type %d != ONF_BCT_OPEN_REPLY") %
  260. reply.type)
  261. return self
  262. except Exception:
  263. self.br.active_bundles.remove(self.active_bundle)
  264. self.active_bundle = None
  265. raise
  266. def __exit__(self, type, value, traceback):
  267. (dp, ofp, ofpp) = self.br._get_dp()
  268. if type is None:
  269. ctrl_type = ofp.ONF_BCT_COMMIT_REQUEST
  270. expected_reply = ofp.ONF_BCT_COMMIT_REPLY
  271. else:
  272. ctrl_type = ofp.ONF_BCT_DISCARD_REQUEST
  273. expected_reply = ofp.ONF_BCT_DISCARD_REPLY
  274. LOG.warning(
  275. "Discarding bundle with ID 0x%(id)x due to an exception",
  276. {'id': self.active_bundle})
  277. try:
  278. msg = ofpp.ONFBundleCtrlMsg(dp, self.active_bundle,
  279. ctrl_type,
  280. self.bundle_flags, [])
  281. reply = self.br._send_msg(msg, reply_cls=ofpp.ONFBundleCtrlMsg)
  282. if reply.type != expected_reply:
  283. # The bundle ID may be in a bad state. Let's leave it
  284. # in active_bundles so that we will never use it again.
  285. raise RuntimeError(_("Unexpected reply type %d") % reply.type)
  286. self.br.active_bundles.remove(self.active_bundle)
  287. finally:
  288. # It is possible the bundle is kept open, but this must be
  289. # cleared or all subsequent __enter__ will fail.
  290. self.active_bundle = None