1
0

synctl 11 KB

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