qdrant/src/settings.rs

use std::borrow::Cow;
use std::{env, io};

use api::grpc::transport_channel_pool::{
    DEFAULT_CONNECT_TIMEOUT, DEFAULT_GRPC_TIMEOUT, DEFAULT_POOL_SIZE,
};
use collection::operations::validation;
use collection::shards::shard::PeerId;
use common::flags::FeatureFlags;
use config::{Config, ConfigError, Environment, File, FileFormat, Source};
use serde::Deserialize;
use storage::types::StorageConfig;
use validator::{Validate, ValidationError};

use crate::common::audit::AuditConfig;
use crate::common::debugger::DebuggerConfig;
use crate::common::inference::config::InferenceConfig;
use crate::tracing;

const MAX_PEER_ID: u64 = (1 << 53) - 1;

const DEFAULT_CONFIG: &str = include_str!("../config/config.yaml");

#[derive(Debug, Deserialize, Validate, Clone)]
pub struct ServiceConfig {
    #[validate(length(min = 1))]
    pub host: String,
    pub http_port: u16,
    pub grpc_port: Option<u16>, // None means that gRPC is disabled

    /// If specified, qdrant will serve a separate service for `/metrics` on this port.
    /// Separate port is not protected by API keys and dedicated for internal monitoring systems.
    /// This port should not be exposed to untrusted networks.
    #[serde(default)]
    pub metrics_port: Option<u16>,

    pub max_request_size_mb: usize,
    pub max_workers: Option<usize>,
    /// Keep-alive timeout for incoming HTTP connections in seconds.
    #[serde(default = "default_http_keep_alive_timeout_sec")]
    #[validate(range(min = 1))]
    pub http_keep_alive_timeout_sec: u64,
    /// Timeout for reading HTTP request data from clients in seconds.
    #[serde(default = "default_http_client_request_timeout_sec")]
    #[validate(range(min = 1))]
    pub http_client_request_timeout_sec: u64,
    /// Timeout for client disconnect handling in seconds.
    #[serde(default = "default_http_client_disconnect_timeout_sec")]
    #[validate(range(min = 1))]
    pub http_client_disconnect_timeout_sec: u64,
    #[serde(default = "default_cors")]
    pub enable_cors: bool,
    #[serde(default)]
    pub enable_tls: bool,
    #[serde(default)]
    pub verify_https_client_certificate: bool,
    pub api_key: Option<String>,

    /// Same as `api_key`, can be used for rolling key rotation.
    pub alt_api_key: Option<String>,

    pub read_only_api_key: Option<String>,
    #[serde(default)]
    pub jwt_rbac: Option<bool>,

    /// Enforce API key / JWT authentication on the internal (p2p) gRPC API.
    ///
    /// The regular API key is always forwarded on internal gRPC requests, but
    /// the receiving side only verifies it when this flag is enabled. This is
    /// opt-in to keep rolling upgrades safe: during an upgrade some nodes may
    /// run a version that does not attach the key yet, so enabling enforcement
    /// before every peer is upgraded would break intra-cluster communication.
    #[serde(default)]
    pub enforce_internal_auth: Option<bool>,

    #[serde(default)]
    pub hide_jwt_dashboard: Option<bool>,

    /// Directory where static files are served from.
    /// For example, the Web-UI should be placed here.
    #[serde(default)]
    pub static_content_dir: Option<String>,

    /// If serving of the static content is enabled.
    /// This includes the Web-UI. True by default.
    #[serde(default)]
    pub enable_static_content: Option<bool>,

    /// How much time is considered too long for a query to execute.
    pub slow_query_secs: Option<f32>,

    /// Whether to enable reporting of measured hardware utilization in API responses.
    #[serde(default)]
    pub hardware_reporting: Option<bool>,

    /// Global prefix for metrics.
    #[serde(default)]
    #[validate(custom(function = validate_metrics_prefix))]
    pub metrics_prefix: Option<String>,

    /// Whether to allow snapshot recovery from remote URLs (http/https).
    /// If disabled, snapshot recovery will only work with local files and uploads.
    /// Disabling this can mitigate SSRF risks in environments where the Qdrant node
    /// has access to internal resources that should not be reachable by users.
    #[serde(default = "default_snapshot_url_recovery")]
    pub enable_snapshot_url_recovery: bool,
}

impl ServiceConfig {
    pub fn hardware_reporting(&self) -> bool {
        self.hardware_reporting.unwrap_or_default()
    }
}

