frugalos 1.2.0

//! Frugal Object Storage.
#![warn(missing_docs)]
#![allow(clippy::new_ret_no_self)]
extern crate atomic_immut;
extern crate bytecodec;
extern crate byteorder;
extern crate cannyls;
extern crate cannyls_rpc;
extern crate fibers;
extern crate fibers_http_server;
extern crate fibers_rpc;
extern crate fibers_tasque;
extern crate frugalos_config;
extern crate frugalos_core;
extern crate frugalos_mds;
extern crate frugalos_raft;
extern crate frugalos_segment;
extern crate futures;
extern crate httpcodec;
extern crate jemalloc_ctl;
extern crate libfrugalos;
extern crate num_cpus;
extern crate prometrics;
extern crate raftlog;
extern crate rustracing;
extern crate rustracing_jaeger;
extern crate serde;
#[macro_use]
extern crate serde_derive;
extern crate serde_yaml;
extern crate siphasher;
extern crate url;
#[macro_use]
extern crate slog;
#[cfg(test)]
extern crate tempdir;
#[macro_use]
extern crate trackable;
extern crate clap;
extern crate sloggers;

use cannyls::device::LongQueuePolicy;
use fibers_http_server::metrics::BucketConfig;
use std::fs::File;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;

macro_rules! dump {
    ($($e:expr),*) => {
        format!(concat!($(stringify!($e), "={:?}; "),*), $($e),*)
    }
}

pub use crate::error::{Error, ErrorKind};

pub mod command;
pub mod daemon;

/// The following module is automatically generated by build.rs .
pub mod build_information;

mod bucket;
mod client;
mod codec;
mod config_server;
mod device;
mod error;
mod http;
mod many_objects;
mod recovery;
mod rpc_server;
mod server;
mod service;

/// クレート固有の`Result`型。
pub type Result<T> = ::std::result::Result<T, Error>;

/// ファイルに書き出した時のフォーマットを調整する。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
struct FrugalosConfigWrapper {
    #[serde(rename = "frugalos")]
    #[serde(default)]
    config: FrugalosConfig,
}

/// frugalos の設定を表す struct。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrugalosConfig {
    /// データ用ディレクトリのパス。
    #[serde(default)]
    pub data_dir: String,
    /// ログをファイルに出力する場合の出力先ファイルパス。
    #[serde(default)]
    pub log_file: Option<PathBuf>,
    /// 出力するログレベルの下限。
    #[serde(default = "default_loglevel")]
    pub loglevel: sloggers::types::Severity,
    /// 同時に処理できるログの最大値。
    #[serde(default = "default_max_concurrent_logs")]
    pub max_concurrent_logs: usize,
    /// デーモン向けの設定。
    #[serde(default)]
    pub daemon: FrugalosDaemonConfig,
    /// サービス向けの設定。
    #[serde(default)]
    pub service: FrugalosServiceConfig,
    /// HTTP server 向けの設定。
    #[serde(default)]
    pub http_server: FrugalosHttpServerConfig,
    /// RPC client 向けの設定。
    #[serde(default)]
    pub rpc_client: FrugalosRpcClientConfig,
    /// frugalos_mds 向けの設定。
    #[serde(default)]
    pub mds: frugalos_mds::FrugalosMdsConfig,
    /// frugalos_segment 向けの設定。
    #[serde(default)]
    pub segment: frugalos_segment::FrugalosSegmentConfig,
    /// fibers_http_server 向けの設定。
    #[serde(default)]
    pub fibers_http_server: FibersHttpServerConfig,
    /// frugalos_config 向けの設定。
    #[serde(default)]
    pub config: frugalos_config::FrugalosConfigConfig,
}

