How to stream files generated by executor during task execution

[cjao]

The Covalent dispatcher currently retrieves stdout and stderr from a task only after the task has finished running. In addition, it does not provide a standard way to retrieve other log files from the executor that may be of interest, such as Slurm logs or profiling data.

The purpose of this discussion is to sketch out some possible APIs for monitoring a specified set of files in the executor during task execution and streaming their updates to the dispatcher, either continuously or according to a specified polling frequency. These executors will usually be remote. For simplicity, let us also assume that the files to be monitored are append-only. This is a reasonable assumption for most log files.

Possible approaches:

• Pull: The dispatcher periodically polls the executor for file updates through some executor-specific logic.
• Push: We ship a file monitoring program together with the task, and executors POST file updates to the dispatcher using an API endpoint like (/api/update_result). Each POST would reference the dispatch_id and node_id, and contain the file delta in its body. This approach assumes that the executor has outgoing network connectivity (e.g. can reach the dispatcher).

Pull:

• Pros: security would not a problem as all communication can occur over SSH or whatever secure channel is already used to communicate between the dispatcher and executor.
• Cons: The usual inefficiencies with a polling approach.

Push:

• Pros: Avoids needless polling when there are no file updates.
• Cons: Security is a major concern because we would be exposing a public HTTP endpoint that can modify data. The dispatcher would need to authenticate the other party and ensure that it can only edit a specific node of a specific dispatch.

A hybrid approach might be for the executor to merely notify that the dispatcher that an update has occurred, which prompts the dispatcher to then pull the updates from the executor. This would avoid the periodic polling while reducing the security requirements.

[cjao]

@santoshkumarradha @wjcunningham7

[cjao]

The branch monitor-execinfo-exp proposes an executor API to implement either the pure-polling or the hybrid notify-and-pull approaches (the pure-push model wouldn't involve the executor at all). Two public methods are to be implemented by concrete executors:

    # Compute list of files to monitor based on dispatch id and node id
    def get_files_to_monitor(self) -> List[str]:
        raise NotImplementedError

    # The user overrides this method to retrieve at most `size` bytes from a
    # remote file starting at the specified position (in bytes).
    async def poll_file(self, path: str, starting_pos: int, size: int = -1) -> str:
        raise NotImplementedError

The executor plugin then starts monitoring each specified file at the beginning of execute() and terminates the monitoring before returning. The placeholder method update_node_info would write the contents of each pull to the result object (for now it just writes to the covalent log file).

[cjao]

Demo script:

import covalent as ct

import time

from dask.distributed import LocalCluster

from covalent.executor import DaskExecutor

lc = LocalCluster()
dask_exec = DaskExecutor(lc.scheduler_address)

@ct.electron(executor=dask_exec)
def write_msg(n_iter):
    for i in range(n_iter):
        with open("/tmp/stdout.log", "a") as f:
            print("Hello", file=f)
        time.sleep(3)
    return 0

@ct.lattice(workflow_executor="local")
def workflow(n_iter):
    return write_msg(n_iter)

Try calling ct.dispatch(workflow)(5) in Jupyter, and observe in the covalent logs:

base.py: Line 485 in create_file_monitor:
DEBUG - create_file_monitor: read 6 bytes from /tmp/stdout.log
base.py: Line 489 in update_node_info:
DEBUG - update_node_info: /tmp/stdout.log: Hello

base.py: Line 517 in watch_file_periodically:
DEBUG - watch_file_periodically: Checking for messages
--
base.py: Line 485 in create_file_monitor:
DEBUG - create_file_monitor: read 12 bytes from /tmp/stdout.log
base.py: Line 489 in update_node_info:
DEBUG - update_node_info: /tmp/stdout.log: Hello

base.py: Line 517 in watch_file_periodically:
DEBUG - watch_file_periodically: Checking for messages
--
base.py: Line 485 in create_file_monitor:
DEBUG - create_file_monitor: read 18 bytes from /tmp/stdout.log
base.py: Line 489 in update_node_info:
DEBUG - update_node_info: /tmp/stdout.log: Hello

base.py: Line 517 in watch_file_periodically:
DEBUG - watch_file_periodically: Checking for messages
--
base.py: Line 485 in create_file_monitor:
DEBUG - create_file_monitor: read 24 bytes from /tmp/stdout.log
base.py: Line 489 in update_node_info:
DEBUG - update_node_info: /tmp/stdout.log: Hello

base.py: Line 517 in watch_file_periodically:
DEBUG - watch_file_periodically: Checking for messages
--
base.py: Line 485 in create_file_monitor:
DEBUG - create_file_monitor: read 30 bytes from /tmp/stdout.log
base.py: Line 489 in update_node_info:
DEBUG - update_node_info: /tmp/stdout.log: Hello

[wjcunningham7]

@cjao I like the pure polling method; it's the simplest to implement in the short term

[wjcunningham7]

CC @santoshkumarradha

[Andrew-S-Rosen]

I definitely think this is a great idea!