Merge branch 'main' into bundle-mariadb

# Conflicts:
#	tools/docker-images/clp-package/Dockerfile

Author: junhaoliao

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

@@ -2,7 +2,17 @@ name: "clp-rust-checks"
 
 on:
   pull_request:
+    paths: &monitored_paths
+      - ".cargo/**"
+      - "Cargo.lock"
+      - "Cargo.toml"
+      - ".github/workflows/clp-rust-checks.yaml"
+      - "components/clp-rust-utils/**"
+      - "components/log-ingestor/**"
+      - "taskfile.yaml"
+      - "taskfiles/**"
   push:
+    paths: *monitored_paths
   schedule:
     # Run daily at 00:15 UTC (the 15 is to avoid periods of high load)
     - cron: "15 0 * * *"
@@ -31,6 +41,10 @@ jobs:
         shell: "bash"
         run: "npm install -g @go-task/[email protected]"
 
+      - name: "Validate lock files"
+        shell: "bash"
+        run: "task deps:lock:check-rust"
+
       - name: "Lint"
         shell: "bash"
         run: "task lint:check-rust"

Cargo.lock

@@ -5,6 +5,7 @@
 
 import aiomysql
 import msgpack
+from clp_py_utils.clp_config import CLP_DEFAULT_DATASET_NAME
 from pymongo import AsyncMongoClient
 
 from .constants import (
@@ -37,27 +38,32 @@ def __init__(self, clp_config: Any) -> None:
             "db": clp_config.database.name,
         }
 
-    async def submit_query(self, query: str, begin_ts: int, end_ts: int) -> str:
+        self._webui_addr = f"http://{clp_config.webui.host}:{clp_config.webui.port}"
+
+    async def submit_query(
+        self, query: str, begin_ts: int | None = None, end_ts: int | None = None
+    ) -> str:
         """
         Submits a query to the CLP database and returns the ID of the query.
 
         :param query: The query string.
         :param begin_ts: The beginning timestamp of the query range.
         :param end_ts: The end timestamp of the query range.
         :return: The ID assigned to the query.
-        :raise ValueError: If ``end_ts`` is smaller than ``begin_ts``.
-        :raise aiomysql.Error: If there is an error connecting to or querying MariaDB.
-        :raise pymongo.errors.PyMongoError: If there is an error interacting with MongoDB.
-        :raise Exception: For any other unexpected errors.
+        :raise: ValueError if `end_ts` is smaller than `begin_ts`.
+        :raise: RuntimeError if it fails to retrieve the ID of the submitted query.
+        :raise: aiomysql.Error if there is an error connecting to or querying MariaDB.
+        :raise: pymongo.errors.PyMongoError if there is an error interacting with MongoDB.
+        :raise: Exception for any other unexpected errors.
         """
