feat(core-clp): Add LZMA Compressor implementation and LZMA dependency.

.github/workflows/clp-pr-title-checks.yaml

@@ -2,9 +2,16 @@ name: "clp-pr-title-checks"
 
 on:
   pull_request_target:
+    # NOTE: Workflows triggered by this event give the workflow access to secrets and grant the
+    # `GITHUB_TOKEN` read/write repository access by default. So we need to ensure:
+    # - This workflow doesn't inadvertently check out, build, or execute untrusted code from the
+    #   pull request triggered by this event.
+    # - Each job has `permissions` set to only those necessary.
     types: ["edited", "opened", "reopened"]
     branches: ["main"]
 
+permissions: {}
+
 concurrency:
   group: "${{github.workflow}}-${{github.ref}}"
 

components/clp-package-utils/clp_package_utils/__init__.py

@@ -0,0 +1,13 @@
+import logging
+
+# Set up console logging
+logging_console_handler = logging.StreamHandler()
+logging_formatter = logging.Formatter(
+    "%(asctime)s.%(msecs)03d %(levelname)s [%(module)s] %(message)s", datefmt="%Y-%m-%dT%H:%M:%S"
+)
+logging_console_handler.setFormatter(logging_formatter)
+
+# Set up root logger
+root_logger = logging.getLogger()
+root_logger.setLevel(logging.INFO)
+root_logger.addHandler(logging_console_handler)

components/clp-package-utils/clp_package_utils/general.py

@@ -20,7 +20,9 @@
     REDIS_COMPONENT_NAME,
     REDUCER_COMPONENT_NAME,
     RESULTS_CACHE_COMPONENT_NAME,
+    StorageType,
     WEBUI_COMPONENT_NAME,