impl FrugalosConfig {
    /// Reads `FrugalosConfig` from a YAML file.
    ///
    /// Return Value:
    ///  * The first component is an obtained value of `FrugalosConfig`.
    ///  * The second component is unknown fields that appear in the `path` with respect to the definition of `FrugalosConfig`.
    pub fn from_yaml<P: AsRef<Path>>(path: P) -> Result<(FrugalosConfig, std::vec::Vec<String>)> {
        let file = File::open(path).map_err(|e| track!(Error::from(e)))?;
        let value: serde_yaml::Value =
            serde_yaml::from_reader(file).map_err(|e| track!(Error::from(e)))?;

        let mut unknowns = std::vec::Vec::new();
        let wrapped: FrugalosConfigWrapper =
            serde_ignored::deserialize(value, |path| unknowns.push(path.to_string()))
                .map_err(|e| track!(Error::from(e)))?;
        Ok((wrapped.config, unknowns))
    }
}

impl Default for FrugalosConfig {
    fn default() -> Self {
        Self {
            data_dir: Default::default(),
            log_file: Default::default(),
            loglevel: default_loglevel(),
            max_concurrent_logs: default_max_concurrent_logs(),
            daemon: Default::default(),
            service: Default::default(),
            http_server: Default::default(),
            rpc_client: Default::default(),
            mds: Default::default(),
            segment: Default::default(),
            fibers_http_server: Default::default(),
            config: Default::default(),
        }
    }
}

/// `FrugalosDaemon` 向けの設定。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrugalosDaemonConfig {
    /// 実行スレッド数。
    #[serde(default = "default_executor_threads")]
    pub executor_threads: usize,

    /// Jaegerのトレースのサンプリング確率。
    #[serde(default = "default_sampling_rate")]
    pub sampling_rate: f64,
}

impl Default for FrugalosDaemonConfig {
    fn default() -> FrugalosDaemonConfig {
        Self {
            executor_threads: default_executor_threads(),
            sampling_rate: default_sampling_rate(),
        }
    }
}

/// `service::Service` 向けの設定。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
pub struct FrugalosServiceConfig {
    /// デバイスの起動についての設定。
    pub device: DeviceBuildingConfig,
}

mod cannyls_remote {
    use cannyls::device::LongQueuePolicy;
    #[derive(Serialize, Deserialize)]
    #[serde(remote = "LongQueuePolicy")]
    pub enum LongQueuePolicyDef {
        #[serde(rename = "refuse_new_requests")]
        RefuseNewRequests { ratio: f64 },
        #[serde(rename = "stop")]
        Stop,
        #[serde(rename = "drop")]
        Drop { ratio: f64 },
    }
}

/// デバイスの起動についての設定。
/// Option 型になっているフィールドは、None にすると cannyls のデフォルト値が採用される。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
pub struct DeviceBuildingConfig {
    /// デバイスが暇だと判定するための閾値(時間)。
    #[serde(default)]
    pub idle_threshold: Option<Duration>,
    /// デバイスの最大キュー長。
    #[serde(default)]
    pub max_queue_len: Option<usize>,
    /// デバイスが最大継続ビジー時間。
    #[serde(default)]
    pub max_keep_busy_duration: Option<Duration>,
    /// デバイスがビジー状態かどうかを判定するための閾値。
    #[serde(default)]
    pub busy_threshold: Option<usize>,
    /// キューが長くなった場合にどうするかの設定。
    #[serde(with = "cannyls_remote::LongQueuePolicyDef")]
    #[serde(default)]
    pub long_queue_policy: LongQueuePolicy,
}

/// HTTP server 向けの設定。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrugalosHttpServerConfig {
    /// bind するアドレス。
    #[serde(default = "default_http_server_bind_addr")]
    pub bind_addr: SocketAddr,
}

impl Default for FrugalosHttpServerConfig {
    fn default() -> Self {
        Self {
            bind_addr: default_http_server_bind_addr(),
        }
    }
}