#[derive(Debug, Deserialize, Clone, Default, Validate)]
pub struct ClusterConfig {
    pub enabled: bool, // disabled by default
    #[serde(default)]
    #[validate(range(min = 1, max = MAX_PEER_ID))]
    pub peer_id: Option<PeerId>,
    #[serde(default = "default_timeout_ms")]
    #[validate(range(min = 1))]
    pub grpc_timeout_ms: u64,
    #[serde(default = "default_connection_timeout_ms")]
    #[validate(range(min = 1))]
    pub connection_timeout_ms: u64,
    #[serde(default)]
    #[validate(nested)]
    pub p2p: P2pConfig,
    #[serde(default)]
    #[validate(nested)]
    pub consensus: ConsensusConfig,
    #[serde(default)]
    pub resharding_enabled: bool, // disabled by default
}

#[derive(Debug, Deserialize, Clone, Validate)]
pub struct P2pConfig {
    #[serde(default)]
    pub port: Option<u16>,
    #[serde(default = "default_connection_pool_size")]
    #[validate(range(min = 1))]
    pub connection_pool_size: usize,
    #[serde(default)]
    pub enable_tls: bool,
}

impl Default for P2pConfig {
    fn default() -> Self {
        P2pConfig {
            port: None,
            connection_pool_size: default_connection_pool_size(),
            enable_tls: false,
        }
    }
}

#[derive(Debug, Deserialize, Clone, Validate)]
pub struct ConsensusConfig {
    #[serde(default = "default_max_message_queue_size")]
    pub max_message_queue_size: usize, // controls the back-pressure at the Raft level
    #[serde(default = "default_tick_period_ms")]
    #[validate(range(min = 1))]
    pub tick_period_ms: u64,
    #[serde(default = "default_bootstrap_timeout_sec")]
    #[validate(range(min = 1))]
    pub bootstrap_timeout_sec: u64,
    #[validate(range(min = 1))]
    #[serde(default = "default_message_timeout_tics")]
    pub message_timeout_ticks: u64,
    /// Compact WAL when it grows to enough applied entries
    #[serde(default = "default_compact_wal_entries")]
    pub compact_wal_entries: u64,
}

impl Default for ConsensusConfig {
    fn default() -> Self {
        ConsensusConfig {
            max_message_queue_size: default_max_message_queue_size(),
            tick_period_ms: default_tick_period_ms(),
            bootstrap_timeout_sec: default_bootstrap_timeout_sec(),
            message_timeout_ticks: default_message_timeout_tics(),
            compact_wal_entries: default_compact_wal_entries(),
        }
    }
}

#[derive(Debug, Deserialize, Clone, Validate)]
pub struct TlsConfig {
    pub cert: String,
    pub key: String,
    pub ca_cert: Option<String>,
    #[serde(default = "default_tls_cert_ttl")]
    #[validate(range(min = 1))]
    pub cert_ttl: Option<u64>,
}

#[derive(Clone, Debug, Deserialize, Validate)]
pub struct GpuConfig {
    /// Enable GPU indexing.
    #[serde(default)]
    pub indexing: bool,
    /// Force half precision for `f32` values while indexing.
    /// `f16` conversion will take place only inside GPU memory and won't affect storage type.
    #[serde(default)]
    pub force_half_precision: bool,
    /// Used vulkan "groups" of GPU. In other words, how many parallel points can be indexed by GPU.
    /// Optimal value might depend on the GPU model.
    /// Proportional, but doesn't necessary equal to the physical number of warps.
    /// Do not change this value unless you know what you are doing.
    /// Default: 512
    #[serde(default)]
    #[validate(range(min = 1))]
    pub groups_count: Option<usize>,
    /// Filter for GPU devices by hardware name. Case insensitive.
    /// Comma-separated list of substrings to match against the gpu device name.
    /// Example: "nvidia"
    /// Default: "" - all devices are accepted.
    #[serde(default)]
    pub device_filter: String,
    /// List of explicit GPU devices to use.
    /// If host has multiple GPUs, this option allows to select specific devices
    /// by their index in the list of found devices.
    /// If `device_filter` is set, indexes are applied after filtering.
    /// By default, all devices are accepted.
    #[serde(default)]
    pub devices: Option<Vec<usize>>,
    /// How many parallel indexing processes are allowed to run.
    /// Default: 1
    #[serde(default)]
    pub parallel_indexes: Option<usize>,
    /// Allow to use integrated GPUs.
    /// Default: false
    #[serde(default)]
    pub allow_integrated: bool,
    /// Allow to use emulated GPUs like LLVMpipe. Useful for CI.
    /// Default: false
    #[serde(default)]
    pub allow_emulated: bool,
}