+    WorkerConfig,
 )
 from clp_py_utils.core import (
     get_config_value,
@@ -107,7 +109,7 @@ def get_clp_home():
     return clp_home.resolve()
 
 
-def generate_container_name(job_type: JobType) -> str:
+def generate_container_name(job_type: str) -> str:
     """
     :param job_type:
     :return: A unique container name for the given job type.
@@ -239,17 +241,17 @@ def generate_container_config(
             DockerMountType.BIND, clp_config.logs_directory, container_clp_config.logs_directory
         )
 
-    container_clp_config.archive_output.directory = pathlib.Path("/") / "mnt" / "archive-output"
+    container_clp_config.archive_output.set_directory(pathlib.Path("/") / "mnt" / "archive-output")
     if not is_path_already_mounted(
         clp_home,
         CONTAINER_CLP_HOME,
-        clp_config.archive_output.directory,
-        container_clp_config.archive_output.directory,
+        clp_config.archive_output.get_directory(),
+        container_clp_config.archive_output.get_directory(),
     ):
         docker_mounts.archives_output_dir = DockerMount(
             DockerMountType.BIND,
-            clp_config.archive_output.directory,
-            container_clp_config.archive_output.directory,
+            clp_config.archive_output.get_directory(),
+            container_clp_config.archive_output.get_directory(),
         )
 
     container_clp_config.stream_output.directory = pathlib.Path("/") / "mnt" / "stream-output"
@@ -268,6 +270,18 @@ def generate_container_config(
     return container_clp_config, docker_mounts
 
 
+def generate_worker_config(clp_config: CLPConfig) -> WorkerConfig:
+    worker_config = WorkerConfig()
+    worker_config.package = clp_config.package.copy(deep=True)
+    worker_config.archive_output = clp_config.archive_output.copy(deep=True)
+    worker_config.data_directory = clp_config.data_directory
+
+    worker_config.stream_output_dir = clp_config.stream_output.directory
+    worker_config.stream_collection_name = clp_config.results_cache.stream_collection_name
+
+    return worker_config
+
+
 def dump_container_config(
     container_clp_config: CLPConfig, clp_config: CLPConfig, container_name: str
 ) -> Tuple[pathlib.Path, pathlib.Path]:
@@ -482,7 +496,7 @@ def validate_results_cache_config(
 
 def validate_worker_config(clp_config: CLPConfig):
     clp_config.validate_input_logs_dir()
-    clp_config.validate_archive_output_dir()
+    clp_config.validate_archive_output_config()
     clp_config.validate_stream_output_dir()
 
 

components/clp-package-utils/clp_package_utils/scripts/compress.py

@@ -18,15 +18,7 @@
     validate_and_load_db_credentials_file,
 )
 
-# Setup logging
-# Create logger
 logger = logging.getLogger(__file__)
-logger.setLevel(logging.INFO)
-# Setup console logging
-logging_console_handler = logging.StreamHandler()
-logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] [%(name)s] %(message)s")
-logging_console_handler.setFormatter(logging_formatter)
-logger.addHandler(logging_console_handler)
 
 
 def main(argv):
@@ -66,7 +58,7 @@ def main(argv):
         logger.exception("Failed to load config.")
         return -1
 
-    container_name = generate_container_name(JobType.COMPRESSION)
+    container_name = generate_container_name(str(JobType.COMPRESSION))
 
     container_clp_config, mounts = generate_container_config(clp_config, clp_home)
     generated_config_path_on_container, generated_config_path_on_host = dump_container_config(

components/clp-package-utils/clp_package_utils/scripts/decompress.py

@@ -5,7 +5,7 @@
 import sys
 from typing import Optional
 
-from clp_py_utils.clp_config import CLPConfig
+from clp_py_utils.clp_config import CLPConfig, StorageType
 
 from clp_package_utils.general import (
     CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH,
@@ -25,15 +25,7 @@
     validate_path_could_be_dir,
 )
 
-# Setup logging
-# Create logger
-logger = logging.getLogger("clp")
-logger.setLevel(logging.DEBUG)
-# Setup console logging
-logging_console_handler = logging.StreamHandler()
-logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] [%(name)s] %(message)s")
-logging_console_handler.setFormatter(logging_formatter)
-logger.addHandler(logging_console_handler)
+logger = logging.getLogger(__file__)
 
 
 def validate_and_load_config(
@@ -89,7 +81,12 @@ def handle_extract_file_cmd(
     if clp_config is None:
         return -1
 
-    container_name = generate_container_name(JobType.FILE_EXTRACTION)
+    storage_type = clp_config.archive_output.storage.type
+    if StorageType.FS != storage_type:
+        logger.error(f"File extraction is not supported for archive storage type: {storage_type}.")
+        return -1
+
+    container_name = generate_container_name(str(JobType.FILE_EXTRACTION))
     container_clp_config, mounts = generate_container_config(clp_config, clp_home)
     generated_config_path_on_container, generated_config_path_on_host = dump_container_config(
         container_clp_config, clp_config, container_name
@@ -164,7 +161,14 @@ def handle_extract_stream_cmd(
     if clp_config is None:
         return -1
 
-    container_name = generate_container_name(JobType.IR_EXTRACTION)
+    storage_type = clp_config.archive_output.storage.type
+    if StorageType.FS != storage_type:
+        logger.error(
+            f"Stream extraction is not supported for archive storage type: {storage_type}."
+        )
+        return -1
+
+    container_name = generate_container_name(str(JobType.IR_EXTRACTION))
     container_clp_config, mounts = generate_container_config(clp_config, clp_home)
     generated_config_path_on_container, generated_config_path_on_host = dump_container_config(
         container_clp_config, clp_config, container_name

components/clp-package-utils/clp_package_utils/scripts/del_archives.py

@@ -0,0 +1,110 @@
+import argparse
+import logging
+import subprocess
+import sys
+from pathlib import Path
+
+from clp_py_utils.clp_config import StorageType
+
+from clp_package_utils.general import (
+    CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH,
+    dump_container_config,
+    generate_container_config,
+    generate_container_name,
+    generate_container_start_cmd,
+    get_clp_home,
+    load_config_file,
+    validate_and_load_db_credentials_file,
+)
+
+logger = logging.getLogger(__file__)
+
+
+def main(argv):
+    clp_home = get_clp_home()
+    default_config_file_path = clp_home / CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH
+
+    args_parser = argparse.ArgumentParser(
+        description="Deletes archives that fall within the specified time range."
+    )
+    args_parser.add_argument(
+        "--config",
+        "-c",
+        default=str(default_config_file_path),
+        help="CLP package configuration file.",
+    )
+    args_parser.add_argument(
+        "--begin-ts",
+        type=int,
+        default=0,
+        help="Time-range lower-bound (inclusive) as milliseconds from the UNIX epoch.",
+    )
+    args_parser.add_argument(
+        "--end-ts",
+        type=int,
+        required=True,
+        help="Time-range upper-bound (include) as milliseconds from the UNIX epoch.",
+    )
+    parsed_args = args_parser.parse_args(argv[1:])
+
+    # Validate and load config file
+    try:
+        config_file_path = Path(parsed_args.config)
+        clp_config = load_config_file(config_file_path, default_config_file_path, clp_home)
+        clp_config.validate_logs_dir()
+
+        # Validate and load necessary credentials
+        validate_and_load_db_credentials_file(clp_config, clp_home, False)
+    except:
+        logger.exception("Failed to load config.")
+        return -1
+
+    storage_type = clp_config.archive_output.storage.type
+    if StorageType.FS != storage_type:
+        logger.error(f"Archive deletion is not supported for storage type: {storage_type}.")
+        return -1
+
+    # Validate the input timestamp
+    begin_ts = parsed_args.begin_ts
+    end_ts = parsed_args.end_ts
+    if begin_ts > end_ts:
+        logger.error("begin-ts must be <= end-ts")
+        return -1
+    if end_ts < 0 or begin_ts < 0:
+        logger.error("begin_ts and end_ts must be non-negative.")
+        return -1
+
+    container_name = generate_container_name("del-archives")
+
+    container_clp_config, mounts = generate_container_config(clp_config, clp_home)
+    generated_config_path_on_container, generated_config_path_on_host = dump_container_config(
+        container_clp_config, clp_config, container_name
+    )
+
+    necessary_mounts = [mounts.clp_home, mounts.logs_dir, mounts.archives_output_dir]
+    container_start_cmd = generate_container_start_cmd(
+        container_name, necessary_mounts, clp_config.execution_container
+    )
+
+    # fmt: off
+    del_archive_cmd = [
+        "python3",
+        "-m", "clp_package_utils.scripts.native.del_archives",
+        "--config", str(generated_config_path_on_container),
+        str(begin_ts),
+        str(end_ts)
+
+    ]
+    # fmt: on
+
+    cmd = container_start_cmd + del_archive_cmd
+    subprocess.run(cmd, check=True)
+
+    # Remove generated files
+    generated_config_path_on_host.unlink()
+
+    return 0
+
+
+if "__main__" == __name__:
+    sys.exit(main(sys.argv))

components/clp-package-utils/clp_package_utils/scripts/native/compress.py

@@ -23,15 +23,7 @@
     load_config_file,
 )
 
-# Setup logging
-# Create logger
 logger = logging.getLogger(__file__)
-logger.setLevel(logging.INFO)
-# Setup console logging
-logging_console_handler = logging.StreamHandler()
-logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] [%(name)s] %(message)s")
-logging_console_handler.setFormatter(logging_formatter)
-logger.addHandler(logging_console_handler)
 
 
 def print_compression_job_status(job_row, current_time):

components/clp-package-utils/clp_package_utils/scripts/native/decompress.py

@@ -32,15 +32,7 @@
     wait_for_query_job,
 )
 
-# Setup logging
-# Create logger
 logger = logging.getLogger(__file__)
-logger.setLevel(logging.INFO)
-# Setup console logging
-logging_console_handler = logging.StreamHandler()
-logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] [%(name)s] %(message)s")
-logging_console_handler.setFormatter(logging_formatter)
-logger.addHandler(logging_console_handler)
 
 
 def get_orig_file_id(db_config: Database, path: str) -> Optional[str]:
@@ -175,7 +167,7 @@ def validate_and_load_config_file(
     """
     try:
         clp_config = load_config_file(config_file_path, default_config_file_path, clp_home)
-        clp_config.validate_archive_output_dir()
+        clp_config.validate_archive_output_config()
         clp_config.validate_logs_dir()
         return clp_config
     except Exception:
@@ -215,7 +207,7 @@ def handle_extract_file_cmd(
     list_path = parsed_args.files_from
 
     logs_dir = clp_config.logs_directory
-    archives_dir = clp_config.archive_output.directory
+    archives_dir = clp_config.archive_output.get_directory()
 
     # Generate database config file for clp
     db_config_file_path = logs_dir / f".decompress-db-config-{uuid.uuid4()}.yml"