[wip]

Author: jonavellecuerdo

Makefile

@@ -1,5 +1,6 @@
 SHELL=/bin/bash
 DATETIME:=$(shell date -u +%Y%m%dT%H%M%SZ)
+MINIO_COMPOSE_FILE=abdiff/helpers/minio/docker-compose.yaml
 
 help: # Preview Makefile commands
 	@awk 'BEGIN { FS = ":.*#"; print "Usage:  make <target>\n\nTargets:" } \
@@ -54,3 +55,10 @@ black-apply: # Apply changes with 'black'
 
 ruff-apply: # Resolve 'fixable errors' with 'ruff'
 	pipenv run ruff check --fix .
+
+# Development commands
+start-minio-server:
+	docker compose --env-file .env -f $(MINIO_COMPOSE_FILE) up -d
+
+stop-minio-server:
+	docker compose -f $(MINIO_COMPOSE_FILE) stop

Pipfile

@@ -19,6 +19,7 @@ boto3 = "*"
 
 [dev-packages]
 black = "*"
+boto3-stubs = {version = "*", extras = ["s3"]}
 coveralls = "*"
 freezegun = "*"
 ipython = "*"

Pipfile.lock

@@ -15,6 +15,23 @@ Compare transformed TIMDEX records from two versions (A,B) of Transmogrifier.
 - To lint the repo: `make lint`
 - To run the app: `pipenv run abdiff --help`
 
