diff --git a/src/qlever/commands/rebuild_index.py b/src/qlever/commands/rebuild_index.py index 057ee959..f6516837 100644 --- a/src/qlever/commands/rebuild_index.py +++ b/src/qlever/commands/rebuild_index.py @@ -1,10 +1,7 @@ from __future__ import annotations -import copy -import os -import shlex +import json import shutil -import socket import subprocess import time from pathlib import Path @@ -12,105 +9,8 @@ from termcolor import colored from qlever.command import QleverCommand -from qlever.commands.start import StartCommand -from qlever.commands.stop import StopCommand from qlever.log import log -from qlever.util import ( - get_existing_index_files, - run_command, -) - - -def validate_index(args, index_dir: str) -> bool: - """ - Validate a newly built index by starting a temporary server on a - different port, sending a simple query, and checking that it succeeds. - Returns True if the index is usable, False otherwise. - """ - # Find a free port and set the container name accordingly. - s = socket.socket() - s.bind(("", 0)) - validation_port = s.getsockname()[1] - s.close() - validation_container = f"qlever-server.validate.{validation_port}" - - # Create args for the validation server: minimal resources, different - # port and container name, working directory is the new index dir. - validation_args = copy.copy(args) - validation_args.port = validation_port - validation_args.server_container = validation_container - validation_args.memory_for_queries = "100M" - validation_args.cache_max_size = "10M" - validation_args.cache_max_size_single_entry = "10M" - validation_args.timeout = "1s" - validation_args.warmup_cmd = "" - # No resource-usage log for the throwaway validation server; its file - # would later be moved over the live server's log. - validation_args.resource_usage_log = "no" - validation_args.show = False - # Additional arguments expected by StartCommand and StopCommand - # (normally set by argparse, but we call execute() directly). - validation_args.kill_existing_with_same_port = False - validation_args.no_warmup = True - validation_args.run_in_foreground = False - validation_args.runtime_parameters = [] - validation_args.cmdline_regex = "qlever-server.* -i [^ ]*%%NAME%%" - validation_args.no_containers = False - validation_args.restart_policy = "no" - - log.info( - f'Validating new index in "{index_dir}" (port {validation_port}) ...' - ) - - # Start the validation server from the new index directory. - original_dir = Path.cwd() - try: - os.chdir(index_dir) - if not StartCommand().execute(validation_args): - log.error("Validation failed: could not start server") - return False - except Exception as e: - log.error(f"Validation failed: {e}") - return False - finally: - os.chdir(original_dir) - - # Send a simple query to check the index works. - try: - endpoint = f"{args.host_name}:{validation_port}" - query_cmd = ( - f"curl -s {endpoint}" - f' --data-urlencode "query=SELECT * WHERE' - f' {{ ?s ?p ?o }} LIMIT 1"' - f' -o /dev/null -w "%{{http_code}}"' - ) - result = run_command(query_cmd, return_output=True) - query_ok = result.strip() == "200" - except Exception: - query_ok = False - - # Stop the validation server. - try: - os.chdir(index_dir) - StopCommand().execute(validation_args) - except Exception: - pass - finally: - os.chdir(original_dir) - - if query_ok: - log.info("Validation successful: new index is usable") - else: - log.error("Validation failed: server started but query failed") - - # Remove the log files written by the validation server inside - # `index_dir`. Otherwise the subsequent `mv {index_dir}/* .` would - # clobber the live `{name}.metrics-log.jsonl` and `{name}.server-log.txt` - # of the running server in the parent directory. - for suffix in (".metrics-log.jsonl", ".server-log.txt"): - (Path(index_dir) / f"{args.name}{suffix}").unlink(missing_ok=True) - - return query_ok +from qlever.util import run_command class RebuildIndexCommand(QleverCommand): @@ -129,60 +29,30 @@ def should_have_qleverfile(self) -> bool: def relevant_qleverfile_arguments(self) -> dict[str, list[str]]: return { - "data": ["name", "description", "text_description"], - "server": [ - "server_binary", - "host_name", - "port", - "access_token", - "memory_for_queries", - "cache_max_size", - "cache_max_size_single_entry", - "cache_max_num_entries", - "num_threads", - "timeout", - "persist_updates", - "only_pso_and_pos_permutations", - "use_patterns", - "use_text_index", - "metrics_log", - "warmup_cmd", - ], - "runtime": [ - "system", - "image", - "server_container", - "restart_policy", - ], + "data": ["name"], + "server": ["host_name", "port", "access_token"], } def additional_arguments(self, subparser) -> None: subparser.add_argument( - "--new-index-dir", + "--rebuild-tmp-dir", type=str, - help="Target directory for the new index (default: not set, " - "move the old index instead; see `--old-index-dir`)", + help="Directory in which the server builds the new index " + "(default: chosen by the server, `rebuild..tmp`)", ) subparser.add_argument( - "--old-index-dir", + "--rebuild-previous-index-dir", type=str, - help="Directory where to move the current index once the rebuild " - "is finished (default: subdirectory `previous.YYYY-MM-DDTHH:MM`, " - "where the timestamp is the time of the earliest index file)", + help="Directory to which the server moves the current index when " + "the new index is swapped in (default: chosen by the server, " + "`previous.`)", ) subparser.add_argument( - "--new-index-dir-basename", - type=str, - default="rebuild.", - help="Basename prefix for the new index directory when " - "`--new-index-dir` is not specified (default: `rebuild.`)", - ) - subparser.add_argument( - "--old-index-dir-basename", - type=str, - default="previous.", - help="Basename prefix for the old index directory when " - "`--old-index-dir` is not specified (default: `previous.`)", + "--test-query", + action="store_true", + default=False, + help="After the rebuild, send a simple test query to the server " + "and report whether it succeeds", ) subparser.add_argument( "--keep-previous-index-dirs", @@ -194,7 +64,8 @@ def additional_arguments(self, subparser) -> None: "original-and-most-recent", ], default="original-and-most-recent", - help="Which previous index directories to keep: " + help="Which `previous.*` index directories to keep after a " + "successful rebuild: " "all (keep all), " "none (delete all), " "original-only (keep only the very first), " @@ -202,177 +73,51 @@ def additional_arguments(self, subparser) -> None: "original-and-most-recent (keep both) " "(default: original-and-most-recent)", ) - subparser.add_argument( - "--index-name", - type=str, - help="Base name of the files of the new index (default: use " - "the same basename as for the current index)", - ) - subparser.add_argument( - "--restart-when-finished", - action="store_true", - default=False, - help="When the rebuild is finished, stop the server with the old " - "index and start it again with the new index", - ) - subparser.add_argument( - "--leave-new-index-in-temporary-directory", - action="store_true", - default=False, - help="Leave the successfully rebuilt (and validated) index in the " - "temporary directory instead of moving it into place (same logic " - "as when the index validation fails)", - ) def execute(self, args) -> bool: - # Either `--new-index-dir` or `--old-index-dir`. - if args.new_index_dir is not None and args.old_index_dir is not None: - log.error( - "Please specify either --new-index-dir (the target directory " - "for the new index) or --old-index-dir (the directory where " - "to move the current index), but not both" - ) - return False - - # Get the list of all files from the current index and get the date of - # the earliest one (in UTC). Add the `Qleverfile` as well. - old_index_files = get_existing_index_files( - args.name, add_non_essential=True - ) - old_index_date = time.strftime( - "%Y-%m-%dT%H:%M:%SZ", - time.gmtime(min(Path(f).stat().st_mtime for f in old_index_files)), - ) - new_index_date = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - old_index_files.append("Qleverfile") - - # Default values for arguments. - # - # NOTE 1: When `--old-index-dir` is specified but not `--new-index-dir`, - # we nevertheless first build the new index in a temporary directory, - # and only when that is successful do we move the current index to the - # directory specified by `--old-index-dir` and move the new index to - # the current index directory. That way, if the rebuild fails, we still - # have the current index in its original location. - # - # NOTE 2: As a consequence of this logic, `args.new_index_dir` is - # always defined after this block, even when it was not specified on - # the command line. - if args.index_name is None: - args.index_name = args.name - if args.new_index_dir is None: - args.new_index_dir = ( - f"{args.new_index_dir_basename}{new_index_date}.tmp" - ) - if args.old_index_dir is None: - # Check if this is the first rebuild (no previous.* directories exist) - existing_previous_dirs = list( - Path(".").glob(f"{args.old_index_dir_basename}*") - ) - is_first_rebuild = len(existing_previous_dirs) == 0 - - args.old_index_dir = ( - f"{args.old_index_dir_basename}{old_index_date}" - + (".ORIGINAL" if is_first_rebuild else "") - ) - if args.new_index_dir.endswith("/"): - args.new_index_dir = args.new_index_dir[:-1] - - # Check that the new index directory either does not exist or is empty. - # Same for the old index directory, if specified. - new_index_path = Path(args.new_index_dir) - if new_index_path.exists() and any(new_index_path.iterdir()): - log.error( - f"The target directory '{args.new_index_dir}' for the new " - "index already exists and is not empty; please specify an " - "empty or non-existing directory" - ) - return False - if args.old_index_dir is not None: - old_index_path = Path(args.old_index_dir) - if old_index_path.exists() and any(old_index_path.iterdir()): - log.error( - f"The target directory '{args.old_index_dir}' for the " - "old index already exists and is not empty; please " - "specify an empty or non-existing directory" - ) - return False - - # Split `new_index_dir` into path and dir name. For example, if - # `new_index_dir` is `path/to/index`, then the path is `path/to` and - # the dir name is `index`. - # - # NOTE: We keep this separate because we can always create a - # subdirectory in the current directory (even when running in a - # container), but not necessarily a directory at an arbitrary path. If - # a path outside the current directory is desired, we move the index - # there after it has been built. - new_index_dir_path = str(Path(args.new_index_dir).parent) - new_index_dir_name = str(Path(args.new_index_dir).name) - log_file_name = f"{args.index_name}.rebuild-index-log.txt" - - # Note which indexes we have to move when done. - move_new_index_when_done = new_index_dir_path != "." - move_old_index_when_done = args.old_index_dir is not None - - # Command for rebuilding the index. - mkdir_cmd = ( - f"mkdir -p {new_index_dir_name} && " - f"cp -a Qleverfile {new_index_dir_name}" - ) + # The server does all the work: it builds the new index in a temporary + # directory (from the current data, including updates), then moves the + # current index to the previous-index directory, moves the new index + # into its place, and swaps it in without a restart. The two + # directories can optionally be chosen via the command parameters. rebuild_index_cmd = ( f"curl -s -w '\\n%{{http_code}}' {args.host_name}:{args.port} " - f"-d cmd=rebuild-index " - f"-d index-name={new_index_dir_name}/{args.index_name} " - f"-d access-token={args.access_token}" + f"-d cmd=rebuild-index" ) - move_new_index_cmd = f"mv {new_index_dir_name} {new_index_dir_path}" - move_old_index_cmd = ( - f"mkdir -p {shlex.quote(args.old_index_dir)} && " - f"mv {' '.join(shlex.quote(f) for f in old_index_files)} " - f"{shlex.quote(args.old_index_dir)} && " - f"mv {shlex.quote(new_index_dir_name)}/* . && " - f"rmdir {shlex.quote(new_index_dir_name)}" - ) - restart_server_cmd = "qlever stop && qlever start" - if not move_old_index_when_done: - restart_server_cmd = ( - f"cd {args.new_index_dir} && ${restart_server_cmd}" + if args.rebuild_tmp_dir is not None: + rebuild_index_cmd += f" -d rebuild-tmp-dir={args.rebuild_tmp_dir}" + if args.rebuild_previous_index_dir is not None: + rebuild_index_cmd += ( + f" -d rebuild-previous-index-dir=" + f"{args.rebuild_previous_index_dir}" ) + rebuild_index_cmd += f" -d access-token={args.access_token}" - # Show the command lines. - cmds_to_show = [mkdir_cmd, rebuild_index_cmd] - if not args.leave_new_index_in_temporary_directory: - if move_old_index_when_done: - cmds_to_show.append(move_old_index_cmd) - if move_new_index_when_done: - cmds_to_show.append(move_new_index_cmd) - if args.restart_when_finished: - cmds_to_show.append(restart_server_cmd) - self.show("\n".join(cmds_to_show), only_show=args.show) + # Show the command line. + self.show(rebuild_index_cmd, only_show=args.show) if args.show: return True - # Create the index directory and the log file. - try: - run_command(mkdir_cmd) - except Exception as e: - log.error(f"Creating the index directory failed: {e}") - return False - - # Show the server log while rebuilding the index. - # - # NOTE: This will only work satisfactorily when no other queries are - # being processed at the same time. It would be better if QLever - # logged the rebuild-index output to a separate log file. + # Show the rebuild log while the rebuild is running. The log is + # written to `/.rebuild-index-log.txt`; when + # the server chooses the temporary directory (the default), its name + # contains the start time of the rebuild, so wait for the most + # recently modified match. + log_file_glob = ( + args.rebuild_tmp_dir + if args.rebuild_tmp_dir is not None + else "rebuild.*.tmp" + ) + f"/{args.name}.rebuild-index-log.txt" tail_cmd = ( - f"while [ ! -f {new_index_dir_name}/{log_file_name} ]; " - f"do sleep 0.1; done && " - f"exec tail -f {new_index_dir_name}/{log_file_name}" + f"while true; do LOG_FILE=$(ls -t {log_file_glob} 2> /dev/null " + f'| head -1); [ -n "$LOG_FILE" ] && break; sleep 0.1; done && ' + f'exec tail -f "$LOG_FILE"' ) tail_proc = subprocess.Popen(tail_cmd, shell=True) - # Run the index rebuild command (and time it). + # Trigger the rebuild and wait for it to finish (the server keeps the + # HTTP request open until the new index has been swapped in, which + # can take from minutes to hours, depending on the size of the index). try: time_start = time.monotonic() try: @@ -386,123 +131,93 @@ def execute(self, args) -> bool: except Exception as e: log.error(f"Rebuilding the index failed: {e}") return False - time_end = time.monotonic() - duration_seconds = round(time_end - time_start) - log.info("") - rebuild_done_msg = f"Rebuilt index in {duration_seconds:,} seconds" - if new_index_dir_path == ".": - rebuild_done_msg += ( - f", in the new directory '{args.new_index_dir}'" - ) - log.info(rebuild_done_msg) + duration_seconds = round(time.monotonic() - time_start) finally: tail_proc.terminate() tail_proc.wait() - # Validate the new index before moving anything. - if not validate_index(args, new_index_dir_name): - log.error( - "The new index is dysfunctional, aborting. " - f'Files are in "{new_index_dir_name}"' + # Report the result, in particular, where the server moved the + # previous index (its directory name is derived from the build date of + # the previous index, so it is not known in advance). + log.info("") + log.info( + f"Rebuilt the index and swapped it in " + f"({duration_seconds:,} seconds)" + ) + try: + previous_index_dir = json.loads(response_body)[ + "previous-index-dir" + ] + log.info(f'The previous index is now in "{previous_index_dir}"') + except (json.JSONDecodeError, KeyError): + log.warning( + f"Could not parse the server response: {response_body}" ) - return False - # If requested, leave the new index in the temporary directory - # instead of moving it into place (same as the validation-failure - # path above, except that the index is fine). - if args.leave_new_index_in_temporary_directory: - log.info( - "Leaving the new index in the temporary directory as " - f'requested; files are in "{new_index_dir_name}"' + # If requested, send a simple test query to the server (which now + # serves the new index). + if args.test_query: + test_query_cmd = ( + f"curl -s {args.host_name}:{args.port}" + f' --data-urlencode "query=SELECT * WHERE' + f' {{ ?s ?p ?o }} LIMIT 1"' + f' -o /dev/null -w "%{{http_code}}"' ) - return True - - # Move the old index to the specified directory, if needed. - if move_old_index_when_done: try: - log.info(f"Moving the old index to {args.old_index_dir}") - run_command(move_old_index_cmd) - except Exception as e: - log.error(f"Moving the old index failed: {e}") - return False - - # Move the new index to the specified directory, if needed. - if move_new_index_when_done: - try: - log.info(f"Moving the new index to {args.new_index_dir}") - run_command(move_new_index_cmd) - except Exception as e: - log.error(f"Moving the new index failed: {e}") - return False - - # Restart the server with the new index, if requested. - if args.restart_when_finished: - try: - log.info("Restarting the server with the new index ...") - log.info("") - log.info(colored("Command: start", attrs=["bold"])) - log.info("") - run_command(restart_server_cmd, show_output=True) - except Exception as e: - log.error(f"Restarting the server failed: {e}") + query_ok = ( + run_command(test_query_cmd, return_output=True).strip() + == "200" + ) + except Exception: + query_ok = False + if query_ok: + log.info("Test query succeeded") + else: + log.error("Test query failed") return False # Clean up previous index directories according to - # `--keep-previous-index-dirs`. Find all subdirectories starting - # with `old_index_dir_basename`, ordered from oldest to newest - # (by creation time), and keep or delete them according to the - # specified policy. - if move_old_index_when_done: - old_index_dirs = sorted( + # `--keep-previous-index-dirs`. Find all subdirectories starting with + # `previous.`, ordered from oldest to newest (by creation time), and + # keep or delete them according to the specified policy. + if args.keep_previous_index_dirs != "all": + previous_index_dirs = sorted( [ dir for dir in Path(".").iterdir() - if dir.is_dir() - and dir.name.startswith(args.old_index_dir_basename) + if dir.is_dir() and dir.name.startswith("previous.") ], key=lambda dir: dir.stat().st_ctime, ) - if old_index_dirs: - policy = args.keep_previous_index_dirs - log.info("") - log.info( - colored( - f"Iterate over previous index directories (oldest" - f" to newest), and check which ones to keep or" - f" delete (keep_previous_index_dirs = {policy}):", - color="blue", - ) + policy = args.keep_previous_index_dirs + log.info("") + log.info( + colored( + f"Iterate over previous index directories (oldest" + f" to newest), and check which ones to keep or" + f" delete (keep_previous_index_dirs = {policy}):", + color="blue", ) - for i, dir in enumerate(old_index_dirs): - is_original = i == 0 - is_most_recent = i == len(old_index_dirs) - 1 - if policy == "all": - action = "KEEP" - elif policy == "none": - action = "DELETE" - elif policy == "original-only": - action = "KEEP" if is_original else "DELETE" - elif policy == "most-recent-only": - action = "KEEP" if is_most_recent else "DELETE" - elif policy == "original-and-most-recent": - action = ( - "KEEP" - if is_original or is_most_recent - else "DELETE" - ) - - log.info(f" {dir.name:<50} {action}") - - # Actually perform the deletion - if action == "DELETE": - try: - shutil.rmtree(dir) - log.info(f" → Deleted {dir.name}") - except Exception as e: - log.error( - f" → Failed to delete {dir.name}: {e}" - ) - - log.info("") + ) + log.info("") + for i, dir in enumerate(previous_index_dirs): + is_original = i == 0 + is_most_recent = i == len(previous_index_dirs) - 1 + if policy == "none": + action = "DELETE" + elif policy == "original-only": + action = "KEEP" if is_original else "DELETE" + elif policy == "most-recent-only": + action = "KEEP" if is_most_recent else "DELETE" + elif policy == "original-and-most-recent": + action = ( + "KEEP" if is_original or is_most_recent else "DELETE" + ) + log.info(f"{dir.name} -> {action}") + if action == "DELETE": + try: + shutil.rmtree(dir) + except Exception as e: + log.error(f"Failed to delete {dir.name}: {e}") return True