refactor(helm)!: Remove all host-path volume mounts. (y-scope#2023)

Co-authored-by: hoophalab <200652805+hoophalab@users.noreply.github.com>

Author: junhaoliao

Cargo.lock

@@ -36,7 +36,5 @@ thiserror = "2.0.18"
 tokio = { version = "1.49.0", features = ["full"] }
 tower-http = { version = "0.6.8", features = ["cors"] }
 tracing = "0.1.44"
-tracing-appender = "0.2.4"
-tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter", "fmt", "std"] }
 utoipa = { version = "5.4.0", features = ["axum_extras"] }
 utoipa-axum = "0.2.0"

components/api-server/Cargo.toml

@@ -1,11 +1,6 @@
 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};
 
 #[derive(Parser)]
 #[command(version, about = "API Server for CLP.")]
@@ -40,29 +35,6 @@ 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);
-    tracing_subscriber::fmt()
-        .event_format(
-            tracing_subscriber::fmt::format()
-                .with_level(true)
-                .with_target(false)
-                .with_file(true)
-                .with_line_number(true)
-                .json(),
-        )
-        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
-        .with_ansi(false)
-        .with_writer(std::io::stdout.and(non_blocking_writer))
-        .init();
-    Ok(guard)
-}
-
 async fn shutdown_signal() {
     let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
         .expect("failed to listen for SIGTERM");
@@ -79,7 +51,7 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
     let (config, credentials) = read_config_and_credentials(&args)?;
-    let _guard = set_up_logging()?;
+    let _guard = clp_rust_utils::logging::set_up_logging("api_server.log");
 
     let api_server_config = config
         .api_server

components/api-server/src/bin/api_server.rs

@@ -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 configure_logging, get_logger
 from clp_py_utils.core import read_yaml_config_file
 from pydantic import ValidationError
 
@@ -38,12 +36,8 @@ 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(logger, os.getenv("CLP_LOGGING_LEVEL"))
+    # Setup optional file logging and logging level.
+    configure_logging(logger, "mcp_server")
 
     exit_code = 0
 

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

@@ -1,4 +1,6 @@
 import logging
+import os
+import pathlib
 from typing import get_args, Literal
 
 LoggingLevel = Literal[
@@ -26,10 +28,36 @@ def get_logger(name: str):
     return logger
 
 
-def set_logging_level(logger: logging.Logger, level: str):
+def set_logging_level(logger: logging.Logger, level: str | None):
+    if level is None:
+        logger.setLevel(logging.INFO)
+        return
     if level not in get_args(LoggingLevel):
         logger.warning(f"Invalid logging level: {level}, using INFO as default")
         logger.setLevel(logging.INFO)
         return
 
     logger.setLevel(level)
+
+
+def configure_logging(
+    logger: logging.Logger,
+    component_name: str,
+):
+    """
+    Configures file logging and the logging level for a logger using environment variables.
+
+    If ``CLP_LOGS_DIR`` is set, a :class:`logging.FileHandler` writing to
+    ``<CLP_LOGS_DIR>/<component_name>.log`` is added to the logger.
+    ``CLP_LOGGING_LEVEL`` is used to set the logging level (defaults to INFO if unset/invalid).
+
+    :param logger: The logger to configure.
+    :param component_name: Used as the log filename stem and in log messages.
+    """
+    logs_dir = os.getenv("CLP_LOGS_DIR")
+    if logs_dir:
+        log_file = pathlib.Path(logs_dir) / f"{component_name}.log"
+        logging_file_handler = logging.FileHandler(filename=log_file, encoding="utf-8")
+        logging_file_handler.setFormatter(get_logging_formatter())
+        logger.addHandler(logging_file_handler)
+    set_logging_level(logger, os.getenv("CLP_LOGGING_LEVEL"))

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

@@ -17,6 +17,8 @@ serde_yaml = "0.9.34"
 sqlx = { version = "0.8.6", features = ["runtime-tokio", "mysql"] }
 strum = "0.28.0"
 thiserror = "2.0.18"
+tracing-appender = "0.2.4"
+tracing-subscriber = { version = "0.3.22", features = ["json", "env-filter", "fmt", "std"] }
 utoipa = { version = "5.4.0" }
 
 [dev-dependencies]

components/clp-rust-utils/Cargo.toml

@@ -3,6 +3,7 @@ pub mod clp_config;
 pub mod database;
 mod error;
 pub mod job_config;
+pub mod logging;
 pub mod s3;
 pub mod serde;
 pub mod sqs;

components/clp-rust-utils/src/lib.rs

@@ -0,0 +1,39 @@
+use tracing_appender::{
+    non_blocking::WorkerGuard,
+    rolling::{RollingFileAppender, Rotation},
+};
+use tracing_subscriber::{self, fmt::writer::MakeWriterExt};
+
+/// Initializes the global tracing subscriber with JSON-formatted output to stdout.
+///
+/// If the `CLP_LOGS_DIR` environment variable is set, logs are also written to a
+/// rolling file (`{log_filename}`) in that directory. The returned [`WorkerGuard`]
+/// must be held for the lifetime of the program to ensure file logs are flushed.
+pub fn set_up_logging(log_filename: &str) -> Option<WorkerGuard> {
+    let subscriber = tracing_subscriber::fmt()
+        .event_format(
+            tracing_subscriber::fmt::format()
+                .with_level(true)
+                .with_target(false)
+                .with_file(true)
+                .with_line_number(true)
+                .json(),
+        )
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .with_ansi(false);
+
+    if let Ok(logs_directory) = std::env::var("CLP_LOGS_DIR") {
+        let logs_directory = std::path::Path::new(logs_directory.as_str());
+        let file_appender =
+            RollingFileAppender::new(Rotation::HOURLY, logs_directory, log_filename);
+        let (non_blocking_writer, guard) = tracing_appender::non_blocking(file_appender);
+
+        subscriber
+            .with_writer(std::io::stdout.and(non_blocking_writer))
+            .init();
+        Some(guard)
+    } else {
+        subscriber.with_writer(std::io::stdout).init();
+        None
+    }
+}

components/clp-rust-utils/src/logging.rs

@@ -10,7 +10,7 @@
     QUERY_JOBS_TABLE_NAME,
     StorageEngine,
 )
-from clp_py_utils.clp_logging import get_logger
+from clp_py_utils.clp_logging import configure_logging, get_logger
 from clp_py_utils.clp_metadata_db_utils import (
     delete_archives_from_metadata_db,
     fetch_existing_datasets,
@@ -24,7 +24,6 @@
     SECOND_TO_MILLISECOND,
 )
 from job_orchestration.garbage_collector.utils import (
-    configure_logger,
     DeletionCandidatesBuffer,
     execute_deletion,
     validate_storage_type,
@@ -192,10 +191,8 @@ def _collect_and_sweep_expired_archives(
             raise ValueError(f"Unsupported Storage engine: {storage_engine}.")
 
 
-async def archive_garbage_collector(
-    clp_config: ClpConfig, log_directory: pathlib.Path, logging_level: str
-) -> None:
-    configure_logger(logger, logging_level, log_directory, ARCHIVE_GARBAGE_COLLECTOR_NAME)
+async def archive_garbage_collector(clp_config: ClpConfig) -> None:
+    configure_logging(logger, ARCHIVE_GARBAGE_COLLECTOR_NAME)
 
     archive_output_config = clp_config.archive_output
     storage_engine = clp_config.package.storage_engine

components/job-orchestration/job_orchestration/garbage_collector/archive_garbage_collector.py

@@ -2,7 +2,6 @@
 
 import argparse
 import asyncio
-import os
 import sys
 from collections.abc import Callable
 from pathlib import Path
@@ -11,7 +10,7 @@
     ClpConfig,
     GARBAGE_COLLECTOR_COMPONENT_NAME,
 )
-from clp_py_utils.clp_logging import get_logger
+from clp_py_utils.clp_logging import configure_logging, get_logger
 from clp_py_utils.core import read_yaml_config_file
 from pydantic import ValidationError
 
@@ -23,7 +22,6 @@
 from job_orchestration.garbage_collector.search_result_garbage_collector import (
     search_result_garbage_collector,
 )
-from job_orchestration.garbage_collector.utils import configure_logger
 
 logger = get_logger(GARBAGE_COLLECTOR_COMPONENT_NAME)
 
@@ -35,10 +33,8 @@ async def main(argv: list[str]) -> int:
     args_parser.add_argument("--config", "-c", required=True, help="CLP configuration file.")
     parsed_args = args_parser.parse_args(argv[1:])
 
-    # Setup logging to file
-    logs_directory = Path(os.getenv("CLP_LOGS_DIR"))
-    logging_level = os.getenv("CLP_LOGGING_LEVEL")
-    configure_logger(logger, logging_level, logs_directory, GARBAGE_COLLECTOR_COMPONENT_NAME)
+    # Setup optional file logging and logging level.
+    configure_logging(logger, GARBAGE_COLLECTOR_COMPONENT_NAME)
 
     # Load configuration
     config_path = Path(parsed_args.config)
@@ -52,7 +48,7 @@ async def main(argv: list[str]) -> int:
         logger.exception("Failed to parse CLP configuration file.")
         return 1
 
-    gc_task_configs: dict[str, tuple[int | None, Callable]] = {
+    gc_task_configs: dict[str, tuple[int | None, Callable[[ClpConfig], None]]] = {
         ARCHIVE_GARBAGE_COLLECTOR_NAME: (
             clp_config.archive_output.retention_period,
             archive_garbage_collector,
@@ -70,11 +66,7 @@ async def main(argv: list[str]) -> int:
             logger.info(f"Retention period is not configured, skip creating {gc_name}.")
             continue
         logger.info(f"Creating {gc_name} with retention period = {retention_period} minutes")
-        gc_tasks.append(
-            asyncio.create_task(
-                task_method(clp_config, logs_directory, logging_level), name=gc_name
-            )
-        )
+        gc_tasks.append(asyncio.create_task(task_method(clp_config), name=gc_name))
 
     # Poll and report any task that finished unexpectedly
     while len(gc_tasks) != 0: