OVSDB application library
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.

164 lines
5.7KB

  1. # Copyright (c) 2015 Red Hat, Inc.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may
  4. # not use this file except in compliance with the License. You may obtain
  5. # a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11. # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12. # License for the specific language governing permissions and limitations
  13. # under the License.
  14. import logging
  15. import os
  16. import threading
  17. import traceback
  18. from ovs.db import idl
  19. from ovs import poller
  20. from six.moves import queue as Queue
  21. from ovsdbapp.backend.ovs_idl import idlutils
  22. from ovsdbapp import exceptions
  23. if os.name == 'nt':
  24. from ovsdbapp.backend.ovs_idl.windows import connection_utils
  25. else:
  26. from ovsdbapp.backend.ovs_idl.linux import connection_utils
  27. LOG = logging.getLogger(__name__)
  28. class TransactionQueue(Queue.Queue, object):
  29. def __init__(self, *args, **kwargs):
  30. super(TransactionQueue, self).__init__(*args, **kwargs)
  31. self._wait_queue = connection_utils.WaitQueue(
  32. max_queue_size=self.maxsize)
  33. def get_nowait(self, *args, **kwargs):
  34. try:
  35. result = super(TransactionQueue, self).get_nowait(*args, **kwargs)
  36. except Queue.Empty:
  37. return None
  38. self._wait_queue.alert_notification_consume()
  39. return result
  40. def put(self, *args, **kwargs):
  41. super(TransactionQueue, self).put(*args, **kwargs)
  42. self._wait_queue.alert_notify()
  43. @property
  44. def alert_fileno(self):
  45. return self._wait_queue.alert_fileno
  46. class Connection(object):
  47. def __init__(self, idl, timeout):
  48. """Create a connection to an OVSDB server using the OVS IDL
  49. :param timeout: The timeout value for OVSDB operations
  50. :param idl: A newly created ovs.db.Idl instance (run never called)
  51. """
  52. self.timeout = timeout
  53. self.txns = TransactionQueue(1)
  54. self.lock = threading.Lock()
  55. self.idl = idl
  56. self.thread = None
  57. self._is_running = None
  58. def start(self):
  59. """Start the connection."""
  60. with self.lock:
  61. if self.thread is not None:
  62. return False
  63. if not self.idl.has_ever_connected():
  64. idlutils.wait_for_change(self.idl, self.timeout)
  65. try:
  66. self.idl.post_connect()
  67. except AttributeError:
  68. # An ovs.db.Idl class has no post_connect
  69. pass
  70. self.poller = poller.Poller()
  71. self._is_running = True
  72. self.thread = threading.Thread(target=self.run)
  73. self.thread.setDaemon(True)
  74. self.thread.start()
  75. def run(self):
  76. errors = 0
  77. while self._is_running:
  78. # If we fail in an Idl call, we could have missed an update
  79. # from the server, leaving us out of sync with ovsdb-server.
  80. # It is not safe to continue without restarting the connection,
  81. # though it is likely that the error is unrecoverable, so only try
  82. # a few times before bailing completely.
  83. try:
  84. self.idl.wait(self.poller)
  85. self.poller.fd_wait(self.txns.alert_fileno, poller.POLLIN)
  86. # TODO(jlibosva): Remove next line once losing connection to
  87. # ovsdb is solved.
  88. self.poller.timer_wait(self.timeout * 1000)
  89. self.poller.block()
  90. self.idl.run()
  91. except Exception as e:
  92. # This shouldn't happen, but is possible if there is a bug
  93. # in python-ovs
  94. errors += 1
  95. LOG.exception(e)
  96. if errors <= 3:
  97. self.idl.force_reconnect()
  98. idlutils.wait_for_change(self.idl, self.timeout)
  99. continue
  100. self._is_running = False
  101. break
  102. errors = 0
  103. txn = self.txns.get_nowait()
  104. if txn is not None:
  105. try:
  106. txn.results.put(txn.do_commit())
  107. except Exception as ex:
  108. er = idlutils.ExceptionResult(ex=ex,
  109. tb=traceback.format_exc())
  110. txn.results.put(er)
  111. self.txns.task_done()
  112. def stop(self, timeout=None):
  113. if not self._is_running:
  114. return True
  115. self._is_running = False
  116. self.txns.put(None)
  117. self.thread.join(timeout)
  118. if self.thread.is_alive():
  119. return False
  120. self.thread = None
  121. return True
  122. def queue_txn(self, txn):
  123. # Even if we aren't started, we can queue a transaction and it will
  124. # run when we are started
  125. try:
  126. self.txns.put(txn, timeout=self.timeout)
  127. except Queue.Full:
  128. raise exceptions.TimeoutException(commands=txn.commands,
  129. timeout=self.timeout)
  130. class OvsdbIdl(idl.Idl):
  131. @classmethod
  132. def from_server(cls, connection_string, schema_name):
  133. """Create the Idl instance by pulling the schema from OVSDB server"""
  134. helper = idlutils.get_schema_helper(connection_string, schema_name)
  135. helper.register_all()
  136. return cls(connection_string, helper)
  137. def post_connect(self):
  138. """Operations to execute after the Idl has connected to the server
  139. An example would be to set up Idl notification handling for watching
  140. and unwatching certain OVSDB change events
  141. """