use std::{
env,
net::{
SocketAddr,
ToSocketAddrs,
},
path::{
Path,
PathBuf,
},
time::Duration,
};
use buck2_resources::Buck2Resources;
use derive_builder::Builder;
use serde::{
Deserialize,
Serialize,
};
use si_crypto::VeritechCryptoConfig;
use si_data_nats::NatsConfig;
use si_pool_noodle::{
Instance,
instance::cyclone::{
LocalHttpInstance,
LocalHttpInstanceSpec,
LocalHttpSocketStrategy,
LocalUdsInstance,
LocalUdsInstanceSpec,
LocalUdsRuntimeStrategy,
LocalUdsSocketStrategy,
},
};
use si_service_endpoints::ServiceEndpointsConfig;
pub use si_settings::{
StandardConfig,
StandardConfigFile,
};
use si_std::CanonicalFileError;
use telemetry::prelude::*;
use thiserror::Error;
use ulid::Ulid;
const DEFAULT_VERITECH_REQUESTS_CONCURRENCY_LIMIT: usize = 1000;
const DEFAULT_POOL_SIZE: u32 = 50;
const DEFAULT_CYCLONE_CLIENT_EXECUTION_TIMEOUT_SECS: u64 = 60 * 35;
const DEFAULT_CYCLONE_CLIENT_EXECUTION_TIMEOUT: Duration =
Duration::from_secs(DEFAULT_CYCLONE_CLIENT_EXECUTION_TIMEOUT_SECS);
const DEFAULT_HEARTBEAT_APP_SLEEP_SECONDS: u64 = 15;
const DEFAULT_HEARTBEAT_APP_PUBLISH_TIMEOUT_SECONDS: u64 = 10;
const DEFAULT_HEARTBEAT_APP_SLEEP_DURATION: Duration =
Duration::from_secs(DEFAULT_HEARTBEAT_APP_SLEEP_SECONDS);
const DEFAULT_HEARTBEAT_APP_PUBLISH_TIMEOUT_DURATION: Duration =
Duration::from_secs(DEFAULT_HEARTBEAT_APP_PUBLISH_TIMEOUT_SECONDS);
#[remain::sorted]
#[derive(Debug, Error)]
pub enum ConfigError {
#[error("builder error: {0}")]
Builder(#[from] ConfigBuilderError),
#[error("canonical file error: {0}")]
CanonicalFile(#[from] CanonicalFileError),
#[error("cyclone spec build error")]
CycloneSpecBuild(#[source] Box<dyn std::error::Error + 'static + Sync + Send>),
#[error("no socket addrs where resolved")]
NoSocketAddrResolved,
#[error("settings error: {0}")]
Settings(#[from] si_settings::SettingsError),
#[error("failed to resolve socket addrs")]
SocketAddrResolve(#[source] std::io::Error),
}
impl ConfigError {
fn cyclone_spec_build(err: impl std::error::Error + 'static + Sync + Send) -> Self {
Self::CycloneSpecBuild(Box::new(err))
}
}
type Result<T> = std::result::Result<T, ConfigError>;
#[derive(Debug, Builder, Serialize)]
pub struct Config {
#[builder(default = "NatsConfig::default()")]
nats: NatsConfig,
#[serde(skip_serializing)]
cyclone_spec: CycloneSpec,
#[builder(default = "VeritechCryptoConfig::default()")]
crypto: VeritechCryptoConfig,
#[builder(default = "default_healthcheck_pool()")]
healthcheck_pool: bool,
#[builder(default = "default_cyclone_client_execution_timeout()")]
cyclone_client_execution_timeout: Duration,
#[builder(default = "default_veritech_requests_concurrency_limit()")]
veritech_requests_concurrency_limit: usize,
#[builder(default = "random_instance_id()")]
instance_id: String,
#[builder(default = "default_heartbeat_app()")]
heartbeat_app: bool,
#[builder(default = "default_heartbeat_app_sleep_duration()")]
heartbeat_app_sleep_duration: Duration,
#[builder(default = "default_heartbeat_app_publish_timeout_duration()")]
heartbeat_app_publish_timeout_duration: Duration,
#[builder(default = "default_service_endpoints_config()")]
service_endpoints: ServiceEndpointsConfig,
}
impl StandardConfig for Config {
type Builder = ConfigBuilder;
}
impl Config {
/// Gets a reference to the config's cyclone spec.
pub fn cyclone_spec(&self) -> &CycloneSpec {
&self.cyclone_spec
}
/// Gets a reference to the config's nats.
#[must_use]
pub fn nats(&self) -> &NatsConfig {
&self.nats
}
/// Gets a reference to the config's subject prefix.
pub fn subject_prefix(&self) -> Option<&str> {
self.nats.subject_prefix.as_deref()
}
/// Gets a reference to the config's cyclone public key path.
pub fn crypto(&self) -> &VeritechCryptoConfig {
&self.crypto
}
/// Gets the config's healthcheck value.
pub fn healthcheck_pool(&self) -> bool {
self.healthcheck_pool
}
/// Consumes into a [`CycloneSpec`].
pub fn into_cyclone_spec(self) -> CycloneSpec {
self.cyclone_spec
}
/// Gets the config's cyclone client execution timeout.
pub fn cyclone_client_execution_timeout(&self) -> Duration {
self.cyclone_client_execution_timeout
}
/// Gets the config's veritech requests concurrency limit.
pub fn veritech_requests_concurrency_limit(&self) -> usize {
self.veritech_requests_concurrency_limit
}
/// Gets the config's instance ID.
pub fn instance_id(&self) -> &str {
self.instance_id.as_ref()
}
/// Indicates if the heartbeat app will be enabled.
pub fn heartbeat_app(&self) -> bool {
self.heartbeat_app
}
/// Gets the config's sleep duration.
pub fn heartbeat_app_sleep_duration(&self) -> Duration {
self.heartbeat_app_sleep_duration
}
/// Gets the config's publish timeout duration.
pub fn heartbeat_app_publish_timeout_duration(&self) -> Duration {
self.heartbeat_app_publish_timeout_duration
}
/// Gets a reference to the config's service endpoints configuration.
#[must_use]
pub fn service_endpoints(&self) -> &ServiceEndpointsConfig {
&self.service_endpoints
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct ConfigFile {
#[serde(default)]
pub nats: NatsConfig,
pub cyclone: CycloneConfig,
#[serde(default)]
pub crypto: VeritechCryptoConfig,
#[serde(default = "default_healthcheck_pool")]
healthcheck_pool: bool,
#[serde(default = "default_cyclone_client_execution_timeout_secs")]
cyclone_client_execution_timeout_secs: u64,
#[serde(default = "default_veritech_requests_concurrency_limit")]
veritech_requests_concurrency_limit: usize,
#[serde(default = "random_instance_id")]
instance_id: String,
#[serde(default = "default_heartbeat_app")]
pub heartbeat_app: bool,
#[serde(default = "default_heartbeat_app_sleep_secs")]
heartbeat_app_sleep_secs: u64,
#[serde(default = "default_heartbeat_app_publish_timeout_secs")]
heartbeat_app_publish_timeout_secs: u64,
#[serde(default = "default_service_endpoints_config")]
service_endpoints: ServiceEndpointsConfig,
}
impl Default for ConfigFile {
fn default() -> Self {
Self::default_local_uds()
}
}
impl ConfigFile {
pub fn default_local_http() -> Self {
Self {
nats: Default::default(),
cyclone: CycloneConfig::default_local_http(),
crypto: Default::default(),
healthcheck_pool: default_healthcheck_pool(),
cyclone_client_execution_timeout_secs: default_cyclone_client_execution_timeout_secs(),
veritech_requests_concurrency_limit: default_veritech_requests_concurrency_limit(),
instance_id: random_instance_id(),
heartbeat_app: default_heartbeat_app(),
heartbeat_app_sleep_secs: default_heartbeat_app_sleep_secs(),
heartbeat_app_publish_timeout_secs: default_heartbeat_app_publish_timeout_secs(),
service_endpoints: default_service_endpoints_config(),
}
}
pub fn default_local_uds() -> Self {
Self {
nats: Default::default(),
cyclone: CycloneConfig::default_local_uds(),
crypto: Default::default(),
healthcheck_pool: default_healthcheck_pool(),
cyclone_client_execution_timeout_secs: default_cyclone_client_execution_timeout_secs(),
veritech_requests_concurrency_limit: default_veritech_requests_concurrency_limit(),
instance_id: random_instance_id(),
heartbeat_app: default_heartbeat_app(),
heartbeat_app_sleep_secs: default_heartbeat_app_sleep_secs(),
heartbeat_app_publish_timeout_secs: default_heartbeat_app_publish_timeout_secs(),
service_endpoints: default_service_endpoints_config(),
}
}
}
impl StandardConfigFile for ConfigFile {
type Error = ConfigError;
}
impl TryFrom<ConfigFile> for Config {
type Error = ConfigError;
fn try_from(mut value: ConfigFile) -> Result<Self> {
detect_and_configure_development(&mut value)?;
let mut config = Config::builder();
config.nats(value.nats);
config.cyclone_spec(value.cyclone.try_into()?);
config.crypto(value.crypto);
config.cyclone_client_execution_timeout(Duration::from_secs(
value.cyclone_client_execution_timeout_secs,
));
config.veritech_requests_concurrency_limit(value.veritech_requests_concurrency_limit);
config.instance_id(value.instance_id);
config.heartbeat_app(value.heartbeat_app);
config.heartbeat_app_sleep_duration(Duration::from_secs(value.heartbeat_app_sleep_secs));
config.heartbeat_app_publish_timeout_duration(Duration::from_secs(
value.heartbeat_app_publish_timeout_secs,
));
config.service_endpoints(value.service_endpoints);
config.build().map_err(Into::into)
}
}
#[remain::sorted]
#[derive(Clone, Debug)]
pub enum CycloneSpec {
LocalHttp(LocalHttpInstanceSpec),
LocalUds(LocalUdsInstanceSpec),
}
#[remain::sorted]
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CycloneStream {
HttpSocket(SocketAddr),
UnixDomainSocket(PathBuf),
}
impl Default for CycloneStream {
fn default() -> Self {
Self::HttpSocket(SocketAddr::from(([0, 0, 0, 0], 5157)))
}
}
impl CycloneStream {
pub fn http_socket(socket_addrs: impl ToSocketAddrs) -> Result<Self> {
let socket_addr = socket_addrs
.to_socket_addrs()
.map_err(ConfigError::SocketAddrResolve)?
.next()
.ok_or(ConfigError::NoSocketAddrResolved)?;
Ok(Self::HttpSocket(socket_addr))
}
pub fn unix_domain_socket(path: impl Into<PathBuf>) -> Self {
let pathbuf = path.into();
Self::UnixDomainSocket(pathbuf)
}
}
#[remain::sorted]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(tag = "kind")]
pub enum CycloneConfig {
LocalHttp {
#[serde(default = "default_cyclone_cmd_path")]
cyclone_cmd_path: String,
#[serde(default = "default_lang_server_cmd_path")]
lang_server_cmd_path: String,
#[serde(default)]
lang_server_function_timeout: Option<usize>,
#[serde(default)]
socket_strategy: LocalHttpSocketStrategy,
#[serde(default)]
watch_timeout: Option<Duration>,
#[serde(default = "default_limit_requests")]
limit_requets: Option<u32>,
#[serde(default = "default_enable_endpoint")]
ping: bool,
#[serde(default = "default_enable_endpoint")]
resolver: bool,
#[serde(default = "default_enable_endpoint")]
action: bool,
},
LocalUds {
#[serde(default = "default_cyclone_cmd_path")]
cyclone_cmd_path: String,
#[serde(default = "default_lang_server_cmd_path")]
lang_server_cmd_path: String,
#[serde(default)]
lang_server_function_timeout: Option<usize>,
#[serde(default)]
socket_strategy: LocalUdsSocketStrategy,
#[serde(default)]
runtime_strategy: LocalUdsRuntimeStrategy,
#[serde(default)]
watch_timeout: Option<Duration>,
#[serde(default = "default_limit_requests")]
limit_requets: Option<u32>,
#[serde(default = "default_enable_endpoint")]
ping: bool,
#[serde(default = "default_enable_endpoint")]
resolver: bool,
#[serde(default = "default_enable_endpoint")]
action: bool,
#[serde(default)]
pool_size: u32,
#[serde(default)]
connect_timeout: u64,
#[serde(default = "default_create_firecracker_setup_scripts")]
create_firecracker_setup_scripts: bool,
},
}
impl CycloneConfig {
pub fn default_local_http() -> Self {
Self::LocalHttp {
cyclone_cmd_path: default_cyclone_cmd_path(),
lang_server_cmd_path: default_lang_server_cmd_path(),
lang_server_function_timeout: Default::default(),
socket_strategy: Default::default(),
watch_timeout: Default::default(),
limit_requets: default_limit_requests(),
ping: default_enable_endpoint(),
resolver: default_enable_endpoint(),
action: default_enable_endpoint(),
}
}
pub fn default_local_uds() -> Self {
Self::LocalUds {
cyclone_cmd_path: default_cyclone_cmd_path(),
lang_server_cmd_path: default_lang_server_cmd_path(),
lang_server_function_timeout: Default::default(),
socket_strategy: Default::default(),
runtime_strategy: default_runtime_strategy(),
watch_timeout: Default::default(),
limit_requets: default_limit_requests(),
ping: default_enable_endpoint(),
resolver: default_enable_endpoint(),
action: default_enable_endpoint(),
pool_size: default_pool_size(),
connect_timeout: default_connect_timeout(),
create_firecracker_setup_scripts: default_create_firecracker_setup_scripts(),
}
}
pub fn cyclone_cmd_path(&self) -> &str {
match self {
CycloneConfig::LocalUds {
cyclone_cmd_path, ..
} => cyclone_cmd_path,
CycloneConfig::LocalHttp {
cyclone_cmd_path, ..
} => cyclone_cmd_path,
}
}
pub fn set_cyclone_cmd_path(&mut self, value: String) {
match self {
CycloneConfig::LocalUds {
cyclone_cmd_path, ..
} => *cyclone_cmd_path = value,
CycloneConfig::LocalHttp {
cyclone_cmd_path, ..
} => *cyclone_cmd_path = value,
};
}
pub fn lang_server_cmd_path(&self) -> &str {
match self {
CycloneConfig::LocalUds {
lang_server_cmd_path,
..
} => lang_server_cmd_path,
CycloneConfig::LocalHttp {
lang_server_cmd_path,
..
} => lang_server_cmd_path,
}
}
pub fn set_lang_server_cmd_path(&mut self, value: String) {
match self {
CycloneConfig::LocalUds {
lang_server_cmd_path,
..
} => *lang_server_cmd_path = value,
CycloneConfig::LocalHttp {
lang_server_cmd_path,
..
} => *lang_server_cmd_path = value,
};
}
pub fn set_limit_requests(&mut self, value: impl Into<Option<u32>>) {
match self {
CycloneConfig::LocalUds { limit_requets, .. } => *limit_requets = value.into(),
CycloneConfig::LocalHttp { limit_requets, .. } => *limit_requets = value.into(),
};
}
pub fn set_ping(&mut self, value: bool) {
match self {
CycloneConfig::LocalUds { ping, .. } => *ping = value,
CycloneConfig::LocalHttp { ping, .. } => *ping = value,
};
}
pub fn set_resolver(&mut self, value: bool) {
match self {
CycloneConfig::LocalUds { resolver, .. } => *resolver = value,
CycloneConfig::LocalHttp { resolver, .. } => *resolver = value,
};
}
pub fn set_action(&mut self, value: bool) {
match self {
CycloneConfig::LocalUds { action, .. } => *action = value,
CycloneConfig::LocalHttp { action, .. } => *action = value,
};
}
pub fn set_pool_size(&mut self, value: u32) {
if let CycloneConfig::LocalUds { pool_size, .. } = self {
*pool_size = value
};
}
}
impl Default for CycloneConfig {
fn default() -> Self {
Self::default_local_uds()
}
}
impl TryFrom<CycloneConfig> for CycloneSpec {
type Error = ConfigError;
fn try_from(value: CycloneConfig) -> std::result::Result<Self, Self::Error> {
match value {
CycloneConfig::LocalUds {
cyclone_cmd_path,
lang_server_cmd_path,
lang_server_function_timeout,
socket_strategy,
runtime_strategy,
watch_timeout,
limit_requets,
ping,
resolver,
action,
pool_size,
connect_timeout,
create_firecracker_setup_scripts,
} => {
let mut builder = LocalUdsInstance::spec();
//we only need these if running local process. Maybe the builder should handle
//this?
if matches!(runtime_strategy, LocalUdsRuntimeStrategy::LocalProcess) {
builder
.try_cyclone_cmd_path(cyclone_cmd_path)
.map_err(ConfigError::cyclone_spec_build)?;
builder
.try_lang_server_cmd_path(lang_server_cmd_path)
.map_err(ConfigError::cyclone_spec_build)?;
}
builder.lang_server_function_timeout(lang_server_function_timeout);
builder.socket_strategy(socket_strategy);
builder.runtime_strategy(runtime_strategy);
if let Some(watch_timeout) = watch_timeout {
builder.watch_timeout(watch_timeout);
}
builder.limit_requests(limit_requets);
if ping {
builder.ping();
}
if resolver {
builder.resolver();
}
if action {
builder.action();
}
builder.pool_size(pool_size);
builder.connect_timeout(connect_timeout);
builder.create_firecracker_setup_scripts(create_firecracker_setup_scripts);
Ok(Self::LocalUds(
builder.build().map_err(ConfigError::cyclone_spec_build)?,
))
}
CycloneConfig::LocalHttp {
cyclone_cmd_path,
lang_server_cmd_path,
lang_server_function_timeout,
socket_strategy,
watch_timeout,
limit_requets,
ping,
resolver,
action,
} => {
let mut builder = LocalHttpInstance::spec();
builder
.try_cyclone_cmd_path(cyclone_cmd_path)
.map_err(ConfigError::cyclone_spec_build)?;
builder
.try_lang_server_cmd_path(lang_server_cmd_path)
.map_err(ConfigError::cyclone_spec_build)?;
builder.lang_server_function_timeout(lang_server_function_timeout);
builder.socket_strategy(socket_strategy);
if let Some(watch_timeout) = watch_timeout {
builder.watch_timeout(watch_timeout);
}
builder.limit_requests(limit_requets);
if ping {
builder.ping();
}
if resolver {
builder.resolver();
}
if action {
builder.action();
}
Ok(Self::LocalHttp(
builder.build().map_err(ConfigError::cyclone_spec_build)?,
))
}
}
}
}
fn random_instance_id() -> String {
Ulid::new().to_string()
}
fn default_cyclone_cmd_path() -> String {
"/usr/local/bin/cyclone".to_string()
}
fn default_lang_server_cmd_path() -> String {
"/usr/local/bin/lang-js".to_string()
}
fn default_limit_requests() -> Option<u32> {
Some(1)
}
fn default_enable_endpoint() -> bool {
true
}
fn default_runtime_strategy() -> LocalUdsRuntimeStrategy {
LocalUdsRuntimeStrategy::default()
}
fn default_pool_size() -> u32 {
DEFAULT_POOL_SIZE
}
fn default_connect_timeout() -> u64 {
10
}
fn default_healthcheck_pool() -> bool {
true
}
fn default_cyclone_client_execution_timeout() -> Duration {
DEFAULT_CYCLONE_CLIENT_EXECUTION_TIMEOUT
}
fn default_cyclone_client_execution_timeout_secs() -> u64 {
DEFAULT_CYCLONE_CLIENT_EXECUTION_TIMEOUT_SECS
}
fn default_veritech_requests_concurrency_limit() -> usize {
DEFAULT_VERITECH_REQUESTS_CONCURRENCY_LIMIT
}
fn default_heartbeat_app() -> bool {
true
}
fn default_heartbeat_app_sleep_duration() -> Duration {
DEFAULT_HEARTBEAT_APP_SLEEP_DURATION
}
fn default_heartbeat_app_sleep_secs() -> u64 {
DEFAULT_HEARTBEAT_APP_SLEEP_SECONDS
}
fn default_heartbeat_app_publish_timeout_duration() -> Duration {
DEFAULT_HEARTBEAT_APP_PUBLISH_TIMEOUT_DURATION
}
fn default_heartbeat_app_publish_timeout_secs() -> u64 {
DEFAULT_HEARTBEAT_APP_PUBLISH_TIMEOUT_SECONDS
}
fn default_create_firecracker_setup_scripts() -> bool {
true
}
fn default_service_endpoints_config() -> ServiceEndpointsConfig {
ServiceEndpointsConfig::new(0)
}
#[allow(clippy::disallowed_methods)] // Used to determine if running in development
pub fn detect_and_configure_development(config: &mut ConfigFile) -> Result<()> {
if env::var("BUCK_RUN_BUILD_ID").is_ok() || env::var("BUCK_BUILD_ID").is_ok() {
buck2_development(config)
} else if let Ok(dir) = env::var("CARGO_MANIFEST_DIR") {
cargo_development(dir, config)
} else {
Ok(())
}
}
fn buck2_development(config: &mut ConfigFile) -> Result<()> {
let resources = Buck2Resources::read().map_err(ConfigError::cyclone_spec_build)?;
let cyclone_cmd_path = resources
.get_ends_with("cyclone")
.map_err(ConfigError::cyclone_spec_build)?
.to_string_lossy()
.to_string();
let decryption_key_path = resources
.get_ends_with("dev.decryption.key")
.map_err(ConfigError::cyclone_spec_build)?
.to_string_lossy()
.to_string();
let lang_server_cmd_path = resources
.get_ends_with("lang-js")
.map_err(ConfigError::cyclone_spec_build)?
.to_string_lossy()
.to_string();
warn!(
cyclone_cmd_path = cyclone_cmd_path.as_str(),
decryption_key_path = decryption_key_path.as_str(),
lang_server_cmd_path = lang_server_cmd_path,
"detected development run",
);
config.cyclone.set_cyclone_cmd_path(cyclone_cmd_path);
config.crypto.decryption_key_file = decryption_key_path.parse().ok();
config
.cyclone
.set_lang_server_cmd_path(lang_server_cmd_path.to_string());
Ok(())
}
fn cargo_development(dir: String, config: &mut ConfigFile) -> Result<()> {
let cyclone_cmd_path = Path::new(&dir)
.join("../../target/debug/cyclone")
.canonicalize()
.expect("failed to canonicalize local dev build of <root>/target/debug/cyclone")
.to_string_lossy()
.to_string();
let decryption_key_path = Path::new(&dir)
.join("../../lib/veritech-server/src/dev.decryption.key")
.canonicalize()
.expect(
"failed to canonicalize local key at <root>/lib/veritech-server/src/dev.decryption.key",
)
.to_string_lossy()
.to_string();
let lang_server_cmd_path = Path::new(&dir)
.join("../../bin/lang-js/target/lang-js")
.canonicalize()
.expect("failed to canonicalize local dev build of <root>/bin/lang-js/target/lang-js")
.to_string_lossy()
.to_string();
warn!(
cyclone_cmd_path = cyclone_cmd_path.as_str(),
decryption_key_path = decryption_key_path.as_str(),
lang_server_cmd_path = lang_server_cmd_path.as_str(),
"detected development run",
);
config.cyclone.set_cyclone_cmd_path(cyclone_cmd_path);
config.crypto.decryption_key_file = decryption_key_path.parse().ok();
config
.cyclone
.set_lang_server_cmd_path(lang_server_cmd_path);
Ok(())
}
