backend.workers.restart_4cat

  1"""
  2Restart 4CAT and optionally upgrade it to the latest release
  3"""
  4import subprocess
  5import requests
  6import hashlib
  7import shlex
  8import json
  9import time
 10import uuid
 11import sys
 12
 13from pathlib import Path
 14
 15from backend.lib.worker import BasicWorker
 16from common.lib.exceptions import WorkerInterruptedException
 17from common.config_manager import config
 18
 19
 20class FourcatRestarterAndUpgrader(BasicWorker):
 21    """
 22    Restart 4CAT and optionally upgrade it to the latest release
 23
 24    Why implement this as a worker? Trying to have 4CAT restart itself leads
 25    to an interesting conundrum: it will not be able to report the outcome of
 26    the restart, because whatever bit of code is keeping track of that will be
 27    interrupted by restarting 4CAT.
 28
 29    Using a worker has the benefit of it restarting after 4CAT restarts, so it
 30    can then figure out that 4CAT was just restarted and report the outcome. It
 31    then uses a log file to keep track of the results. The log file can then be
 32    used by other parts of 4CAT to see if the restart was successful.
 33
 34    It does lead to another conundrum - what if due to some error, 4CAT never
 35    restarts? Then this worker will not be run again to report its own failure.
 36    There seem to be no clean ways around this, so anything watching the
 37    outcome of the worker probably needs to implement some timeout after which
 38    it is assumed that the restart/upgrade process failed catastrophically.
 39    """
 40    type = "restart-4cat"
 41    max_workers = 1
 42
 43    def work(self):
 44        """
 45        Restart 4CAT and optionally upgrade it to the latest release
 46        """
 47        # figure out if we're starting the restart or checking the result
 48        # after 4cat has been restarted
 49        is_resuming = self.job.data["attempts"] > 0
 50
 51        # prevent multiple restarts running at the same time which could blow
 52        # up really fast
 53        lock_file = Path(config.get("PATH_ROOT")).joinpath("config/restart.lock")
 54
 55        # this file has the log of the restart worker itself and is checked by
 56        # the frontend to see how far we are
 57        log_file_restart = Path(config.get("PATH_ROOT")).joinpath(config.get("PATH_LOGS")).joinpath("restart.log")
 58        log_stream_restart = log_file_restart.open("a")
 59
 60        if not is_resuming:
 61            log_stream_restart.write("Initiating 4CAT restart worker\n")
 62            self.log.info("New restart initiated.")
 63
 64            # this lock file will ensure that people don't start two
 65            # simultaneous upgrades or something
 66            with lock_file.open("w") as outfile:
 67                hasher = hashlib.blake2b()
 68                hasher.update(str(uuid.uuid4()).encode("utf-8"))
 69                outfile.write(hasher.hexdigest())
 70
 71            # trigger a restart and/or upgrade
 72            # returns a JSON with a 'status' key and a message, the message
 73            # being the process output
 74
 75            if self.job.data["remote_id"].startswith("upgrade"):
 76                command = sys.executable + " helper-scripts/migrate.py --repository %s --yes --restart --output %s" % \
 77                          (shlex.quote(config.get("4cat.github_url")), shlex.quote(str(log_file_restart)))
 78                if self.job.details and self.job.details.get("branch"):
 79                    # migrate to code in specific branch
 80                    command += f" --branch {shlex.quote(self.job.details['branch'])}"
 81                else:
 82                    # migrate to latest release
 83                    command += " --release"
 84
 85            else:
 86                command = sys.executable + " 4cat-daemon.py --no-version-check force-restart"
 87
 88            try:
 89                # flush any writes before the other process starts writing to
 90                # the stream
 91                self.log.info(f"Running command {command}")
 92                log_stream_restart.flush()
 93
 94                # the tricky part is that this command will interrupt the
 95                # daemon, i.e. this worker!
 96                # so we'll never get to actually send a response, if all goes
 97                # well. but the file descriptor that stdout is piped to remains
 98                # open, somehow, so we can use that to keep track of the output
 99                # stdin needs to be /dev/null here because else when 4CAT