/// RPC client 向けの設定。
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct FrugalosRpcClientConfig {
    /// RPC の接続タイムアウト時間。
    #[serde(
        rename = "tcp_connect_timeout_millis",
        default = "default_tcp_connect_timeout",
        with = "frugalos_core::serde_ext::duration_millis"
    )]
    pub tcp_connect_timeout: Duration,

    /// RPC の書き込みタイムアウト時間。
    #[serde(
        rename = "tcp_write_timeout_millis",
        default = "default_tcp_write_timeout",
        with = "frugalos_core::serde_ext::duration_millis"
    )]
    pub tcp_write_timeout: Duration,
}

impl FrugalosRpcClientConfig {
    fn channel_options(&self) -> fibers_rpc::channel::ChannelOptions {
        let mut options = fibers_rpc::channel::ChannelOptions::default();
        options.tcp_connect_timeout = self.tcp_connect_timeout;
        options.tcp_write_timeout = self.tcp_write_timeout;
        options
    }
}

impl Default for FrugalosRpcClientConfig {
    fn default() -> Self {
        Self {
            tcp_connect_timeout: default_tcp_connect_timeout(),
            tcp_write_timeout: default_tcp_write_timeout(),
        }
    }
}

/// fibers_http_server の設定。
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct FibersHttpServerConfig {
    request_duration_bucket_config: HttpRequestDurationHistogramBucketConfig,
}

/// histogram メトリクスにおける、バケツの upper_bound の設定。単調増加である必要がある。
///
/// 設定がない場合は fibers_http_server のデフォルト値が使われる。
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct HttpRequestDurationHistogramBucketConfig(pub Option<Vec<f64>>);

impl From<HttpRequestDurationHistogramBucketConfig> for BucketConfig {
    /// BucketConfig を作って返す
    fn from(config: HttpRequestDurationHistogramBucketConfig) -> Self {
        match config.0 {
            Some(buckets) => BucketConfig::new(buckets),
            None => BucketConfig::default(),
        }
    }
}

fn default_executor_threads() -> usize {
    num_cpus::get()
}

fn default_sampling_rate() -> f64 {
    0.001
}

fn default_http_server_bind_addr() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], 3000))
}

fn default_tcp_connect_timeout() -> Duration {
    Duration::from_secs(5)
}

fn default_tcp_write_timeout() -> Duration {
    Duration::from_secs(5)
}

fn default_loglevel() -> sloggers::types::Severity {
    sloggers::types::Severity::Info
}

fn default_max_concurrent_logs() -> usize {
    4096
}

#[cfg(test)]
mod tests {
    use super::*;
    use frugalos_segment::config::MdsRequestPolicy;
    use libfrugalos::time::Seconds;
    use std::fs::File;
    use std::io::Write;
    use tempdir::TempDir;
    use trackable::result::TestResult;