#[derive(Debug, Deserialize, Clone, Validate)]
pub struct Settings {
    #[serde(default)]
    pub log_level: Option<String>,
    #[serde(default)]
    pub logger: tracing::LoggerConfig,
    #[validate(nested)]
    pub storage: StorageConfig,
    #[validate(nested)]
    pub service: ServiceConfig,
    #[serde(default)]
    #[validate(nested)]
    pub cluster: ClusterConfig,
    #[serde(default = "default_telemetry_disabled")]
    pub telemetry_disabled: bool,
    #[validate(nested)]
    pub tls: Option<TlsConfig>,
    #[serde(default)]
    pub debugger: DebuggerConfig,
    /// A list of messages for errors that happened during loading the configuration. We collect
    /// them and store them here while loading because then our logger is not configured yet.
    /// We therefore need to log these messages later, after the logger is ready.
    #[serde(default, skip)]
    pub load_errors: Vec<LogMsg>,
    #[serde(default)]
    pub inference: Option<InferenceConfig>,
    #[serde(default)]
    #[validate(nested)]
    pub gpu: Option<GpuConfig>,
    #[serde(default)]
    pub feature_flags: FeatureFlags,
    /// Audit logging configuration.
    #[serde(default)]
    pub audit: Option<AuditConfig>,
}

impl Settings {
    pub fn new(custom_config_path: Option<String>) -> Result<Self, ConfigError> {
        let mut load_errors = vec![];
        let config_exists = |path| File::with_name(path).collect().is_ok();

        // Check if custom config file exists, report error if not
        if let Some(path) = &custom_config_path
            && !config_exists(path)
        {
            load_errors.push(LogMsg::Error(format!(
                "Config file via --config-path is not found: {path}"
            )));
        }

        let env = env::var("RUN_MODE").unwrap_or_else(|_| "development".into());
        let config_path_env = format!("config/{env}");

        // Report error if main or env config files exist, report warning if not
        // Check if main and env configuration file
        load_errors.extend(
            ["config/config", &config_path_env]
                .into_iter()
                .filter(|path| !config_exists(path))
                .map(|path| LogMsg::Warn(format!("Config file not found: {path}"))),
        );

        // Configuration builder: define different levels of configuration files
        let mut config = Config::builder()
            // Start with compile-time base config
            .add_source(File::from_str(DEFAULT_CONFIG, FileFormat::Yaml))
            // Merge main config: config/config
            .add_source(File::with_name("config/config").required(false))
            // Merge env config: config/{env}
            // Uses RUN_MODE, defaults to 'development'
            .add_source(File::with_name(&config_path_env).required(false))
            // Merge local config, not tracked in git: config/local
            .add_source(File::with_name("config/local").required(false));

        #[cfg(feature = "deb")]
        {
            // Read config, installed with deb package
            config = config.add_source(File::with_name("/etc/qdrant/config").required(false));
        }

        // Merge user provided config with --config-path
        if let Some(path) = custom_config_path {
            config = config.add_source(File::with_name(&path).required(false));
        }

        // Merge environment settings
        // E.g.: `QDRANT_DEBUG=1 ./target/app` would set `debug=true`
        config = config.add_source(Environment::with_prefix("QDRANT").separator("__"));

        // Build and merge config and deserialize into Settings, attach any load errors we had
        let mut settings: Settings = config.build()?.try_deserialize()?;
        settings.load_errors.extend(load_errors);
        Ok(settings)
    }

    pub fn tls(&self) -> io::Result<&TlsConfig> {
        self.tls
            .as_ref()
            .ok_or_else(Self::tls_config_is_undefined_error)
    }

    pub fn tls_config_is_undefined_error() -> io::Error {
        io::Error::other("TLS config is not defined in the Qdrant config file")
    }