100                # restarts and we re-attempt to make a daemon, it will fail
101                # when trying to close the stdin file descriptor of the
102                # subprocess (man, that was a fun bug to hunt down)
103                process = subprocess.Popen(shlex.split(command), cwd=str(config.get("PATH_ROOT")),
104                                           stdout=log_stream_restart, stderr=log_stream_restart,
105                                           stdin=subprocess.DEVNULL)
106
107                while not self.interrupted:
108                    # basically wait for either the process to quit or 4CAT to
109                    # be restarted (hopefully the latter)
110                    try:
111                        # now see if the process is finished - if not a
112                        # TimeoutExpired will be raised
113                        process.wait(1)
114                        break
115                    except subprocess.TimeoutExpired:
116                        pass
117
118                if process.returncode is not None:
119                    # if we reach this, 4CAT was never restarted, and so the job failed
120                    log_stream_restart.write(
121                        f"\nUnexpected outcome of restart call ({process.returncode})\n")
122
123                    raise RuntimeError()
124                else:
125                    # interrupted before the process could finish (as it should)
126                    self.log.info("Restart triggered. Restarting 4CAT.\n")
127                    raise WorkerInterruptedException()
128
129            except (RuntimeError, subprocess.CalledProcessError) as e:
130                log_stream_restart.write(str(e))
131                log_stream_restart.write(
132                    "[Worker] Error while restarting 4CAT. The script returned a non-standard error code "
133                    "(see above). You may need to restart 4CAT manually.\n")
134                self.log.error(f"Error restarting 4CAT. See {log_stream_restart.name} for details.")
135                lock_file.unlink()
136                self.job.finish()
137
138            finally:
139                log_stream_restart.close()
140
141        else:
142            # 4CAT back-end was restarted - now check the results and make the
143            # front-end restart or upgrade too
144            self.log.info("Restart worker resumed after restarting 4CAT, restart successful.")
145            log_stream_restart.write("4CAT restarted.\n")
146            with Path(config.get("PATH_ROOT")).joinpath("config/.current-version").open() as infile:
147                log_stream_restart.write(f"4CAT is now running version {infile.readline().strip()}.\n")
148
149            # we're gonna use some specific Flask routes to trigger this, i.e.
150            # we're interacting with the front-end through HTTP
151            api_host = "https://" if config.get("flask.https") else "http://"
152            if config.get("USING_DOCKER"):
153                import os
154                docker_exposed_port = os.environ['PUBLIC_PORT']
155                api_host += f"host.docker.internal{':' + docker_exposed_port if docker_exposed_port != '80' else ''}"
156            else:
157                api_host += config.get("flask.server_name")
158
159            if self.job.data["remote_id"].startswith("upgrade") and config.get("USING_DOCKER"):
160                # when using Docker, the front-end needs to update separately
161                log_stream_restart.write("Telling front-end Docker container to upgrade...\n")
162                log_stream_restart.close()  # close, because front-end will be writing to it
163                upgrade_ok = False
164                upgrade_timeout = False
165                try:
166                    upgrade_url = api_host + "/admin/trigger-frontend-upgrade/"
167                    with lock_file.open() as infile:
168                        frontend_upgrade = requests.post(upgrade_url, data={"token": infile.read()}, timeout=(10 * 60))
169                    upgrade_ok = frontend_upgrade.json()["status"] == "OK"
170                except requests.RequestException:
171                    pass
172                except TimeoutError:
173                    upgrade_timeout = True
174
175                log_stream_restart = log_file_restart.open("a")
176                if not upgrade_ok:
177                    if upgrade_timeout:
178                        log_stream_restart.write("Upgrade timed out.")
179                    log_stream_restart.write("Error upgrading front-end container. You may need to upgrade and restart"
180                                             "containers manually.\n")
181                    lock_file.unlink()
182                    return self.job.finish()
183
184            # restart front-end
185            log_stream_restart.write("Asking front-end to restart itself...\n")
186            log_stream_restart.flush()
187            try:
188                restart_url = api_host + "/admin/trigger-frontend-restart/"
189                with lock_file.open() as infile:
190                    response = requests.post(restart_url, data={"token": infile.read()}, timeout=5).json()
191
192                if response.get("message"):
193                    log_stream_restart.write(response.get("message") + "\n")
194            except (json.JSONDecodeError, requests.RequestException):
195                # this may happen because the server restarts and interrupts
196                # the request
197                pass
198
199            # wait for front-end to come online after a restart
200            time.sleep(3)  # give some time for the restart to trigger
201            start_time = time.time()
202            frontend_ok = False
203            while time.time() < start_time + 60:
204                try:
205                    frontend = requests.get(api_host + "/", timeout=5)
206                    if frontend.status_code > 401:
207                        time.sleep(2)
208                        continue
209                    frontend_ok = True
210                    break
211                except requests.RequestException as e:
212                    time.sleep(1)
213                    continue
214
215            # too bad
216            if not frontend_ok:
217                log_stream_restart.write("Timed out waiting for front-end to restart. You may need to restart it "
218                                         "manually.\n")
219                self.log.error("Front-end did not come back online after restart")
220            else:
221                log_stream_restart.write("Front-end is available. Restart complete.")
222                self.log.info("Front-end is available. Restart complete.")
223
224            log_stream_restart.close()
225            lock_file.unlink()
226
227            self.job.finish()