    #[test]
    fn config_works() -> TestResult {
        let content = r##"
---
frugalos:
  data_dir: "/var/lib/frugalos"
  log_file: ~
  loglevel: critical
  max_concurrent_logs: 30
  fibers_http_server:
    request_duration_bucket_config:
      - 1.5
      - 2.0
      - 3.0
      - 4.0
  daemon:
    executor_threads: 3
    sampling_rate: 0.1
  service:
    device:
      long_queue_policy:
        refuse_new_requests:
          ratio: 0.5
  http_server:
    bind_addr: "127.0.0.1:2222"
  rpc_client:
    tcp_connect_timeout_millis: 8000
    tcp_write_timeout_millis: 10000
  mds:
    commit_timeout_threshold: 20
    large_proposal_queue_threshold: 250
    large_leader_waiting_queue_threshold: 400
    leader_waiting_timeout_threshold: 12
    node_polling_interval_millis: 200
    reelection_threshold: 48
    snapshot_threshold_min: 100
    snapshot_threshold_max: 200
    staled_object_threshold: 5000
  config:
    leader_waiters_threshold: 20000
  segment:
    dispersed_client:
      get_timeout_millis: 4000
      cannyls_device_max_queue_len: 64
      cannyls_rpc_max_queue_len: 128
    replicated_client:
      cannyls_device_max_queue_len: 2048
      cannyls_rpc_max_queue_len: 32
    mds_client:
      get_request_policy:
        type: 'conservative'
      head_request_policy:
        type: 'conservative'
      default_request_policy:
        type: 'speculative'
        timeout_millis: 3000
      put_content_timeout_secs: 32"##;
        let dir = track_any_err!(TempDir::new("frugalos_test"))?;
        let filepath = dir.path().join("frugalos1.yml");
        let mut file = track_any_err!(File::create(filepath.clone()))?;

        track_any_err!(file.write(content.as_bytes()))?;

        let (actual, _) = track!(FrugalosConfig::from_yaml(filepath))?;
        let mut expected = FrugalosConfig::default();
        expected.data_dir = "/var/lib/frugalos".to_owned();
        expected.max_concurrent_logs = 30;
        expected.loglevel = sloggers::types::Severity::Critical;
        expected.fibers_http_server.request_duration_bucket_config =
            HttpRequestDurationHistogramBucketConfig(Some(vec![1.5, 2.0, 3.0, 4.0]));
        expected.daemon.sampling_rate = 0.1;
        expected.daemon.executor_threads = 3;
        expected.service.device.long_queue_policy =
            LongQueuePolicy::RefuseNewRequests { ratio: 0.5 };
        expected.http_server.bind_addr = SocketAddr::from(([127, 0, 0, 1], 2222));
        expected.rpc_client.tcp_connect_timeout = Duration::from_secs(8);
        expected.rpc_client.tcp_write_timeout = Duration::from_secs(10);
        expected.mds.commit_timeout_threshold = 20;
        expected.mds.large_proposal_queue_threshold = 250;
        expected.mds.large_leader_waiting_queue_threshold = 400;
        expected.mds.leader_waiting_timeout_threshold = 12;
        expected.mds.node_polling_interval = Duration::from_millis(200);
        expected.mds.reelection_threshold = 48;
        expected.mds.snapshot_threshold_min = 100;
        expected.mds.snapshot_threshold_max = 200;
        expected.mds.staled_object_threshold = 5000;
        expected.config.leader_waiters_threshold = 20000;
        expected.segment.dispersed_client.get_timeout = Duration::from_secs(4);
        expected
            .segment
            .dispersed_client
            .cannyls
            .device_max_queue_len = 64;
        expected.segment.dispersed_client.cannyls.rpc_max_queue_len = 128;
        expected
            .segment
            .replicated_client
            .cannyls
            .device_max_queue_len = 2048;
        expected.segment.replicated_client.cannyls.rpc_max_queue_len = 32;
        expected.segment.mds_client.get_request_policy = MdsRequestPolicy::Conservative;
        expected.segment.mds_client.head_request_policy = MdsRequestPolicy::Conservative;
        expected.segment.mds_client.default_request_policy = MdsRequestPolicy::Speculative {
            timeout: Duration::from_secs(3),
        };
        expected.segment.mds_client.put_content_timeout = Seconds(32);

        assert_eq!(expected, actual);

        Ok(())
    }

    #[test]
    fn default_config_values_is_used() -> TestResult {
        let content = r##"---
        frugalos: {}
        "##;
        let dir = track_any_err!(TempDir::new("frugalos_test"))?;
        let filepath = dir.path().join("frugalos2.yml");
        let mut file = track_any_err!(File::create(filepath.clone()))?;

        track_any_err!(file.write(content.as_bytes()))?;

        let (actual, _) = track!(FrugalosConfig::from_yaml(filepath))?;
        assert_eq!(FrugalosConfig::default(), actual);

        Ok(())
    }

    #[test]
    fn it_works_even_if_mds_config_is_missing() -> TestResult {
        let content = r##"---
        frugalos:
          segment: {}
        "##;
        let dir = track_any_err!(TempDir::new("frugalos_test"))?;
        let filepath = dir.path().join("frugalos3.yml");
        let mut file = track_any_err!(File::create(filepath.clone()))?;

        track_any_err!(file.write(content.as_bytes()))?;

        let (actual, _) = track!(FrugalosConfig::from_yaml(filepath))?;
        assert_eq!(FrugalosConfig::default(), actual);

        Ok(())
    }

