Write container logs to separate text files as runs complete (#69)

Why these changes are being introduced:
* Improve observability of 'run_ab_transforms' while running
by creating individual log files per container as a run is completed.

How this addresses that need:
* Write logs to a persistent folder in the run directory
* Add 'write_log_file' sub-function to 'run_ab_transforms'
* Deprecate 'aggregate_logs' sub-function
* Simplify logs in 'run_ab_transforms' to improve readability
* Remove list of containers from return value of 'collect_container_results'
* Update unit tests

Side effects of this change:
* This introduces a new 'logs' folder in the run directory,
which is now created as part of the 'init_run' core function.

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-382

Author: jonavellecuerdo

README.md

@@ -188,10 +188,15 @@ Usage: -c run-diff [OPTIONS]
 
 Options:
   -d, --job-directory TEXT  Job directory to create.  [required]
-  -i, --input-files TEXT    Input files to transform.  This may be a comma
+  -i, --input-files TEXT    Input files to transform. This may be a comma
                             separated list of input files, or a local CSV file
                             that provides a list of files.  [required]
   -m, --message TEXT        Message to describe Run.
+  --download-files          Pass to download input files (or use previously
+                            downloaded input files) from AWS S3. The
+                            downloaded files are stored in a local MinIO S3
+                            server and made available for Transmogrifier to
+                            use.
   -h, --help                Show this message and exit.
 ```
 

abdiff/cli.py

@@ -141,7 +141,7 @@ def init_job(
     type=str,
     required=True,
     help=(
-        "Input files to transform.  This may be a comma separated list of input files, "
+        "Input files to transform. This may be a comma separated list of input files, "
         "or a local CSV file that provides a list of files."
     ),
 )
@@ -157,8 +157,9 @@ def init_job(
     "--download-files",
     is_flag=True,
     help=(
-        "Pass to download input files from AWS S3 to a local Minio S3 server "
-        "for Transmogrifier to use."
+        "Pass to download input files (or use previously downloaded input files) "
+        "from AWS S3. The downloaded files are stored in a local MinIO S3 server "
+        "and made available for Transmogrifier to use."
     ),
 )
 def run_diff(

abdiff/core/init_run.py

@@ -20,8 +20,11 @@ def init_run(
     """Function to initialize a new Run as part of a parent Job."""
     run_timestamp = datetime.datetime.now(tz=datetime.UTC).strftime("%Y-%m-%d_%H-%M-%S")
     run_directory = str(Path(job_directory) / "runs" / run_timestamp)
+    logs_directory = str(Path(run_directory) / "logs")
     os.makedirs(run_directory)
+    os.makedirs(logs_directory)
     logger.info(f"Run directory created: {run_directory}")
+    logger.info(f"Logs directory created: {logs_directory}")
 
     # clone job data and update with run information
     run_data = read_job_json(job_directory)

abdiff/core/run_ab_transforms.py

@@ -40,18 +40,11 @@ def run_ab_transforms(
 ) -> tuple[list[str], ...]:
     """Run Docker containers with versioned images of Transmogrifier.
 
-    The following steps are performed in sequential order:
-        1. Directories are created to capture the transformed files.
-        2. For all input files, an A and B version of Transmogrifier is run.
-        3. Wait for all containers to complete.
-        4. Aggregate logs from all containers.
-        5. Validate output.
-        6. Update run.json with lists describing input and transformed files.
-        7. Return a tuple containing two lists representing all A and B transformed files.
-
     Parallelization is handled by invoking the Docker containers via threads, limited by
     the ThreadPoolExecutor.max_workers argument.  Each thread invokes a detached Docker
-    container and manages its lifecycle until completion.
+    container and manages its lifecycle until completion. As each container exits, logs
+    are written to a text file in the run directory; log filenames follow the format:
+    "{source}-{run-date}-{run-type}-{container-short-id}-logs.txt".
 
     Args:
         run_directory (str): Run directory.
@@ -88,8 +81,10 @@ def run_ab_transforms(
     # initialize environment
     if not docker_client:
         docker_client = docker.from_env()
+
     transformed_directory_a, transformed_directory_b = create_subdirectories(
-        base_directory=run_directory, subdirectories=["transformed/a", "transformed/b"]
+        base_directory=run_directory,
+        subdirectories=["transformed/a", "transformed/b"],
     )
     logger.info(
         "Transformed directories created: "
@@ -102,19 +97,17 @@ def run_ab_transforms(
 
     # run containers and collect results
     futures = run_all_docker_containers(
-        docker_client, input_files, run_configs, use_local_s3=use_local_s3
+        docker_client, input_files, run_configs, run_directory, use_local_s3=use_local_s3
     )
-    containers, exceptions = collect_container_results(futures)
+    exceptions = collect_container_results(futures)
     logger.info(
-        f"Successful containers: {len(containers)}, failed containers: {len(exceptions)}"
+        f"Successful containers: {len(futures)}, failed containers: {len(exceptions)}"
     )
 
     # process results
-    log_file = aggregate_logs(run_directory, containers)
-    logger.info(f"Log file created: {log_file}")
     if not CONFIG.allow_failed_transmogrifier_containers and exceptions:
         raise RuntimeError(  # noqa: TRY003
-            f"{len(exceptions)} / {len(containers)} containers failed "
+            f"{len(exceptions)} / {len(futures)} containers failed "
             "to complete successfully."
         )
     ab_transformed_file_lists = get_transformed_files(run_directory)
@@ -137,6 +130,7 @@ def run_all_docker_containers(
     docker_client: docker.client.DockerClient,
     input_files: list[str],
     run_configs: list[tuple],
+    run_directory: str,
     *,
     use_local_s3: bool = False,
 ) -> list[Future]:
@@ -157,6 +151,7 @@ def run_all_docker_containers(
             for docker_image, transformed_directory in run_configs:
                 args = (
                     docker_image,
+                    run_directory,
                     transformed_directory,
                     str(filename_details["source"]),
                     input_file,
@@ -175,6 +170,7 @@ def run_all_docker_containers(
 
 def run_docker_container(
     docker_image: str,
+    run_directory: str,
     transformed_directory: str,
     source: str,
     input_file: str,
@@ -218,41 +214,52 @@ def run_docker_container(
         },
         volumes=[f"{os.path.abspath(transformed_directory)}:/tmp"],
     )
-    logger.info(
-        f"Container '{container.id}' (Docker image: {docker_image}) "
-        f"RUNNING transform for '{source}' input_file: {input_file}."
-    )
+    logger.info(f"Transmogrifier container ({container.short_id}) STARTED: {input_file}")
 
     exception = None
     try:
         start_time = perf_counter()
         while True:
             time.sleep(0.5)
             container.reload()
+            elapsed_time = time.perf_counter() - start_time
             if container.status == "exited":
-                logger.info(f"Container {container.id} exited.")
+                logger.info(
+                    f"Transmogrifier container ({container.short_id}) EXITED, "
+                    f"elapsed {timedelta(seconds=elapsed_time)}: {input_file}"
+                )
+                write_log_file(run_directory, input_file, container)
                 break
 
             if time.perf_counter() - start_time > timeout:
                 logger.error(
-                    f"Container {container.id} timed out after {timeout} seconds"
+                    f"Transmogrifier container ({container.short_id}) TIMED OUT, "
+                    f"elapsed {timedelta(seconds=elapsed_time)}: {input_file}"
                 )
                 container.stop()
                 exception = DockerContainerTimeoutError(
                     container_id=container.id, timeout=timeout
                 )
+                write_log_file(
+                    run_directory,
+                    input_file,
+                    container,
+                    extra_messages=[
+                        f"Transmogrifier container ({container.short_id}) TIMED OUT"
+                    ],
+                )
                 break
 
     except Exception as e:
         exception = e  # type: ignore[assignment]
-        logger.exception("Unhandled exception while waiting for container to complete.")
+        logger.exception(
+            f"Transmogrifier container ({container.short_id}) UNHANDLED EXCEPTION: {input_file}"  # noqa: E501
+        )
 
     return container, exception
 
 
-def collect_container_results(
-    futures: list[Future],
-) -> tuple[list[Container], list[Exception]]:
+def collect_container_results(futures: list[Future]) -> list[Exception]:
     """Collect results of container executions.
 
     Each future will contain a tuple of (Container, Exception) where the exception may
@@ -261,39 +268,51 @@ def collect_container_results(
 
     Returns a tuple of (Containers (success), Exceptions (failure)) from all executions.
     """
-    containers = []
     exceptions = []
     for future in futures:
         container, exception = future.result()
-        containers.append(container)
 
         if exception:
             exceptions.append(exception)
         if container.attrs["State"]["ExitCode"] != 0:
             exceptions.append(DockerContainerRuntimeError(container.id))
 
     logger.info(
-        f"Container results collected: {len(containers) - len(exceptions)} successes, "
+        f"Container results collected: {len(futures) - len(exceptions)} successes, "
         f"{len(exceptions)} failures"
     )
-    return containers, exceptions
-
-
-def aggregate_logs(run_directory: str, containers: list[Container]) -> str:
-    """Retrieve logs for containers in a list, aggregating to a single log file."""
-    log_file = str(Path(run_directory) / "transformed/logs.txt")
-    with open(log_file, "w") as file:
-        for container in containers:
-            file.write(f"container: {container.id}\n")
-            descriptors = (
-                f"docker_image: {container.labels['docker_image']} | "
-                f"source: {container.labels['source']} | "
-                f"input_file: {container.labels['input_file']}\n"
-            )
-            file.write(descriptors)
-            file.write(container.logs().decode())
-            file.write("\n\n")
-    return log_file
+    return exceptions
+
+
+def write_log_file(
+    run_directory: str,
+    input_file: str,
+    container: Container,
+    extra_messages: list[str] | None = None,
+) -> None:
+    """Write logs for a given container to a text file."""
+    filename_details = parse_timdex_filename(input_file)
+    log_filename = "{source}-{run_date}-{run_type}-{container_id}-logs.txt".format(
+        source=filename_details["source"],
+        run_date=filename_details["run-date"],
+        run_type=filename_details["run-type"],
+        container_id=container.short_id,
+    )
+    log_filepath = str(Path(run_directory) / "logs" / log_filename)
+    header = (
+        f"docker_image: {container.labels['docker_image']} | "
+        f"source: {container.labels['source']} | "
+        f"input_file: {container.labels['input_file']}"
+    )
+    container_desc = f"container: {container.id}"
+    with open(log_filepath, "w") as file:
+        file.write(header + "\n")
+        file.write(container_desc + "\n")
+        for log in container.logs(stream=True):
+            file.write(log.decode())
+
+        if extra_messages:
+            file.write("\n".join(extra_messages))
 
 
 def get_transformed_files(run_directory: str) -> tuple[list[str], ...]:

tests/conftest.py

@@ -32,8 +32,9 @@
 class Container:
     """Stub for docker.models.container.Container object."""
 
-    def __init__(self, id, labels, attrs: dict | None = None):  # noqa: A002
+    def __init__(self, id, short_id, labels, attrs: dict | None = None):  # noqa: A002
         self.id = id
+        self.short_id = short_id
         self.labels = labels
         self.status = "created"
         self.run_duration = 0.5
@@ -56,8 +57,10 @@ def reload(self):
         if time.time() - self.start_time >= self.run_duration:
             self.status = "exited"
 
-    def logs(self):
+    def logs(self, *, stream: bool = True):
         with open("tests/fixtures/transmogrifier-logs.txt", "rb") as file:
+            if stream:
+                yield from file
             return file.read()
 
     def stop(self):
@@ -110,6 +113,7 @@ def create_transformed_files(
         if self.errors:
             return Container(
                 id=container_id,
+                short_id=container_id[:3],
                 labels={
                     "docker_image": image_name,
                     "source": "source",
@@ -119,6 +123,7 @@ def create_transformed_files(
             )
         return Container(
             id=container_id,
+            short_id=container_id[:3],
             labels={
                 "docker_image": image_name,
                 "source": "source",
@@ -277,6 +282,7 @@ def mocked_docker_container_and_image(
 def mocked_docker_container_a():
     return Container(
         id="abc123",
+        short_id="abc",
         labels={
             "docker_image": "transmogrifier-example-job-1-abc123:latest",
             "source": "source",
@@ -289,6 +295,7 @@ def mocked_docker_container_a():
 def mocked_docker_container_b():
     return Container(
         id="def456",
+        short_id="def",
         labels={
             "docker_image": "transmogrifier-example-job-1-def456:latest",
             "source": "source",