pub mod cache;
pub mod error;
pub mod resolver;
use crate::error::FlagdError;
#[cfg(feature = "in-process")]
use crate::resolver::in_process::resolver::{FileResolver, InProcessResolver};
use async_trait::async_trait;
#[cfg(feature = "rpc")]
use open_feature::EvaluationContextFieldValue;
use open_feature::provider::{FeatureProvider, ProviderMetadata, ResolutionDetails};
use open_feature::{EvaluationContext, EvaluationError, StructValue, Value};
#[cfg(feature = "rest")]
use resolver::rest::RestResolver;
use tracing::debug;
use tracing::instrument;
#[cfg(feature = "rpc")]
use std::collections::BTreeMap;
use std::sync::Arc;
pub use cache::{CacheService, CacheSettings, CacheType};
#[cfg(feature = "rpc")]
pub use resolver::rpc::RpcResolver;
#[cfg(any(feature = "rpc", feature = "in-process"))]
pub mod flagd {
#[cfg(feature = "rpc")]
pub mod evaluation {
pub mod v1 {
include!(concat!(env!("OUT_DIR"), "/flagd.evaluation.v1.rs"));
}
}
#[cfg(feature = "in-process")]
pub mod sync {
pub mod v1 {
include!(concat!(env!("OUT_DIR"), "/flagd.sync.v1.rs"));
}
}
}
#[derive(Debug, Clone)]
pub struct FlagdOptions {
pub host: String,
pub port: u16,
pub target_uri: Option<String>,
pub resolver_type: ResolverType,
pub tls: bool,
pub cert_path: Option<String>,
pub deadline_ms: u32,
pub cache_settings: Option<CacheSettings>,
pub retry_backoff_ms: u32,
pub retry_backoff_max_ms: u32,
pub retry_grace_period: u32,
pub selector: Option<String>,
pub socket_path: Option<String>,
pub source_configuration: Option<String>,
pub stream_deadline_ms: u32,
pub keep_alive_time_ms: u64,
pub offline_poll_interval_ms: Option<u32>,
pub provider_id: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub enum ResolverType {
#[cfg(feature = "rpc")]
Rpc,
#[cfg(feature = "rest")]
Rest,
#[cfg(feature = "in-process")]
InProcess,
#[cfg(feature = "in-process")]
File,
}
impl Default for FlagdOptions {
fn default() -> Self {
let resolver_type = Self::default_resolver_type();
let port = Self::default_port(&resolver_type);
#[allow(unused_mut)]
let mut options = Self {
host: std::env::var("FLAGD_HOST").unwrap_or_else(|_| "localhost".to_string()),
port: std::env::var("FLAGD_PORT")
.ok()
.and_then(|p| p.parse().ok())
.unwrap_or(port),
target_uri: std::env::var("FLAGD_TARGET_URI").ok(),
resolver_type,
tls: std::env::var("FLAGD_TLS")
.map(|v| v.to_lowercase() == "true")
.unwrap_or(false),
cert_path: std::env::var("FLAGD_SERVER_CERT_PATH").ok(),
deadline_ms: std::env::var("FLAGD_DEADLINE_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(500),
retry_backoff_ms: std::env::var("FLAGD_RETRY_BACKOFF_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(1000),
retry_backoff_max_ms: std::env::var("FLAGD_RETRY_BACKOFF_MAX_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(120000),
retry_grace_period: std::env::var("FLAGD_RETRY_GRACE_PERIOD")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(5),
stream_deadline_ms: std::env::var("FLAGD_STREAM_DEADLINE_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(600000),
keep_alive_time_ms: std::env::var("FLAGD_KEEP_ALIVE_TIME_MS")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0), socket_path: std::env::var("FLAGD_SOCKET_PATH").ok(),
selector: std::env::var("FLAGD_SOURCE_SELECTOR").ok(),
cache_settings: Some(CacheSettings::default()),
source_configuration: std::env::var("FLAGD_OFFLINE_FLAG_SOURCE_PATH").ok(),
offline_poll_interval_ms: Some(
std::env::var("FLAGD_OFFLINE_POLL_MS")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(5000),
),
provider_id: std::env::var("FLAGD_PROVIDER_ID").ok(),
};
#[cfg(feature = "in-process")]
{
let resolver_env_set = std::env::var("FLAGD_RESOLVER").is_ok();
if options.source_configuration.is_some() && !resolver_env_set {
options.resolver_type = ResolverType::File;
}
if matches!(
options.resolver_type,
ResolverType::InProcess | ResolverType::File
) {
options.cache_settings = None;
}
}
options
}
}
impl FlagdOptions {
fn default_resolver_type() -> ResolverType {
if let Ok(r) = std::env::var("FLAGD_RESOLVER") {
match r.to_uppercase().as_str() {
#[cfg(feature = "rpc")]
"RPC" => return ResolverType::Rpc,
#[cfg(feature = "rest")]
"REST" => return ResolverType::Rest,
#[cfg(feature = "in-process")]
"IN-PROCESS" | "INPROCESS" => return ResolverType::InProcess,
#[cfg(feature = "in-process")]
"FILE" | "OFFLINE" => return ResolverType::File,
_ => {}
}
}
#[cfg(feature = "rpc")]
return ResolverType::Rpc;
#[cfg(all(feature = "rest", not(feature = "rpc")))]
return ResolverType::Rest;
#[cfg(all(feature = "in-process", not(feature = "rpc"), not(feature = "rest")))]
return ResolverType::InProcess;
#[cfg(not(any(feature = "rpc", feature = "rest", feature = "in-process")))]
compile_error!("At least one resolver feature must be enabled: rpc, rest, or in-process");
}
fn default_port(resolver_type: &ResolverType) -> u16 {
match resolver_type {
#[cfg(feature = "rpc")]
ResolverType::Rpc => 8013,
#[cfg(feature = "in-process")]
ResolverType::InProcess => 8015,
#[cfg(feature = "rest")]
ResolverType::Rest => 8016,
#[allow(unreachable_patterns)]
_ => 8013,
}
}
}
#[derive(Clone)]
pub struct FlagdProvider {
provider: Arc<dyn FeatureProvider + Send + Sync>,
cache: Option<Arc<CacheService<Value>>>,
}
impl FlagdProvider {
#[instrument(skip(options))]
pub async fn new(options: FlagdOptions) -> Result<Self, FlagdError> {
debug!("Initializing FlagdProvider with options: {:?}", options);
#[cfg(feature = "in-process")]
if options.resolver_type == ResolverType::File && options.source_configuration.is_none() {
return Err(FlagdError::Config(
"File resolver requires 'source_configuration' (FLAGD_OFFLINE_FLAG_SOURCE_PATH) to be set".to_string()
));
}
let provider: Arc<dyn FeatureProvider + Send + Sync> = match options.resolver_type {
#[cfg(feature = "rpc")]
ResolverType::Rpc => {
debug!("Using RPC resolver");
Arc::new(RpcResolver::new(&options).await?)
}
#[cfg(feature = "rest")]
ResolverType::Rest => {
debug!("Using REST resolver");
Arc::new(RestResolver::new(&options).await?)
}
#[cfg(feature = "in-process")]
ResolverType::InProcess => {
debug!("Using in-process resolver");
Arc::new(InProcessResolver::new(&options).await?)
}
#[cfg(feature = "in-process")]
ResolverType::File => {
debug!("Using file resolver");
Arc::new(
FileResolver::new(
options
.source_configuration
.expect("source_configuration validated above"),
options.cache_settings.clone(),
)
.await?,
)
}
};
Ok(Self {
provider,
cache: options
.cache_settings
.map(|settings| Arc::new(CacheService::new(settings))),
})
}
}
impl std::fmt::Debug for FlagdProvider {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlagdProvider")
.field("cache", &self.cache)
.finish()
}
}
#[cfg(feature = "rpc")]
pub(crate) fn convert_context(context: &EvaluationContext) -> Option<prost_types::Struct> {
let mut fields = BTreeMap::new();
if let Some(targeting_key) = &context.targeting_key {
fields.insert(
"targetingKey".to_string(),
prost_types::Value {
kind: Some(prost_types::value::Kind::StringValue(targeting_key.clone())),
},
);
}
for (key, value) in &context.custom_fields {
let prost_value = match value {
EvaluationContextFieldValue::String(s) => prost_types::Value {
kind: Some(prost_types::value::Kind::StringValue(s.clone())),
},
EvaluationContextFieldValue::Bool(b) => prost_types::Value {
kind: Some(prost_types::value::Kind::BoolValue(*b)),
},
EvaluationContextFieldValue::Int(i) => prost_types::Value {
kind: Some(prost_types::value::Kind::NumberValue(*i as f64)),
},
EvaluationContextFieldValue::Float(f) => prost_types::Value {
kind: Some(prost_types::value::Kind::NumberValue(*f)),
},
EvaluationContextFieldValue::DateTime(dt) => prost_types::Value {
kind: Some(prost_types::value::Kind::StringValue(dt.to_string())),
},
EvaluationContextFieldValue::Struct(s) => prost_types::Value {
kind: Some(prost_types::value::Kind::StringValue(format!("{:?}", s))),
},
};
fields.insert(key.clone(), prost_value);
}
Some(prost_types::Struct { fields })
}
#[cfg(feature = "rpc")]
pub(crate) fn convert_proto_struct_to_struct_value(
proto_struct: prost_types::Struct,
) -> StructValue {
let fields = proto_struct
.fields
.into_iter()
.map(|(key, value)| {
(
key,
match value.kind.unwrap() {
prost_types::value::Kind::NullValue(_) => Value::String(String::new()),
prost_types::value::Kind::NumberValue(n) => Value::Float(n),
prost_types::value::Kind::StringValue(s) => Value::String(s),
prost_types::value::Kind::BoolValue(b) => Value::Bool(b),
prost_types::value::Kind::StructValue(s) => Value::String(format!("{:?}", s)),
prost_types::value::Kind::ListValue(l) => Value::String(format!("{:?}", l)),
},
)
})
.collect();
StructValue { fields }
}
impl FlagdProvider {
async fn get_cached_value<T>(
&self,
flag_key: &str,
context: &EvaluationContext,
value_converter: impl Fn(Value) -> Option<T>,
) -> Option<T> {
if let Some(cache) = &self.cache
&& let Some(cached_value) = cache.get(flag_key, context).await
{
return value_converter(cached_value);
}
None
}
}
#[async_trait]
impl FeatureProvider for FlagdProvider {
fn metadata(&self) -> &ProviderMetadata {
self.provider.metadata()
}
async fn resolve_bool_value(
&self,
flag_key: &str,
context: &EvaluationContext,
) -> Result<ResolutionDetails<bool>, EvaluationError> {
if let Some(value) = self
.get_cached_value(flag_key, context, |v| match v {
Value::Bool(b) => Some(b),
_ => None,
})
.await
{
return Ok(ResolutionDetails::new(value));
}
let result = self.provider.resolve_bool_value(flag_key, context).await?;
if let Some(cache) = &self.cache {
cache
.add(flag_key, context, Value::Bool(result.value))
.await;
}
Ok(result)
}
async fn resolve_int_value(
&self,
flag_key: &str,
context: &EvaluationContext,
) -> Result<ResolutionDetails<i64>, EvaluationError> {
if let Some(value) = self
.get_cached_value(flag_key, context, |v| match v {
Value::Int(i) => Some(i),
_ => None,
})
.await
{
return Ok(ResolutionDetails::new(value));
}
let result = self.provider.resolve_int_value(flag_key, context).await?;
if let Some(cache) = &self.cache {
cache.add(flag_key, context, Value::Int(result.value)).await;
}
Ok(result)
}
async fn resolve_float_value(
&self,
flag_key: &str,
context: &EvaluationContext,
) -> Result<ResolutionDetails<f64>, EvaluationError> {
if let Some(value) = self
.get_cached_value(flag_key, context, |v| match v {
Value::Float(f) => Some(f),
_ => None,
})
.await
{
return Ok(ResolutionDetails::new(value));
}
let result = self.provider.resolve_float_value(flag_key, context).await?;
if let Some(cache) = &self.cache {
cache
.add(flag_key, context, Value::Float(result.value))
.await;
}
Ok(result)
}
async fn resolve_string_value(
&self,
flag_key: &str,
context: &EvaluationContext,
) -> Result<ResolutionDetails<String>, EvaluationError> {
if let Some(value) = self
.get_cached_value(flag_key, context, |v| match v {
Value::String(s) => Some(s),
_ => None,
})
.await
{
return Ok(ResolutionDetails::new(value));
}
let result = self
.provider
.resolve_string_value(flag_key, context)
.await?;
if let Some(cache) = &self.cache {
cache
.add(flag_key, context, Value::String(result.value.clone()))
.await;
}
Ok(result)
}
async fn resolve_struct_value(
&self,
flag_key: &str,
context: &EvaluationContext,
) -> Result<ResolutionDetails<StructValue>, EvaluationError> {
if let Some(value) = self
.get_cached_value(flag_key, context, |v| match v {
Value::Struct(s) => Some(s),
_ => None,
})
.await
{
return Ok(ResolutionDetails::new(value));
}
let result = self
.provider
.resolve_struct_value(flag_key, context)
.await?;
if let Some(cache) = &self.cache {
cache
.add(flag_key, context, Value::Struct(result.value.clone()))
.await;
}
Ok(result)
}
}