y-scope · junhaoliao · Nov 14, 2025 · Nov 16, 2025 · Nov 16, 2025 · Nov 16, 2025
@@ -1,11 +1,7 @@
 use anyhow::Context;
 use clap::Parser;
 use clp_rust_utils::{clp_config::package, serde::yaml};
-use tracing_appender::{
-    non_blocking::WorkerGuard,
-    rolling::{RollingFileAppender, Rotation},
-};
-use tracing_subscriber::{self, fmt::writer::MakeWriterExt};
+use tracing_subscriber;
 
 #[derive(Parser)]
 #[command(version, about = "API Server for CLP.")]
@@ -42,13 +38,8 @@ fn read_config_and_credentials(
     Ok((config, credentials))
 }
 
-fn set_up_logging() -> anyhow::Result<WorkerGuard> {
-    let logs_directory =
-        std::env::var("CLP_LOGS_DIR").context("Expect `CLP_LOGS_DIR` environment variable.")?;
-    let logs_directory = std::path::Path::new(logs_directory.as_str());
-    let file_appender =
-        RollingFileAppender::new(Rotation::HOURLY, logs_directory, "api_server.log");
-    let (non_blocking_writer, guard) = tracing_appender::non_blocking(file_appender);
+fn set_up_logging() {
+    // Logs to stdout only - Docker's fluentd driver captures and forwards to Fluent Bit
     tracing_subscriber::fmt()
         .event_format(
             tracing_subscriber::fmt::format()
@@ -60,9 +51,8 @@ fn set_up_logging() -> anyhow::Result<WorkerGuard> {
         )
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .with_ansi(false)
-        .with_writer(std::io::stdout.and(non_blocking_writer))
+        .with_writer(std::io::stdout)
         .init();
-    Ok(guard)
 }
 
 async fn shutdown_signal() {
@@ -81,7 +71,7 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
     let (config, credentials) = read_config_and_credentials(&args)?;
-    let _guard = set_up_logging()?;
+    set_up_logging();
 
     let addr = format!(
         "{}:{}",

@@ -1,15 +1,13 @@
 """CLP MCP Server entry point."""
 
 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.clp_logging import get_logger, set_logging_level
 from clp_py_utils.core import read_yaml_config_file
 from pydantic import ValidationError
 
@@ -38,11 +36,7 @@ def main(host: str, port: int, config_path: Path) -> int:
     :param config_path: The path to server's configuration file.
     :return: Exit code (0 for success, non-zero for failure).
     """
-    # 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 from environment
     set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
 
     exit_code = 0

@@ -32,6 +32,7 @@
     DatabaseEngine,
     DB_COMPONENT_NAME,
     DeploymentType,
+    FLUENT_BIT_COMPONENT_NAME,
     GARBAGE_COLLECTOR_COMPONENT_NAME,
     LOG_INGESTOR_COMPONENT_NAME,
     MCP_SERVER_COMPONENT_NAME,
@@ -711,6 +712,7 @@ def _set_up_env_for_webui(self, container_clp_config: ClpConfig) -> EnvVarsDict:
             "LogViewerDir": str(container_webui_dir / "yscope-log-viewer"),
             "StreamTargetUncompressedSize": self._clp_config.stream_output.target_uncompressed_size,
             "ClpQueryEngine": self._clp_config.package.query_engine,
+            "OperationalLogsDir": "/var/log/clp",
         }
 
         stream_storage = self._clp_config.stream_output.storage
@@ -852,6 +854,28 @@ def _set_up_env_for_garbage_collector(self) -> EnvVarsDict:
 
         return env_vars
 
+    def _set_up_env_for_fluent_bit(self) -> EnvVarsDict:
+        """
+        Sets up environment variables for the Fluent Bit component.
+
+        Fluent Bit receives logs from Docker's fluentd logging driver and writes
+        them to the logs directory for hot tier access.
+
+        :return: Dictionary of environment variables necessary to launch the component.
+        """
+        component_name = FLUENT_BIT_COMPONENT_NAME
+        logger.info(f"Setting up environment for {component_name}...")
+
+        env_vars = EnvVarsDict()
+
+        # Paths
+        fluent_bit_conf_dir = self._conf_dir / "fluent-bit"
+        env_vars |= {
+            "CLP_FLUENT_BIT_CONF_DIR_HOST": str(fluent_bit_conf_dir),
+        }
+
+        return env_vars
+
     def _read_and_update_settings_json(
         self, settings_file_path: pathlib.Path, updates: dict[str, Any]
     ) -> dict[str, Any]:
@@ -997,6 +1021,7 @@ def set_up_env(self) -> None:
         env_vars |= self._set_up_env_for_webui(container_clp_config)
         env_vars |= self._set_up_env_for_mcp_server()
         env_vars |= self._set_up_env_for_garbage_collector()
+        env_vars |= self._set_up_env_for_fluent_bit()
 
         # Write the environment variables to the `.env` file.
         with open(f"{self._clp_home}/.env", "w") as env_file:

@@ -45,6 +45,7 @@
 WEBUI_COMPONENT_NAME = "webui"
 MCP_SERVER_COMPONENT_NAME = "mcp_server"
 GARBAGE_COLLECTOR_COMPONENT_NAME = "garbage_collector"
+FLUENT_BIT_COMPONENT_NAME = "fluent_bit"
 
 # Action names
 ARCHIVE_MANAGER_ACTION_NAME = "archive_manager"

@@ -347,7 +347,6 @@ def run_clp(
     worker_config: WorkerConfig,
     clp_config: ClpIoConfig,
     clp_home: pathlib.Path,
-    logs_dir: pathlib.Path,
     job_id: int,
     task_id: int,
     tag_ids: list[int],
@@ -362,7 +361,6 @@ def run_clp(
     :param worker_config: WorkerConfig
     :param clp_config: ClpIoConfig
     :param clp_home:
-    :param logs_dir:
     :param job_id:
     :param task_id:
     :param tag_ids:
@@ -444,17 +442,22 @@ def cleanup_temporary_files():
         if converted_inputs_dir is not None:
             shutil.rmtree(converted_inputs_dir)
 
-    # Open stderr log file
-    stderr_log_path = logs_dir / f"{instance_id_str}-stderr.log"
-    stderr_log_file = open(stderr_log_path, "w")
+    def log_stderr(stderr_data: bytes, process_name: str):
+        """Log stderr content through the logger (captured by Docker's fluentd driver)."""
+        stderr_text = stderr_data.decode("utf-8").strip()
+        if stderr_text:
+            for line in stderr_text.split("\n"):
+                logger.info(f"[{process_name} stderr] {line}")
 
     conversion_return_code = 0
     if conversion_cmd is not None:
         logger.debug("Execute log-converter with command: %s", conversion_cmd)
         conversion_proc = subprocess.Popen(
-            conversion_cmd, stdout=subprocess.DEVNULL, stderr=stderr_log_file, env=conversion_env
+            conversion_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=conversion_env
         )
-        conversion_return_code = conversion_proc.wait()
+        _, stderr_data = conversion_proc.communicate()
+        conversion_return_code = conversion_proc.returncode
+        log_stderr(stderr_data, "log-converter")
 
     if conversion_return_code != 0:
         cleanup_temporary_files()
@@ -464,16 +467,15 @@ def cleanup_temporary_files():
         worker_output = {
             "total_uncompressed_size": 0,
             "total_compressed_size": 0,
-            "error_message": f"Check logs in {stderr_log_path}",
+            "error_message": "See worker logs (captured by Fluent Bit)",
         }
-        stderr_log_file.close()
         return CompressionTaskStatus.FAILED, worker_output
 
     # Start compression
     logger.debug("Compressing...")
     compression_successful = False
     proc = subprocess.Popen(
-        compression_cmd, stdout=subprocess.PIPE, stderr=stderr_log_file, env=compression_env
+        compression_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=compression_env
     )
 
     # Compute the total amount of data compressed
@@ -550,23 +552,27 @@ def cleanup_temporary_files():
                     )
 
                     try:
-                        subprocess.run(
+                        result = subprocess.run(
                             indexer_cmd,
                             stdout=subprocess.DEVNULL,
-                            stderr=stderr_log_file,
+                            stderr=subprocess.PIPE,
                             check=True,
                             env=indexer_env,
                         )
-                    except subprocess.CalledProcessError:
+                        log_stderr(result.stderr, "indexer")
+                    except subprocess.CalledProcessError as e:
+                        log_stderr(e.stderr if e.stderr else b"", "indexer")
                         logger.exception("Failed to index archive.")
 
             if enable_s3_write:
                 archive_path.unlink()
 
         last_archive_stats = stats
 
-    # Wait for compression to finish
+    # Wait for compression to finish and capture stderr
+    compression_stderr = proc.stderr.read()
     return_code = proc.wait()
+    log_stderr(compression_stderr, "compression")
 
     if 0 != return_code:
         logger.error(f"Failed to compress, return_code={return_code!s}")
@@ -576,9 +582,6 @@ def cleanup_temporary_files():
 
     logger.debug("Compressed.")
 
-    # Close stderr log file
-    stderr_log_file.close()
-
     worker_output = {
         "total_uncompressed_size": total_uncompressed_size,
         "total_compressed_size": total_compressed_size,
@@ -588,7 +591,7 @@ def cleanup_temporary_files():
         return CompressionTaskStatus.SUCCEEDED, worker_output
     error_msgs = []
     if compression_successful is False:
-        error_msgs.append(f"See logs {stderr_log_path}")
+        error_msgs.append("See worker logs (captured by Fluent Bit)")
     if s3_error is not None:
         error_msgs.append(s3_error)
     worker_output["error_message"] = "\n".join(error_msgs)
@@ -606,10 +609,8 @@ def compression_entry_point(
 ):
     clp_home = pathlib.Path(os.getenv("CLP_HOME"))
 
-    # Set logging level
-    logs_dir = pathlib.Path(os.getenv("CLP_LOGS_DIR"))
-    clp_logging_level = str(os.getenv("CLP_LOGGING_LEVEL"))
-    set_logging_level(logger, clp_logging_level)
+    # Set logging level from environment
+    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
 
     # Load configuration
     try:
@@ -637,7 +638,6 @@ def compression_entry_point(
         worker_config,
         clp_io_config,
         clp_home,
-        logs_dir,
         job_id,
         task_id,
         tag_ids,

@@ -191,10 +191,8 @@ def extract_stream(
 ) -> dict[str, Any]:
     task_name = "Stream Extraction"
 
-    # Setup logging to file
-    clp_logs_dir = Path(os.getenv("CLP_LOGS_DIR"))
-    clp_logging_level = os.getenv("CLP_LOGGING_LEVEL")
-    set_logging_level(logger, clp_logging_level)
+    # Set logging level from environment
+    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
 
     logger.info(f"Started {task_name} task for job {job_id}")
 
@@ -242,7 +240,6 @@ def extract_stream(
     task_results, task_stdout_str = run_query_task(
         sql_adapter=sql_adapter,
         logger=logger,
-        clp_logs_dir=clp_logs_dir,
         task_command=task_command,
         env_vars=core_clp_env_vars,
         task_name=task_name,
@@ -289,7 +286,7 @@ def extract_stream(
 
         if upload_error:
             task_results.status = QueryTaskStatus.FAILED
-            task_results.error_log_path = str(os.getenv("CLP_WORKER_LOG_PATH"))
+            task_results.error_log_path = "See worker logs (captured by Fluent Bit)"
         else:
             logger.info("Finished uploading streams.")
 

@@ -199,10 +199,8 @@ def search(
 ) -> dict[str, Any]:
     task_name = "search"
 
-    # Setup logging to file
-    clp_logs_dir = Path(os.getenv("CLP_LOGS_DIR"))
-    clp_logging_level = os.getenv("CLP_LOGGING_LEVEL")
-    set_logging_level(logger, clp_logging_level)
+    # Set logging level from environment
+    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))
 
     logger.info(f"Started {task_name} task for job {job_id}")
 
@@ -242,7 +240,6 @@ def search(
     task_results, _ = run_query_task(
         sql_adapter=sql_adapter,
         logger=logger,
-        clp_logs_dir=clp_logs_dir,
         task_command=task_command,
         env_vars=core_clp_env_vars,
         task_name=task_name,
@@ -268,7 +265,7 @@ def search(
         except Exception as err:
             logger.error(f"Failed to upload query results {dest_path}: {err}")
             task_results.status = QueryTaskStatus.FAILED
-            task_results.error_log_path = str(os.getenv("CLP_WORKER_LOG_PATH"))
+            task_results.error_log_path = "See worker logs (captured by Fluent Bit)"
 
         src_file.unlink()