diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md index 2d2997d03..77a489829 100644 --- a/.github/pull_request_template.md +++ b/.github/pull_request_template.md @@ -46,13 +46,13 @@ Unit tests - some tests can't be ran on the CI. This is [documented](https://git Integration tests - the `daqsystemtest_integtest_bundle` requires a lot of resources, and connections to the EHN1 infrastructure. Check the [cross referenced list](https://github.com/DUNE-DAQ/drunc/wiki#users-with-access-to-clusters-for-running-daqsystemtest_integtest_bundlesh) if you can't run these. The developer needs to run at least the [.](https://github.com/DUNE-DAQ/daqsystemtest/blob/develop/integtest/minimal_system_quick_test.py) - Unit tests (`pytest --marker`) passed - - [ ] With relevant marker - - [ ] Without marker + - [ ] With relevant marker `_INSERT MAKER NAME HERE_` + - [ ] Relying on the CI workflow - Integration tests passed - [ ] Only `daqsystemtest_integtest_bundle.sh -k minimal_system_quick_test.py` - [ ] Full `daqsystemtest_integtest_bundle.sh` - [ ] Testing skipped as there are no core code changes in this PR, this only relates to documentation/CI workflows -- [ ] Drunc integration tests pass (`./scripts/drunc_integtest_bundle.sh`) +- [ ] Drunc integration tests pass (`dunedaq_integtest_bundle.sh -r drunc`) ## Final checklist prior to marking this as "Ready for Review" diff --git a/.github/workflows/run_pytest.yml b/.github/workflows/run_pytest.yml index 5a8dc81cc..9ff24e86b 100644 --- a/.github/workflows/run_pytest.yml +++ b/.github/workflows/run_pytest.yml @@ -1,10 +1,27 @@ name: Run pytest on: + # Run on any pushes to the main branch which releases are generated from + push: + branches: + - develop + - 'prep-release/**' + pull_request: + branches: + - develop + - 'prep-release/**' + # If taking a PR out of draft mode, run the workflow on the PR event as well + types: [opened, synchronize, reopened, ready_for_review] + +# Cancel any running jobs for this same branch if a new commit is pushed +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true jobs: install_and_test: + if: github.event.pull_request.draft == false runs-on: ubuntu-latest diff --git a/config/tests/deep-segments-config.data.xml b/config/drunc/deep-segments-config.data.xml similarity index 100% rename from config/tests/deep-segments-config.data.xml rename to config/drunc/deep-segments-config.data.xml diff --git a/config/drunc/failure-testing.data.xml b/config/drunc/failure-testing.data.xml new file mode 100644 index 000000000..0b7ca9fe0 --- /dev/null +++ b/config/drunc/failure-testing.data.xml @@ -0,0 +1,467 @@ + + + + + + + + + + + + + + + + + + + + + + + + +]> + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/config/tests/nestedConfig.data.xml b/config/drunc/nestedConfig.data.xml similarity index 62% rename from config/tests/nestedConfig.data.xml rename to config/drunc/nestedConfig.data.xml index c29497dc8..45518f89a 100644 --- a/config/tests/nestedConfig.data.xml +++ b/config/drunc/nestedConfig.data.xml @@ -71,29 +71,6 @@ - - - - - - - - - - - - - - - - - - - - - - - @@ -120,6 +97,16 @@ + + + + + + + + + + @@ -183,6 +170,7 @@ + @@ -206,7 +194,6 @@ - @@ -236,70 +223,4 @@ - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/config/tests/one-controller-config.data.xml b/config/drunc/one-controller-config.data.xml similarity index 100% rename from config/tests/one-controller-config.data.xml rename to config/drunc/one-controller-config.data.xml diff --git a/conftest.py b/conftest.py index f45c6c2fe..2966bdf83 100644 --- a/conftest.py +++ b/conftest.py @@ -4,6 +4,14 @@ """ +def pytest_configure(config): + """Block coverage reporting for integration tests.""" + if any("integtest" in str(arg) for arg in config.args): + plugin = config.pluginmanager.get_plugin("_cov") + if plugin: + plugin.options.no_cov = True + + def pytest_addoption(parser): """Register custom command-line options for pytest""" parser.addoption( diff --git a/pyproject.toml b/pyproject.toml index a5d6b26e5..b3f1cafe4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,17 +74,17 @@ source = ["drunc"] # * See https://docs.astral.sh/ruff/rules/ for details on Ruff's linting options [tool.ruff.lint] -ignore = [ - "E501", # Don't enforce line lengths within a linting context -] select = [ - "E", # pycodestyle errors "F", # check for errors using PyFlakes "I", # best practices for import calls - "UP", # suggestions for code modernization - "RUF", # build in Ruff warnings - "R", # refactoring suggestions ] +# "UP", # suggestions for code modernization +# "RUF", # build in Ruff warnings +# "R", # refactoring suggestions +# "E", # pycodestyle errors + +[tool.ruff.lint.per-file-ignores] +"**/integtest/*.py" = ["F401"] # Ignore unused imports in all integration test files, e.g. pytestmark which is used but not explicitly [tool.mypy] check_untyped_defs = true diff --git a/scripts/setup_drunc_config_path.sh b/scripts/setup_drunc_config_path.sh new file mode 100644 index 000000000..9c6667085 --- /dev/null +++ b/scripts/setup_drunc_config_path.sh @@ -0,0 +1 @@ +export DUNEDAQ_DB_PATH=$DUNEDAQ_DB_PATH:$DBT_AREA_ROOT"/pythoncode/drunc" \ No newline at end of file diff --git a/src/drunc/apps/fake_daqapp_rest.py b/src/drunc/apps/fake_daqapp_rest.py index 7a87dc7fc..5687f0f49 100644 --- a/src/drunc/apps/fake_daqapp_rest.py +++ b/src/drunc/apps/fake_daqapp_rest.py @@ -1,13 +1,20 @@ -"""This is a fake DAQ application that doesn't do anything, but should talk in the same way to the run control""" +""" +This is a fake DAQ application that doesn't do anything, but should talk in the same way +to the run control. + +It is primarily used for the testing of the Run Control. +""" -import argparse import copy as cp import os import random +import signal import threading import time from urllib.parse import urlparse +from wsgiref.simple_server import make_server +import click import conffwk import requests from flask import Flask, Response, request @@ -22,26 +29,64 @@ ) __version__ = "1.0.0" + +# Set up logging get_root_logger("info") -log = get_logger("fake_daqapp_rest", rich_handler=True) +log = get_logger("fake_daqapp_rest", stream_handlers=True, log_level="INFO") class AppState: + """ + Tracks state of the apps, and simulates the behaviour of daq_applicaitons with + stateful commands. + """ + def __init__(self, app_name: str): + """ + Initialize the app state. + + Args: + app_name (str): The name of the app, used for logging and in the responses + + Returns: + None + + Raises: + None + """ self.appname = app_name self.state = "INITIAL" self.executing_command = False - self.log = get_logger("fake_daqapp_rest.AppState") + self.log = get_logger("fake_daqapp_rest.AppState", log_level="INFO") def send_response_to_response_listener( self, address: str, txt: str, success: bool = True, data: dict = {} ): + """ + Send a response to the response listener. + + Args: + address (str): The address of the response listener + txt (str): The text to send in the response + success (bool): Whether the command was executed successfully or not + data (dict): Additional data to send in the response + + Returns: + None + + Raises: + None + """ + + # The response is sent as a POST request to the response listener, with the + # following contents in the body: data_to_send = { "success": success, "result": txt, "appname": self.appname, "data": data, } + self.log.info(f"Sending RESPONSE to {address}, data: {data_to_send}") try: response = requests.post( @@ -57,31 +102,68 @@ def send_response_to_response_listener( self.log.exception(e) def execute_command( - self, req_data, answer_port, answer_host, remote_host + self, + req_data: dict[str, str | dict], + answer_port: str, + answer_host: str | None, + remote_host: str, ) -> Response: - # The following block simulates a failure of the app while executing a stateful - # command. Thisserves uniquely to test the robustness of the Run Control when an - # app exits upon running an applciation, and should not be used for any other - # purpose. The environment variable is set in the configuration file that tests - # this behaviour. - if os.getenv("DRUNC_FAILURE_TESTING_CMD", None): - log.info("Simulating failure during initialization") - exit(1) - + """ + Execute a command received from the command facility. + + Args: + req_data (dict): The data received in the command, should contain at least + the following keys: + - id: The id of the command, used for logging and in the responses + - entry_state: The state the app should be in to execute the command, or + "*" to ignore the state + - exit_state: The state the app will be in after executing the command + - data: A dictionary with additional data for the command, can contain: + - execution-time: An integer with the time the command should take + to execute, in seconds + - seg_fault: An integer that if present will cause the app to exit + with that code + - throw: If present, the app will throw an exception instead of + executing the command + answer_port (str): The port to send the response to + answer_host (str | None): The host to send the response to, if None, the + remote_host will be used + remote_host (str): The host that sent the command, used for logging and as a + fallback for the answer_host + + Returns: + Response: A Flask response object with the result of the command execution + + Raises: + None + """ + self.log.debug("Received command with the following data:") + self.log.debug(f"{req_data=}") + self.log.debug(f"{answer_port=}") + self.log.debug(f"{answer_host=}") + self.log.debug(f"{remote_host=}") + + # Construct the address to send the response to reply_address = ( f"http://{answer_host}:{answer_port}/response" if answer_host else f"{remote_host}:{answer_port}/response" ) + # Extract the relevant information from the command data entry_state = req_data["entry_state"] exit_state = req_data["exit_state"] command_id = req_data["id"] data = req_data.get("data", {}) + # If the app is already executing a command, it should not execute another one. + # Send a response to the response listener indicating that it is busy if self.executing_command: - response_txt = "Already executing a command!!" - self.log.info(response_txt) + response_txt = "Already executing a command!" + self.log.info( + "Application is already executing a command, cannot execute another " + "one simultaneously." + ) self.send_response_to_response_listener( address=reply_address, txt=response_txt, @@ -89,12 +171,24 @@ def execute_command( ) return - time_spent = data.get("execution-time", random.randint(1, 5)) - - worries = random.randint(0, time_spent) - + # Determine the time the command should take to execute. If not specified in the + # data, it will be a random time between 1 and 5 seconds. We also determine a + # random time for the worries, which is the time the app will wait before + # failing the command in case of a seg_fault or throw, to simulate the time it + # takes for the app to fail after starting the execution of the command. + cmd_exec_time = data.get("execution-time", random.randint(1, 5)) + worries = random.randint(0, cmd_exec_time) + + # Validate that the app is in the correct state to execute the command. If not, + # send a response to the response listener indicating that the command cannot + # be executed due to the state of the app. The wildcard "*" can be used to + # indicate that the command can be executed in any state. if entry_state != "*" and self.state != entry_state.upper(): - info = f"DAQ Application is in state {self.state} and command {command_id} requires to be in state {entry_state.upper()} to execute. Not executing" + info = ( + f"DAQ Application is in state {self.state} and command {command_id} " + f"requires to be in state {entry_state.upper()} to execute. Not " + "executing." + ) self.log.info(info) self.send_response_to_response_listener( success=False, @@ -103,48 +197,103 @@ def execute_command( ) return + # Execute the command, and mark the app as busy executing a command to prevent + # concurrent executions. self.log.info(f"Executing {command_id}") - self.executing_command = True + # Failure testing through payload if data.get("seg_fault"): time.sleep(worries) - info = "" - self.log.info(info) + app_execution_info = "" + self.log.info(app_execution_info) self.send_response_to_response_listener( success=False, address=reply_address, - txt=info, + txt=app_execution_info, ) self.executing_command = False exit(data["seg_fault"]) if data.get("throw"): time.sleep(worries) - what = ( - "This is an eRrOr, YoU hAvE bEeN vErY nAuGhTy (aka task failed successfully)", + app_execution_info = ( + "This is an eRrOr, YoU hAvE bEeN vErY nAuGhTy (aka task failed " + "successfully)", ) - self.log.info(what) + self.log.info(app_execution_info) self.send_response_to_response_listener( success=False, address=reply_address, - txt=what, + txt=app_execution_info, ) self.executing_command = False return - print(f"Sleeping for {time_spent} seconds") + # FAILURE TESTING - CMD TIMEOUT + # For testing purposes, we can delay the execution of the command to simulate a + # long running command and test timeouts in the run control + ft_fsm_timeout = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT") + if ft_fsm_timeout: + ft_fsm_timeout = int(ft_fsm_timeout) + ft_fsm_timeout_cmd = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT_CMD") + ft_fsm_timeout_app_name = os.getenv("DRUNC_FT_FSM_CMD_TIMEOUT_APP_NAME") + if ( + ft_fsm_timeout + and ft_fsm_timeout_cmd == command_id + and ft_fsm_timeout_app_name == self.appname + ): + self.log.warning( + f"Delaying execution of {command_id} in {ft_fsm_timeout_app_name} by " + f"{ft_fsm_timeout} seconds" + ) + time.sleep(ft_fsm_timeout) + + # FAILURE TESTING - CMD PROCESS DEATH + # The following block simulates a failure of the app while executing a stateful + # command. Thisserves uniquely to test the robustness of the Run Control when an + # app exits upon running an applciation, and should not be used for any other + # purpose. + ft_fsm_death_cmd: bool = os.getenv("DRUNC_FT_FSM_CMD_DEATH_CMD", False) + if ft_fsm_death_cmd: + ft_fsm_death_cmd = ft_fsm_death_cmd.strip('"').strip("'") == req_data["id"] + self.log.debug(f"{ft_fsm_death_cmd=}") + + ft_fsm_death_app_name: bool = os.getenv( + "DRUNC_FT_FSM_CMD_DEATH_APP_NAME", False + ) + if ft_fsm_death_app_name: + ft_fsm_death_app_name = ( + ft_fsm_death_app_name.strip('"').strip("'") == self.appname + ) + self.log.debug(f"{ft_fsm_death_app_name=}") - time.sleep(time_spent) + if ft_fsm_death_cmd and ft_fsm_death_app_name: + self.log.debug("'Worries' sleeping prior to simulating process death") + time.sleep(worries) + self.log.warning( + f"Simulating death of {self.appname} during FSM cmd execution" + ) + # This requires a more agressive exit than sys.exit(), as this process is + # running in a separate thread. + os._exit(1) - info = f"Executed {command_id} successfully, after {time_spent} seconds" - self.log.info(info) + # "Execute" the command by sleeping for the determined time + self.log.info(f"Sleeping for {cmd_exec_time} seconds") + time.sleep(cmd_exec_time) + # Notify command success + app_execution_info = ( + f"Executed {command_id} successfully, after {cmd_exec_time} seconds" + ) + self.log.info(app_execution_info) self.send_response_to_response_listener( success=True, address=reply_address, - txt=info, + txt=app_execution_info, ) + + # Update app state, and mark as not busy self.state = exit_state.upper() self.executing_command = False return @@ -156,14 +305,40 @@ def execute_command( class AppCommand(Resource): + """ + Flask interface for the fake daq app. + + Receives the commands from the command facility and passes them to the AppState to + be executed, and sends the response back to the response listener. + """ + @classmethod - def pass_daq_app(cls, daq_app): + def pass_daq_app(cls, daq_app) -> type["AppCommand"]: + """ + Interface to pass the daq_app instance to the Flask resource, since Flask + doesn't allow to pass arguments to the resource constructor. + """ cls.daq_app = daq_app return cls - def post(self): + def post(self) -> (str, int): + """ + Endpoint to receive commands from the command facility. The command data should + be sent in a JSON format, with the following structure: + { + "id": "command_id", + "entry_state": "state the app should be in to execute the command, or *", + "exit_state": "state the app will be in after executing the command", + "data": { optional parameters: + "execution-time": "time the command should take to execute", + "seg_fault": "app will exit with this code to simulate a failure", + "throw": "app will throw an exception to simulate a failure" + } + } + """ global app_state + # Validate that the request contains JSON data try: data = request.get_json(force=True) except: @@ -171,6 +346,11 @@ def post(self): log = get_logger("fake_daqapp_rest.AppCommand") log.info(f"GET request with args: {data}") + + # Execute the command in a separate thread to not block the Flask app and to + # allow concurrent command executions, since the app can receive multiple + # commands while executing one command, and to allow the simulation of long + # running commands without blocking the Flask app. thread = threading.Thread( target=self.daq_app.execute_command, kwargs={ @@ -185,7 +365,31 @@ def post(self): return "Command received\n", 202 -def update_connectivity_service(name, connectivity_service, interval, url): +# Helper functions +def update_connectivity_service( + name: str, connectivity_service: ConnectivityServiceClient, interval: int, url: str +): + """ + Function to continuously update the connectivity service with the address of the + app, to simulate the behaviour of a real DAQ application that is continuously + publishing its address to the connectivity service. This is necessary for the Run + Control to be able to send commands to the app, since the Run Control gets the + address of the app from the connectivity service. The function runs in a separate + thread to not block the main thread of the app, which is running the Flask app to + receive commands from the command facility. + + Args: + name: The name of the publishing app + connectivity_service: the client to publish to the connectivity service + interval: Interval in seconds to update the connectivity service + url: The app address to publish to the connectivity service + + Returns: + None + + Raises: + None + """ while True: connectivity_service.publish( name + "_control", @@ -196,95 +400,136 @@ def update_connectivity_service(name, connectivity_service, interval, url): def index(): + """ + Endpoint to check if the app is running, can be used in the tests to wait for the + app to be ready before sending commands to it. + + Args: + None + + Returns: + str: A string indicating the app is running. + + Raises: + None + """ return f"Fake DAQ app v{__version__}" -def get_address_for_conn_srv(hostname): +def get_address(hostname: str): + """ + Gets a new address for the application, by finding an available port. + + Args: + hostname: The hostname to use in the address + + Returns: + str: URI with the given hostname and a new available port + + Raises: + None + """ return f"rest://{hostname}:{get_new_port()}" -def main(): +@click.command() +@click.option("-n", "--name", required=True, help="The name of the app in the response") +@click.option( + "-d", + "--configurationservice", + required=True, + help="This is a dummy argument in this case", +) +@click.option( + "-c", + "--commandfacility", + required=False, + help="Where the fake app should get its command from", +) +@click.option( + "-i", + "--informationservice", + default="stdout://flat", + help="This is a dummy argument in this case", +) +@click.option( + "-l", "--log_level", default="info", help="Logging level minimum threshold" +) +@click.option( + "-p", + "--partition", + default="global", + help="This is a dummy argument in this case", +) +@click.option("-s", "--session", default="test", help="name of session") +@click.option("-k", "--configurationid", default="test-config", help="ID of session") +def main( + name: str, + configurationservice: str, + commandfacility: str, + informationservice: str, + log_level: str, + partition: str, + session: str, + configurationid: str, +) -> None: # The following block simulates a failure during the initialization of the app. This # serves uniquely to test the robustness of the Run Control when an app fails to # initialize, and should not be used for any other purpose. The environment variable # is set in the configuration file that tests this behaviour. - if os.getenv("DRUNC_FAILURE_TESTING_INIT", None): - log.info("Simulating failure during initialization") + if os.getenv("DRUNC_PROCESS_DEATH_ON_BOOT", None): + log.info("Sleeping to allow intiialization timeout") + time.sleep(20) + log.warning("Simulating failure during initialization") exit(1) - - parser = argparse.ArgumentParser( - prog="FakeApplication", - description="This is a fake application that communicate in the same way with the RunControl as the DAQApplication (thru REST)", - ) - parser.add_argument( - "-n", "--name", required=True, help="The name of the app in the response" - ) - parser.add_argument( - "-d", - "--configurationService", - required=True, - help="This is a dummy argument in this case", - ) - parser.add_argument( - "-c", - "--commandFacility", - required=False, - help="Where the fake app should get its command from", - ) - parser.add_argument( - "-i", - "--informationService", - default="stdout://flat", - help="This is a dummy argument in this case", - ) - parser.add_argument( - "-l", "--log_level", default="info", help="Logging level minimum threshold" - ) - parser.add_argument( - "-p", - "--partition", - default="global", - help="This is a dummy argument in this case", - ) - parser.add_argument("-s", "--session", default="test", help="name of session") - parser.add_argument( - "-k", "--configurationID", default="test-config", help="ID of session" - ) - - args = parser.parse_args() - - name = args.name - print(f"Name: {name}") + log.info(f"Starting application {name}") app_state = AppState(name) - conf = conffwk.Configuration(args.configurationService) - session = conf.get_dal( + # Set up and parse configuration + conf = conffwk.Configuration(configurationservice) + session_dal = conf.get_dal( class_name="Session", - uid=args.configurationID, + uid=configurationid, ) connectivity_service_address = ( - session.connectivity_service.host + session_dal.connectivity_service.host + ":" - + str(session.connectivity_service.service.port) + + str(session_dal.connectivity_service.service.port) ) - if not args.commandFacility: - log.error("No command facility passed, exiting") + + # Validate command facility argument + if not commandfacility: + log.critical("No command facility passed, exiting") exit(1) - url = urlparse(resolve_localhost_and_127_ip_to_network_ip(args.commandFacility)) + # Resolve the command facility URL and validate the scheme + url = urlparse(resolve_localhost_and_127_ip_to_network_ip(commandfacility)) if url.scheme != "rest": log.exception("DAQApplication communication scheme must be rest") exit(1) log.debug(f"Initializing fake_daq_application with address {url}") if url.port == 0: - url = get_address_for_conn_srv(url.hostname) + url = get_address(url.hostname) log.info(f"Communication address is {url}") interval = 2 + # FAILURE TESTING - DEATH ON BOOT + # The following block simulates a failure on initialization of the app. This + # serves uniquely to test the robustness of the Run Control when an app fails to + # complete initialization, and should not be used for any other purpose. The + # environment variable is set in the configuration file that tests this behaviour. + ft_die_on_boot: bool = ( + os.getenv("DRUNC_FT_PROCESS_DEATH_ON_BOOT", "false").lower() == "true" + ) + ft_app_to_die_boot: str = os.getenv("DRUNC_FT_PROCESS_DEATH_BOOT_APP_NAME", None) + if ft_die_on_boot and ft_app_to_die_boot == name: + log.warning(f"Simulating death of {name} on boot") + exit(1) + connectivity_service = ConnectivityServiceClient( - session=args.session, + session=session, address=connectivity_service_address, ) @@ -294,35 +539,83 @@ def main(): name="connectivity_service_updating_thread", ) - # Doesn't do what is expected, probably flask - # def terminate(signum, sigframe): - # connectivity_service_thread.join() - # log.info("Connectivity service terminated") - # exit(1) - # for sig in [signal.SIGINT, signal.SIGHUP, signal.SIGTERM, signal.SIGQUIT]: - # signal.signal(sig, terminate) + def terminate(*args): # Accept args for signal handlers + for s in [signal.SIGTERM, signal.SIGQUIT]: + if signal.getsignal(s) in args: + log.warning(f"Received termination signal {s}, shutting down {name}...") + log.info(f"Terminating application {name}...") + shutdown_event.set() + + # 2. Close connections explicitly + if "server" in server_container: + try: + server_container["server"].server_close() + except: + pass + + # 3. Give threads a tiny buffer to stop gracefully + time.sleep(0.1) + + # 4. Final hard exit + log.info("Shutdown complete. Exiting.") + os._exit(1) + + def terminate_signal_process(signum, sigframe): + log.warning(f"Received signal {signum}, terminating process") + terminate() + + for sig in [signal.SIGTERM, signal.SIGQUIT]: + signal.signal(sig, terminate) app = Flask(__name__) api = Api(app) DAQAppCMD = AppCommand.pass_daq_app(app_state) api.add_resource(DAQAppCMD, "/command", methods=["POST"]) app.add_url_rule("/", "index", index) + server_ready = threading.Event() + shutdown_event = threading.Event() + + def run_flask_app(app, host, port, event, server_container): + server = make_server(host, port, app) + server.timeout = 0.5 + event.set() + + # Don't use serve_forever() directly if you need external control + # Use a loop that checks the shutdown event + while not shutdown_event.is_set(): + server.handle_request() # Handles one request at a time + + server.shutdown() + server.server_close() url = urlparse(url) flask_url = url.geturl().replace("rest://", "http://") - log.info(f"Starting FakeDAQ app on {flask_url}") + server_container = {} flask_thread = threading.Thread( - target=app.run, - kwargs={"host": url.hostname, "port": url.port, "debug": False}, + target=run_flask_app, + kwargs={ + "app": app, + "host": url.hostname, + "port": url.port, + "event": server_ready, + "server_container": server_container, + }, name="flask_thread", + daemon=True, # Ensure the thread exits when the main program exits ) - flask_thread.start() + if not server_ready.wait(timeout=10): + log.error("Timed out waiting for FakeDAQ app to start") + exit(1) + + time.sleep(1) for i in range(10): + log.debug(f"Trying to connect to Flask app, attempt {i + 1}/10") response = requests.get(flask_url + "/") - log.info(f"Response: {response.status_code}") + log.debug(f"Response: {response.status_code}") if response.status_code == 200: + log.info("Fake DAQ app started successfully and is responding to requests") break if i == 9: log.error("Failed to start fake DAQ app") @@ -331,13 +624,21 @@ def main(): connectivity_service_thread.start() + # FAILURE TESTING LOGIC BLOCK - DEATH POST BOOT # The following block simulates a failure after the initialization of the app. This # serves uniquely to test the robustness of the Run Control when an app fails to # complete initialization, and should not be used for any other purpose. The # environment variable is set in the configuration file that tests this behaviour. - if os.getenv("DRUNC_FAILURE_TESTING_POST_BOOT", None): - log.info("Simulating failure after initialization") - exit(1) + ft_die_post_boot: bool = ( + os.getenv("DRUNC_FT_PROCESS_DEATH_POST_BOOT", "false").lower() == "true" + ) + if ft_die_post_boot and ft_app_to_die_boot == name: + log.warning(f"Simulating death of {name} post boot") + terminate() + + log.info( + "Fake DAQ application is running and publishing to connectivity service. Press Ctrl+C to exit." + ) if __name__ == "__main__": diff --git a/src/drunc/controller/children_interface/child_node.py b/src/drunc/controller/children_interface/child_node.py index 5a6891461..21b5e55c1 100644 --- a/src/drunc/controller/children_interface/child_node.py +++ b/src/drunc/controller/children_interface/child_node.py @@ -1,5 +1,6 @@ from abc import ABC, abstractmethod +from druncschema.common_pb2 import LogOnServerResponse from druncschema.controller_pb2 import ( DescribeFSMResponse, DescribeResponse, @@ -155,3 +156,14 @@ def to_error( execute_on_all_subsequent_children_in_path: bool = True, ) -> ToErrorResponse: pass + + @abstractmethod + def log_on_server( + self, + text: str, + severity: str = "INFO", + target: str = "", + execute_along_path: bool = False, + execute_on_all_subsequent_children_in_path: bool = True, + ) -> LogOnServerResponse: + pass diff --git a/src/drunc/controller/children_interface/grpc_child.py b/src/drunc/controller/children_interface/grpc_child.py index ee216e573..9c177c460 100644 --- a/src/drunc/controller/children_interface/grpc_child.py +++ b/src/drunc/controller/children_interface/grpc_child.py @@ -3,6 +3,7 @@ from typing import NoReturn, cast import grpc +from druncschema.common_pb2 import LogOnServerRequest, LogOnServerResponse from druncschema.controller_pb2 import ( DescribeFSMRequest, DescribeFSMResponse, @@ -133,7 +134,9 @@ def _setup_connection(self): time.sleep(5) else: - self.log.info(f"Connected to the controller ({self.uri})!") + self.log.info( + f"Application {self.name} connected to the parent application ({self.uri})!" + ) break def _attempt_reconnection(self, retry_call): @@ -606,3 +609,48 @@ def handle_child_grpc_error(self, error: grpc.RpcError) -> NoReturn: self.log.error(text) raise error + + def log_on_server( + self, + text: str, + severity: str = "INFO", + target: str = "", + execute_along_path: bool = False, + execute_on_all_subsequent_children_in_path: bool = True, + ) -> LogOnServerResponse: + """ + Log a message on the server with the specified severity. + + Args: + text (str): The message to log. + severity (str): The severity level of the log message (default: "INFO"). + target (str): The target for the log message (default: ""). + execute_along_path (bool): Whether to execute along the path (default: False). + execute_on_all_subsequent_children_in_path (bool): Whether to execute on all subsequent children in the path (default: True). + + Returns: + LogOnServerResponse: The response from the server after logging the message. + """ + request = LogOnServerRequest( + token=None, + text=text, + severity=severity, + target=target, + execute_along_path=execute_along_path, + execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path, + ) + + try: + response = self.stub.log_on_server(request) + except grpc.RpcError as e: + try: + self.handle_child_grpc_error(e) + except ServerUnreachable: + self.log.info( + f"Connection to {self.name} at {self.uri} failed during log_on_server, attempting to reconnect..." + ) + response = self._attempt_reconnection( + lambda: self.stub.log_on_server(request) + ) + + return response diff --git a/src/drunc/controller/children_interface/rest_api_child.py b/src/drunc/controller/children_interface/rest_api_child.py index aa28a5951..02df44be5 100644 --- a/src/drunc/controller/children_interface/rest_api_child.py +++ b/src/drunc/controller/children_interface/rest_api_child.py @@ -9,6 +9,7 @@ import requests import socks +from druncschema.common_pb2 import LogOnServerResponse from druncschema.controller_pb2 import ( DescribeFSMResponse, DescribeResponse, @@ -855,3 +856,34 @@ def to_error( name=self.name, flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) + + def log_on_server( + self, + text: str, + severity: str = "INFO", + target: str = "", + execute_along_path: bool = False, + execute_on_all_subsequent_children_in_path: bool = True, + ) -> LogOnServerResponse: + """ + Log a message on the server with the specified severity. + + Right now, this is not implemented in the daq_applications. It is unlikely that + this will be implemented, thus this simply returns the correctly formatted + message with a NOT_EXECUTED_NOT_IMPLEMENTED flag. + + Args: + text (str): The message to log. + severity (str): The severity level of the log message (default: "INFO"). + target (str): The target for the log message (default: ""). + execute_along_path (bool): Whether to execute along the path (default: False). + execute_on_all_subsequent_children_in_path (bool): Whether to execute on all subsequent children in the path (default: True). + + Returns: + LogOnServerResponse: The response from the server after logging the message. + """ + return LogOnServerResponse( + token=None, + name=self.name, + flag=ResponseFlag.NOT_EXECUTED_NOT_IMPLEMENTED, + ) diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index de3ac41c2..cf950dfd0 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -6,6 +6,7 @@ from daqpytools.logging import LogHandlerConf, setup_daq_ers_logger from druncschema.authoriser_pb2 import ActionType, SystemType +from druncschema.common_pb2 import LogOnServerRequest, LogOnServerResponse from druncschema.controller_pb2 import ( DescribeFSMRequest, DescribeFSMResponse, @@ -65,6 +66,7 @@ DotDruncJsonNotFound, ) from drunc.fsm.utils import convert_fsm_transition +from drunc.utils.grpc_utils import ServerTimeout from drunc.utils.utils import get_logger T = TypeVar("T") @@ -222,6 +224,9 @@ def init_controller(self) -> None: for response in child_responses: children_states[response.name] = response.status.state if response.status.in_error: + self.log.error( + f"Child {response.name} is in error state. Placing controller in error state." + ) self.stateful_node.to_error() if any([c.lower() != "initial" for c in children_states.values()]): @@ -232,7 +237,8 @@ def init_controller(self) -> None: bad_children = [k for k, v in children_states.items() if v.lower() != "initial"] if bad_children: log_init_controller.error( - f"Children that did not initialise in time: {bad_children}" + f"Children that did not initialise in time: [red]{', '.join(bad_children).rstrip(', ')}[/]. Placing " + "controller in error state." ) self.stateful_node.to_error() @@ -819,13 +825,14 @@ def execute_fsm_command( flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) + # Parse and validate target. try: - # Parse and validate target. request.target = self.parse_target_string(request.target) except ValueError: response.flag = ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT return response + # Extract command information. command = request.command command_name = command.command_name @@ -857,6 +864,7 @@ def execute_fsm_command( ) ) + # Extract FSM transition. transition = self.stateful_node.get_fsm_transition(command_name) self.log.debug(f"FSM transition: {transition}") @@ -893,7 +901,7 @@ def execute_fsm_command( response.fsm_flag = FSMResponseFlag.FSM_INVALID_TRANSITION return response - # This node. + # Execute FSM transition on this node. if request.target == self.name or request.execute_along_path: fsm_args = self.stateful_node.decode_fsm_arguments(command) fsm_data = self.stateful_node.prepare_transition( @@ -937,17 +945,23 @@ def execute_fsm_command( child_command = FSMCommand() child_command.CopyFrom(command) child_command.data = fsm_data - child_responses = self.propagate_concurrently( - lambda child, target: child.execute_fsm_command( - child_command, - target, - request.execute_along_path, - request.execute_on_all_subsequent_children_in_path, - ), - child_list, - indices=connected_indices, - ) - response.children.extend(child_responses) + try: + child_responses = self.propagate_concurrently( + lambda child, target: child.execute_fsm_command( + child_command, + target, + request.execute_along_path, + request.execute_on_all_subsequent_children_in_path, + ), + child_list, + indices=connected_indices, + ) + response.children.extend(child_responses) + except ServerTimeout as e: + response.fsm_flag = FSMResponseFlag.FSM_FAILED + self.stateful_node.to_error() + self.log.error(f"FSM command '{command_name}' failed: {e}") + return response # Finish propagating FSM transition to children. self.stateful_node.finish_propagating_transition_mark(transition) @@ -1526,7 +1540,6 @@ def to_error( name=self.name, flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) - try: # Parse and validate target. request.target = self.parse_target_string(request.target) @@ -1568,5 +1581,86 @@ def to_error( # This node. if request.target == self.name or request.execute_along_path: self.stateful_node.to_error() + self.log.critical( + f"Error state for this node: {self.stateful_node.node_is_in_error()}" + ) + + self.log.critical(f"Returning to_error response: {response}") + return response + + @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) + @publish_command_time + def log_on_server( + self, + request: LogOnServerRequest, + context: ServicerContext, + ) -> LogOnServerResponse: + """ + Logs a message on the server with the specified severity level. + + Args: + request (LogOnServerRequest): The request containing the log message, severity level, and target information. + context (ServicerContext): The gRPC context for the request. + + Returns: + LogOnServerResponse: The response indicating the result of the logging operation. + + Raises: + None + """ + response = LogOnServerResponse( + token=None, + flag=ResponseFlag.EXECUTED_SUCCESSFULLY, + ) + + try: + # Parse and validate target. + request.target = self.parse_target_string(request.target) + except ValueError: + response.flag = ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT + return response + + # This node. + if request.target == self.name or request.execute_along_path: + request.target = "" + + # Children nodes (ignore exclusion). + child_list = self.address_target_path( + request.target, + request.execute_on_all_subsequent_children_in_path, + include_excluded_nodes=True, + ) + connected_indices, disconnected_indices = self._partition_connected_children( + child_list, + operation_name="who_is_in_charge", + ) + child_responses = self.propagate_concurrently( + lambda child, target: child.log_on_server( + request.text, + request.severity, + request.target, + request.execute_along_path, + request.execute_on_all_subsequent_children_in_path, + ), + child_list, + indices=connected_indices, + ) + child_responses.extend( + [ + LogOnServerResponse( + token=None, + name=child_list[i][0].name, + flag=ResponseFlag.NOT_EXECUTED_NOT_READY, + ) + for i in disconnected_indices + ] + ) + response.children.extend(child_responses) + + # This node. + if request.target in [self.name, ""] or request.execute_along_path: + level = request.severity.lower() + log_method = getattr(self.log, level, self.log.info) + log_method(request.text) return response diff --git a/src/drunc/controller/controller_driver.py b/src/drunc/controller/controller_driver.py index 376530660..6b242575c 100644 --- a/src/drunc/controller/controller_driver.py +++ b/src/drunc/controller/controller_driver.py @@ -2,6 +2,7 @@ import socket import grpc +from druncschema.common_pb2 import LogOnServerRequest, LogOnServerResponse from druncschema.controller_pb2 import ( DescribeFSMRequest, DescribeFSMResponse, @@ -356,6 +357,47 @@ def to_error( return response + def log_on_server( + self, + text: str, + severity: str = "INFO", + target: str = "", + execute_along_path: bool = False, + execute_on_all_subsequent_children_in_path: bool = True, + timeout: int | float = 60, + ) -> LogOnServerResponse: + """ + Logs a message to the server's log system. + + Args: + text (str): The message to log. + target (str, optional): The target node for the log message. Defaults to "". + execute_along_path (bool, optional): Whether to execute the log command along the path. Defaults to False. + execute_on_all_subsequent_children_in_path (bool, optional): Whether to execute the log command on all subsequent children in the path. Defaults to True. + timeout (int | float, optional): The timeout for the gRPC request in seconds. Defaults to 60. + + Returns: + None + + Raises: + grpc.RpcError: If the gRPC request fails. + """ + request = LogOnServerRequest( + token=self.token, + text=text, + severity=severity, + target=target, + execute_along_path=execute_along_path, + execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path, + ) + request.token.CopyFrom(self.token) + try: + response = self.stub.log_on_server(request, timeout=timeout) + except grpc.RpcError as e: + handle_grpc_error(e) + + return response + def handle_response(self, response, command, outformat): dr = DecodedResponse( name=response.name, diff --git a/src/drunc/controller/interface/commands.py b/src/drunc/controller/interface/commands.py index 26f60972b..570dbca56 100644 --- a/src/drunc/controller/interface/commands.py +++ b/src/drunc/controller/interface/commands.py @@ -84,6 +84,12 @@ def status( execute_on_all_subsequent_children_in_path: bool, extended: bool, ) -> None: + log_msg = ( + f"Getting status for target '{target}'..." + if target + else "Getting status for all targets..." + ) + obj.log.info(log_msg) obj.print( render_status_table( obj, @@ -255,6 +261,39 @@ def echo(obj, text: str | None) -> None: log_echo.info(text or "") +@click.command("log") +@click.argument("text", required=True) +@click.option("--target", type=str, help="The target to address", default="") +@click.option( + "--execute-along-path/--dont-execute-along-path", + is_flag=True, + show_default=True, + help="Execute the command along the path", + default=False, +) +@click.option( + "--execute-on-all-subsequent-children-in-path/--dont-execute-on-all-subsequent-children-in-path", + is_flag=True, + show_default=True, + help="Execute the command on all subsequent children in the path", + default=False, +) +@click.pass_obj +def log_on_server( + obj: ControllerContext, + text: str, + target: str, + execute_along_path: bool, + execute_on_all_subsequent_children_in_path: bool, +) -> None: + obj.get_driver("controller").log_on_server( + text=text, + target=target, + execute_along_path=execute_along_path, + execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path, + ) + + @click.command("who-is-in-charge") @click.option("--target", type=str, help="The target to address", default="") @click.option( diff --git a/src/drunc/controller/interface/shell_utils.py b/src/drunc/controller/interface/shell_utils.py index ec5529baa..f30890449 100644 --- a/src/drunc/controller/interface/shell_utils.py +++ b/src/drunc/controller/interface/shell_utils.py @@ -39,6 +39,7 @@ from rich.table import Table from drunc.controller.interface.context import ControllerContext +from drunc.controller.utils import get_all_apps_with_named_substate from drunc.exceptions import DruncSetupException, DruncShellException from drunc.unified_shell.context import UnifiedShellContext, UnifiedShellMode from drunc.utils.grpc_utils import ( @@ -590,6 +591,7 @@ def run_one_fsm_command( if ( obj.running_mode in [UnifiedShellMode.BATCH, UnifiedShellMode.SEMIBATCH] and obj.get_driver("controller").status().status.in_error + and not obj.no_stop_error_batch_mode ): obj.get_driver("controller").status() log.error( @@ -690,14 +692,43 @@ class DummyCommand: str(ae) ) # TODO: Manually raise exception, see if the str declaration is needed with rich handling return - except ServerTimeout as e: - log.error(e) + except ServerTimeout: log.error( "The command timed out, unfortunately this means the server is in undefined state, and [red]your best option at this stage is to [bold]terminate[/bold] and [bold]boot[/bold][/]." ) - log.error( - "Alternatively, if you are patient, you can try to wait a bit longer and send [yellow]'status'[/yellow] to check if the command ends up being executed (you may want to check the logs of the controller and application with the [yellow]'logs'[/yellow] command)." + # The following line is outdated, but in the future when error states and their + # recovery are better defined, we can provide better options to the user. + # log.error( + # "Alternatively, if you are patient, you can try to wait a bit longer and send [yellow]'status'[/yellow] to check if the command ends up being executed (you may want to check the logs of the controller and application with the [yellow]'logs'[/yellow] command)." + # ) + + # Mark the controller as in error state, so that if the user tries to run + # another command, it will be prevented, and they will be encouraged to check + # the error application logs + status_response = obj.get_driver("controller").status() + apps_that_timed_out = get_all_apps_with_named_substate( + status_response, "executing_cmd" + ) + apps_that_timed_out_str = ", ".join(apps_that_timed_out) + err_str = ( + "The session did not complete the stateful transition in the specified " + f"time of {timeout} seconds. To investigate the cause, [yellow]check the " + f"logs of {apps_that_timed_out_str} with the logs command[/] as:" + ) + log.error(err_str) + for app in apps_that_timed_out: + log.error(f"\t[yellow]logs -n {app}[/]") + obj.get_driver("controller").log_on_server(err_str, severity="ERROR") + obj.get_driver("controller").to_error( + execute_on_all_subsequent_children_in_path=False ) + + statuses = obj.get_driver("controller").status() + descriptions = obj.get_driver("controller").describe() + t = get_status_table(statuses, descriptions) + obj.print(t) + obj.print_status_summary() + return if not result: diff --git a/src/drunc/controller/utils.py b/src/drunc/controller/utils.py index 82970e6ea..75bf90962 100644 --- a/src/drunc/controller/utils.py +++ b/src/drunc/controller/utils.py @@ -1,7 +1,7 @@ import time from dataclasses import dataclass -from druncschema.controller_pb2 import RunInfo, Status +from druncschema.controller_pb2 import RunInfo, Status, StatusResponse from drunc.utils.utils import get_logger @@ -46,6 +46,70 @@ def get_status_message(controller): return msg +def count_processes_in_status_response(response: StatusResponse) -> int: + """ + Count the number of processes in the status table, including all children. + + This function is recursive to allow for the counting of processes following a nested + structure of StatusResponse objects through the `child` attribute. + + Args: + response (StatusResponse): The StatusResponse object returrned from a controller + servicer status request. + + Returns: + int: The total number of processes in the status table. + + Raises: + None + """ + processes_found = 0 + + # 1. Count the processes in the current node + if response.status: + processes_found += 1 + + for child in response.children: + processes_found += count_processes_in_status_response(child) + + return processes_found + + +def get_all_states(response: StatusResponse) -> list[str]: + """ + Recursively extracts 'state' from StatusResponse and its children. + """ + states = [] + + # 1. Get the state of the current node + if response.status: + states.append(response.status.state) + + # 2. Recurse through all children + for child in response.children: + states.extend(get_all_states(child)) + + return states + + +def get_all_apps_with_named_substate( + response: StatusResponse, substate_query: str +) -> list[str]: + """ + Recursively searches for app names with a specific substate in StatusResponse and its children. + """ + matching_apps = [] + + if response.status and response.status.sub_state == substate_query: + if response.name: + matching_apps.append(response.name) + + for child in response.children: + matching_apps.extend(get_all_apps_with_named_substate(child, substate_query)) + + return matching_apps + + def get_detector_name(configuration) -> str: detector_name = None log = get_logger("controller.core.get_detector_name") diff --git a/src/drunc/integtest/failure_mode_death_on_boot_nest_app_test.py b/src/drunc/integtest/failure_mode_death_on_boot_nest_app_test.py new file mode 100644 index 000000000..3cffa5bd3 --- /dev/null +++ b/src/drunc/integtest/failure_mode_death_on_boot_nest_app_test.py @@ -0,0 +1,195 @@ +""" +Run a session with a nested segment application dying at the start of boot. +Check that the application is correctly reported as dead, and that the session is in an +error state after boot. The application that dies is ft-nested-segment-2-application. +""" + +import os +import re + +import integrationtest.data_classes as data_classes +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-death-on-boot-nest-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate the drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_death_on_boot_nest_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps -w 300 + +echo status-post-boot +status +""".split() + +dead_app_name = "ft-nested-segment-2-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_boot_failure_logfile(run_dunerc) -> None: + """ + Checks that the application that dies has a logfile, and that the defined logfile + contains the expected message indicating that the application simulated dying on + boot, prior to registering itself on the connectivity service. + """ + + # Retrieve the log file for the application that is configured to die on boot + simulated_death_app_logfile = next( + (log for log in run_dunerc.log_files if dead_app_name in str(log)), + None, + ) + + assert simulated_death_app_logfile is not None, ( + f"Expected to find a log file for {dead_app_name}, but did not." + ) + + # Check that the expected boot failure message is in the log file for the + # application that dies on boot + app_death_str = [f"Simulating death of {dead_app_name} on boot"] + line_found = check_file_containing(app_death_str, simulated_death_app_logfile) + assert line_found == True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + +def test_process_dead_in_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is not present in the ps table after + boot. + """ + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + # Get the ps table + ps_table = get_ps_table_after_echo(lines, "ps-post-boot") + + # Format the entry rows into a parsable list of dicts, and check that the dead + # application is not present in the ps + ps_table_dead_app_entry = get_rows_by_friendly_name_from_ps_table( + ps_table, dead_app_name + ) + + # Check that the dead application is present in the ps table + assert ps_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + + # Check that the app that simulated death is in fact dead in the ps table + aliveness_state = ps_table_dead_app_entry[0]["alive"] + assert aliveness_state == "False", ( + f"Expected to see {dead_app_name} marked as dead in the ps table, but it was not." + ) + + +def test_boot_failure_cli(run_dunerc) -> None: + """ + Checks that the application that dies on boot causes the session to go into an error + state, and that the expected message is printed to stdout. + """ + # Get the stdout and format it + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Check the stdout for the expected error message. + search_str = "Booted, but the session is in an error state." + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + # Check the status table for the root controller error state + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + + # Get the entry for the root controller, and make sure it exists + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_boot, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check that the root controller is in an error state + assert root_controller_row[0]["in_error"] == "Yes", ( + "Expected root controller to be in error, but it is not." + ) diff --git a/src/drunc/integtest/failure_mode_death_on_boot_top_app_test.py b/src/drunc/integtest/failure_mode_death_on_boot_top_app_test.py new file mode 100644 index 000000000..e2601196c --- /dev/null +++ b/src/drunc/integtest/failure_mode_death_on_boot_top_app_test.py @@ -0,0 +1,203 @@ +""" +Run a session with the top segment application dying at the start of boot. +Check that the application is correctly reported as dead, and that the session is in an +error state after boot. The application that dies is ft-top-segment-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-death-on-boot-top-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_death_on_boot_top_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps -w 140 + +echo status-post-boot +status +""".split() + +dead_app_name = "ft-top-segment-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_boot_failure_logfile(run_dunerc) -> None: + """ + Checks that the application that dies has a logfile, and that the defined logfile + contains the expected message indicating that the application simulated dying on + boot, prior to registering itself on the connectivity service. + """ + + # Retrieve the log file for the application that is configured to die on boot + simulated_death_app_logfile = next( + ( + log + for log in run_dunerc.log_files + if "ft-top-segment-application" in str(log) + ), + None, + ) + + assert simulated_death_app_logfile is not None, ( + "Expected to find a log file for ft-top-segment-application, but did not." + ) + + # Check that the expected boot failure message is in the log file for the + # application that dies on boot + app_death_str = [f"Simulating death of {dead_app_name} on boot"] + line_found = check_file_containing(app_death_str, simulated_death_app_logfile) + assert line_found == True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + +def test_process_dead_in_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is not present in the ps table after + boot. + """ + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table + ps_table = get_ps_table_after_echo(lines, "ps-post-boot") + + # Format the entry rows into a parsable list of dicts, and check that the dead + # application is not present in the ps + ps_table_dead_app_entry = get_rows_by_friendly_name_from_ps_table( + ps_table, dead_app_name + ) + + # Check that the dead application is present in the ps table + assert ps_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + + # Check that the app that simulated death is in fact dead in the ps table + aliveness_state = ps_table_dead_app_entry[0]["alive"] + assert aliveness_state == "False", ( + f"Expected to see {dead_app_name} marked as dead in the ps table, but it was not." + ) + + +def test_boot_failure_cli(run_dunerc) -> None: + """ + Checks that the application that dies on boot causes the session to go into an error + state, and that the expected message is printed to stdout. + """ + # Get the stdout and format it + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Check the stdout for the expected error message. + search_str = "Booted, but the session is in an error state." + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + # Check the status table for the root controller error state + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + + # Get the entry for the root controller, and make sure it exists + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_boot, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check that the root controller is in an error state + assert root_controller_row[0]["in_error"] == "Yes", ( + "Expected root controller to be in error, but it is not." + ) diff --git a/src/drunc/integtest/failure_mode_death_post_boot_nest_app_test.py b/src/drunc/integtest/failure_mode_death_post_boot_nest_app_test.py new file mode 100644 index 000000000..29db74e7e --- /dev/null +++ b/src/drunc/integtest/failure_mode_death_post_boot_nest_app_test.py @@ -0,0 +1,246 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the application is marked as dead in the ps table, and disconnected in the +status table, and that the session is in an error state. The application that dies is +ft-nested-segment-2-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-death-post-boot-nest-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_death_post_boot_nest_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps + +echo status-post-boot +status +""".split() + +dead_app_name = "ft-nested-segment-2-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_boot_failure_logfile(run_dunerc) -> None: + """ + Checks that the application that dies has a logfile, and that the defined logfile + contains the expected message indicating that the application simulated dying on + boot, prior to registering itself on the connectivity service. + """ + + # Retrieve the log file for the application that is configured to die on boot + simulated_death_app_logfile = next( + (log for log in run_dunerc.log_files if dead_app_name in str(log)), + None, + ) + + assert simulated_death_app_logfile is not None, ( + f"Expected to find a log file for {dead_app_name}, but did not." + ) + + # Check that the expected boot failure message is in the log file for the + # application that dies on boot + app_death_str = [f"Simulating death of {dead_app_name} post boot"] + line_found = check_file_containing(app_death_str, simulated_death_app_logfile) + assert line_found == True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + +def test_expected_log_message_in_terminal(run_dunerc) -> None: + """ + Checks that the expected message indicating that the application died on boot is + printed to stdout, as a summary after the controller check is complete. + """ + # Get and format the stdout + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Check that the expected boot failure message is in stdout for the application that + # dies on boot + search_str = "Booted, but there are disconnected applications/controllers." + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the misaligned process count record in stdout, but did not." + ) + + +def test_process_dead_in_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is not present in the ps table after + boot. + """ + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table + ps_table = get_ps_table_after_echo(lines, "ps-post-boot") + + # Format the entry rows into a parsable list of dicts, and check that the dead + # application is not present in the ps + ps_table_dead_app_entry = get_rows_by_friendly_name_from_ps_table( + ps_table, dead_app_name + ) + + # Check that the dead application is present in the ps table + assert ps_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + + # Check that the app that simulated death is in fact dead in the ps table + aliveness_state = ps_table_dead_app_entry[0]["alive"] + assert aliveness_state == "False", ( + f"Expected to see {dead_app_name} marked as dead in the ps table, but it was not." + ) + + +def test_process_disconnected_in_status_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is marked with a disconnected status + in the status table after boot. + """ + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the status table post boot + status_table = get_status_table_after_echo(lines, "status-post-boot") + + # Check that the dead application is present in the status table + status_table_dead_app_entry = get_rows_by_name_from_status_table( + status_table, dead_app_name + ) + assert status_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + + # Check that the app that simulated death is in fact marked as disconnected in the status table + status_table_state = status_table_dead_app_entry[0]["state"] + assert status_table_state == "disconnected", ( + f"Expected to see {dead_app_name} marked with state 'disconnected' in the status table, but it was not." + ) + status_table_state = status_table_dead_app_entry[0]["substate"] + assert status_table_state == "disconnected", ( + f"Expected to see {dead_app_name} marked with substate 'disconnected' in the status table, but it was not." + ) + + +def test_boot_failure_cli(run_dunerc) -> None: + """ + Checks that the application that dies on boot causes the session to go into an error + state, and that the expected message is printed to stdout. + """ + # Get the stdout and format it + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Check the stdout for the expected error message. + search_str = "Booted, but the session is in an error state." + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + # Check the status table for the root controller error state + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + + # Get the entry for the root controller, and make sure it exists + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_boot, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check that the root controller is in an error state + assert root_controller_row[0]["in_error"] == "Yes", ( + "Expected root controller to be in error, but it is not." + ) diff --git a/src/drunc/integtest/failure_mode_death_post_boot_top_app_test.py b/src/drunc/integtest/failure_mode_death_post_boot_top_app_test.py new file mode 100644 index 000000000..9534e0532 --- /dev/null +++ b/src/drunc/integtest/failure_mode_death_post_boot_top_app_test.py @@ -0,0 +1,241 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the application is marked as dead in the ps table, and disconnected in the +status table, and that the session is in an error state. The application that dies is +ft-top-segment-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-death-post-boot-top-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_death_post_boot_top_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps + +echo status-post-boot +status +""".split() + + +def test_dunerc_success(run_dunerc) -> None: + """Checks that the drunc integration command sequence completes successfully.""" + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected process log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_boot_failure_logfile(run_dunerc) -> None: + """ + Checks that the application that dies has a logfile, and that the defined logfile + contains the expected message indicating that the application died on boot. + """ + # Retrieve the log file for the application that is configured to die on boot + simulated_death_app_logfile = next( + ( + log + for log in run_dunerc.log_files + if "ft-top-segment-application" in str(log) + ), + None, + ) + assert simulated_death_app_logfile is not None, ( + "Expected to find a log file for ft-top-segment-application, but did not." + ) + + # Check that the expected boot failure message is in the log file for the + # application that dies on boot + app_death_str = ["Simulating death of ft-top-segment-application post boot"] + line_found = check_file_containing(app_death_str, simulated_death_app_logfile) + assert line_found == True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + +def test_expected_log_message_in_terminal(run_dunerc) -> None: + """ + Checks that the expected message indicating that the application died on boot is + printed to stdout. + """ + # Check that the expected boot failure message is in stdout for the application that + # dies on boot + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + search_str = "Booted, but there are disconnected applications/controllers." + + str_found = any(search_str in line for line in lines) + assert str_found, ( + "Expected to see the misaligned process count record in stdout, but did not." + ) + + +def test_process_dead_in_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is not present in the ps table after + boot. + """ + # Check that the application that dies on boot is not present in the ps table after + # boot. + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + ps_table = get_ps_table_after_echo(lines, "ps-post-boot") + dead_app_name = "ft-top-segment-application" + ps_table_dead_app_entry = get_rows_by_friendly_name_from_ps_table( + ps_table, dead_app_name + ) + assert ps_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + assert dead_app_name not in ps_table, ( + f"Expected to see {dead_app_name} missing from the ps table, but it was found." + ) + aliveness_state = ps_table_dead_app_entry[0]["alive"] + assert aliveness_state == "False", ( + f"Expected to see {dead_app_name} marked as dead in the ps table, but it was not." + ) + + +def test_process_disconnected_in_status_table(run_dunerc) -> None: + """ + Checks that the application that dies on boot is marked with a disconnected status + in the status table after boot. + """ + # Check that the application that dies on boot is not present in the ps table after + # boot. + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + status_table = get_status_table_after_echo(lines, "status-post-boot") + dead_app_name = "ft-top-segment-application" + status_table_dead_app_entry = get_rows_by_name_from_status_table( + status_table, dead_app_name + ) + assert status_table_dead_app_entry, ( + f"Expected to see {dead_app_name} in the ps table, but it was not found" + ) + assert dead_app_name not in status_table, ( + f"Expected to see {dead_app_name} missing from the ps table, but it was found." + ) + status_table_state = status_table_dead_app_entry[0]["state"] + assert status_table_state == "disconnected", ( + f"Expected to see {dead_app_name} marked with state 'disconnected' in the status table, but it was not." + ) + status_table_state = status_table_dead_app_entry[0]["substate"] + assert status_table_state == "disconnected", ( + f"Expected to see {dead_app_name} marked with substate 'disconnected' in the status table, but it was not." + ) + + +def test_boot_failure_cli(run_dunerc) -> None: + """ + Checks that the application that dies on boot causes the session to go into an error + state, and that the expected message is printed to stdout. + """ + # Check that the session is correctly put in error state if an appliucation dies on + # boot. + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + search_str = "Booted, but the session is in an error state." + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the boot failure message in stdout, but did not." + ) + + +def test_fsm_in_error_status_table(run_dunerc) -> None: + """ + Checks that the session FSM is in an error state after boot if an application dies on + boot. + """ + # Check that the session FSM is correctly put in error state if an appliucation dies + # on boot. + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + status_table = get_status_table_after_echo(lines, "status-post-boot") + root_controller_status = get_rows_by_name_from_status_table( + status_table, "ft-root-controller" + ) + assert root_controller_status, ( + "Expected to see the ft-root-controller in the status table, but it was not found" + ) + error_state = root_controller_status[0]["in_error"] + assert error_state == "Yes", ( + "Expected to see the session FSM marked as in error in the status table, but it was not." + ) diff --git a/src/drunc/integtest/failure_mode_fsm_cmd_death_nest_app_test.py b/src/drunc/integtest/failure_mode_fsm_cmd_death_nest_app_test.py new file mode 100644 index 000000000..b4ed78537 --- /dev/null +++ b/src/drunc/integtest/failure_mode_fsm_cmd_death_nest_app_test.py @@ -0,0 +1,271 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the session goes into an error state, and that the expected messages are +printed to stdout and logged to the application log file. The application that dies is +ft-nested-segment-2-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-fsm-cmd-death-nest-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_fsm_cmd_death_nest_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps + +echo status-post-boot +status + +echo pre-conf +conf + +echo status-post-conf +status + +echo ps-post-conf +ps +""".split() + +dead_app_name = "ft-nested-segment-2-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_all_apps_alive_and_no_initial_error(run_dunerc) -> None: + """Checks that all expected applications are alive after boot, and that no errors are encountered.""" + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table + ps_table_post_boot = get_ps_table_after_echo(lines, "ps-post-boot") + assert ps_table_post_boot, "Expected ps table after boot, but did not find it." + + # Check that all expected applications are alive after boot + alive_processes = [ + row["friendly_name"] for row in ps_table_post_boot if row["alive"] == "True" + ] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + assert app_name in alive_processes, ( + f"Expected {app_name} to be alive after boot, but it was not." + ) + + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + assert status_table_post_boot, ( + "Expected status table after boot, but did not find it." + ) + + # Check that the session is not in an error state after boot + all_application_error_state_query = [ + app["in_error"] for app in status_table_post_boot + ] + assert all(state == "No" for state in all_application_error_state_query), ( + "Expected all applications to not be in error state after boot, but found some in error state." + ) + + +def test_fsm_cmd_application_death_log_file(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution logs the expected message to its log file. + """ + # Get the dead application name log file + simulated_death_app_logfile = next( + (log for log in run_dunerc.log_files if dead_app_name in str(log)), + None, + ) + assert simulated_death_app_logfile is not None, ( + "Expected to find a log file for ft-nested-segment-2-application, but did not." + ) + + # Check the logfile for the entry about the simulated death of the application during FSM command execution + search_str = ( + "Simulating death of ft-nested-segment-2-application during FSM cmd execution" + ) + lines = strip_ansi(simulated_death_app_logfile.read_text()).splitlines() + + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the error message in stdout, but did not." + ) + + +def test_fsm_cmd_application_death_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution is marked as dead in the ps table. + """ + # Get the ps table after the fsm cmd execution + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + ps_table_post_conf = get_ps_table_after_echo(lines, "ps-post-conf") + assert ps_table_post_conf, "Expected ps table after conf, but did not find it." + + # Get te row for the dead application + dead_app_ps_row = get_rows_by_friendly_name_from_ps_table( + ps_table_post_conf, dead_app_name + ) + assert dead_app_ps_row, ( + f"Expected to find a row for the application {dead_app_name} in the ps table, but did not." + ) + + # Ensure the app is marked as dead in the ps table + assert dead_app_ps_row[0]["alive"] == "False", ( + f"Expected application {dead_app_name} to be dead after fsm cmd execution, but found it alive." + ) + + +def test_session_in_error_cli(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution causes the session + to go into an error state, and that the expected message is printed to stdout. + """ + # Get the status table after the command execution + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + + status_table_post_conf = get_status_table_after_echo(lines, "status-post-conf") + + # Get the root controller row + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check the substate never made it to the target state. + assert root_controller_row[0]["substate"] == "propagating-conf", ( + f"Expected root controller substate to be 'propagating-conf', but found '{root_controller_row[0]['substate']}'." + ) + + # Check the state of a segment controller which does not time out reaches the target state + nested_segment_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-controller" + ) + assert nested_segment_controller_row, ( + "Expected to find a row for the nested segment controller in the status table, but did not." + ) + assert nested_segment_controller_row[0]["substate"] == "configured", ( + f"Expected nested segment controller state to be 'configured', but found '{nested_segment_controller_row[0]['state']}'." + ) + + # Check the state of a segment application which does not time out reaches the target state + nested_segment_application_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-application" + ) + assert nested_segment_application_row, ( + "Expected to find a row for the nested segment application in the status table, but did not." + ) + assert nested_segment_application_row[0]["substate"] == "idle", ( + f"Expected nested segment application state to be 'idle', but found '{nested_segment_application_row[0]['state']}'." + ) + + # Check the stdout for the cmd timeout message + expected_timeout_message = "The command timed out," + assert expected_timeout_message in stdout, ( + "Expected to find the timeout message in stdout, but did not." + ) + + # Checked that the error state is explicitly logged + search_str = "FSM is in error" + lines = strip_ansi(stdout).splitlines() + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the FSM error report message in stdout, but did not." + ) diff --git a/src/drunc/integtest/failure_mode_fsm_cmd_death_top_app_test.py b/src/drunc/integtest/failure_mode_fsm_cmd_death_top_app_test.py new file mode 100644 index 000000000..2d3f35835 --- /dev/null +++ b/src/drunc/integtest/failure_mode_fsm_cmd_death_top_app_test.py @@ -0,0 +1,271 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the session goes into an error state, and that the expected messages are +printed to stdout and logged to the log files. The application that dies is +ft-top-segment-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + get_ps_table_after_echo, + get_rows_by_friendly_name_from_ps_table, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-fsm-cmd-death-top-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_fsm_cmd_death_top_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps + +echo status-post-boot +status + +echo pre-conf +conf + +echo status-post-conf +status + +echo ps-post-conf +ps +""".split() + +dead_app_name = "ft-top-segment-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_all_apps_alive_and_no_initial_error(run_dunerc) -> None: + """Checks that all expected session applications are alive after boot in both the ps and status tables.""" + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table after boot + ps_table_post_boot = get_ps_table_after_echo(lines, "ps-post-boot") + assert ps_table_post_boot, "Expected ps table after boot, but did not find it." + + # Check that all expected applications are alive after boot + alive_processes = [ + row["friendly_name"] for row in ps_table_post_boot if row["alive"] == "True" + ] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert app_name in alive_processes, ( + f"Expected {app_name} to be alive after boot, but it was not." + ) + + # Get the status table after boot + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + assert status_table_post_boot, ( + "Expected status table after boot, but did not find it." + ) + + # Check that the session is not in an error state after boot + all_application_error_state_query = [ + app["in_error"] for app in status_table_post_boot + ] + assert all(state == "No" for state in all_application_error_state_query), ( + "Expected all applications to not be in error state after boot, but found some in error state." + ) + + +def test_fsm_cmd_application_death_log_file(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution logs the expected message to its log file. + """ + # Get the dead application name log file + simulated_death_app_logfile = next( + (log for log in run_dunerc.log_files if dead_app_name in str(log)), + None, + ) + assert simulated_death_app_logfile is not None, ( + "Expected to find a log file for ft-top-segment-application, but did not." + ) + + # Check the logfile for the entry about the simulated death of the application during FSM command execution + search_str = ( + "Simulating death of ft-top-segment-application during FSM cmd execution" + ) + lines = strip_ansi(simulated_death_app_logfile.read_text()).splitlines() + + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the error message in stdout, but did not." + ) + + +def test_fsm_cmd_application_death_ps_table(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution is marked as dead in the ps table. + """ + # Get the ps table after the fsm cmd execution + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + ps_table_post_conf = get_ps_table_after_echo(lines, "ps-post-conf") + assert ps_table_post_conf, "Expected ps table after conf, but did not find it." + + # Get te row for the dead application + dead_app_ps_row = get_rows_by_friendly_name_from_ps_table( + ps_table_post_conf, dead_app_name + ) + assert dead_app_ps_row, ( + f"Expected to find a row for the application {dead_app_name} in the ps table, but did not." + ) + + # Ensure the app is marked as dead in the ps table + assert dead_app_ps_row[0]["alive"] == "False", ( + f"Expected application {dead_app_name} to be dead after fsm cmd execution, but found it alive." + ) + + +def test_session_in_error_cli(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution causes the session + to go into an error state, and that the expected message is printed to stdout. + """ + # Get the status table after the command execution + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + status_table_post_conf = get_status_table_after_echo(lines, "status-post-conf") + + # Get the root controller row + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check the substate never made it to the target state. + assert root_controller_row[0]["substate"] == "propagating-conf", ( + f"Expected root controller substate to be 'propagating-conf', but found '{root_controller_row[0]['substate']}'." + ) + + # Check the state of a segment controller which does not time out reaches the target state + nested_segment_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-controller" + ) + assert nested_segment_controller_row, ( + "Expected to find a row for the nested segment controller in the status table, but did not." + ) + assert nested_segment_controller_row[0]["substate"] == "configured", ( + f"Expected nested segment controller state to be 'configured', but found '{nested_segment_controller_row[0]['state']}'." + ) + + # Check the state of a segment application which does not time out reaches the target state + nested_segment_application_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-application" + ) + assert nested_segment_application_row, ( + "Expected to find a row for the nested segment application in the status table, but did not." + ) + assert nested_segment_application_row[0]["substate"] == "idle", ( + f"Expected nested segment application state to be 'idle', but found '{nested_segment_application_row[0]['state']}'." + ) + + # Check the stdout for the cmd timeout message + expected_timeout_message = "The command timed out," + assert expected_timeout_message in stdout, ( + "Expected to find the timeout message in stdout, but did not." + ) + + # Checked that the error state is explicitly logged + search_str = "FSM is in error" + lines = strip_ansi(stdout).splitlines() + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the FSM error report message in stdout, but did not." + ) diff --git a/src/drunc/integtest/failure_mode_fsm_cmd_timeout_nest_app_test.py b/src/drunc/integtest/failure_mode_fsm_cmd_timeout_nest_app_test.py new file mode 100644 index 000000000..2a15bfef3 --- /dev/null +++ b/src/drunc/integtest/failure_mode_fsm_cmd_timeout_nest_app_test.py @@ -0,0 +1,260 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the session goes into an error state, and that the expected messages are +printed to stdout. The application that dies is ft-nested-segment-2-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-fsm-cmd-timeout-nest-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_fsm_cmd_timeout_nest_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps -w 200 + +echo status-post-boot +status + +echo pre-conf +conf + +echo status-post-conf +status +""".split() + +timeout_app_name = "ft-nested-segment-2-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_all_apps_alive_and_no_initial_error(run_dunerc) -> None: + """Checks that all expected applications are alive after boot.""" + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table + ps_table_post_boot = get_ps_table_after_echo(lines, "ps-post-boot") + assert ps_table_post_boot, "Expected ps table after boot, but did not find it." + + # Check that all expected applications are alive after boot + alive_processes = [ + row["friendly_name"] for row in ps_table_post_boot if row["alive"] == "True" + ] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert app_name in alive_processes, ( + f"Expected {app_name} to be alive after boot, but it was not." + ) + + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + assert status_table_post_boot, ( + "Expected status table after boot, but did not find it." + ) + + # Check that the session is not in an error state after boot + all_application_error_state_query = [ + app["in_error"] for app in status_table_post_boot + ] + assert all(state == "No" for state in all_application_error_state_query), ( + "Expected all applications to not be in error state after boot, but found some in error state." + ) + + +def test_fsm_cmd_timeout_logfile(run_dunerc) -> None: + """ + Checks that the application that times out on stateful command execution has a + logfile, and that the defined logfile contains the expected message indicating that + the stateful command induced delay is being simulated. + """ + # Retrieve the log file for the application that is configured to timeout on fsm command execution + simulated_fsm_cmd_delay_logfile = next( + (log for log in run_dunerc.log_files if timeout_app_name in str(log)), + None, + ) + assert simulated_fsm_cmd_delay_logfile is not None, ( + f"Expected to find a log file for {timeout_app_name}, but did not." + ) + + # Check that the expected delay message is present in the log file + fsm_cmd_delay_str = [ + f"Delaying execution of conf in {timeout_app_name} by 100 seconds" + ] + line_found = check_file_containing( + fsm_cmd_delay_str, simulated_fsm_cmd_delay_logfile + ) + assert line_found == True, ( + "Expected to see the fsm conf delay message in stdout, but did not." + ) + + +def test_session_in_error_cli(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution causes the session + to go into an error state, and that the expected message is printed to stdout. + """ + # Get the status table shown during the command execution + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + status_table_post_conf = get_status_table_after_echo(lines, "status-post-conf") + + # Get the root contorller row in the status table + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check that the root controller did not reach the target state + assert root_controller_row[0]["substate"] == "propagating-conf", ( + f"Expected root controller substate to be 'propagating-conf', but found '{root_controller_row[0]['substate']}'." + ) + + # Check the state of a segment controller which does not time out reaches the target state + nested_segment_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-controller" + ) + assert nested_segment_controller_row, ( + "Expected to find a row for the nested segment controller in the status table, but did not." + ) + assert nested_segment_controller_row[0]["substate"] == "configured", ( + f"Expected nested segment controller state to be 'configured', but found '{nested_segment_controller_row[0]['state']}'." + ) + + # Check the state of a segment application which does not time out reaches the target state + nested_segment_application_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-application" + ) + assert nested_segment_application_row, ( + "Expected to find a row for the nested segment application in the status table, but did not." + ) + assert nested_segment_application_row[0]["substate"] == "idle", ( + f"Expected nested segment application state to be 'idle', but found '{nested_segment_application_row[0]['state']}'." + ) + + # Check the stdout for the cmd timeout message + expected_timeout_message = "The command timed out," + assert expected_timeout_message in stdout, ( + "Expected to find the timeout message in stdout, but did not." + ) + + # Checked that this is explicitly logged too + search_str = "FSM is in error" + lines = strip_ansi(stdout).splitlines() + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the FSM error report message in stdout, but did not." + ) + + +def test_suggestion_to_check_logs_is_present(run_dunerc) -> None: + """ + Checks that the suggestion to check the log files is present in stdout. + """ + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + + expected_suggestion = f"logs -n {timeout_app_name}" + suggestion_found = any(expected_suggestion in line for line in lines) + assert suggestion_found, ( + "Expected to find the suggestion to check log files in stdout, but did not." + ) diff --git a/src/drunc/integtest/failure_mode_fsm_cmd_timeout_top_app_test.py b/src/drunc/integtest/failure_mode_fsm_cmd_timeout_top_app_test.py new file mode 100644 index 000000000..346442a02 --- /dev/null +++ b/src/drunc/integtest/failure_mode_fsm_cmd_timeout_top_app_test.py @@ -0,0 +1,261 @@ +""" +Run a session with a nested segment application dying at the end of boot. +Check that the session goes into an error state, and that the expected messages are +printed to stdout. The application that dies is ft-top-segment-application. +""" + +import os +import re + +# from datetime import datetime +import integrationtest.data_classes as data_classes + +# import integrationtest.log_file_checks as log_file_checks +from integ_test_utils import ( + check_file_containing, + get_ps_table_after_echo, + get_rows_by_name_from_status_table, + get_status_table_after_echo, + require_drunc, + strip_ansi, +) + +pytestmark = require_drunc + + +pytest_plugins = "integrationtest.integrationtest_drunc" + +check_for_logfile_errors = True + +ignored_logfile_problems = { + "-controller": [ + "Worker with pid \\d+ was terminated due to signal", + "Connection '.*' not found on the application registry", + ], + "connectivity-service": [ + "errorlog: -", + ], +} + +# Point to the drunc config file for this test +conf_dict = data_classes.integtest_params_for_predefined_dunedaq_config() +conf_dict.predefined_config_db = "config/drunc/failure-testing.data.xml" +conf_dict.config_session_name = "ft-fsm-cmd-timeout-top-app" +conf_dict.dunerc_cmd_args = ["--no-stop-error-batch-mode"] + +# Define the operational environment for this test +conf_dict.op_env = "test" + +# Connectivity service configuration +# Allow drunc to manage ConnectivityService (default is False, integrationtest manages +# the Connectivity Service) +conf_dict.drunc_connsvc = True +# Specify connectivity service port (default is 0, a random port is chosen for the +# Connectivity Service) +# conf_dict.connsvc_port = 12345 + +# Collate tthe drunc config arguments into a dict to pass to the fixture +confgen_arguments = {"test_failure_mode_fsm_cmd_timeout_top_app": conf_dict} + +# Run these commands in the run control +dunerc_command_list = """ +boot + +echo ps-post-boot +ps -w 200 + +echo status-post-boot +status + +echo pre-conf +conf + +echo status-post-conf +status +""".split() + +timeout_app_name = "ft-top-segment-application" + + +def test_dunerc_success(run_dunerc) -> None: + """ + Checks that the drunc integration command sequence completes successfully without + any unexpected failures. + """ + + # print the name of the current test + current_test = os.environ.get("PYTEST_CURRENT_TEST") + match_obj = re.search(r".*\[(.+)-run_.*rc.*\d].*", current_test) + if match_obj: + current_test = match_obj.group(1) + + banner_line = re.sub(".", "=", current_test) + print(banner_line) + print(current_test) + print(banner_line) + + # Check that dunerc completed correctly + assert run_dunerc.completed_process.returncode == 0 + + +def test_log_files_are_present(run_dunerc) -> None: + """Checks that expected session log files exist.""" + generated_log_files = [str(log.name) for log in run_dunerc.log_files] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert any( + f"{run_dunerc.daq_session_name}_{app_name}" in logfile + for logfile in generated_log_files + ) + + +def test_all_apps_alive_and_no_initial_error(run_dunerc) -> None: + """Checks that all expected applications are alive after boot.""" + lines = strip_ansi(run_dunerc.completed_process.stdout).splitlines() + + # Get the ps table + ps_table_post_boot = get_ps_table_after_echo(lines, "ps-post-boot") + assert ps_table_post_boot, "Expected ps table after boot, but did not find it." + + # Check that all expected applications are alive after boot + alive_processes = [ + row["friendly_name"] for row in ps_table_post_boot if row["alive"] == "True" + ] + for app_name in [ + "ft-root-controller", + "ft-top-segment-controller", + "ft-nested-segment-1-controller", + "ft-nested-segment-1-application", + "ft-nested-segment-2-controller", + "ft-nested-segment-2-application", + "ft-nested-segment-2.1-application", + "ft-top-segment-application", + ]: + print(f"Checking for log file for {app_name}...") + assert app_name in alive_processes, ( + f"Expected {app_name} to be alive after boot, but it was not." + ) + + # Get the status table + status_table_post_boot = get_status_table_after_echo(lines, "status-post-boot") + assert status_table_post_boot, ( + "Expected status table after boot, but did not find it." + ) + + # Check that the session is not in an error state after boot + all_application_error_state_query = [ + app["in_error"] for app in status_table_post_boot + ] + assert all(state == "No" for state in all_application_error_state_query), ( + "Expected all applications to not be in error state after boot, but found some in error state." + ) + + +def test_fsm_cmd_timeout_logfile(run_dunerc) -> None: + """ + Checks that the application that times out on stateful command execution has a + logfile, and that the defined logfile contains the expected message indicating that + the stateful command induced delay is being simulated. + """ + # Retrieve the log file for the application that is configured to timeout on fsm command execution + simulated_fsm_cmd_delay_logfile = next( + (log for log in run_dunerc.log_files if timeout_app_name in str(log)), + None, + ) + assert simulated_fsm_cmd_delay_logfile is not None, ( + f"Expected to find a log file for {timeout_app_name}, but did not." + ) + + # Check that the expected delay message is present in the log file + fsm_cmd_delay_str = [ + f"Delaying execution of conf in {timeout_app_name} by 100 seconds" + ] + line_found = check_file_containing( + fsm_cmd_delay_str, simulated_fsm_cmd_delay_logfile + ) + assert line_found == True, ( + "Expected to see the fsm conf delay message in stdout, but did not." + ) + + +def test_session_in_error_cli(run_dunerc) -> None: + """ + Checks that the application that dies on fsm cmd execution causes the session + to go into an error state, and that the expected message is printed to stdout. + """ + # Get the status table shown during the command execution + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + status_table_post_conf = get_status_table_after_echo(lines, "status-post-conf") + + # Get the root contorller row in the status table + root_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-root-controller" + ) + assert root_controller_row, ( + "Expected to find a row for the root controller in the status table, but did not." + ) + + # Check that the root controller did not reach the target state + assert root_controller_row[0]["substate"] == "propagating-conf", ( + f"Expected root controller substate to be 'propagating-conf', but found '{root_controller_row[0]['substate']}'." + ) + + # Check the state of a segment controller which does not time out reaches the target state + nested_segment_controller_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-controller" + ) + assert nested_segment_controller_row, ( + "Expected to find a row for the nested segment controller in the status table, but did not." + ) + assert nested_segment_controller_row[0]["substate"] == "configured", ( + f"Expected nested segment controller state to be 'configured', but found '{nested_segment_controller_row[0]['state']}'." + ) + + # Check the state of a segment application which does not time out reaches the target state + nested_segment_application_row = get_rows_by_name_from_status_table( + status_table_post_conf, "ft-nested-segment-1-application" + ) + assert nested_segment_application_row, ( + "Expected to find a row for the nested segment application in the status table, but did not." + ) + assert nested_segment_application_row[0]["substate"] == "idle", ( + f"Expected nested segment application state to be 'idle', but found '{nested_segment_application_row[0]['state']}'." + ) + + # Check the stdout for the cmd timeout message + expected_timeout_message = "The command timed out," + assert expected_timeout_message in stdout, ( + "Expected to find the timeout message in stdout, but did not." + ) + + # Checked that this is explicitly logged too + search_str = "FSM is in error" + lines = strip_ansi(stdout).splitlines() + str_found = any(search_str in line for line in lines) + assert str_found is True, ( + "Expected to see the FSM error report message in stdout, but did not." + ) + + +def test_suggestion_to_check_logs_is_present(run_dunerc) -> None: + """ + Checks that the suggestion to check the log files is present in stdout. + """ + stdout = run_dunerc.completed_process.stdout + lines = strip_ansi(stdout).splitlines() + + expected_suggestion = f"logs -n {timeout_app_name}" + suggestion_found = any(expected_suggestion in line for line in lines) + assert suggestion_found, ( + "Expected to find the suggestion to check log files in stdout, but did not." + ) diff --git a/src/drunc/integtest/integ_test_utils.py b/src/drunc/integtest/integ_test_utils.py index 570190569..cf29029f8 100644 --- a/src/drunc/integtest/integ_test_utils.py +++ b/src/drunc/integtest/integ_test_utils.py @@ -14,11 +14,32 @@ reported through `assert` with context-rich messages. """ +import os import re from collections.abc import Callable +from pathlib import PosixPath + +import pytest ANSI_ESCAPE_RE = re.compile(r"\x1B\[[0-9;]*[A-Za-z]") +# Define a regex for parsing UUIDs from the ps table in the drunc logsx +UUID_RE = re.compile( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" +) + +# For the failure mode testing, reuqire drunc to be a part of DUNEDAQ_DB_PATH +db_path_env = os.getenv("DUNEDAQ_DB_PATH", "") +drunc_missing = not any( + "drunc" == segment for path in db_path_env.split(":") for segment in path.split("/") +) + +# Define the exportable marker +require_drunc = pytest.mark.skipif( + drunc_missing, + reason="drunc is not present in DUNEDAQ_DB_PATH, skipping drunc integration tests", +) + def strip_ansi(text: str) -> str: """Remove ANSI escape codes from a text block.""" @@ -78,6 +99,46 @@ def require_line_index( return line_idx +def check_file_containing( + lines: list[str], + file: PosixPath, +) -> bool: + """ + For each line in `lines`, check if the file contains the line. + + Example: + >>> file = PosixPath("test_file.txt") + >>> file.write_text("Hello\\nWorld\\n") + >>> check_file_containing(["Hello", "World"], file) + True + >>> check_file_containing(["Hello", "Missing"], file) + False + + Args: + lines: List of strings to check for presence in the file. + file: Path to the file to read and check against the lines. + + Returns: + True if all lines are found in the file, False otherwise. + + Raises: + None. + """ + # Read in the file and split it by lines + try: + file_lines = file.read_text(encoding="utf-8").splitlines() + except FileNotFoundError: + print(f"Error: {file} not found.") + return False + + # Check if the passed strings are present in the file lines + for target in lines: + if not any(target in file_line for file_line in file_lines): + return False + + return True + + def require_line_containing( lines: list[str], text: str, @@ -284,8 +345,34 @@ def get_status_table_after_echo( If no status table is found after the marker, returns an empty list. + The table is structured as + | Name | Info | State | Substate | In error | Included | Endpoint | + + Example: + >>> stdout = ( + ... "[2026/03/17 10:48:15 UTC] INFO drunc.echo test_status_marker\n" + ... "unit-test status\n" + ... "│ root-controller │ │ initial │ initial │ No | Yes │ grpc://np04-srv-029.cern.ch:30006 │\n" + ... "└" + ... ) + >>> table = get_status_table_after_echo(stdout, "test_status_marker") + >>> expected_row = { + ... "Name": "root-controller", + ... "Info": "", + ... "State": "initial", + ... "Substate": "initial", + ... "In error": "No", + ... "Included": "Yes", + ... "Endpoint": "grpc://np04-srv-029.cern.ch:30006", + ... } + >>> table[0] == expected_row + True + Returns: Parsed rows with keys: name, info, state, substate, in_error, included, endpoint. + + Raises: + None """ return _get_table_after_echo(lines, echo_marker, "status", _STATUS_COLUMNS) @@ -327,11 +414,39 @@ def get_column_for_friendly_name( ) -def get_rows_for_friendly_name( +#! Replace this with a generic one +def get_rows_from_table( + table: list[dict[str, str]], column: str, value: str +) -> list[dict[str, str]]: + """ + Return all rows whose `column` matches `value`exactly after stripping. + + Args: + table: List of dictionaries representing the table rows. + column: The column name to match against. + value: The value to match in the specified column. + + Returns: + List of dictionaries representing the matching rows. + + Raises: + KeyError: If the specified column does not exist in the table rows. + """ + return [row for row in table if row[column].strip() == value] + + +def get_rows_by_friendly_name_from_ps_table( ps_table: list[dict[str, str]], friendly_name: str ) -> list[dict[str, str]]: """Return all rows whose `friendly_name` matches exactly after stripping.""" - return [row for row in ps_table if row["friendly_name"].strip() == friendly_name] + return get_rows_from_table(ps_table, "friendly_name", friendly_name) + + +def get_rows_by_name_from_status_table( + status_table: list[dict[str, str]], name: str +) -> list[dict[str, str]]: + """Return all rows whose `Name` matches exactly after stripping.""" + return get_rows_from_table(status_table, "name", name) def assert_process_presence( @@ -374,7 +489,7 @@ def assert_process_presence( ... expected_present=False, ... ) """ - matching_rows = get_rows_for_friendly_name(ps_table, friendly_name) + matching_rows = get_rows_by_friendly_name_from_ps_table(ps_table, friendly_name) if expected_present: assert matching_rows, ( diff --git a/src/drunc/process_manager/interface/commands.py b/src/drunc/process_manager/interface/commands.py index 643cbdd61..6cc8e5341 100644 --- a/src/drunc/process_manager/interface/commands.py +++ b/src/drunc/process_manager/interface/commands.py @@ -390,8 +390,6 @@ def ps_impl( getattr(query, "session", None) or getattr(obj, "session_name", None) or "" ) title = f"Processes running{f' in session {session_name}' if session_name else ''}" - log_msg = f"No processes running{f' in session [green]{session_name}[/]' if session_name else ''}" - # If there are processes running, tabulate them, otherwise log that there are no # processes running. if results.values: @@ -406,4 +404,21 @@ def ps_impl( soft_wrap=True, ) else: - log.info(log_msg) + log.info(f"No processes running in session [green]{obj.session_name}[/]") + + +@click.command("log") +@click.argument("text", required=True) +@click.option( + "-s", + "--severity", + type=str, + default="INFO", + help=( + "Severity level of the log message (default INFO). Options: DEBUG, INFO, " + "WARNING, ERROR, CRITICAL" + ), +) +@click.pass_obj +def log_on_server(obj: ProcessManagerContext, text: str, severity: str) -> None: + obj.get_driver("process_manager").log_on_server(text=text, severity=severity) diff --git a/src/drunc/process_manager/interface/shell.py b/src/drunc/process_manager/interface/shell.py index b933ffcc1..2d9a3d91d 100644 --- a/src/drunc/process_manager/interface/shell.py +++ b/src/drunc/process_manager/interface/shell.py @@ -60,7 +60,7 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> # process_manager_shell_log.error(e.message) # TODO: Keep this for production branch, remove this from dev branch exit(1) - ctx.obj.get_driver("process_manager").send_msg( + ctx.obj.get_driver("process_manager").log_on_server( f"{getpass.getuser()} connected from {ctx.obj.shell_id}" ) @@ -78,7 +78,7 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> ) def cleanup(): - ctx.obj.get_driver("process_manager").send_msg( + ctx.obj.get_driver("process_manager").log_on_server( f"{getpass.getuser()} disconnecting from {ctx.obj.shell_id}" ) ctx.obj.terminate() diff --git a/src/drunc/process_manager/k8s_process_manager.py b/src/drunc/process_manager/k8s_process_manager.py index fbd5d9405..766d210df 100644 --- a/src/drunc/process_manager/k8s_process_manager.py +++ b/src/drunc/process_manager/k8s_process_manager.py @@ -1968,15 +1968,6 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines: lines=[f"Could not retrieve logs: {e.reason}"], ) - def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: - # Note: currently exact same implementation as ssh manager - # Although there is room here to change as necessary - try: - self.log.info(f"{msg}; from {peer}") - except Exception as e: - self.log.error(f"Failed to receive message with exception {e}") - return OutcomeStatus(flag=OutcomeFlag.FAIL) - return OutcomeStatus(flag=OutcomeFlag.SUCCESS) def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: diff --git a/src/drunc/process_manager/process_manager.py b/src/drunc/process_manager/process_manager.py index c6414968c..c61898ef6 100644 --- a/src/drunc/process_manager/process_manager.py +++ b/src/drunc/process_manager/process_manager.py @@ -6,12 +6,11 @@ from daqpytools.logging import LogHandlerConf, exceptions, setup_daq_ers_logger from druncschema.authoriser_pb2 import ActionType, SystemType +from druncschema.common_pb2 import LogOnServerRequest, LogOnServerResponse from druncschema.description_pb2 import CommandDescription, Description -from druncschema.generic_pb2 import OutcomeStatus from druncschema.opmon.process_manager_pb2 import ProcessStatus from druncschema.process_manager_pb2 import ( BootRequest, - GenericNotificationMessage, LogLines, LogRequest, ProcessInstance, @@ -38,7 +37,7 @@ ProcessManagerRunningMode, ProcessManagerTypes, ) -from drunc.utils.utils import get_logger, pid_info_str, resolve_context_peer +from drunc.utils.utils import get_logger, pid_info_str class BadQuery(DruncCommandException): @@ -488,57 +487,36 @@ def logs(self, request: LogRequest, context: ServicerContext) -> LogLines: return response - @abc.abstractmethod - def _send_msg_impl( - self, msg: str | None = None, peer: str | None = None - ) -> OutcomeStatus: - raise NotImplementedError - - @authentified_and_authorised( - action=ActionType.READ, system=SystemType.PROCESS_MANAGER - ) - def send_msg(self, request: Request, context: ServicerContext) -> OutcomeStatus: - self.log.debug(f"{self.name} running send_msg") - - try: - peer = context.peer() - peer_display = resolve_context_peer(peer) - except Exception: - self.log.warning("Could not determine caller peer", exc_info=True) - peer_display = "unknown" + @authentified_and_authorised(action=ActionType.READ, system=SystemType.CONTROLLER) + def log_on_server( + self, + request: LogOnServerRequest, + context: ServicerContext, + ) -> LogOnServerResponse: + """ + Log a message on the server with the specified severity. - # Try to extract an optional GenericNotificationMessage from request.data - try: - if ( - request is not None - and hasattr(request, "data") - and request.data is not None - ): - gm = GenericNotificationMessage() - request.data.Unpack(gm) - msg_value = gm.message - except Exception as e: - self.log.debug( - f"Error while extracting send_msg payload: {e}", exc_info=True - ) - msg_value = "unknown payload" + Args: + request: LogOnServerRequest containing the log message and severity. + context: gRPC ServicerContext (not used). - try: - response = self._send_msg_impl(msg_value, peer_display) - except NotImplementedError: - raise DruncNotImplementedException( - message="Implementation missing", - domain="ProcessManager.send_msg", - ) - except Exception as e: - context_msg = f"Unhandled exception in ProcessManager.send_msg: {e}" - self.log.exception(context_msg) + Returns: + LogOnServerResponse indicating the result of the logging operation. - raise DruncCommandException( - message=context_msg, - domain="ProcessManager.send_msg", - ) + Raises: + None + """ + # Construct the default response indicating successful execution + response = LogOnServerResponse( + token=None, + flag=ResponseFlag.EXECUTED_SUCCESSFULLY, + ) + # Get the log method corresponding to the severity level (e.g., debug, info, + # warning, error), and log the message + level = request.severity.lower() + log_method = getattr(self.log, level, self.log.info) + log_method(request.text) return response def _ensure_one_process( diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index a218e1b5e..89306c14b 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -14,10 +14,10 @@ from daqconf.set_connectivity_service_port import set_connectivity_service_port from daqconf.set_rc_controller_port import set_rc_controller_port from daqconf.utils import find_free_port +from druncschema.common_pb2 import LogOnServerRequest, LogOnServerResponse from druncschema.description_pb2 import Description from druncschema.process_manager_pb2 import ( BootRequest, - GenericNotificationMessage, LogLines, LogRequest, ProcessDescription, @@ -86,34 +86,20 @@ def close(self) -> None: except Exception as e: self.log.error(f"Error closing gRPC channel: {e}", exc_info=True) - def send_msg(self, msg): - request = Request(token=copy_token(self.token)) - - if msg is not None: - try: - gm = GenericNotificationMessage(message=str(msg)) - request.data.Pack(gm) - except Exception: - self.log.critical("Failed to pack send_msg payload", exc_info=True) - - timeout = 10 + def update_controller_logs(self, ctrl_dal, level): + """ + Update the log level of the controller in the DAL. - try: - response = self.stub.send_msg(request, timeout=timeout) - except grpc.RpcError as e: - try: - error_details = extract_grpc_rich_error(e) - self.log.error(error_details) - except Exception as extraction_error: - self.log.critical( - f"Could not extract rich error details from gRPC error: {extraction_error}", - exc_info=True, - ) - handle_grpc_error(e) + Args: + ctrl_dal: The controller DAL object. + level: The new log level to set. - return response + Returns: + The updated controller DAL object. - def update_controller_logs(self, ctrl_dal, level): + Raises: + None + """ ctrl_dal.controller_log_level = level return ctrl_dal @@ -637,8 +623,8 @@ def get_controller_address(session_dal, session_name): # 1: Try dynamic lookup via Connectivity Service if csc: - self.log.debug( - f"Attempting to discover controller '{top_controller_name}' via connectivity service at {connection_server}:{connection_port}" + self.log.info( + f"Looking for top controller '{top_controller_name}' in the connectivity service at http://{connection_server}:{connection_port}" ) try: timeout = ( @@ -1107,3 +1093,39 @@ def _log_controller_interrupt( [yellow]connect {{controller_address}}:{{controller_port}}>[/] """ ) + + def log_on_server( + self, + text: str, + severity: str = "INFO", + timeout: int | float = 60, + ) -> LogOnServerResponse: + """ + Logs a message to the server's log system. + + Args: + text (str): The message to log. + severity (str, optional): The severity level of the log message. Defaults to "INFO". + timeout (int | float, optional): The timeout for the gRPC request in seconds. Defaults to 60. + + Returns: + None + + Raises: + grpc.RpcError: If the gRPC request fails. + """ + request = LogOnServerRequest( + token=self.token, + text=text, + severity=severity, + target="", + execute_along_path=False, + execute_on_all_subsequent_children_in_path=False, + ) + request.token.CopyFrom(self.token) + try: + response = self.stub.log_on_server(request, timeout=timeout) + except grpc.RpcError as e: + handle_grpc_error(e) + + return response diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index f7fd85712..58d3b6352 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -3,7 +3,6 @@ import uuid from typing import List, Optional -from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -510,15 +509,6 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: ) return ret_fmt - def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: - try: - self.log.info(f"{msg}; from {peer}") - except Exception as e: - self.log.error(f"Failed to receive message with exception {e}") - return OutcomeStatus(flag=OutcomeFlag.FAIL) - - return OutcomeStatus(flag=OutcomeFlag.SUCCESS) - def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: self.log.debug(f"{self.name} running boot command") this_uuid = str(uuid.uuid4()) diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index e81b11e48..fb3a674ac 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -6,6 +6,7 @@ from druncschema.process_manager_pb2 import ProcessInstance, ProcessQuery from drunc.controller.interface.shell_utils import controller_setup +from drunc.controller.utils import count_processes_in_status_response, get_all_states from drunc.exceptions import DruncSetupException from drunc.process_manager.interface.cli_argument import add_query_options_no_session from drunc.process_manager.interface.commands import ( @@ -120,31 +121,186 @@ def boot( log.error("Could not understand where the controller is!") return + # Determine whether the session should be placed into an error state, regardless of + # the outcome of the `boot` command. + # This variable catches all cases for which the session is not booted or reported + # correctly, and a relevant log message is shown. + # This catches additional issues that the `boot` process does not, as `boot` only + # deploys the processes, and not e.g. whether the processes have successfully + # registered on the connevity service, or if the process has died shortly after + # booting. + # Example - the process booted, but after booting it died before all the apps were + # registered on the connectivity service. The process manager would report this as a + # success, which is valid for process management, but it is a failure from the + # session perspective, as an error has occured with other services which the process + # manager does not interface with. + put_in_error_state: bool = False + + # If the session applications are not found on the connectivity serivce, then the + # session is not booted correctly. This is a critical error, the user should be + # informed, and the session should be placed in error state. + ps_response = obj.get_driver("process_manager").ps( + ProcessQuery(session=session_name) + ) + ps_process_count = len(ps_response.values) + + status_response = obj.get_driver("controller").status() + status_process_count = count_processes_in_status_response(status_response) + + # Local connectivity serivces are not reported in the status table, but they should + # be. Increment the status_process_count by 1 if using the LCS. + # TODO: Remove this once the LCS is reported in the status table (issue 745). + if obj.session_uses_local_connectivity_service: + status_process_count += 1 + + if ps_process_count != status_process_count: + log.debug( # TODO - once issue 793 is resolved, this should be a log.error + f"Booted, but the number of processes registered with the process manager " + f"({ps_process_count}) does not match the number of processes registered " + f"with the top segment (root) controller ({status_process_count}). Use the " + "[yellow]ps[/] command to determine which applications did not correctly " + "register themselves on the connectivity service by comparing against the " + "status table, and the [yellow]logs[/] command to find out more about this " + "failure." + ) + # TODO: Uncomment this once the cause of inconsistent status table printing is + # understood (issue 793) + # put_in_error_state = True + + # Check if session booted correctly, if not put it in error state + session_states = get_all_states(status_response) + if "disconnected" in session_states: + log.error( + "Booted, but there are disconnected applications/controllers. Use the " + "[yellow]logs[/] command to find out more about this failure." + ) + put_in_error_state = True + # If any processes died immediately, place the controller in error. alive_process_count = len( [p for p in processes.values if p.status_code == ProcessInstance.RUNNING] ) - dead_process_count = expected_booted_processes - alive_process_count + if dead_process_count > 0: + log.error( + f"Booted, but {dead_process_count} processes died. Use the [yellow]ps[/] " + "command to find out which applications are dead, and [yellow]logs[/] " + "command to find out more about this failure on a per-application basis." + ) + put_in_error_state = True + + # Check if there is or should be an error state. If not, then the boot was + # successful and we can return, otherwise, we will log the error and place the + # session in an error state if required. + in_error_state = obj.get_driver("controller").status().status.in_error + if not in_error_state and not put_in_error_state: + log.info("Booted successfully") + return + # An error state has been detected, or should be placed. Log the error and place the + # session in an error state if required. + log.info( + "Booted, but the session is in an error state. Use the [yellow]status[/] " + "command to find out more about this failure, and check the logs of the " + "applications that are in an error state with the [yellow]logs[/] command." + ) + if put_in_error_state and not in_error_state: + log.error("Placing the session into an error state due to boot issues") + obj.get_driver("controller").to_error() + in_error_state = obj.get_driver("controller").status().status.in_error + + # If the unified shell is running in batch or semibatch mode, exit with a non-zero + # exit code unless bypassed with the --no-stop-error-batch-mode option in the + # unified shell. if ( - not obj.get_driver("controller").status().status.in_error - and dead_process_count == 0 + in_error_state + and obj.running_mode in [UnifiedShellMode.BATCH, UnifiedShellMode.SEMIBATCH] + and not obj.no_stop_error_batch_mode ): - log.info("Booted successfully") - elif dead_process_count != 0: - log.error(f"Booted, but {dead_process_count} processes died after booting.") - # The following line has been commented out as there are issues with the k8s PM - # booting process, which terminates processes and immediately reboots them. The - # current cause of this issue is unknown, and has been listed in the issue list. - # obj.get_driver("controller").to_error() - elif obj.get_driver("controller").status().status.in_error: - log.error("Booted, but the top controller is in error") - if obj.running_mode in [UnifiedShellMode.BATCH, UnifiedShellMode.SEMIBATCH]: - log.error( - "Unified shell: Running in batch mode, and because error state is detected, exiting." - ) - sys.exit(1) + log.error( + "Running in batch mode, and because error state is detected, exiting." + ) + sys.exit(1) + + +@click.command("log") +@click.argument("text", required=True) +@click.option( + "--target-server", + type=str, + default="", + help="Server to use the log command on. Default value of '' will send the log message to all the servers, e.g. the process manager and the root controller.", +) +@click.option( + "-s", + "--severity", + type=str, + default="INFO", + help=( + "Severity level of the log message (default INFO). Options: DEBUG, INFO, " + "WARNING, ERROR, CRITICAL" + ), +) +@click.option("--target", type=str, help="The session target to address", default="") +@click.option( + "--execute-along-path/--dont-execute-along-path", + is_flag=True, + show_default=True, + help="Execute the command along the session application path", + default=False, +) +@click.option( + "--execute-on-all-subsequent-children-in-path/--dont-execute-on-all-subsequent-children-in-path", + is_flag=True, + show_default=True, + help="Execute the command on all subsequent children in the session application path", + default=True, +) +@click.pass_obj +def log_on_server( + obj: ProcessManagerContext, + text: str, + target_server: str, + severity: str, + target: str, + execute_along_path: bool, + execute_on_all_subsequent_children_in_path: bool, +) -> None: + """ + Log a message to the specified server. + + This command allows you to send a log message to a specific server or to all servers + in the system. You can specify the severity level of the log message. + + Args: + obj (ProcessManagerContext): The context object containing session information. + text (str): The log message text. + target_server (str): The server to send the log message to. Default is '' (all servers). + severity (str): The severity level of the log message. Default is 'INFO'. + + Returns: + None + + Raises: + None + """ + log = get_logger("unified_shell.log_on_server") + log.debug("Logging message to server(s)...") + + if target_server in ["", "process_manager"]: + obj.get_driver("process_manager").log_on_server( + text=text, + severity=severity, + ) + + if target_server in ["", "controller"] and obj.has_driver("controller"): + obj.get_driver("controller").log_on_server( + text=text, + severity=severity, + target=target, + execute_along_path=execute_along_path, + execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path, + ) @click.command("terminate") diff --git a/src/drunc/unified_shell/context.py b/src/drunc/unified_shell/context.py index 1e28ed084..fbccd1b72 100644 --- a/src/drunc/unified_shell/context.py +++ b/src/drunc/unified_shell/context.py @@ -33,6 +33,8 @@ def __init__(self): self.override_logs = True self.running_mode = UnifiedShellMode.INTERACTIVE self.batch_commands: list(str) = [] + self.no_stop_error_batch_mode = False + self.session_uses_local_connectivity_service: bool | None = None super(UnifiedShellContext, self).__init__() def reset(self, address_pm: str = ""): diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 557d5030a..871d3ee77 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -55,6 +55,7 @@ boot, flush, kill, + log_on_server, logs, ps, restart, @@ -115,6 +116,17 @@ "will be useful for hardware operations." ), ) # For production, change default to true/remove it +@click.option( + "-nsb", + "--no-stop-error-batch-mode", + is_flag=True, + default=False, + help=( + "The default behaviour of the unified shell is to exit if the root-controller " + "of the sessions is in error. This option will allow the unified shell to " + "continue executing commands, mainly for use in testing scenarios." + ), +) @click.pass_context def unified_shell( ctx: click.core.Context, @@ -126,6 +138,7 @@ def unified_shell( override_logs: bool, log_path: str, safe_mode: bool, + no_stop_error_batch_mode: bool, ) -> None: """ The unified shell is a command line interface to interact with the process manager @@ -208,6 +221,7 @@ def unified_shell( ctx.obj.configuration_id = configuration_id ctx.obj.session_name = session_name + ctx.obj.no_stop_error_batch_mode = no_stop_error_batch_mode # Get the session DAL db = conffwk.Configuration(ctx.obj.configuration_file) @@ -296,6 +310,11 @@ def unified_shell( ctx.obj.log.info("Setting up the controller interface") + # Keep track of whether the session uses a local connectivity service + ctx.obj.session_uses_local_connectivity_service = ( + session_dal.connectivity_service.host == "localhost" + ) + # Run a simple command (describe) to check the connection with the process manager try: ctx.obj.get_driver().describe() @@ -325,13 +344,13 @@ def unified_shell( sys.exit(1) ctx.obj.log.debug("Communication with the process manager verified successfully") - ctx.obj.get_driver("process_manager").send_msg( + ctx.obj.get_driver("process_manager").log_on_server( f"{getpass.getuser()} connected from unified shell" ) # Add the unified shell Click commands to the CLI ctx.obj.log.debug("Adding [green]unified_shell[/green] commands") - unified_shell_commands = [boot, ps, terminate] + unified_shell_commands: list[click.Command] = [boot, log_on_server, ps, terminate] for cmd in unified_shell_commands: ctx.command.add_command(cmd, format_name_for_cli(cmd.name)) ctx.obj.dynamic_commands.add(format_name_for_cli(cmd.name)) @@ -521,7 +540,7 @@ def cleanup(): ) # Remove the connection to the process manager - ctx.obj.get_driver("process_manager").send_msg( + ctx.obj.get_driver("process_manager").log_on_server( f"{getpass.getuser()} disconnected from unified shell" ) ctx.obj.get_driver("process_manager").close() diff --git a/src/drunc/utils/shell_utils.py b/src/drunc/utils/shell_utils.py index b0a6a0b16..8c23f773d 100644 --- a/src/drunc/utils/shell_utils.py +++ b/src/drunc/utils/shell_utils.py @@ -380,7 +380,7 @@ def log_pm_cmd(obj: ShellContext): of the message while still recording the command name, optional session name, and shell identity. - These are sent over via send_msg so that it can be displayed in the process manager + These are sent over via so that it can be displayed in the process manager shell Args: @@ -398,4 +398,4 @@ def log_pm_cmd(obj: ShellContext): args = f" with arguments {parms_dict}" if parms_dict else "" session = f" for session {obj.session_name}" if hasattr(obj, "session_name") else "" msg = f"{getpass.getuser()} sent {cmd_name}{args}{session} via {obj.get_shell_id()}" - obj.get_driver("process_manager").send_msg(msg) + obj.get_driver("process_manager").log_on_server(msg) diff --git a/src/drunc/utils/utils.py b/src/drunc/utils/utils.py index d24711b2c..f57b2e6ad 100644 --- a/src/drunc/utils/utils.py +++ b/src/drunc/utils/utils.py @@ -583,8 +583,8 @@ def get_control_type_and_uri_from_connectivity_service( ApplicationLookupUnsuccessful: If the URI cannot be resolved. """ uris: list[dict[str, object]] = [] - logger = get_logger("utils.get_control_type_and_uri_from_connectivity_service") - shared_console = get_shared_rich_console(logger) + log = get_logger("utils.get_control_type_and_uri_from_connectivity_service") + shared_console = get_shared_rich_console(log) start = time.time() elapsed = 0.0 @@ -617,7 +617,7 @@ def get_control_type_and_uri_from_connectivity_service( except ApplicationLookupUnsuccessful: elapsed = time.time() - start - logger.debug( + log.debug( f"Could not resolve '{name}_control' elapsed {elapsed:.2f}s/{timeout}s" ) time.sleep(retry_wait) @@ -637,7 +637,7 @@ def get_control_type_and_uri_from_connectivity_service( except ApplicationLookupUnsuccessful: elapsed = time.time() - start - logger.debug( + log.debug( f"Could not resolve '{name}_control' elapsed {elapsed:.2f}s/{timeout}s" ) time.sleep(retry_wait) diff --git a/tests/conftest.py b/tests/conftest.py index e2e8705a0..ea46e664c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -33,7 +33,7 @@ def load_test_config() -> None: # Determine the path to the test configurations cwd = Path(os.path.abspath(__file__)) - test_configs = cwd.parent / ".." / "config" / "tests" + test_configs = cwd.parent / ".." / "config" / "drunc" test_configs = test_configs.resolve() print(f"{test_configs=}") @@ -41,6 +41,10 @@ def load_test_config() -> None: os.makedirs(consolidated_conf_path, exist_ok=True) DUNEDAQ_DB_PATH += f":{test_configs!s}:{consolidated_conf_path!s}" + # Add the drunc root too + drunc_root = cwd.parent.parent + DUNEDAQ_DB_PATH += f":{drunc_root!s}" + # For debugging, print the DUNEDAQ_DB_PATH entries print("DUNEDAQ_DB_PATH entries:") for entry in DUNEDAQ_DB_PATH.split(":"): diff --git a/tests/controller/test_utils.py b/tests/controller/test_utils.py index 680ce54eb..efa5a0b75 100644 --- a/tests/controller/test_utils.py +++ b/tests/controller/test_utils.py @@ -1,6 +1,49 @@ import pytest +@pytest.fixture +def mock_status_tree(): + """Builds and returns a mock StatusResponse tree for testing.""" + from druncschema.controller_pb2 import StatusResponse + from druncschema.request_response_pb2 import ResponseFlag + + def create_status( + name: str, state: str, sub_state: str, children: list = None + ) -> StatusResponse: + resp = StatusResponse(name=name, flag=ResponseFlag.EXECUTED_SUCCESSFULLY) + resp.status.state = state + resp.status.sub_state = sub_state + resp.status.in_error = False + resp.status.included = True + if children: + for child in children: + resp.children.add().CopyFrom(child) + return resp + + ftns1_app = create_status("ft-nested-segment-1-application", "initial", "idle") + ftns1_ctrl = create_status( + "ft-nested-segment-1-controller", "initial", "idle", [ftns1_app] + ) + ftns2_app = create_status( + "ft-nested-segment-2-application", "disconnected", "disconnected" + ) + ftns21_app = create_status("ft-nested-segment-2.1-application", "initial", "idle") + ftns2_ctrl = create_status( + "ft-nested-segment-2-controller", + "initialising", + "initialising", + [ftns2_app, ftns21_app], + ) + ftts_app = create_status("ft-top-segment-application", "initial", "idle") + + return create_status( + "ft-top-segment-controller", + "initialising", + "initialising", + [ftns1_ctrl, ftns2_ctrl, ftts_app], + ) + + def test_get_segment_lookup_timeout(load_test_config): from drunc.utils.configuration import parse_conf_url @@ -34,3 +77,20 @@ def test_get_segment_lookup_timeout(load_test_config): segment_6 = db.get_dal(class_name="Segment", uid="segment-6") assert get_segment_lookup_timeout(segment_6, base_timeout=60) == 60 * 1 + + +# Now your tests become beautifully short: + + +def test_get_all_states(mock_status_tree): + from drunc.controller.utils import get_all_states + + top_segment_states = set(get_all_states(mock_status_tree)) + assert top_segment_states == {"initialising", "disconnected", "initial"} + + +def test_count_processes_in_status_response(mock_status_tree): + from drunc.controller.utils import count_processes_in_status_response + + process_count = count_processes_in_status_response(mock_status_tree) + assert process_count == 7 diff --git a/tests/issues/test_issue363.py b/tests/issues/test_issue363.py index ff477a5e7..8d7148420 100644 --- a/tests/issues/test_issue363.py +++ b/tests/issues/test_issue363.py @@ -1,5 +1,7 @@ # https://github.com/DUNE-DAQ/drunc/issues/363 +import os + from drunc.controller.configuration import ControllerConfHandler from drunc.utils.configuration import OKSKey from drunc.utils.utils import get_root_logger @@ -7,11 +9,24 @@ def test_issue363(load_test_config): get_root_logger("INFO") - conf_path = "oksconflibs:nestedConfig.data.xml" + conf_path = "config/drunc/nestedConfig.data.xml" + + path_found: bool = False + for path in os.getenv("DUNEDAQ_DB_PATH", "").split(":"): + if os.path.exists(os.path.join(path, conf_path)): + print(f"Found nestedConfig.data.xml in {path}") + path_found = True + break + + if not path_found: + raise FileNotFoundError( + "nestedConfig.data.xml not found in any of the paths specified in DUNEDAQ_DB_PATH" + ) + controller_id = "nested-segment-controller" controller_configuration = ControllerConfHandler.from_oks( - url=conf_path, + url="oksconflibs:" + conf_path, oks_key=OKSKey( schema_file="schema/confmodel/dunedaq.schema.xml", class_name="RCApplication", diff --git a/tests/process_manager/interface/test_commands.py b/tests/process_manager/interface/test_commands.py index 9c843bdcb..49923829e 100644 --- a/tests/process_manager/interface/test_commands.py +++ b/tests/process_manager/interface/test_commands.py @@ -102,7 +102,11 @@ def logs(self, log_request): mock_result.lines = [] return mock_result - def send_msg(self, msg: str) -> None: + def log(self, msg: str) -> None: + # simulate sending a message; tests don't assert on this, so store it + self._last_sent_msg = msg + + def log_on_server(self, msg: str) -> None: # simulate sending a message; tests don't assert on this, so store it self._last_sent_msg = msg diff --git a/tests/process_manager/process_manager_mock_impls.py b/tests/process_manager/process_manager_mock_impls.py index 2ab0773b4..55b4a125e 100644 --- a/tests/process_manager/process_manager_mock_impls.py +++ b/tests/process_manager/process_manager_mock_impls.py @@ -9,7 +9,6 @@ from typing import Optional from unittest.mock import Mock -from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -100,13 +99,4 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines: ) def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList: - return self._not_implemented_response() - - def _send_msg_impl( - self, msg: str | None = None, peer: str | None = None - ) -> OutcomeStatus: - """ - Returns an empty response to indicate communication is working. - Accepts an optional message parameter for compatibility with new API. - """ - return OutcomeStatus(flag=OutcomeFlag.SUCCESS) + return self._not_implemented_response() \ No newline at end of file diff --git a/tests/utils/test_utils.py b/tests/utils/test_utils.py index 5094824b0..85addfa28 100644 --- a/tests/utils/test_utils.py +++ b/tests/utils/test_utils.py @@ -208,9 +208,12 @@ def parent_process(): pids = psutil.pids() child_pid_still_exists = False for pid in pids: - if psutil.Process(pid).name() == "tester_child_process": - child_pid_still_exists = True - break + try: + if psutil.Process(pid).name() == "tester_child_process": + child_pid_still_exists = True + break + except psutil.NoSuchProcess: + continue assert not child_pid_still_exists