    pub fn validate_and_warn(&self) {
        //
        // JWT RBAC
        //
        // Using HMAC-SHA256, recommended secret size is 32 bytes
        const JWT_RECOMMENDED_SECRET_LENGTH: usize = 256 / 8;

        let all_keys_are_empty = self
            .service
            .api_key
            .as_deref()
            .unwrap_or_default()
            .is_empty()
            && self
                .service
                .alt_api_key
                .as_deref()
                .unwrap_or_default()
                .is_empty();

        let min_length = [
            self.service.api_key.as_ref(),
            self.service.alt_api_key.as_ref(),
        ]
        .into_iter()
        .flatten()
        .map(|key| key.len())
        .min()
        .unwrap_or_default();

        let any_api_key_is_short = min_length < JWT_RECOMMENDED_SECRET_LENGTH;

        // Log if JWT RBAC is enabled but no API key is set
        if self.service.jwt_rbac.unwrap_or_default() {
            if all_keys_are_empty {
                log::warn!("JWT RBAC configured but no API key set, JWT RBAC is not enabled")
            // Log if JWT RAC is enabled, API key is set but smaller than recommended size for JWT secret
            } else if any_api_key_is_short {
                log::warn!(
                    "It is highly recommended to use an API key of {JWT_RECOMMENDED_SECRET_LENGTH} bytes when JWT RBAC is enabled",
                )
            }
        }

        // Print any load error messages we had
        self.load_errors.iter().for_each(LogMsg::log);

        if let Err(ref errs) = self.validate() {
            validation::warn_validation_errors("Settings configuration file", errs);
        }
    }
}

/// Returns the number of maximum actix workers.
pub fn max_web_workers(settings: &Settings) -> usize {
    match settings.service.max_workers {
        Some(0) => {
            let num_cpu = common::cpu::get_num_cpus();
            std::cmp::max(1, num_cpu - 1)
        }
        Some(max_workers) => max_workers,
        None => settings.storage.performance.max_search_threads,
    }
}

#[derive(Clone, Debug)]
pub enum LogMsg {
    Warn(String),
    Error(String),
}

impl LogMsg {
    fn log(&self) {
        match self {
            Self::Warn(msg) => log::warn!("{msg}"),
            Self::Error(msg) => log::error!("{msg}"),
        }
    }
}

const fn default_telemetry_disabled() -> bool {
    false
}

const fn default_cors() -> bool {
    true
}

const fn default_snapshot_url_recovery() -> bool {
    true
}

const fn default_http_keep_alive_timeout_sec() -> u64 {
    5
}

const fn default_http_client_request_timeout_sec() -> u64 {
    5
}

const fn default_http_client_disconnect_timeout_sec() -> u64 {
    5
}

const fn default_timeout_ms() -> u64 {
    DEFAULT_GRPC_TIMEOUT.as_millis() as u64
}

const fn default_connection_timeout_ms() -> u64 {
    DEFAULT_CONNECT_TIMEOUT.as_millis() as u64
}

const fn default_tick_period_ms() -> u64 {
    100
}

// Should not be less than `DEFAULT_META_OP_WAIT` as bootstrapping perform sync. consensus meta operations.
const fn default_bootstrap_timeout_sec() -> u64 {
    15
}

const fn default_max_message_queue_size() -> usize {
    100
}

const fn default_connection_pool_size() -> usize {
    DEFAULT_POOL_SIZE
}

const fn default_message_timeout_tics() -> u64 {
    10
}

const fn default_compact_wal_entries() -> u64 {
    128
}

#[allow(clippy::unnecessary_wraps)] // Used as serde default
const fn default_tls_cert_ttl() -> Option<u64> {
    // Default one hour
    Some(3600)
}

/// Custom validation function for metrics prefixes.
fn validate_metrics_prefix(prefix: &str) -> Result<(), ValidationError> {
    // Prefix is not required
    if prefix.is_empty() {
        return Ok(());
    }

    // Only allow alphanumeric characters or '_'
    if !prefix
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_')
    {
        return Err(
            ValidationError::new("invalid_metrics_prefix").with_message(Cow::Borrowed(
                "Metrics prefix must be of all alphanumeric characters, with an exception for '_'",
            )),
        );
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use fs_err as fs;
    use sealed_test::prelude::*;

    use super::*;

    /// Ensure we can successfully deserialize into [`Settings`] with just the default configuration.
    #[test]
    fn test_default_config() {
        let config = Config::builder()
            .add_source(File::from_str(DEFAULT_CONFIG, FileFormat::Yaml))
            .build()
            .expect("failed to build default config")
            .try_deserialize::<Settings>()
            .expect("failed to deserialize default config");

        assert_eq!(
            config.service.http_keep_alive_timeout_sec,
            default_http_keep_alive_timeout_sec()
        );
        assert_eq!(
            config.service.http_client_request_timeout_sec,
            default_http_client_request_timeout_sec()
        );
        assert_eq!(
            config.service.http_client_disconnect_timeout_sec,
            default_http_client_disconnect_timeout_sec()
        );

        config
            .validate()
            .expect("failed to validate default config");
    }

    #[expect(
        clippy::disallowed_methods,
        reason = "#[sealed_test] uses std::fs::copy"
    )]
    #[expect(clippy::disallowed_types, reason = "#[sealed_test] uses std::fs::File")]
    #[sealed_test(files = ["config/config.yaml", "config/development.yaml"])]
    fn test_runtime_development_config() {
        unsafe { env::set_var("RUN_MODE", "development") };

        // `sealed_test` copies files into the same directory as the test runs in.
        // We need them in a subdirectory.
        fs::create_dir("config").expect("failed to create `config` subdirectory.");
        fs::copy("config.yaml", "config/config.yaml").expect("failed to copy `config.yaml`.");
        fs::copy("development.yaml", "config/development.yaml")
            .expect("failed to copy `development.yaml`.");

        // Read config
        let config = Settings::new(None).expect("failed to load development config at runtime");

        // Validate
        config
            .validate()
            .expect("failed to validate development config at runtime");
        assert!(config.load_errors.is_empty(), "must not have load errors")
    }

    #[expect(clippy::disallowed_types, reason = "#[sealed_test] uses std::fs::File")]
    #[sealed_test]
    fn test_no_config_files() {
        let non_existing_config_path = "config/non_existing_config".to_string();

        // Read config
        let config = Settings::new(Some(non_existing_config_path))
            .expect("failed to load with non-existing runtime config");

        // Validate
        config
            .validate()
            .expect("failed to validate with non-existing runtime config");
        assert!(!config.load_errors.is_empty(), "must have load errors")
    }

    #[expect(clippy::disallowed_types, reason = "#[sealed_test] uses std::fs::File")]
    #[sealed_test]
    fn test_custom_config() {
        let path = "config/custom.yaml";

        // Create custom config file
        {
            fs::create_dir("config").unwrap();
            let mut custom = fs::File::create(path).unwrap();
            write!(&mut custom, "service:\n    http_port: 9999").unwrap();
            custom.flush().unwrap();
        }

        // Load settings with custom config
        let config = Settings::new(Some(path.into())).unwrap();

        // Ensure our custom config is the most important
        assert_eq!(config.service.http_port, 9999);
        assert_eq!(
            config.service.http_keep_alive_timeout_sec,
            default_http_keep_alive_timeout_sec()
        );
        assert_eq!(
            config.service.http_client_request_timeout_sec,
            default_http_client_request_timeout_sec()
        );
        assert_eq!(
            config.service.http_client_disconnect_timeout_sec,
            default_http_client_disconnect_timeout_sec()
        );
    }

    #[expect(clippy::disallowed_types, reason = "#[sealed_test] uses std::fs::File")]
    #[sealed_test]
    fn test_custom_http_transport_config() {
        let path = "config/custom_http_transport.yaml";

        {
            fs::create_dir("config").unwrap();
            let mut custom = fs::File::create(path).unwrap();
            write!(
                &mut custom,
                "service:\n    http_keep_alive_timeout_sec: 120\n    http_client_request_timeout_sec: 45\n    http_client_disconnect_timeout_sec: 60"
            )
            .unwrap();
            custom.flush().unwrap();
        }

        let config = Settings::new(Some(path.into())).unwrap();
        config
            .validate()
            .expect("custom HTTP transport timeouts must pass validation");

        assert_eq!(config.service.http_keep_alive_timeout_sec, 120);
        assert_eq!(config.service.http_client_request_timeout_sec, 45);
        assert_eq!(config.service.http_client_disconnect_timeout_sec, 60);
    }

    #[expect(clippy::disallowed_types, reason = "#[sealed_test] uses std::fs::File")]
    #[sealed_test]
    fn test_invalid_http_transport_config() {
        let path = "config/invalid_http_transport.yaml";

        {
            fs::create_dir("config").unwrap();
            let mut custom = fs::File::create(path).unwrap();
            write!(
                &mut custom,
                "service:\n    http_keep_alive_timeout_sec: 0\n    http_client_request_timeout_sec: 0\n    http_client_disconnect_timeout_sec: 0"
            )
            .unwrap();
            custom.flush().unwrap();
        }

        let config = Settings::new(Some(path.into())).unwrap();
        assert!(
            config.validate().is_err(),
            "zero timeout values must fail validation"
        );
    }
}