1
0

synctl 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384
  1. #!/usr/bin/env python
  2. # Copyright 2014-2016 OpenMarket Ltd
  3. # Copyright 2018 New Vector Ltd
  4. #
  5. # Licensed under the Apache License, Version 2.0 (the "License");
  6. # you may not use this file except in compliance with the License.
  7. # You may obtain 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,
  13. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14. # See the License for the specific language governing permissions and
  15. # limitations under the License.
  16. import argparse
  17. import collections
  18. import errno
  19. import glob
  20. import os
  21. import os.path
  22. import signal
  23. import subprocess
  24. import sys
  25. import time
  26. import yaml
  27. from synapse.config import find_config_files
  28. SYNAPSE = [sys.executable, "-m", "synapse.app.homeserver"]
  29. GREEN = "\x1b[1;32m"
  30. YELLOW = "\x1b[1;33m"
  31. RED = "\x1b[1;31m"
  32. NORMAL = "\x1b[m"
  33. def pid_running(pid):
  34. try:
  35. os.kill(pid, 0)
  36. return True
  37. except OSError as err:
  38. if err.errno == errno.EPERM:
  39. return True
  40. return False
  41. def write(message, colour=NORMAL, stream=sys.stdout):
  42. # Lets check if we're writing to a TTY before colouring
  43. should_colour = False
  44. try:
  45. should_colour = stream.isatty()
  46. except AttributeError:
  47. # Just in case `isatty` isn't defined on everything. The python
  48. # docs are incredibly vague.
  49. pass
  50. if not should_colour:
  51. stream.write(message + "\n")
  52. else:
  53. stream.write(colour + message + NORMAL + "\n")
  54. def abort(message, colour=RED, stream=sys.stderr):
  55. write(message, colour, stream)
  56. sys.exit(1)
  57. def start(configfile: str, daemonize: bool = True) -> bool:
  58. """Attempts to start synapse.
  59. Args:
  60. configfile: path to a yaml synapse config file
  61. daemonize: whether to daemonize synapse or keep it attached to the current
  62. session
  63. Returns:
  64. True if the process started successfully
  65. False if there was an error starting the process
  66. If deamonize is False it will only return once synapse exits.
  67. """
  68. write("Starting ...")
  69. args = SYNAPSE
  70. if daemonize:
  71. args.extend(["--daemonize", "-c", configfile])
  72. else:
  73. args.extend(["-c", configfile])
  74. try:
  75. subprocess.check_call(args)
  76. write("started synapse.app.homeserver(%r)" % (configfile,), colour=GREEN)
  77. return True
  78. except subprocess.CalledProcessError as e:
  79. write(
  80. "error starting (exit code: %d); see above for logs" % e.returncode,
  81. colour=RED,
  82. )
  83. return False
  84. def start_worker(app: str, configfile: str, worker_configfile: str) -> bool:
  85. """Attempts to start a synapse worker.
  86. Args:
  87. app: name of the worker's appservice
  88. configfile: path to a yaml synapse config file
  89. worker_configfile: path to worker specific yaml synapse file
  90. Returns:
  91. True if the process started successfully
  92. False if there was an error starting the process
  93. """
  94. args = [
  95. sys.executable,
  96. "-m",
  97. app,
  98. "-c",
  99. configfile,
  100. "-c",
  101. worker_configfile,
  102. "--daemonize",
  103. ]
  104. try:
  105. subprocess.check_call(args)
  106. write("started %s(%r)" % (app, worker_configfile), colour=GREEN)
  107. return True
  108. except subprocess.CalledProcessError as e:
  109. write(
  110. "error starting %s(%r) (exit code: %d); see above for logs"
  111. % (app, worker_configfile, e.returncode),
  112. colour=RED,
  113. )
  114. return False
  115. def stop(pidfile: str, app: str) -> bool:
  116. """Attempts to kill a synapse worker from the pidfile.
  117. Args:
  118. pidfile: path to file containing worker's pid
  119. app: name of the worker's appservice
  120. Returns:
  121. True if the process stopped successfully
  122. False if process was already stopped or an error occured
  123. """
  124. if os.path.exists(pidfile):
  125. pid = int(open(pidfile).read())
  126. try:
  127. os.kill(pid, signal.SIGTERM)
  128. write("stopped %s" % (app,), colour=GREEN)
  129. return True
  130. except OSError as err:
  131. if err.errno == errno.ESRCH:
  132. write("%s not running" % (app,), colour=YELLOW)
  133. elif err.errno == errno.EPERM:
  134. abort("Cannot stop %s: Operation not permitted" % (app,))
  135. else:
  136. abort("Cannot stop %s: Unknown error" % (app,))
  137. return False
  138. else:
  139. write(
  140. "No running worker of %s found (from %s)\nThe process might be managed by another controller (e.g. systemd)"
  141. % (app, pidfile),
  142. colour=YELLOW,
  143. )
  144. return False
  145. Worker = collections.namedtuple(
  146. "Worker", ["app", "configfile", "pidfile", "cache_factor", "cache_factors"]
  147. )
  148. def main():
  149. parser = argparse.ArgumentParser()
  150. parser.add_argument(
  151. "action",
  152. choices=["start", "stop", "restart"],
  153. help="whether to start, stop or restart the synapse",
  154. )
  155. parser.add_argument(
  156. "configfile",
  157. nargs="?",
  158. default="homeserver.yaml",
  159. help="the homeserver config file. Defaults to homeserver.yaml. May also be"
  160. " a directory with *.yaml files",
  161. )
  162. parser.add_argument(
  163. "-w", "--worker", metavar="WORKERCONFIG", help="start or stop a single worker"
  164. )
  165. parser.add_argument(
  166. "-a",
  167. "--all-processes",
  168. metavar="WORKERCONFIGDIR",
  169. help="start or stop all the workers in the given directory"
  170. " and the main synapse process",
  171. )
  172. parser.add_argument(
  173. "--no-daemonize",
  174. action="store_false",
  175. dest="daemonize",
  176. help="Run synapse in the foreground for debugging. "
  177. "Will work only if the daemonize option is not set in the config.",
  178. )
  179. options = parser.parse_args()
  180. if options.worker and options.all_processes:
  181. write('Cannot use "--worker" with "--all-processes"', stream=sys.stderr)
  182. sys.exit(1)
  183. if not options.daemonize and options.all_processes:
  184. write('Cannot use "--no-daemonize" with "--all-processes"', stream=sys.stderr)
  185. sys.exit(1)
  186. configfile = options.configfile
  187. if not os.path.exists(configfile):
  188. write(
  189. "No config file found\n"
  190. "To generate a config file, run '%s -c %s --generate-config"
  191. " --server-name=<server name> --report-stats=<yes/no>'\n"
  192. % (" ".join(SYNAPSE), options.configfile),
  193. stream=sys.stderr,
  194. )
  195. sys.exit(1)
  196. config_files = find_config_files([configfile])
  197. config = {}
  198. for config_file in config_files:
  199. with open(config_file) as file_stream:
  200. yaml_config = yaml.safe_load(file_stream)
  201. if yaml_config is not None:
  202. config.update(yaml_config)
  203. pidfile = config["pid_file"]
  204. cache_factor = config.get("synctl_cache_factor")
  205. start_stop_synapse = True
  206. if cache_factor:
  207. os.environ["SYNAPSE_CACHE_FACTOR"] = str(cache_factor)
  208. cache_factors = config.get("synctl_cache_factors", {})
  209. for cache_name, factor in cache_factors.items():
  210. os.environ["SYNAPSE_CACHE_FACTOR_" + cache_name.upper()] = str(factor)
  211. worker_configfiles = []
  212. if options.worker:
  213. start_stop_synapse = False
  214. worker_configfile = options.worker
  215. if not os.path.exists(worker_configfile):
  216. write(
  217. "No worker config found at %r" % (worker_configfile,), stream=sys.stderr
  218. )
  219. sys.exit(1)
  220. worker_configfiles.append(worker_configfile)
  221. if options.all_processes:
  222. # To start the main synapse with -a you need to add a worker file
  223. # with worker_app == "synapse.app.homeserver"
  224. start_stop_synapse = False
  225. worker_configdir = options.all_processes
  226. if not os.path.isdir(worker_configdir):
  227. write(
  228. "No worker config directory found at %r" % (worker_configdir,),
  229. stream=sys.stderr,
  230. )
  231. sys.exit(1)
  232. worker_configfiles.extend(
  233. sorted(glob.glob(os.path.join(worker_configdir, "*.yaml")))
  234. )
  235. workers = []
  236. for worker_configfile in worker_configfiles:
  237. with open(worker_configfile) as stream:
  238. worker_config = yaml.safe_load(stream)
  239. worker_app = worker_config["worker_app"]
  240. if worker_app == "synapse.app.homeserver":
  241. # We need to special case all of this to pick up options that may
  242. # be set in the main config file or in this worker config file.
  243. worker_pidfile = worker_config.get("pid_file") or pidfile
  244. worker_cache_factor = (
  245. worker_config.get("synctl_cache_factor") or cache_factor
  246. )
  247. worker_cache_factors = (
  248. worker_config.get("synctl_cache_factors") or cache_factors
  249. )
  250. # The master process doesn't support using worker_* config.
  251. for key in worker_config:
  252. if key == "worker_app": # But we allow worker_app
  253. continue
  254. assert not key.startswith(
  255. "worker_"
  256. ), "Main process cannot use worker_* config"
  257. else:
  258. worker_pidfile = worker_config["worker_pid_file"]
  259. worker_cache_factor = worker_config.get("synctl_cache_factor")
  260. worker_cache_factors = worker_config.get("synctl_cache_factors", {})
  261. workers.append(
  262. Worker(
  263. worker_app,
  264. worker_configfile,
  265. worker_pidfile,
  266. worker_cache_factor,
  267. worker_cache_factors,
  268. )
  269. )
  270. action = options.action
  271. if action == "stop" or action == "restart":
  272. has_stopped = True
  273. for worker in workers:
  274. if not stop(worker.pidfile, worker.app):
  275. # A worker could not be stopped.
  276. has_stopped = False
  277. if start_stop_synapse:
  278. if not stop(pidfile, "synapse.app.homeserver"):
  279. has_stopped = False
  280. if not has_stopped and action == "stop":
  281. sys.exit(1)
  282. # Wait for synapse to actually shutdown before starting it again
  283. if action == "restart":
  284. running_pids = []
  285. if start_stop_synapse and os.path.exists(pidfile):
  286. running_pids.append(int(open(pidfile).read()))
  287. for worker in workers:
  288. if os.path.exists(worker.pidfile):
  289. running_pids.append(int(open(worker.pidfile).read()))
  290. if len(running_pids) > 0:
  291. write("Waiting for process to exit before restarting...")
  292. for running_pid in running_pids:
  293. while pid_running(running_pid):
  294. time.sleep(0.2)
  295. write("All processes exited; now restarting...")
  296. if action == "start" or action == "restart":
  297. error = False
  298. if start_stop_synapse:
  299. # Check if synapse is already running
  300. if os.path.exists(pidfile) and pid_running(int(open(pidfile).read())):
  301. abort("synapse.app.homeserver already running")
  302. if not start(configfile, bool(options.daemonize)):
  303. error = True
  304. for worker in workers:
  305. env = os.environ.copy()
  306. # Skip starting a worker if its already running
  307. if os.path.exists(worker.pidfile) and pid_running(
  308. int(open(worker.pidfile).read())
  309. ):
  310. print(worker.app + " already running")
  311. continue
  312. if worker.cache_factor:
  313. os.environ["SYNAPSE_CACHE_FACTOR"] = str(worker.cache_factor)
  314. for cache_name, factor in worker.cache_factors.items():
  315. os.environ["SYNAPSE_CACHE_FACTOR_" + cache_name.upper()] = str(factor)
  316. if not start_worker(worker.app, configfile, worker.configfile):
  317. error = True
  318. # Reset env back to the original
  319. os.environ.clear()
  320. os.environ.update(env)
  321. if error:
  322. exit(1)
  323. if __name__ == "__main__":
  324. main()