The Gatekeeper, or a project gating system
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.

fingergw.py 7.0KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216
  1. # Copyright 2017 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 functools
  15. import logging
  16. import socket
  17. import threading
  18. import zuul.rpcclient
  19. from zuul.lib import commandsocket
  20. from zuul.lib import streamer_utils
  21. COMMANDS = ['stop']
  22. class RequestHandler(streamer_utils.BaseFingerRequestHandler):
  23. '''
  24. Class implementing the logic for handling a single finger request.
  25. '''
  26. log = logging.getLogger("zuul.fingergw")
  27. def __init__(self, *args, **kwargs):
  28. self.rpc = kwargs.pop('rpc')
  29. super(RequestHandler, self).__init__(*args, **kwargs)
  30. def _fingerClient(self, server, port, build_uuid):
  31. '''
  32. Open a finger connection and return all streaming results.
  33. :param server: The remote server.
  34. :param port: The remote port.
  35. :param build_uuid: The build UUID to stream.
  36. Both IPv4 and IPv6 are supported.
  37. '''
  38. with socket.create_connection((server, port), timeout=10) as s:
  39. # timeout only on the connection, let recv() wait forever
  40. s.settimeout(None)
  41. msg = "%s\n" % build_uuid # Must have a trailing newline!
  42. s.sendall(msg.encode('utf-8'))
  43. while True:
  44. data = s.recv(1024)
  45. if data:
  46. self.request.sendall(data)
  47. else:
  48. break
  49. def handle(self):
  50. '''
  51. This method is called by the socketserver framework to handle an
  52. incoming request.
  53. '''
  54. try:
  55. build_uuid = self.getCommand()
  56. port_location = self.rpc.get_job_log_stream_address(build_uuid)
  57. if not port_location:
  58. msg = 'Invalid build UUID %s' % build_uuid
  59. self.request.sendall(msg.encode('utf-8'))
  60. return
  61. self._fingerClient(
  62. port_location['server'],
  63. port_location['port'],
  64. build_uuid,
  65. )
  66. except BrokenPipeError: # Client disconnect
  67. return
  68. except Exception:
  69. self.log.exception('Finger request handling exception:')
  70. msg = 'Internal streaming error'
  71. self.request.sendall(msg.encode('utf-8'))
  72. return
  73. class FingerGateway(object):
  74. '''
  75. Class implementing the finger multiplexing/gateway logic.
  76. For each incoming finger request, a new thread is started that will
  77. be responsible for finding which Zuul executor is executing the
  78. requested build (by asking Gearman), forwarding the request to that
  79. executor, and streaming the results back to our client.
  80. '''
  81. log = logging.getLogger("zuul.fingergw")
  82. def __init__(self, gearman, address, user, command_socket, pid_file):
  83. '''
  84. Initialize the finger gateway.
  85. :param tuple gearman: Gearman connection information. This should
  86. include the server, port, SSL key, SSL cert, and SSL CA.
  87. :param tuple address: The address and port to bind to for our gateway.
  88. :param str user: The user to which we should drop privileges after
  89. binding to our address.
  90. :param str command_socket: Path to the daemon command socket.
  91. :param str pid_file: Path to the daemon PID file.
  92. '''
  93. self.gear_server = gearman[0]
  94. self.gear_port = gearman[1]
  95. self.gear_ssl_key = gearman[2]
  96. self.gear_ssl_cert = gearman[3]
  97. self.gear_ssl_ca = gearman[4]
  98. self.address = address
  99. self.user = user
  100. self.pid_file = pid_file
  101. self.rpc = None
  102. self.server = None
  103. self.server_thread = None
  104. self.command_thread = None
  105. self.command_running = False
  106. self.command_socket = command_socket
  107. self.command_map = dict(
  108. stop=self.stop,
  109. )
  110. def _runCommand(self):
  111. while self.command_running:
  112. try:
  113. command = self.command_socket.get().decode('utf8')
  114. if command != '_stop':
  115. self.command_map[command]()
  116. else:
  117. return
  118. except Exception:
  119. self.log.exception("Exception while processing command")
  120. def _run(self):
  121. try:
  122. self.server.serve_forever()
  123. except Exception:
  124. self.log.exception('Abnormal termination:')
  125. raise
  126. def start(self):
  127. self.rpc = zuul.rpcclient.RPCClient(
  128. self.gear_server,
  129. self.gear_port,
  130. self.gear_ssl_key,
  131. self.gear_ssl_cert,
  132. self.gear_ssl_ca)
  133. self.server = streamer_utils.CustomThreadingTCPServer(
  134. self.address,
  135. functools.partial(RequestHandler, rpc=self.rpc),
  136. user=self.user,
  137. pid_file=self.pid_file)
  138. # Start the command processor after the server and privilege drop
  139. if self.command_socket:
  140. self.log.debug("Starting command processor")
  141. self.command_socket = commandsocket.CommandSocket(
  142. self.command_socket)
  143. self.command_socket.start()
  144. self.command_running = True
  145. self.command_thread = threading.Thread(
  146. target=self._runCommand, name='command')
  147. self.command_thread.daemon = True
  148. self.command_thread.start()
  149. # The socketserver shutdown() call will hang unless the call
  150. # to server_forever() happens in another thread. So let's do that.
  151. self.server_thread = threading.Thread(target=self._run)
  152. self.server_thread.daemon = True
  153. self.server_thread.start()
  154. self.log.info("Finger gateway is started")
  155. def stop(self):
  156. if self.server:
  157. try:
  158. self.server.shutdown()
  159. self.server.server_close()
  160. self.server = None
  161. except Exception:
  162. self.log.exception("Error stopping TCP server:")
  163. if self.rpc:
  164. try:
  165. self.rpc.shutdown()
  166. self.rpc = None
  167. except Exception:
  168. self.log.exception("Error stopping RCP client:")
  169. if self.command_socket:
  170. self.command_running = False
  171. try:
  172. self.command_socket.stop()
  173. except Exception:
  174. self.log.exception("Error stopping command socket:")
  175. self.log.info("Finger gateway is stopped")
  176. def wait(self):
  177. '''
  178. Wait on the gateway to shutdown.
  179. '''
  180. self.server_thread.join()