controller_bootstrap.rs

// Copyright 2023 The RocketMQ Rust Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#![recursion_limit = "256"]

#[path = "controller_bootstrap/controller_security.rs"]
mod controller_security;

use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use anyhow::bail;
use anyhow::Context;
use anyhow::Result;
use rocketmq_controller::parse_command_line;
use rocketmq_controller::resolve_controller_raft_bind_addr;
use rocketmq_controller::ControllerCli;
use rocketmq_controller::ControllerConfig;
use rocketmq_controller::ControllerManager;
use rocketmq_controller::Node;
use rocketmq_error::fields;
use rocketmq_error::CliErrorView;
use rocketmq_error::CliVerbosity;
use rocketmq_error::Error;
use rocketmq_error::ErrorContext;
use rocketmq_error::CONTROLLER_CONFIGURATION_INVALID;
use rocketmq_error::CONTROLLER_CONSENSUS_TIMED_OUT;
use rocketmq_error::CONTROLLER_INTERNAL_FAILURE;
use rocketmq_model::common::mq_version::CURRENT_VERSION;
use rocketmq_protocol::protocol::remoting_command_facade::initialize_remoting_defaults;
use rocketmq_runtime::common::parse_config_file;
use rocketmq_runtime::ChildServiceContext;
use rocketmq_runtime::RuntimeComponent;
use rocketmq_runtime::RuntimeConfig;
use rocketmq_runtime::RuntimeOwner;
use rocketmq_runtime::ServiceLifecycle;
use rocketmq_runtime::ServiceLifecycleState;
use rocketmq_runtime::ShutdownDeadline;
use rocketmq_runtime::ShutdownReason;
use rocketmq_runtime::TaskGroup;
use rocketmq_security_api::SecurityBootstrap;
use rocketmq_security_api::SecurityBootstrapConfig;
use rocketmq_security_api::SecurityBootstrapOutcome;
use rocketmq_security_api::SecurityBootstrapProfile;
use tracing::info;

use controller_security::build_controller_security;

/// RocketMQ Controller Bootstrap
///
/// This is the main entry point for starting the RocketMQ Controller.
/// It handles:
/// - Command-line argument parsing
/// - Configuration loading
/// - Controller initialization and startup
///
/// # Examples
///
/// ```bash
/// # Start with default configuration
/// rocketmq-controller-rust
///
/// # Start with configuration file
/// rocketmq-controller-rust --config-file /opt/rocketmq/conf/controller.toml
///
/// # Print configuration and exit
/// rocketmq-controller-rust --print-config-item
/// ```
const CONTROLLER_AUTO_INITIALIZE_CLUSTER_ENV: &str = "ROCKETMQ_CONTROLLER_AUTO_INITIALIZE_CLUSTER";

fn controller_internal(operation: &'static str) -> Error {
    Error::new(&CONTROLLER_INTERNAL_FAILURE)
        .with_context(ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, operation))
}

fn controller_internal_by(operation: &'static str, source: impl std::error::Error + Send + Sync + 'static) -> Error {
    Error::caused_by(&CONTROLLER_INTERNAL_FAILURE, source).with_context(
        ErrorContext::new()
            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
            .with_secret_presence(fields::SOURCE_PRESENT),
    )
}

fn configuration_invalid(key: &'static str) -> Error {
    Error::new(&CONTROLLER_CONFIGURATION_INVALID).with_context(
        ErrorContext::new()
            .with_text(fields::KEY, key)
            .with_secret_presence(fields::REASON_PRESENT),
    )
}

fn configuration_invalid_by(key: &'static str, source: impl std::error::Error + Send + Sync + 'static) -> Error {
    Error::caused_by(&CONTROLLER_CONFIGURATION_INVALID, source).with_context(
        ErrorContext::new()
            .with_text(fields::KEY, key)
            .with_secret_presence(fields::REASON_PRESENT)
            .with_secret_presence(fields::SOURCE_PRESENT),
    )
}

fn consensus_timed_out(operation: &'static str, timeout_ms: u64) -> Error {
    Error::new(&CONTROLLER_CONSENSUS_TIMED_OUT).with_context(
        ErrorContext::new()
            .with_text(fields::OPERATION_DIAGNOSTIC, operation)
            .with_u64(fields::TIMEOUT_MS, timeout_ms),
    )
}

#[derive(Debug)]
struct ControllerProcessShutdownFailure {
    controller: Option<anyhow::Error>,
    service_tasks_unhealthy: bool,
}

#[derive(Debug)]
struct ControllerTelemetryShutdownFailure {
    controller: anyhow::Error,
    telemetry: anyhow::Error,
}

impl std::fmt::Display for ControllerTelemetryShutdownFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let _ = &self.telemetry;
        formatter.write_str("controller and telemetry shutdown failed")
    }
}

