Merge branch 'main' into improve-deps-main-taskfile

Author: davidlion

.github/workflows/clp-uv-checks.yaml

@@ -0,0 +1,52 @@
+name: "clp-uv-checks"
+
+on:
+  pull_request:
+    paths: &monitored_paths
+      - ".github/workflows/clp-uv-checks.yaml"
+      - "components/clp-mcp-server/pyproject.toml"
+      - "components/clp-mcp-server/uv.lock"
+      - "components/clp-package-utils/pyproject.toml"
+      - "components/clp-package-utils/uv.lock"
+      - "components/clp-py-utils/pyproject.toml"
+      - "components/clp-py-utils/uv.lock"
+      - "components/job-orchestration/pyproject.toml"
+      - "components/job-orchestration/uv.lock"
+      - "integration-tests/pyproject.toml"
+      - "integration-tests/uv.lock"
+  push:
+    paths: *monitored_paths
+  schedule:
+    # Run daily at 00:15 UTC (the 15 is to avoid periods of high load)
+    - cron: "15 0 * * *"
+  workflow_dispatch:
+
+concurrency:
+  group: "${{github.workflow}}-${{github.ref}}"
+  # Cancel in-progress jobs for efficiency
+  cancel-in-progress: true
+
+jobs:
+  uv-checks:
+    strategy:
+      matrix:
+        os:
+          - "macos-15"
+          - "ubuntu-22.04"
+          - "ubuntu-24.04"
+    runs-on: "${{matrix.os}}"
+    steps:
+      - uses: "actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8"  # v5.0.0
+        with:
+          submodules: "recursive"
+
+      - name: "Install task"
+        shell: "bash"
+        run: "npm install -g @go-task/[email protected]"
+
+      - name: "Install the latest version of uv"
+        uses: "astral-sh/setup-uv@85856786d1ce8acfbcc2f13a5f3fbd6b938f9f41"  # v7.1.2
+
+      - name: "Validate lock files"
+        shell: "bash"
+        run: "task deps:lock:check-uv"

Cargo.lock

@@ -743,8 +743,7 @@ def start(self) -> None:
         logger.info(f"Starting CLP using Docker Compose ({deployment_type} deployment)...")
 
         cmd = ["docker", "compose", "--project-name", self._project_name]
-        if deployment_type == DeploymentType.BASE:
-            cmd += ["--file", "docker-compose.base.yaml"]
+        cmd += ["--file", self._get_docker_file_name()]
         if self._clp_config.mcp_server is not None:
             cmd += ["--profile", "mcp"]
         cmd += ["up", "--detach", "--wait"]
@@ -795,6 +794,15 @@ def _get_num_workers() -> int:
         # This will change when we move from single to multi-container workers. See y-scope/clp#1424
         return multiprocessing.cpu_count() // 2
 