-        if end_ts < begin_ts:
+        if begin_ts is not None and end_ts is not None and end_ts < begin_ts:
             err_msg = f"end_ts {end_ts} is smaller than begin_ts {begin_ts}."
             raise ValueError(err_msg)
 
         job_config = msgpack.packb(
             {
                 "begin_timestamp": begin_ts,
-                "dataset": None,
+                "dataset": CLP_DEFAULT_DATASET_NAME,
                 "end_timestamp": end_ts,
                 "ignore_case": True,
                 "max_num_results": SEARCH_MAX_NUM_RESULTS,
@@ -118,9 +124,10 @@ async def wait_query_completion(self, query_id: str, timeout: float | None = Non
 
         :param query_id: The ID of the query.
         :param timeout: Maximum time to wait in seconds, or None for no timeout.
-        :raise aiomysql.Error: If there is an error connecting to or querying MariaDB.
-        :raise ValueError: When the query is not found.
-        :raise RuntimeError: When the query fails or is cancelled.
+        :raise: aiomysql.Error if there is an error connecting to or querying MariaDB.
+        :raise: ValueError if the query is not found.
+        :raise: RuntimeError if the query fails or is cancelled.
+        :raise: TimeoutError if the timeout is reached before the query completes.
         """
         waiting_states = {QueryJobStatus.PENDING, QueryJobStatus.RUNNING, QueryJobStatus.CANCELLING}
         error_states = {QueryJobStatus.FAILED, QueryJobStatus.CANCELLED, QueryJobStatus.KILLED}
@@ -132,7 +139,10 @@ async def wait_query_completion(self, query_id: str, timeout: float | None = Non
             if status == QueryJobStatus.SUCCEEDED:
                 break
             if status in error_states:
-                err_msg = f"Query job with ID {query_id} ended in status {status.name}."
+                err_msg = (
+                    f"Query job with ID {query_id} ended in "
+                    f"status {QueryJobStatus(status).name}."
+                )
                 raise RuntimeError(err_msg)
             if status not in waiting_states:
                 err_msg = f"Query job with ID {query_id} has unknown status {status}."
@@ -155,6 +165,13 @@ async def read_results(self, query_id: str) -> list[dict]:
         results = []
 
         async for doc in collection.find({}, limit=SEARCH_MAX_NUM_RESULTS):
+            doc["link"] = (
+                f"{self._webui_addr}/streamFile?type=json"
+                f'&streamId={doc["archive_id"]}'
+                f"&dataset={CLP_DEFAULT_DATASET_NAME}"
+                f'&logEventIdx={doc["log_event_ix"]}'
+            )
+            doc["_id"] = None
             results.append(doc)
 
         return results

components/clp-mcp-server/clp_mcp_server/clp_connector.py

@@ -2,35 +2,55 @@
 
 import ipaddress
 import logging
+import os
 import socket
 import sys
+from pathlib import Path
 
 import click
+from clp_py_utils.clp_config import CLPConfig, MCP_SERVER_COMPONENT_NAME
+from clp_py_utils.clp_logging import get_logger, get_logging_formatter, set_logging_level
+from clp_py_utils.core import read_yaml_config_file
+from pydantic import ValidationError
 
 from .server import create_mcp_server
 
+logger = get_logger(MCP_SERVER_COMPONENT_NAME)
+
 
 @click.command()
 @click.option(
     "--host", type=str, default="127.0.0.1", help="The server's host address (default: 127.0.0.1)."
 )
 @click.option("--port", type=int, default=8000, help="The server's port number (default: 8000).")
-def main(host: str, port: int) -> None:
+@click.option(
+    "--config-path",
+    type=click.Path(exists=True),
+    default="/etc/clp-config.yml",
+    help="The path to server's configuration file (default: /etc/clp-config.yml).",
+)
+def main(host: str, port: int, config_path: Path) -> int:
     """
     Runs the CLP MCP server with HTTP transport.
 
     :param host: The server's host address (IP address or hostname).
     :param port: The server's port number (1-65535).
+    :param config_path: The path to server's configuration file.
+    :return: Exit code (0 for success, non-zero for failure).
     """
-    logging.basicConfig(
-        level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-    )
-    logger = logging.getLogger(__name__)
+    # Setup logging to file
+    log_file_path = Path(os.getenv("CLP_LOGS_DIR")) / "mcp_server.log"
+    logging_file_handler = logging.FileHandler(filename=log_file_path, encoding="utf-8")
+    logging_file_handler.setFormatter(get_logging_formatter())
+    logger.addHandler(logging_file_handler)
+    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
+
+    exit_code = 0
 
     # Validate host and port
     if len(host.strip()) == 0:
         logger.error("Host cannot be empty.")
-        sys.exit(1)
+        exit_code = 1
 
     # Validate host format (IP address or resolvable hostname)
     try:
@@ -44,21 +64,32 @@ def main(host: str, port: int) -> None:
                 "Host validation failed: '%s' is not a valid IP address and DNS resolution failed.",
                 host,
             )
-            sys.exit(1)
+            exit_code = 1
 
     max_port = 65535
     if port <= 0 or port > max_port:
         logger.error("Port must be between 1 and %d, got: %d.", max_port, port)
-        sys.exit(1)
+        exit_code = 1
 
     try:
-        mcp = create_mcp_server()
+        clp_config = CLPConfig.model_validate(read_yaml_config_file(config_path))
+    except ValidationError:
+        logger.exception("Configuration validation failed.")
+        exit_code = 1
+    except Exception:
+        logger.exception("Failed to load configuration.")
+        exit_code = 1
+
+    try:
+        mcp = create_mcp_server(clp_config)
         logger.info("Starting CLP MCP Server on %s:%d.", host, port)
         mcp.run(transport="streamable-http", host=host, port=port)
     except Exception:
         logger.exception("Failed to start MCP server.")
-        sys.exit(1)
+        exit_code = 1
+
+    return exit_code
 
 
 if __name__ == "__main__":
-    main()
+    sys.exit(main())

components/clp-mcp-server/clp_mcp_server/clp_mcp_server.py

@@ -6,20 +6,25 @@
 # 10 minutes
 SESSION_TTL_SECONDS = 600
 
+TIMESTAMP_NOT_AVAILABLE = "N/A"
+
 SERVER_NAME = "clp-mcp-server"
 
 # fmt: off
 SYSTEM_PROMPT = (
 "You are an AI assistant that helps users query a log database using KQL (Kibana Query Language)."
 " You should generate a KQL query that accurately expresses the user's intent. The generated KQL"
-" query should be as specific as possible to minimize the number of log messages returned.\n\n"
+" query should be as specific as possible to minimize the number of log messages returned. When "
+"displaying log messages, wrap them in hyperlinks with the `link` field from the search result.\n\n"
 "You should consider the following guidelines to generate KQL queries efficiently:\n"
 "- Use specific field names and values to narrow down the search.\n"
 "- Avoid using wildcards (`*`) unless absolutely necessary, as they can lead to large result"
 " sets.\n"
 "- Use logical operators (`AND`, `OR`, `NOT`) to combine one or more key-value searches.\n"
-"- Consider the time range of the logs you are searching. If the user specifies a time range,"
-" include it in the KQL query.\n"
+"- Consider specifying a time range to narrow down the search. Use"
+" `search_by_kql_with_timestamp_range` with your KQL query and explicit start and end timestamps."
+" Timestamps must follow the ISO 8601 UTC format (`YYYY-MM-DDTHH:mm:ss.fffZ`), where the trailing"
+" `Z` indicates UTC.\n"
 "- If the user query is ambiguous or lacks detail, ask clarifying questions to better understand"
 " their intent before generating the KQL query.\n"
 "- Always ensure that the generated KQL query is syntactically correct and can be executed without"