impl std::error::Error for ControllerTelemetryShutdownFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        Some(self.controller.as_ref())
    }
}

impl std::fmt::Display for ControllerProcessShutdownFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let _ = self.service_tasks_unhealthy;
        formatter.write_str("controller process shutdown was unhealthy")
    }
}

impl std::error::Error for ControllerProcessShutdownFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.controller
            .as_deref()
            .map(|source| source as &(dyn std::error::Error + 'static))
    }
}
const LOGO: &str = r#"
        ______           _        _  ___  ________       ______          _     _____             _             _ _
        | ___ \         | |      | | |  \/  |  _  |      | ___ \        | |   /  __ \           | |           | | |
        | |_/ /___   ___| | _____| |_| .  . | | | |______| |_/ /   _ ___| |_  | /  \/ ___  _ __ | |_ _ __ ___ | | | ___ _ __
        |    // _ \ / __| |/ / _ \ __| |\/| | | | |______|    / | | / __| __| | |    / _ \| '_ \| __| '__/ _ \| | |/ _ \ '__|
        | |\ \ (_) | (__|   <  __/ |_| |  | \ \/' /      | |\ \ |_| \__ \ |_  | \__/\ (_) | | | | |_| | | (_) | | |  __/ |
        \_| \_\___/ \___|_|\_\___|\__\_|  |_/\_/\_\      \_| \_\__,_|___/\__|  \____/\___/|_| |_|\__|_|  \___/|_|_|\___|_|
    "#;

fn print_release_version_if_requested(component: &str) -> bool {
    let arguments: Vec<_> = std::env::args_os().skip(1).collect();
    let version = std::ffi::OsStr::new("--version");
    let verbose = std::ffi::OsStr::new("--verbose");
    let requested = (arguments.len() == 1 && arguments[0].as_os_str() == version)
        || (arguments.len() == 2 && arguments[0].as_os_str() == version && arguments[1].as_os_str() == verbose);
    if !requested {
        return false;
    }
    println!("{component}");
    println!("version={}", env!("CARGO_PKG_VERSION"));
    if arguments.len() == 2 {
        println!(
            "artifact_id={}",
            option_env!("ROCKETMQ_RELEASE_ARTIFACT_ID").unwrap_or("development")
        );
        println!(
            "requested_features={}",
            option_env!("ROCKETMQ_RELEASE_REQUESTED_FEATURES").unwrap_or("default")
        );
        println!(
            "effective_features={}",
            option_env!("ROCKETMQ_RELEASE_EFFECTIVE_FEATURES").unwrap_or("default")
        );
    }
    true
}

pub fn main() {
    if let Err(source) = try_main() {
        let error = Error::new(&CONTROLLER_INTERNAL_FAILURE)
            .with_boxed_source(source.into_boxed_dyn_error())
            .with_context(
                ErrorContext::new()
                    .with_text(fields::OPERATION_DIAGNOSTIC, "run-controller-service")
                    .with_secret_presence(fields::SOURCE_PRESENT),
            );
        let output = CliErrorView::from_error(&error).output(CliVerbosity::Default);
        eprintln!("{}", output.stderr());
        std::process::exit(output.exit_code().as_i32());
    }
}

fn try_main() -> Result<()> {
    if print_release_version_if_requested("rocketmq-controller-rust") {
        return Ok(());
    }
    let owner = RuntimeOwner::plan(controller_runtime_config())
        .expect("controller runtime profile is internally valid")
        .build()
        .map_err(|source| controller_internal_by("build controller runtime", source))?;
    let service_context = owner.root_context().component("rocketmq-controller-runtime");
    let lifecycle = ServiceLifecycle::from_env("rocketmq-controller")
        .map_err(|error| configuration_invalid_by("controller.lifecycle", error))?;

    let run_result = owner.block_on(run(service_context, lifecycle.clone()));
    if run_result.is_err() {
        lifecycle.mark_failed();
    }
    let shutdown_request = lifecycle
        .shutdown_request()
        .unwrap_or_else(|| lifecycle.request_shutdown(ShutdownReason::Internal));
    let shutdown_result = owner
        .shutdown_runtime_blocking_until(shutdown_request.deadline)
        .map_err(|source| controller_internal_by("shutdown controller runtime", source));

    match (run_result, shutdown_result) {
        (Err(error), _) => Err(error),
        (Ok(()), Err(error)) => Err(error.into()),
        (Ok(()), Ok(report)) => {
            if !report.is_healthy() {
                lifecycle.mark_failed();
                tracing::warn!(
                    report = %report.to_json(),
                    "controller runtime shutdown report is unhealthy"
                );
                return Err(controller_internal("shutdown controller runtime cleanly").into());
            }
            Ok(())
        }
    }
}

fn controller_runtime_config() -> RuntimeConfig {
    RuntimeConfig::controller_default()
}

async fn run(service_context: ChildServiceContext, lifecycle: ServiceLifecycle) -> Result<()> {
    initialize_remoting_defaults(CURRENT_VERSION as i32)
        .context("failed to initialize the immutable Controller remoting defaults")?;

    // Parse command line and load configuration
    let (cli, config) = parse_command_line()?;

    if cli.print_config_item {
        ControllerCli::print_config(&config);
        return Ok(());
    }

    let rocketmq_home = &config.rocketmq_home;
    if rocketmq_home.is_empty() {
        bail!("Please set the ROCKETMQ_HOME environment variable. Example: export ROCKETMQ_HOME=/opt/rocketmq");
    }

    let logging_overrides = load_logging_overrides(cli.config_file.as_deref())?;
    let mut telemetry_bootstrap = build_controller_telemetry_bootstrap_config(&config);
    telemetry_bootstrap.logging.reload = logging_overrides.logging.reload;
    let rocketmq_observability::TelemetryResolution {
        bootstrap: bootstrap_config,
        process: process_telemetry,
        prometheus_listener_addr,
        ..
    } = rocketmq_observability::resolve_telemetry_from_env(
        "rocketmq-controller",
        telemetry_bootstrap,
        &config.observability,
        rocketmq_observability::TelemetryEnvironmentSpec::default(),
    )
    .context("failed to resolve Controller telemetry configuration")?;
    let security_bootstrap =
        SecurityBootstrapConfig::from_env().context("failed to load Controller security bootstrap configuration")?;
    let validated_security = validate_controller_security(
        &security_bootstrap,
        &config,
        prometheus_listener_addr,
        lifecycle.config().probe_bind_addr,
    )
    .context("Controller security bootstrap failed before listener bind")?;

    let environment_filter = rocketmq_observability::read_rust_log().context("failed to read RUST_LOG")?;
    let resolved_filter = resolve_startup_log_filter(&cli, &logging_overrides, environment_filter.as_deref())
        .context("failed to resolve controller log filter")?;
    let telemetry_guard = rocketmq_observability::install_global_with_filter_and_service_context(
        &bootstrap_config,
        resolved_filter.clone(),
        &service_context,
    )
    .await
    .context("failed to initialize controller telemetry bootstrap")?;
    let telemetry_handle = telemetry_guard.handle();
    if let Err(error) = register_controller_release_identity(&process_telemetry, &telemetry_handle) {
        let request = lifecycle.request_shutdown(ShutdownReason::Internal);
        if let Err(shutdown_error) = telemetry_guard
            .shutdown_with_service_context(&service_context, request.deadline.remaining())
            .await
            .into_result()
        {
            tracing::warn!(
                error = %shutdown_error,
                "controller telemetry cleanup after release identity failure was unhealthy"
            );
        }
        return Err(error);
    }
    log_telemetry_bootstrap(
        &bootstrap_config,
        &resolved_filter,
        telemetry_guard.subscriber_install_status(),
    );
    log_security_bootstrap(validated_security);

    lifecycle.set_observer(std::sync::Arc::new(
        rocketmq_observability::metrics::runtime::RuntimeMetricsRecorder::from_handle(
            &telemetry_handle,
            RuntimeComponent::Controller,
        ),
    ))?;
    if let Err(error) = lifecycle.start(&service_context).await {
        lifecycle.mark_failed();
        let request = lifecycle.request_shutdown(ShutdownReason::Internal);
        if let Err(shutdown_error) = telemetry_guard
            .shutdown_with_service_context(&service_context, request.deadline.remaining())
            .await
            .into_result()
        {
            tracing::warn!(error = %shutdown_error, "controller telemetry cleanup after lifecycle startup failure was unhealthy");
        }
        return Err(controller_internal_by("start controller lifecycle boundary", error).into());
    }
    let diagnostics_sources = rocketmq_observability::RuntimeDiagnosticsSources::default();
    let diagnostics = rocketmq_observability::RuntimeDiagnosticsService::new(
        &service_context,
        RuntimeComponent::Controller,
        telemetry_handle.clone(),
        std::sync::Arc::new(diagnostics_sources.clone()),
    );
    let diagnostics_start = async {
        diagnostics
            .start(rocketmq_observability::RuntimeDiagnosticsMode::from_env()?)
            .await
    }
    .await;
    if let Err(error) = diagnostics_start {
        lifecycle.mark_failed();
        let request = lifecycle.request_shutdown(ShutdownReason::Internal);
        if let Err(shutdown_error) = telemetry_guard
            .shutdown_with_service_context(&service_context, request.deadline.remaining())
            .await
            .into_result()
        {
            tracing::warn!(error = %shutdown_error, "controller telemetry cleanup after diagnostics startup failure was unhealthy");
        }
        return Err(controller_internal_by("start controller runtime diagnostics", error).into());
    }

    println!("{}", LOGO);

    info!("RocketMQ Controller configuration loaded successfully");
    info!("Node ID: {}, Listen Address: {}", config.node_id, config.listen_addr);
    info!("ROCKETMQ_HOME: {}", rocketmq_home);

    let controller_result = run_controller(
        config,
        service_context.clone(),
        lifecycle.clone(),
        telemetry_handle,
        process_telemetry.metrics_enabled(),
        &diagnostics_sources,
    )
    .await;
    if controller_result.is_err() {
        lifecycle.mark_failed();
        lifecycle.request_shutdown(ShutdownReason::Internal);
    }
    let shutdown_request = lifecycle
        .shutdown_request()
        .unwrap_or_else(|| lifecycle.request_shutdown(ShutdownReason::Internal));
    // The final telemetry flush runs after this task group closes, so its blocking allowance
    // is reserved while the scope is still open.
    let telemetry_flush_lease =
        rocketmq_observability::reserve_telemetry_flush_lease(&service_context, Some(shutdown_request.deadline));
    let controller_result = finish_controller_process_shutdown(
        controller_result,
        service_context.task_group(),
        shutdown_request.deadline,
        &diagnostics_sources,
    )
    .await;
    rocketmq_observability::metrics::runtime::RuntimeMetricsRecorder::from_handle(
        &telemetry_guard.handle(),
        rocketmq_runtime::RuntimeComponent::Controller,
    )
    .record_business_drain(if shutdown_request.deadline.is_expired() {
        rocketmq_observability::metrics::runtime::RuntimeBusinessDrainOutcome::DeadlineExceeded
    } else if controller_result.is_ok() {
        rocketmq_observability::metrics::runtime::RuntimeBusinessDrainOutcome::Drained
    } else {
        rocketmq_observability::metrics::runtime::RuntimeBusinessDrainOutcome::Failed
    });
    let telemetry_report = match telemetry_flush_lease {
        Some(lease) => {
            telemetry_guard
                .shutdown_with_drain_lease(lease, shutdown_request.deadline.remaining())
                .await
        }
        None => {
            telemetry_guard
                .shutdown_with_service_context(&service_context, shutdown_request.deadline.remaining())
                .await
        }
    };
    let shutdown_result = telemetry_report
        .into_result()
        .context("failed to shutdown controller telemetry bootstrap");

    match (controller_result, shutdown_result) {
        (Err(controller_error), Err(telemetry_error)) => Err(controller_internal_by(
            "shutdown controller and telemetry",
            ControllerTelemetryShutdownFailure {
                controller: controller_error,
                telemetry: telemetry_error,
            },
        )
        .into()),
        (Err(error), Ok(_report)) => Err(error),
        (Ok(()), Err(error)) => {
            lifecycle.mark_failed();
            Err(error)
        }
        (Ok(()), Ok(_report)) => {
            lifecycle.mark_stopped();
            Ok(())
        }
    }
}

fn validate_controller_security(
    security_bootstrap: &SecurityBootstrap,
    controller_config: &ControllerConfig,
    prometheus_bind_addr: Option<SocketAddr>,
    probe_bind_addr: Option<SocketAddr>,
) -> Result<SecurityBootstrapOutcome> {
    if security_bootstrap.requires_authentication()
        && (!controller_config.authentication_enabled || !controller_config.authorization_enabled)
    {
        anyhow::bail!("secure-enforced Controller requires both authenticationEnabled and authorizationEnabled");
    }
    if !security_bootstrap.is_enabled() {
        return security_bootstrap.validate(&[]).map_err(anyhow::Error::from);
    }
    let raft_bind_addr = resolve_controller_raft_bind_addr(controller_config.local_raft_addr())
        .context("failed to resolve Controller Raft listener address")?;
    let mut listeners = vec![controller_config.listen_addr, raft_bind_addr];
    if let Some(prometheus_bind_addr) = prometheus_bind_addr {
        listeners.push(prometheus_bind_addr);
    }
    if let Some(probe_bind_addr) = probe_bind_addr {
        listeners.push(probe_bind_addr);
    }
    security_bootstrap.validate(&listeners).map_err(anyhow::Error::from)
}

fn log_security_bootstrap(outcome: SecurityBootstrapOutcome) {
    match outcome {
        SecurityBootstrapOutcome::Disabled => {
            tracing::warn!("Controller security bootstrap is disabled because no security profile is configured")
        }
        SecurityBootstrapOutcome::Validated(validated) => match validated.profile() {
            SecurityBootstrapProfile::DevelopmentInsecureLoopback => tracing::warn!(
                profile = validated.profile().as_str(),
                listener_count = validated.listener_count(),
                "Controller development-insecure security profile is active; every listener is restricted to loopback"
            ),
            SecurityBootstrapProfile::SecureEnforced => info!(
                profile = validated.profile().as_str(),
                listener_count = validated.listener_count(),
                "Controller secure bootstrap completed before listener bind"
            ),
        },
    }
}

async fn run_controller(
    config: ControllerConfig,
    service_context: ChildServiceContext,
    lifecycle: ServiceLifecycle,
    telemetry_handle: rocketmq_observability::TelemetryHandle,
    release_identity_required: bool,
    diagnostics_sources: &rocketmq_observability::RuntimeDiagnosticsSources,
) -> Result<()> {
    // Create controller manager
    info!("Creating Controller Manager...");
    let security = build_controller_security(&config, &service_context).await?;
    let controller_manager = ControllerManager::new_with_security_and_remoting_command_factory(
        config,
        service_context.clone(),
        telemetry_handle.clone(),
        security,
        rocketmq_protocol::protocol::remoting_command_defaults::application_remoting_command_factory(),
    )
    .await?;
    let controller_manager = Arc::new(controller_manager);
    // Initialize controller
    info!("Initializing Controller...");

    let init_result = match controller_manager.initialize().await {
        Ok(result) => result,
        Err(error) => {
            shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
            return Err(error.into());
        }
    };
    if !init_result {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        bail!("Controller initialization failed");
    }

    if let Err(error) = controller_manager.start().await {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        return Err(error.into());
    }
    if let Some(observer) = controller_manager.leadership_watch_observer() {
        diagnostics_sources.set_schedules(observer);
    }
    if let Err(error) = initialize_cluster_if_configured(&controller_manager).await {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        return Err(error);
    }
    if let Err(error) = wait_for_cluster_recovery(&controller_manager).await {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        return Err(error);
    }
    if release_identity_required && !telemetry_handle.release_identity_registered() {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        bail!("Controller release identity was not registered before readiness");
    }
    if let Err(error) = lifecycle.mark_ready() {
        shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
        return Err(error.into());
    }

    info!("Controller started successfully!");
    info!("  Node ID: {}", controller_manager.controller_config().node_id);
    info!("  Listen:  {}", controller_manager.controller_config().listen_addr);
    info!("Controller is running. Press Ctrl+C to stop.");

    let shutdown_request = match lifecycle.wait_for_shutdown_signal().await {
        Ok(request) => request,
        Err(error) => {
            shutdown_controller_after_startup_failure(&controller_manager, &lifecycle).await;
            return Err(error.into());
        }
    };
    info!(
        reason = shutdown_request.reason.as_str(),
        remaining_ms = shutdown_request.deadline.remaining().as_millis(),
        "Received Controller shutdown request"
    );

    // Graceful shutdown
    controller_manager.shutdown_until(shutdown_request.deadline).await?;
    info!("Controller shutdown completed.");
    Ok(())
}

async fn finish_controller_process_shutdown(
    controller_result: Result<()>,
    service_tasks: &TaskGroup,
    deadline: ShutdownDeadline,
    diagnostics_sources: &rocketmq_observability::RuntimeDiagnosticsSources,
) -> Result<()> {
    let service_report = service_tasks.shutdown_until(deadline).await;
    diagnostics_sources.retain_shutdown(&service_report);
    match (controller_result, service_report.is_healthy()) {
        (Ok(()), true) => Ok(()),
        (Err(controller_error), true) => Err(controller_error),
        (Ok(()), false) => Err(controller_internal_by(
            "shutdown controller process",
            ControllerProcessShutdownFailure {
                controller: None,
                service_tasks_unhealthy: true,
            },
        )
        .into()),
        (Err(controller_error), false) => Err(controller_internal_by(
            "shutdown controller process",
            ControllerProcessShutdownFailure {
                controller: Some(controller_error),
                service_tasks_unhealthy: true,
            },
        )
        .into()),
    }
}

async fn shutdown_controller_after_startup_failure(
    controller_manager: &Arc<ControllerManager>,
    lifecycle: &ServiceLifecycle,
) {
    lifecycle.mark_failed();
    let request = lifecycle.request_shutdown(ShutdownReason::Internal);
    if let Err(error) = controller_manager.shutdown_until(request.deadline).await {
        tracing::warn!(error = %error, "Controller startup-failure cleanup was unhealthy");
    }
    debug_assert_eq!(lifecycle.state(), ServiceLifecycleState::Failed);
}

fn build_controller_telemetry_bootstrap_config(
    controller_config: &ControllerConfig,
) -> rocketmq_observability::TelemetryBootstrapConfig {
    let mut observability = rocketmq_observability::ObservabilityConfig {
        service_name: "rocketmq-controller".to_string(),
        service_namespace: "rocketmq".to_string(),
        node_type: "controller".to_string(),
        node_id: controller_config.node_id.to_string(),
        ..rocketmq_observability::ObservabilityConfig::default()
    };
    observability.subscriber_install_policy = rocketmq_observability::SubscriberInstallPolicy::Required;

    let mut logging = rocketmq_observability::LoggingConfig::default();
    logging.file.directory = controller_log_directory(controller_config);
    logging.file.file_name_prefix = "rocketmq-controller".to_string();

    rocketmq_observability::TelemetryBootstrapConfig { observability, logging }
}

fn register_controller_release_identity(
    process_telemetry: &rocketmq_observability::metrics::release_identity::ProcessTelemetryConfig,
    telemetry_handle: &rocketmq_observability::TelemetryHandle,
) -> Result<()> {
    if !process_telemetry.metrics_enabled() {
        return Ok(());
    }

    #[cfg(feature = "metrics")]
    {
        telemetry_handle
            .register_release_identity(process_telemetry.release_identity().clone())
            .context("failed to register Controller release identity before readiness")?;
        Ok(())
    }

    #[cfg(not(feature = "metrics"))]
    {
        let _ = telemetry_handle;
        bail!("Controller metrics were enabled without the `metrics` Cargo feature");
    }
}

fn controller_log_directory(controller_config: &ControllerConfig) -> String {
    if controller_config.rocketmq_home.trim().is_empty() {
        return "logs".to_string();
    }
    std::path::PathBuf::from(controller_config.rocketmq_home.as_str())
        .join("logs")
        .to_string_lossy()
        .into_owned()
}

fn log_telemetry_bootstrap(
    config: &rocketmq_observability::TelemetryBootstrapConfig,
    resolved_filter: &rocketmq_observability::ResolvedLogFilter,
    subscriber_install_status: rocketmq_observability::SubscriberInstallStatus,
) {
    info!(
        service = "rocketmq-controller",
        effective_filter = resolved_filter.filter(),
        filter_source = %resolved_filter.source(),
        metrics_exporter = ?config.observability.metrics.exporter,
        trace_exporter = ?config.observability.traces.exporter,
        log_exporter = ?config.observability.logs.exporter,
        subscriber_installed = subscriber_install_status.installed,
        reload_enabled = config.logging.reload.enabled,
        file_log_enabled = config.logging.file.enabled,
        "controller telemetry bootstrap initialized"
    );
}

fn load_logging_overrides(config_file: Option<&std::path::Path>) -> Result<rocketmq_observability::LoggingOverrides> {
    match config_file {
        Some(path) => parse_config_file::parse_config_file(path.to_path_buf())
            .context("failed to parse Controller logging configuration"),
        None => Ok(rocketmq_observability::LoggingOverrides::default()),
    }
}

fn resolve_startup_log_filter(
    cli: &ControllerCli,
    overrides: &rocketmq_observability::LoggingOverrides,
    environment_filter: Option<&str>,
) -> Result<rocketmq_observability::ResolvedLogFilter, rocketmq_observability::ObservabilityError> {
    rocketmq_observability::LogFilterResolver::resolve(rocketmq_observability::LogFilterInputs {
        runtime: None,
        cli: cli.log_filter.as_deref(),
        environment: environment_filter,
        config: overrides.logging.filter.as_deref(),
        legacy_config: overrides.log_filter.as_deref(),
    })
}

async fn initialize_cluster_if_configured(controller_manager: &Arc<ControllerManager>) -> Result<()> {
    let config = controller_manager.controller_config();
    let peers = config.raft_member_endpoints();
    if peers.is_empty() {
        return Ok(());
    }

    let is_single_node = peers.len() == 1;
    if !is_single_node && !auto_initialize_cluster_enabled()? {
        return Ok(());
    }
    let bootstrap_node_id = peers
        .iter()
        .map(|peer| peer.id)
        .min()
        .ok_or_else(|| configuration_invalid("controller.raft_peers"))?;
    if bootstrap_node_id != config.node_id {
        return Ok(());
    }
    if controller_manager.controller().has_persisted_committed_log().await? {
        info!(
            node_id = config.node_id,
            "Controller cluster already has committed logs; skip bootstrap"
        );
        return Ok(());
    }

    let nodes = peers
        .iter()
        .map(|peer| {
            (
                peer.id,
                Node {
                    node_id: peer.id,
                    rpc_addr: peer.addr.clone(),
                },
            )
        })
        .collect::<BTreeMap<_, _>>();
    info!(
        node_id = config.node_id,
        members = nodes.len(),
        "Bootstrapping configured Controller cluster"
    );
    controller_manager.controller().initialize_cluster(nodes).await?;

    let attempts = if is_single_node { 30 } else { 300 };
    for _ in 0..attempts {
        if controller_manager.is_leader() {
            info!(node_id = config.node_id, "Controller bootstrap node became leader");
            return Ok(());
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    Err(consensus_timed_out("initialize controller cluster", 3_000).into())
}

async fn wait_for_cluster_recovery(controller_manager: &Arc<ControllerManager>) -> Result<()> {
    let config = controller_manager.controller_config();
    if config.raft_member_endpoints().is_empty() {
        return Ok(());
    }

    for _ in 0..300 {
        if controller_manager.controller().has_recovered_cluster_state() {
            info!(
                node_id = config.node_id,
                "Controller observed a Raft leader and recovered committed state"
            );
            return Ok(());
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    Err(consensus_timed_out("recover controller cluster", 30_000).into())
}

fn auto_initialize_cluster_enabled() -> Result<bool> {
    let Some(raw) = std::env::var_os(CONTROLLER_AUTO_INITIALIZE_CLUSTER_ENV) else {
        return Ok(false);
    };
    let raw = raw
        .into_string()
        .map_err(|_| configuration_invalid("controller.auto_initialize_cluster"))?;
    parse_auto_initialize_cluster_flag(&raw)
}

fn parse_auto_initialize_cluster_flag(raw: &str) -> Result<bool> {
    match raw.trim().to_ascii_lowercase().as_str() {
        "1" | "true" => Ok(true),
        "0" | "false" | "" => Ok(false),
        _ => Err(configuration_invalid("controller.auto_initialize_cluster").into()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn secure_profile_rejects_disabled_request_protection_before_material_loading() {
        let security =
            SecurityBootstrap::Enabled(SecurityBootstrapConfig::new(SecurityBootstrapProfile::SecureEnforced));
        for (authentication, authorization) in [(false, false), (true, false), (false, true)] {
            let mut config = ControllerConfig::default();
            config.authentication_enabled = authentication;
            config.authorization_enabled = authorization;
            let error = validate_controller_security(&security, &config, None, None).unwrap_err();
            assert!(error
                .to_string()
                .contains("requires both authenticationEnabled and authorizationEnabled"));
        }
    }

    #[test]
    fn telemetry_shutdown_aggregate_retains_both_typed_failures_with_fixed_display() {
        let failure = ControllerTelemetryShutdownFailure {
            controller: anyhow::anyhow!("private controller failure"),
            telemetry: anyhow::anyhow!("private telemetry failure"),
        };

        assert_eq!(failure.to_string(), "controller and telemetry shutdown failed");
        assert_eq!(failure.controller.to_string(), "private controller failure");
        assert_eq!(failure.telemetry.to_string(), "private telemetry failure");
        let outer = controller_internal_by("shutdown controller and telemetry", failure);
        let retained = std::error::Error::source(&outer)
            .and_then(|source| source.downcast_ref::<ControllerTelemetryShutdownFailure>())
            .expect("telemetry shutdown aggregate must remain the typed source");
        assert_eq!(retained.telemetry.to_string(), "private telemetry failure");
    }

    #[test]
    fn disabled_security_bootstrap_allows_default_controller_listeners() {
        let outcome = validate_controller_security(
            &rocketmq_security_api::SecurityBootstrap::Disabled,
            &ControllerConfig::default(),
            None,
            None,
        )
        .expect("disabled security bootstrap should not restrict Controller listeners");

        assert_eq!(outcome, rocketmq_security_api::SecurityBootstrapOutcome::Disabled);
    }

    #[test]
    fn security_bootstrap_precedes_controller_listener_bind() {
        let security = rocketmq_security_api::SecurityBootstrap::Enabled(SecurityBootstrapConfig::new(
            SecurityBootstrapProfile::DevelopmentInsecureLoopback,
        ));
        let mut controller = ControllerConfig::default()
            .with_node_info(1, SocketAddr::from(([127, 0, 0, 1], 60109)))
            .with_raft_peers(Vec::new());

        validate_controller_security(
            &security,
            &controller,
            None,
            Some(SocketAddr::from(([127, 0, 0, 1], 8088))),
        )
        .expect("loopback-only Controller bootstrap should pass");

        controller.listen_addr = SocketAddr::from(([0, 0, 0, 0], 60109));
        assert!(validate_controller_security(&security, &controller, None, None).is_err());
    }

    #[test]
    fn development_security_rejects_public_prometheus_listener() {
        let security = rocketmq_security_api::SecurityBootstrap::Enabled(SecurityBootstrapConfig::new(
            SecurityBootstrapProfile::DevelopmentInsecureLoopback,
        ));
        let controller = ControllerConfig::default()
            .with_node_info(1, SocketAddr::from(([127, 0, 0, 1], 60109)))
            .with_raft_peers(Vec::new());

        assert!(validate_controller_security(
            &security,
            &controller,
            Some(SocketAddr::from(([0, 0, 0, 0], 5557))),
            None,
        )
        .is_err());
    }

    #[test]
    fn auto_initialize_cluster_flag_accepts_only_explicit_boolean_values() {
        assert!(parse_auto_initialize_cluster_flag("true").unwrap());
        assert!(parse_auto_initialize_cluster_flag("1").unwrap());
        assert!(!parse_auto_initialize_cluster_flag("false").unwrap());
        assert!(!parse_auto_initialize_cluster_flag("0").unwrap());
        assert!(parse_auto_initialize_cluster_flag("yes").is_err());
    }

    #[test]
    fn controller_telemetry_bootstrap_uses_required_logging_defaults() {
        let mut controller_config = ControllerConfig::default();
        controller_config.rocketmq_home = "target/controller-telemetry-bootstrap".to_string();
        controller_config.node_id = 7;
        controller_config.observability.metrics.exporter = Some(rocketmq_observability::MetricsExporter::Prometheus);

        let config = build_controller_telemetry_bootstrap_config(&controller_config);

        assert_eq!(config.observability.service_name, "rocketmq-controller");
        assert_eq!(config.observability.node_id, "7");
        assert_eq!(
            config.observability.subscriber_install_policy,
            rocketmq_observability::SubscriberInstallPolicy::Required
        );
        assert!(!config.observability.enabled);
        assert!(config.logging.enabled);
        assert!(config.logging.console.enabled);
        assert!(!config.logging.file.enabled);
        assert_eq!(config.logging.file.file_name_prefix, "rocketmq-controller");
        assert_eq!(
            config.observability.metrics.exporter,
            rocketmq_observability::MetricsExporter::Disable
        );

        let expected_log_dir = std::path::PathBuf::from("target/controller-telemetry-bootstrap").join("logs");
        assert_eq!(
            std::path::PathBuf::from(config.logging.file.directory.as_str()),
            expected_log_dir
        );
    }

    #[test]
    fn controller_telemetry_bootstrap_validates_file_prometheus_listener_before_startup() {
        let security = rocketmq_security_api::SecurityBootstrap::Enabled(SecurityBootstrapConfig::new(
            SecurityBootstrapProfile::DevelopmentInsecureLoopback,
        ));
        let mut controller = ControllerConfig::default();
        controller.observability = rocketmq_observability::ObservabilityOverrides {
            metrics: rocketmq_observability::MetricsOverrides {
                exporter: Some(rocketmq_observability::MetricsExporter::Prometheus),
                ..Default::default()
            },
            prometheus: rocketmq_observability::PrometheusOverrides {
                host: Some("0.0.0.0".to_string()),
                port: Some(9464),
                path: Some("/rocketmq".to_string()),
            },
            ..Default::default()
        };
        let resolution = rocketmq_observability::resolve_telemetry_values(
            "rocketmq-controller",
            build_controller_telemetry_bootstrap_config(&controller),
            &controller.observability,
            &rocketmq_observability::TelemetryEnvironmentValues::default(),
            rocketmq_observability::TelemetryEnvironmentSpec::default(),
        )
        .expect("file Prometheus listener should resolve");

        assert_eq!(
            resolution.prometheus_listener_addr,
            Some(SocketAddr::from(([0, 0, 0, 0], 9464)))
        );
        validate_controller_security(&security, &controller, resolution.prometheus_listener_addr, None)
            .expect_err("public file Prometheus listener must fail closed before startup");
    }

    #[test]
    fn controller_bootstrap_accepts_standard_otlp_environment_values() {
        let environment = rocketmq_observability::TelemetryEnvironmentValues {
            otlp_endpoint: Some("http://collector:4317".into()),
            otlp_protocol: Some("grpc".into()),
            ..Default::default()
        };
        let controller = ControllerConfig::default();
        let resolution = rocketmq_observability::resolve_telemetry_values(
            "rocketmq-controller",
            build_controller_telemetry_bootstrap_config(&controller),
            &controller.observability,
            &environment,
            rocketmq_observability::TelemetryEnvironmentSpec::default(),
        )
        .expect("valid standard OTLP environment should resolve");
        let config = resolution.bootstrap;

        assert!(config.observability.enabled);
        assert_eq!(config.observability.service_name, "rocketmq-controller");
        assert_eq!(
            config.observability.metrics.exporter,
            rocketmq_observability::MetricsExporter::OtlpGrpc
        );
        assert_eq!(
            config.observability.traces.exporter,
            rocketmq_observability::TraceExporter::OtlpGrpc
        );
        assert_eq!(
            config.observability.logs.exporter,
            rocketmq_observability::LogsExporter::OtlpGrpc
        );
    }
}