    #[test]
    fn frugalos_config_value_must_not_be_unit_type() -> TestResult {
        let content = r##"---
        frugalos:
        "##;
        let dir = track_any_err!(TempDir::new("frugalos_test"))?;
        let filepath = dir.path().join("frugalos4.yml");
        let mut file = track_any_err!(File::create(filepath.clone()))?;

        track_any_err!(file.write(content.as_bytes()))?;
        assert!(FrugalosConfig::from_yaml(filepath).is_err());
        Ok(())
    }

    #[test]
    fn frugalos_config_from_yaml_reports_unknown_fields() -> TestResult {
        // NOTE: serde_ignored で認識されないケースについて
        // https://github.com/frugalos/frugalos/pull/130#issuecomment-476986133
        let content = r##"---
        frugalos:
          this_is_invalid_field:
            inner:  "these member are not in FrugalosConfig"
          daemon:
            executor_threadz: "typo of executor_threads"
          segment:
            mds_client:
              put_content_timeout_millis: "MdsClient only has put_content_timeout_secs"
        "##;
        let dir = track_any_err!(TempDir::new("frugalos_test"))?;
        let filepath = dir.path().join("frugalos5.yml");
        let mut file = track_any_err!(File::create(filepath.clone()))?;

        track_any_err!(file.write(content.as_bytes()))?;
        let (actual, mut unknowns) = track!(FrugalosConfig::from_yaml(filepath))?;
        unknowns.sort();

        assert_eq!(FrugalosConfig::default(), actual);
        assert_eq!(
            unknowns,
            vec![
                "frugalos.daemon.executor_threadz",
                "frugalos.segment.mds_client.put_content_timeout_millis",
                "frugalos.this_is_invalid_field"
            ]
        );

        Ok(())
    }

    #[test]
    fn long_queue_policy_serialize_works() {
        use cannyls::device::LongQueuePolicy;

        // LongQueuePolicy そのものには Serialize は実装されていないので、DeviceBuildingConfig にラップして検査する。
        let value = DeviceBuildingConfig {
            idle_threshold: None,
            max_queue_len: None,
            max_keep_busy_duration: None,
            busy_threshold: None,
            long_queue_policy: LongQueuePolicy::RefuseNewRequests { ratio: 0.5 },
        };
        let expected = r#"---
idle_threshold: ~
max_queue_len: ~
max_keep_busy_duration: ~
busy_threshold: ~
long_queue_policy:
  refuse_new_requests:
    ratio: 0.5"#;
        assert_eq!(serde_yaml::to_string(&value).unwrap(), expected);
    }
    #[test]
    fn long_queue_policy_deserialize_works() {
        use cannyls::device::LongQueuePolicy;

        // LongQueuePolicy そのものには Deserialize は実装されていないので、DeviceBuildingConfig にラップして検査する。
        let str = r#"---
long_queue_policy:
  refuse_new_requests:
    ratio: 0.5"#;
        let expected = LongQueuePolicy::RefuseNewRequests { ratio: 0.5 };
        assert_eq!(
            serde_yaml::from_str::<DeviceBuildingConfig>(&str)
                .unwrap()
                .long_queue_policy,
            expected
        );
        let str = r#"---
long_queue_policy: stop"#;
        let expected = LongQueuePolicy::Stop;
        assert_eq!(
            serde_yaml::from_str::<DeviceBuildingConfig>(&str)
                .unwrap()
                .long_queue_policy,
            expected
        );
        let str = r#"---
long_queue_policy:
  drop:
    ratio: 0.5"#;
        let expected = LongQueuePolicy::Drop { ratio: 0.5 };
        assert_eq!(
            serde_yaml::from_str::<DeviceBuildingConfig>(&str)
                .unwrap()
                .long_queue_policy,
            expected
        );
    }
}