+    def _get_docker_file_name(self) -> str:
+        """
+        :return: The Docker Compose file name to use based on the config.
+        """
+        deployment_type = self._clp_config.get_deployment_type()
+        if deployment_type == DeploymentType.BASE:
+            return "docker-compose-base.yaml"
+        return "docker-compose.yaml"
+
 
 def get_or_create_instance_id(clp_config: CLPConfig) -> str:
     """

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

@@ -94,6 +94,8 @@ def _generate_compress_cmd(
     if parsed_args.timestamp_key is not None:
         compress_cmd.append("--timestamp-key")
         compress_cmd.append(parsed_args.timestamp_key)
+    if parsed_args.unstructured:
+        compress_cmd.append("--unstructured")
     if parsed_args.tags is not None:
         compress_cmd.append("--tags")
         compress_cmd.append(parsed_args.tags)
@@ -148,6 +150,11 @@ def main(argv):
         "--timestamp-key",
         help="The path (e.g. x.y) for the field containing the log event's timestamp.",
     )
+    args_parser.add_argument(
+        "--unstructured",
+        action="store_true",
+        help="Treat all inputs as unstructured text logs.",
+    )
     args_parser.add_argument(
         "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
     )
@@ -198,11 +205,17 @@ def main(argv):
             logger.error(e)
             return -1
 
-        if parsed_args.timestamp_key is None:
+        if parsed_args.timestamp_key is None and not parsed_args.unstructured:
             logger.warning(
                 "`--timestamp-key` not specified. Events will not have assigned timestamps and can "
                 "only be searched from the command line without a timestamp filter."
             )
+        if parsed_args.timestamp_key is not None and parsed_args.unstructured:
+            parsed_args.timestamp_key = None
+            logger.warning(
+                "`--timestamp-key` and `--unstructured` are not compatible. The input logs will be "
+                "treated as unstructured, and the argument to `--timestamp-key` will be ignored."
+            )
     elif dataset is not None:
         logger.error(f"Dataset selection is not supported for storage engine: {storage_engine}.")
         return -1

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

@@ -97,6 +97,8 @@ def _generate_compress_cmd(
     if parsed_args.timestamp_key is not None:
         compress_cmd.append("--timestamp-key")
         compress_cmd.append(parsed_args.timestamp_key)
+    if parsed_args.unstructured:
+        compress_cmd.append("--unstructured")
     if parsed_args.tags is not None:
         compress_cmd.append("--tags")
         compress_cmd.append(parsed_args.tags)
@@ -178,6 +180,11 @@ def main(argv):
         "--timestamp-key",
         help="The path (e.g. x.y) for the field containing the log event's timestamp.",
     )
+    args_parser.add_argument(
+        "--unstructured",
+        action="store_true",
+        help="Treat all inputs as unstructured text logs.",
+    )
     args_parser.add_argument(
         "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
     )
@@ -253,11 +260,17 @@ def main(argv):
         logger.error(e)
         return -1
 
-    if parsed_args.timestamp_key is None:
+    if parsed_args.timestamp_key is None and not parsed_args.unstructured:
         logger.warning(
             "`--timestamp-key` not specified. Events will not have assigned timestamps and can"
             " only be searched from the command line without a timestamp filter."
         )
+    if parsed_args.timestamp_key is not None and parsed_args.unstructured:
+        parsed_args.timestamp_key = None
+        logger.warning(
+            "`--timestamp-key` and `--unstructured` are not compatible. The input logs will be "
+            "treated as unstructured, and the argument to `--timestamp-key` will be ignored."
+        )
 
     if parsed_args.subcommand == S3_OBJECT_COMPRESSION:
         _validate_s3_object_args(parsed_args, args_parser)

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

@@ -157,6 +157,7 @@ def _generate_clp_io_config(
             paths_to_compress=logs_to_compress,
             timestamp_key=parsed_args.timestamp_key,
             path_prefix_to_remove=str(CONTAINER_INPUT_LOGS_ROOT_DIR),
+            unstructured=parsed_args.unstructured,
         )
     elif input_type != "s3":
         raise ValueError(f"Unsupported input type: `{input_type}`.")
@@ -180,6 +181,7 @@ def _generate_clp_io_config(
             keys=keys,
             aws_authentication=aws_authentication,
             timestamp_key=parsed_args.timestamp_key,
+            unstructured=parsed_args.unstructured,
         )
     elif s3_compress_subcommand == S3_KEY_PREFIX_COMPRESSION:
         if len(urls) != 1:
@@ -195,6 +197,7 @@ def _generate_clp_io_config(
             keys=None,
             aws_authentication=aws_authentication,
             timestamp_key=parsed_args.timestamp_key,
+            unstructured=parsed_args.unstructured,
         )
     else:
         raise ValueError(f"Unsupported S3 compress subcommand: `{s3_compress_subcommand}`.")
@@ -337,6 +340,11 @@ def main(argv):
         "--timestamp-key",
         help="The path (e.g. x.y) for the field containing the log event's timestamp.",
     )
+    args_parser.add_argument(
+        "--unstructured",
+        action="store_true",
+        help="Treat all inputs as unstructured text logs.",
+    )
     args_parser.add_argument(
         "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives."
     )

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

@@ -77,7 +77,7 @@ def create_and_monitor_job_in_db(
 
     if do_count_aggregation is None and count_by_time_bucket_size is None:
         return
-    with pymongo.MongoClient(results_cache.get_uri()) as client:
+    with pymongo.MongoClient(results_cache.get_uri(), directConnection=True) as client:
         search_results_collection = client[results_cache.db_name][str(job_id)]
         if do_count_aggregation is not None:
             for document in search_results_collection.find():

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

@@ -48,8 +48,8 @@
 COMPRESSION_TASKS_TABLE_NAME = "compression_tasks"
 
 # Paths
-CONTAINER_AWS_CONFIG_DIRECTORY = pathlib.Path("/") / ".aws"
 CONTAINER_CLP_HOME = pathlib.Path("/") / "opt" / "clp"
+CONTAINER_AWS_CONFIG_DIRECTORY = CONTAINER_CLP_HOME / ".aws"
 CONTAINER_INPUT_LOGS_ROOT_DIR = pathlib.Path("/") / "mnt" / "logs"
 CLP_DEFAULT_CONFIG_FILE_RELATIVE_PATH = pathlib.Path("etc") / "clp-config.yml"
 CLP_DEFAULT_CREDENTIALS_FILE_PATH = pathlib.Path("etc") / "credentials.yml"

components/clp-py-utils/clp_py_utils/clp_config.py

@@ -5,5 +5,6 @@ edition = "2024"
 
 [dependencies]
 aws-config = "1.8.8"
-aws-sdk-s3 = { version = "1.106.0" }
+aws-sdk-s3 = "1.106.0"
+aws-sdk-sqs = "1.86.0"
 secrecy = { version = "0.10.3", features = ["serde"] }

components/clp-rust-utils/Cargo.toml

@@ -1 +1,2 @@
 pub mod s3;
+pub mod sqs;