feat(presto-clp): Update docs and config generation to support reading archives from S3 and Presto split-filtering config.

docs/src/user-guide/guides-using-presto.md

@@ -164,7 +164,6 @@ The Presto CLP integration has the following limitations at present:
 * Nested fields containing special characters cannot be queried (see [y-scope/presto#8]). Allowed
   characters are alphanumeric characters and underscores. To get around this limitation, you'll
   need to preprocess your logs to remove any special characters.
-* Only logs stored on the filesystem, rather than S3, can be queried through Presto.
 
 These limitations will be addressed in a future release of the Presto integration.
 

tools/deployment/presto-clp/scripts/generate-user-env-vars-file.py → tools/deployment/presto-clp/scripts/init.py

@@ -2,7 +2,7 @@
 import logging
 import sys
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 import yaml
 from dotenv import dotenv_values
@@ -28,7 +28,9 @@ def main(argv=None) -> int:
         argv = sys.argv
 
     args_parser = argparse.ArgumentParser(
-        description="Generates an environment variables file for any user-configured properties."
+        description=(
+            "Generates any necessary config files corresponding to user-configured properties."
+        )
     )
     args_parser.add_argument(
         "--clp-package-dir", help="CLP package directory.", required=True, type=Path
@@ -41,43 +43,49 @@ def main(argv=None) -> int:
     clp_package_dir: Path = parsed_args.clp_package_dir.resolve()
     output_file: Path = parsed_args.output_file
 
+    clp_config_file_path = clp_package_dir / "etc" / "clp-config.yml"
+    if not clp_config_file_path.exists():
+        logger.error(
+            "'%s' doesn't exist. Is '%s' the location of the CLP package?",
+            clp_config_file_path,
+            clp_package_dir.resolve(),
+        )
+        return 1
+
+    with open(clp_config_file_path, "r") as clp_config_file:
+        clp_config = yaml.safe_load(clp_config_file)
+
     env_vars: Dict[str, str] = {}
-    if not _add_clp_env_vars(clp_package_dir, env_vars):
+    if not _add_clp_env_vars(clp_config, clp_package_dir, env_vars):
         return 1
 
     script_dir = Path(__file__).parent.resolve()
     if not _add_worker_env_vars(script_dir.parent / "coordinator-common.env", env_vars):
         return 1
 
+    if not _generate_worker_clp_properties(script_dir.parent / "worker" / "config-template", env_vars):
+        return 1
+
     with open(output_file, "w") as output_file_handle:
         for key, value in env_vars.items():
             output_file_handle.write(f"{key}={value}\n")
 
     return 0
 
 
-def _add_clp_env_vars(clp_package_dir: Path, env_vars: Dict[str, str]) -> bool:
+def _add_clp_env_vars(
+    clp_config: Dict[str, Any], clp_package_dir: Path, env_vars: Dict[str, str]
+) -> bool:
     """
     Adds environment variables for CLP config values to `env_vars`.
 
+    :param clp_config:
     :param clp_package_dir:
     :param env_vars:
     :return: Whether the environment variables were successfully added.
     """
     env_vars["PRESTO_COORDINATOR_CLPPROPERTIES_METADATA_TABLE_PREFIX"] = "clp_"
 
-    clp_config_file_path = clp_package_dir / "etc" / "clp-config.yml"
-    if not clp_config_file_path.exists():
-        logger.error(
-            "'%s' doesn't exist. Is '%s' the location of the CLP package?",
-            clp_config_file_path,
-            clp_package_dir.resolve(),
-        )
-        return False
-
-    with open(clp_config_file_path, "r") as clp_config_file:
-        clp_config = yaml.safe_load(clp_config_file)
-
     database_type = _get_config_value(clp_config, "database.type", "mariadb")
     if "mariadb" != database_type and "mysql" != database_type:
         logger.error(
@@ -98,24 +106,61 @@ def _add_clp_env_vars(clp_package_dir: Path, env_vars: Dict[str, str]) -> bool:
     clp_archive_output_storage_type = _get_config_value(
         clp_config, "archive_output.storage.type", "fs"
     )
-    if "fs" != clp_archive_output_storage_type:
+    env_vars["CLP_ARCHIVES_DIR"] = _resolve_archives_dir(
+        clp_package_dir,
+        _get_config_value(clp_config, "archive_output.storage.directory"),
+        clp_package_dir
+        / "var"
+        / "data"
+        / ("archives" if clp_archive_output_storage_type == "fs" else "staged-archives"),
+    )
+    if "fs" == clp_archive_output_storage_type:
+        pass
+    elif "s3" == clp_archive_output_storage_type:
+        s3_config_key_prefix = f"archive_output.storage.s3_config"
+        s3_credentials_key_prefix = f"{s3_config_key_prefix}.aws_authentication.credentials"
+
+        s3_access_key_id = _get_config_value(
+            clp_config, f"{s3_credentials_key_prefix}.access_key_id"
+        )
+
+        s3_bucket = _get_config_value(clp_config, f"{s3_config_key_prefix}.bucket")
+        s3_region_code = _get_config_value(clp_config, f"{s3_config_key_prefix}.region_code")
+
+        s3_secret_access_key = _get_config_value(
+            clp_config, f"{s3_credentials_key_prefix}.secret_access_key"
+        )
+
+        # Validate required S3 fields
+        missing = []
+
+        for k, v in {
+            f"{s3_credentials_key_prefix}.access_key_id": s3_access_key_id,
+            f"{s3_credentials_key_prefix}.secret_access_key": s3_secret_access_key,
+            f"{s3_config_key_prefix}.bucket": s3_bucket,
+            f"{s3_config_key_prefix}.region_code": s3_region_code,
+        }.items():
+            if not v:
+                missing.append(k)
+
+        if missing:
+            logger.error("Missing required S3 config key(s): %s", ", ".join(missing))
+            return False
+
+        env_vars["PRESTO_WORKER_CLPPROPERTIES_STORAGE_TYPE"] = "s3"
+        env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_AUTH_PROVIDER"] = "clp_package"
+        env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_ACCESS_KEY_ID"] = s3_access_key_id
+        s3_end_point = f"https://{s3_bucket}.s3.{s3_region_code}.amazonaws.com/"
+        env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_END_POINT"] = s3_end_point
+        env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_SECRET_ACCESS_KEY"] = s3_secret_access_key
+    else:
         logger.error(
-            "Expected CLP's archive_output.storage.type to be fs but found '%s'. Presto"
-            " currently only supports reading archives from the fs storage type.",
+            "Expected CLP's archive_output.storage.type to be fs or s3 but found '%s'. Presto"
+            " currently only supports reading archives from the fs or s3 storage type.",
             clp_archive_output_storage_type,
         )
         return False
 
-    clp_archives_dir = _get_config_value(
-        clp_config,
-        "archive_output.storage.directory",
-        str(clp_package_dir / "var" / "data" / "archives"),
-    )
-    if Path(clp_archives_dir).is_absolute():
-        env_vars["CLP_ARCHIVES_DIR"] = clp_archives_dir
-    else:
-        env_vars["CLP_ARCHIVES_DIR"] = str(clp_package_dir / clp_archives_dir)
-
     credentials_file_path = clp_package_dir / "etc" / "credentials.yml"
     if not credentials_file_path.exists():
         logger.error("'%s' doesn't exist. Did you start CLP?", credentials_file_path)
@@ -137,6 +182,11 @@ def _add_clp_env_vars(clp_package_dir: Path, env_vars: Dict[str, str]) -> bool:
     return True
 
 
+def _resolve_archives_dir(base_dir: Path, configured: Optional[str], default: Path) -> str:
+    effective = configured or str(default)
+    return effective if Path(effective).is_absolute() else str(base_dir / effective)
+
+
 def _add_worker_env_vars(coordinator_common_env_file_path: Path, env_vars: Dict[str, str]) -> bool:
     """
     Adds environment variables for worker config values to `env_vars`.
@@ -163,6 +213,24 @@ def _add_worker_env_vars(coordinator_common_env_file_path: Path, env_vars: Dict[
     return True
 
 
+def _generate_worker_clp_properties(worker_config_template_path: Path, env_vars: Dict[str, str]) -> bool:
+    """
+    Generates a clp.properties for worker config.
+
+    :param worker_config_template_path:
+    """
+    clp_properties_path = worker_config_template_path / "clp.properties"
+    config_options = ["connector.name=clp"]
+    if "s3" == env_vars["PRESTO_WORKER_CLPPROPERTIES_STORAGE_TYPE"]:
+        config_options.append("clp.storage-type=s3")
+        config_options.append(f'clp.s3-auth-provider={env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_AUTH_PROVIDER"]}')
+        config_options.append(f'clp.s3-access-key-id={env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_ACCESS_KEY_ID"]}')
+        config_options.append(f'clp.s3-end-point={env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_END_POINT"]}')
+        config_options.append(f'clp.s3-secret-access-key={env_vars["PRESTO_WORKER_CLPPROPERTIES_S3_SECRET_ACCESS_KEY"]}')
+    with clp_properties_path.open("w", encoding="utf-8") as f:
+        f.write("\n".join(config_options) + "\n")
+
+
 def _get_config_value(config: dict, key: str, default_value: Optional[str] = None) -> str:
     """
     Gets the value corresponding to `key` from `config` if it exists.

tools/deployment/presto-clp/scripts/set-up-config.sh

@@ -23,6 +23,6 @@ echo "Installing required Python packages..."
 pip3 install -r "${script_dir}/requirements.txt"
 
 echo "Generating environment variables file for user-configured properties..."
-python3 "${script_dir}/generate-user-env-vars-file.py" \
+python3 "${script_dir}/init.py" \
     --clp-package-dir "${clp_package_dir}" \
     --output-file "${script_dir}/../.env"

tools/deployment/presto-clp/worker/scripts/generate-configs.sh

@@ -48,13 +48,19 @@ update_config_file() {
     local file_path=$1
     local key=$2
     local value=$3
+    local sensitive=${4:-false}
 
     if grep --quiet "^${key}=.*$" "$file_path"; then
         sed --in-place "s|^${key}=.*|${key}=${value}|" "$file_path"
     else
         echo "${key}=${value}" >>"$file_path"
     fi
-    log "INFO" "Set ${key}=${value} in ${file_path}"
+    if [[ "$sensitive" == "true" ]]; then
+        local masked="****"
+        log "INFO" "Set ${key}=${masked} in ${file_path}"
+    else
+        log "INFO" "Set ${key}=${value} in ${file_path}"
+    fi
 }
 
 apt-get update && apt-get install --assume-yes --no-install-recommends jq wget