+### Storing Files in a Local Minio Server
+
+TIMDEX extract files from S3 (i.e., input files to use in transformations) can be downloaded to a local MinIO server hosted via Docker container. [MinIO is an object storage solution that provides an Amazon Web Services S3-compatible API and supports all core S3 features](https://min.io/docs/minio/kubernetes/upstream/). Downloading extract files improves the runtime of a diff by reducing the number of requests sent to S3 and avoids repeated downloads of extract files.
+
+1. Create an AWS profile `minio`. When prompted for an "AWS Access Key ID" and "AWS Secret Access Key", pass the values set for the `MINIO_ROOT_USER` and `MINIO_ROOT_PASSWORD` environment variables in the Docker Compose YAML file.
+   ```shell
+   aws configure --profile minio
+   ```
+
+2. Launch a local Minio server via Docker container: `make start-minio-server`.
+   The API is accessible at: http://127.0.0.1:9000.
+   The WebUI is accessible at: http://127.0.0.1:9001.
+
+3. On your browser, navigate to the WebUI and sign into the local Minio server using the credentials set in the Docker Compose YAML file.
+
+4. Through the UI, create a bucket in the local Minio server named after the S3 bucket containing the TIMDEX extract files that will be used in the A/B Diff.
+
 ## Concepts
 
 A **Job** in `abdiff` represents the A/B test for comparing the results from two versions of Transmogrifier.  When a job is first created, a working directory and a JSON file `job.json` with an initial set of configurations is created.
@@ -90,6 +107,11 @@ AWS_SESSION_TOKEN=# passed to Transmogrifier containers for use
 ### Optional
 
 ```text
+MINIO_S3_LOCAL_STORAGE=# full file system path to the directory where MinIO stores its object data on the local disk
+MINIO_S3_URL=# endpoint for MinIO server API; default is "http://localhost:9000/"
+MINIO_S3_CONTAINER_URL=# endpoint for the MinIO server when acccessed from inside a Docker container; default is "http://host.docker.internal:9000/"
+MINIO_ROOT_USER=# username for root user account for MinIO server
+MINIO_ROOT_PASSWORD=# password for root user account MinIO server
 WEBAPP_HOST=# host for flask webapp
 WEBAPP_PORT=# port for flask webapp
 TRANSMOGRIFIER_MAX_WORKERS=# max number of Transmogrifier containers to run in parallel; default is 6

README.md

@@ -11,6 +11,7 @@
 from abdiff.config import Config, configure_logger
 from abdiff.core import (
     build_ab_images,
+    download_input_files,
     calc_ab_diffs,
     calc_ab_metrics,
     collate_ab_transforms,
@@ -148,7 +149,12 @@ def init_job(
     help="Message to describe Run.",
     default="Not provided.",
 )
-def run_diff(job_directory: str, input_files: str, message: str) -> None:
+@click.option(
+    "--download-files", is_flag=True, help="Pass to skip download of extract files"
+)
+def run_diff(
+    job_directory: str, input_files: str, message: str, download_files: bool
+) -> None:
 
     job_data = read_job_json(job_directory)
     run_directory = init_run(job_directory, message=message)
@@ -160,11 +166,15 @@ def run_diff(job_directory: str, input_files: str, message: str) -> None:
     else:
         input_files_list = [filepath.strip() for filepath in input_files.split(",")]
 
+    if download_files:
+        download_input_files(input_files_list)
+
     ab_transformed_file_lists = run_ab_transforms(
         run_directory=run_directory,
         image_tag_a=job_data["image_tag_a"],
         image_tag_b=job_data["image_tag_b"],
         input_files=input_files_list,
+        download_files=download_files,
     )
     collated_dataset_path = collate_ab_transforms(
         run_directory=run_directory,

abdiff/cli.py

@@ -11,6 +11,11 @@ class Config:
         "WORKSPACE",
     )
     OPTIONAL_ENV_VARS = (
+        "MINIO_S3_LOCAL_STORAGE",
+        "MINIO_S3_URL",
+        "MINIO_S3_CONTAINER_URL",
+        "MINIO_ROOT_USER",
+        "MINIO_ROOT_PASSWORD",
         "WEBAPP_HOST",
         "WEBAPP_PORT",
         "TRANSMOGRIFIER_MAX_WORKERS",
@@ -25,6 +30,22 @@ def __getattr__(self, name: str) -> Any:  # noqa: ANN401
         message = f"'{name}' not a valid configuration variable"
         raise AttributeError(message)
 
+    @property
+    def minio_s3_url(self) -> str:
+        return self.MINIO_S3_URL or "http://localhost:9000/"
+
+    @property
+    def minio_s3_container_url(self) -> str:
+        return self.MINIO_S3_CONTAINER_URL or "http://host.docker.internal:9000/"
+
+    @property
+    def minio_root_user(self) -> str:
+        return self.MINIO_ROOT_USER or "minioadmin"
+
+    @property
+    def minio_root_password(self) -> str:
+        return self.MINIO_ROOT_PASSWORD or "minioadmin"
+
     @property
     def webapp_host(self) -> str:
         return self.WEBAPP_HOST or "localhost"

abdiff/config.py

@@ -7,6 +7,7 @@
 from abdiff.core.calc_ab_diffs import calc_ab_diffs
 from abdiff.core.calc_ab_metrics import calc_ab_metrics
 from abdiff.core.collate_ab_transforms import collate_ab_transforms
+from abdiff.core.download_input_files import download_input_files
 from abdiff.core.init_job import init_job
 from abdiff.core.init_run import init_run
 from abdiff.core.run_ab_transforms import run_ab_transforms
@@ -15,6 +16,7 @@
     "init_job",
     "init_run",
     "build_ab_images",
+    "download_input_files",
     "run_ab_transforms",
     "collate_ab_transforms",
     "calc_ab_diffs",

abdiff/core/__init__.py

@@ -0,0 +1,65 @@
+import logging
+import subprocess
+
+import boto3
+from botocore.exceptions import ClientError
+from mypy_boto3_s3.client import S3Client
+
+from abdiff.config import Config
+
+
+logger = logging.getLogger(__name__)
+
+CONFIG = Config()
+
+
+def download_input_files(input_files: str):
+    s3_client = boto3.client(
+        "s3",
+        endpoint_url=CONFIG.minio_s3_url,
+        aws_access_key_id=CONFIG.MINIO_ROOT_USER,
+        aws_secret_access_key=CONFIG.MINIO_ROOT_PASSWORD,
+    )
+
+    for input_file in input_files:
+        if check_object_exists(CONFIG.TIMDEX_BUCKET, input_file, s3_client):
+            logger.info(f"File found for input: {input_file}. Skipping download.")
+            continue
+
+        logger.info(f"Downloading input file from {CONFIG.TIMDEX_BUCKET}: {input_file}")
+        copy_command = ["aws", "s3", "cp", input_file, "-"]
+        upload_command = [
+            "aws",
+            "s3",
+            "cp",
+            "--endpoint-url",
+            CONFIG.minio_s3_url,
+            "--profile",
+            "minio",
+            "-",
+            input_file,
+        ]
+
+        try:
+            copy_process = subprocess.run(
+                args=copy_command, check=True, capture_output=True
+            )
+            subprocess.run(
+                args=upload_command,
+                check=True,
+                input=copy_process.stdout,
+            )
+        except subprocess.CalledProcessError:
+            logger.exception(f"Failed to download input file: {input_file}")
+
+
+def check_object_exists(bucket: str, input_file: str, s3_client: S3Client) -> bool:
+    key = input_file.replace(f"s3://{bucket}/", "")
+    try:
+        s3_client.head_object(Bucket=bucket, Key=key)
+        return True
+    except ClientError as exception:
+        if exception.response["Error"]["Code"] == "404":
+            return False
+        logger.exception(f"Cannot determine if object exists for key {key}.")
+        return False

abdiff/core/download_input_files.py

@@ -34,6 +34,7 @@ def run_ab_transforms(
     image_tag_a: str,
     image_tag_b: str,
     input_files: list[str],
+    download_files: bool,
     docker_client: docker.client.DockerClient | None = None,
 ) -> tuple[list[str], ...]:
     """Run Docker containers with versioned images of Transmogrifier.
@@ -95,7 +96,9 @@ def run_ab_transforms(
     ]
 
     # run containers and collect results
-    futures = run_all_docker_containers(docker_client, input_files, run_configs)
+    futures = run_all_docker_containers(
+        docker_client, input_files, run_configs, download_files
+    )
     containers, exceptions = collect_container_results(futures)
     logger.info(
         f"Successful containers: {len(containers)}, failed containers: {len(exceptions)}"
@@ -129,6 +132,7 @@ def run_all_docker_containers(
     docker_client: docker.client.DockerClient,
     input_files: list[str],
     run_configs: list[tuple],
+    download_files: bool,
 ) -> list[Future]:
     """Invoke Docker containers to run in parallel via threads.
 
@@ -151,6 +155,7 @@ def run_all_docker_containers(
                     input_file,
                     get_transformed_filename(filename_details),
                     docker_client,
+                    download_files,
                 )
                 tasks.append(executor.submit(run_docker_container, *args))
 
@@ -165,13 +170,28 @@ def run_docker_container(
     input_file: str,
     output_file: str,
     docker_client: docker.client.DockerClient,
+    download_files: bool,
     timeout: int = CONFIG.transmogrifier_timeout,
 ) -> tuple[Container, Exception | None]:
     """Run Transmogrifier via Docker container to transform input file.
 
     The container is run in a detached state to capture a container handle for later use
     but this function waits for the container to exit before returning.
     """
+
+    if download_files:
+        environment_variables = {
+            "AWS_ENDPOINT_URL": CONFIG.minio_s3_container_url,
+            "AWS_ACCESS_KEY_ID": CONFIG.MINIO_ROOT_USER,
+            "AWS_SECRET_ACCESS_KEY": CONFIG.MINIO_ROOT_PASSWORD,
+        }
+    else:
+        environment_variables = {
+            "AWS_ACCESS_KEY_ID": CONFIG.AWS_ACCESS_KEY_ID,
+            "AWS_SECRET_ACCESS_KEY": CONFIG.AWS_SECRET_ACCESS_KEY,
+            "AWS_SESSION_TOKEN": CONFIG.AWS_SESSION_TOKEN,
+        }
+
     container = docker_client.containers.run(
         docker_image,
         command=[
@@ -180,11 +200,7 @@ def run_docker_container(
             f"--source={source}",
         ],
         detach=True,
-        environment={
-            "AWS_ACCESS_KEY_ID": CONFIG.AWS_ACCESS_KEY_ID,
-            "AWS_SECRET_ACCESS_KEY": CONFIG.AWS_SECRET_ACCESS_KEY,
-            "AWS_SESSION_TOKEN": CONFIG.AWS_SESSION_TOKEN,
-        },
+        environment=environment_variables,
         labels={
             "docker_image": docker_image,
             "source": source,

abdiff/core/run_ab_transforms.py

@@ -0,0 +1,24 @@
+# Settings and configurations that are common for all containers
+x-minio-common: &minio-common
+  image: quay.io/minio/minio:RELEASE.2024-10-29T16-01-48Z
+  command: server --console-address ":9001" /mnt/data
+  ports:
+    - "9000:9000"  # API port
+    - "9001:9001"  # Console port
+  environment:
+    MINIO_ROOT_USER: ${MINIO_ROOT_USER}
+    MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
+  healthcheck:
+    test: ["CMD", "mc", "ready", "local"]
+    interval: 5s
+    timeout: 5s
+    retries: 5
+
+services:
+  minio:
+    <<: *minio-common
+    volumes:
+      - ${MINIO_S3_LOCAL_STORAGE}:/mnt/data
+
+
+