use std::{collections::HashMap, env, future::Future, net::SocketAddr, sync::Arc, time::Duration};
use anyhow::{Context, Result, ensure};
use tracing::{info, warn};
use crate::{
bucket::{GcsBucket, GrpcReplicaPeers, RuntimeStorage},
postgres::PostgresDatabase,
sandbox::{CommandSandboxProvider, HostSandboxRuntimeConfig},
};
use super::{ActorJwtVerifier, ControlPlaneService};
const DEFAULT_JWT_ISSUER: &str = "durable-actors-control-plane";
const DEFAULT_AUTHORITY_AUDIENCE: &str = "durable-actors-authority";
const DEFAULT_INVOCATION_AUDIENCE: &str = "durable-actors-invoke";
const DEFAULT_JWT_TTL_SECONDS: u64 = 86_400;
const DEFAULT_ACTOR_IDLE_TIMEOUT_SECONDS: u64 = 60;
const DEFAULT_HOST_IDLE_TIMEOUT_MS: u64 = 300_000;
const MAX_IDLE_TIMEOUT_MS: u64 = 86_400_000;
pub struct ControlPlaneProcessConfig {
pub bind: SocketAddr,
pub jwt_signing_key: String,
pub jwt_key_id: String,
pub jwt_issuer: String,
pub authority_audience: String,
pub invocation_audience: String,
pub jwt_max_lifetime: Duration,
pub api_key: Option<String>,
pub storage: ControlPlaneStorageConfig,
pub sandbox_provider: SandboxProviderConfig,
pub socket_event_sink: Option<SocketEventSinkConfig>,
pub region: Option<String>,
}
pub struct ControlPlaneStorageConfig {
pub postgres_url: String,
pub bucket: String,
pub replica_regions: Vec<String>,
}
pub struct SandboxProviderConfig {
pub runtime_image: String,
pub(super) pool: crate::sandbox::pool::PoolConfig,
pub provider_name: String,
pub command: String,
pub environment: HashMap<String, String>,
pub runtime: HostSandboxRuntimeConfig,
}
pub struct SocketEventSinkConfig {
pub url: String,
}
impl ControlPlaneProcessConfig {
pub fn from_env() -> Result<Self> {
Self::from_lookup(|name| env::var(name).ok())
}
}
pub async fn serve_control_plane(
config: ControlPlaneProcessConfig,
shutdown: impl Future<Output = ()> + Send + 'static,
) -> Result<()> {
let bind = config.bind;
warn_if_authentication_disabled(bind, config.api_key.as_deref());
let stop = tokio_util::sync::CancellationToken::new();
let _guard = stop.clone().drop_guard();
let routes = control_plane_routes(config, stop).await?;
info!(bind = %bind, "durable-actors control plane is ready");
let listener = tokio::net::TcpListener::bind(bind)
.await
.context("bind durable-actors control plane")?;
serve_routes(listener, routes, shutdown).await
}
fn warn_if_authentication_disabled(bind: SocketAddr, secret: Option<&str>) {
if secret.is_none() && !bind.ip().is_loopback() {
warn!(
%bind,
"Authentication is disabled. Anyone who can reach this server can access its API. Set DURABLE_ACTORS_SECRET to enable authentication."
);
}
}
async fn serve_routes(
listener: tokio::net::TcpListener,
routes: tonic::service::Routes,
shutdown: impl Future<Output = ()> + Send + 'static,
) -> Result<()> {
axum::serve(listener, routes.into_axum_router())
.with_graceful_shutdown(shutdown)
.await
.context("serve durable-actors control plane")
}
async fn control_plane_routes(
config: ControlPlaneProcessConfig,
stop: tokio_util::sync::CancellationToken,
) -> Result<tonic::service::Routes> {
let issuer = super::ActorJwtIssuer::from_base64_pkcs8(
&config.jwt_signing_key,
config.jwt_key_id,
config.jwt_issuer.clone(),
config.authority_audience.clone(),
config.invocation_audience,
config.jwt_max_lifetime,
)?;
let auth = ActorJwtVerifier::for_scope(
issuer.verifier_keys_json()?,
config.jwt_issuer,
config.authority_audience,
super::ActorTokenPurpose::ControlPlane,
config.jwt_max_lifetime,
)?;
let database = PostgresDatabase::lazy(&config.storage.postgres_url)?;
let authority = Arc::new(GcsBucket::new(&config.storage.bucket).await?);
let registry = Arc::new(super::PostgresAdminRegistry::from_database(
database.clone(),
));
let (fleet, access) = super::replication::fleet(
registry.clone(),
&config.sandbox_provider,
&config.jwt_signing_key,
config.storage.replica_regions,
database.clone(),
stop.clone(),
)?;
let storage = Arc::new(RuntimeStorage::new(
authority,
fleet.clone(),
Arc::new(GrpcReplicaPeers::new(access.clone())?),
access.clone(),
config.sandbox_provider.runtime.control_plane_url.clone(),
std::sync::Arc::new(crate::clock::SystemClock),
)?);
let runtime_access = Arc::new(crate::bucket::access::RuntimeAccess::new(
crate::bucket::access::BucketLocation::Gcs {
bucket: config.storage.bucket.clone(),
},
fleet.clone(),
access,
storage.clone(),
)?);
let placements = storage.clone();
fleet.start(storage.clone(), stop.clone());
let provisioner = sandbox_provisioner(
config.sandbox_provider,
&issuer,
runtime_access.clone(),
database,
registry.clone(),
stop,
)?;
let socket_events = config
.socket_event_sink
.map(|sink| {
super::event_sink::HttpSocketMessageEventSink::new(sink.url, config.api_key.clone())
})
.transpose()?
.map(|sink| Arc::new(sink) as Arc<dyn super::event_sink::SocketMessageEventSink>);
let mut service = ControlPlaneService::new(
placements.clone(),
auth,
registry.clone(),
issuer.clone(),
provisioner,
)
.with_runtime_access(runtime_access)
.with_socket_event_sink(socket_events);
service.region = config.region;
let admin = super::admin::AdminService::new(config.api_key, registry, issuer)?;
let inspector =
super::inspection::ActorInspector::new(storage.clone(), service.changes.clone())
.with_traces(service.traces.clone());
let public_api = super::public_api::router(service.clone(), admin.clone())
.merge(super::inspection::router(inspector, admin))
.merge(storage.router());
let internal_api = service.into_internal_service();
Ok(tonic::service::Routes::from(public_api).add_service(internal_api))
}
fn sandbox_provisioner(
config: SandboxProviderConfig,
issuer: &super::ActorJwtIssuer,
access: Arc<crate::bucket::access::RuntimeAccess>,
database: PostgresDatabase,
registry: Arc<dyn super::admin::AdminRegistry>,
stop: tokio_util::sync::CancellationToken,
) -> Result<Arc<dyn super::service::HostProvisioner>> {
let provider = Arc::new(CommandSandboxProvider::new(
config.provider_name,
config.command,
config.environment,
)?);
let pool = crate::sandbox::pool::SparePool::new(database, provider.clone(), config.pool);
pool.start(registry, stop);
Ok(Arc::new(
super::service::SandboxHostProvisioner::new(
provider,
config.runtime,
issuer.clone(),
Some(config.runtime_image),
)
.with_runtime_access(access)
.with_pool(pool),
))
}
impl ControlPlaneProcessConfig {
fn from_lookup(mut get: impl FnMut(&str) -> Option<String>) -> Result<Self> {
let bind = get("DURABLE_ACTORS_CONTROL_PLANE_BIND")
.unwrap_or_else(|| "127.0.0.1:7100".into())
.parse()
.context("DURABLE_ACTORS_CONTROL_PLANE_BIND must be a socket address")?;
let jwt_signing_key = required(&mut get, "DURABLE_ACTORS_JWT_SIGNING_KEY")?;
let jwt_key_id = get("DURABLE_ACTORS_JWT_KEY_ID").unwrap_or_else(|| "primary".into());
let jwt_issuer =
get("DURABLE_ACTORS_JWT_ISSUER").unwrap_or_else(|| DEFAULT_JWT_ISSUER.into());
let authority_audience = get("DURABLE_ACTORS_AUTHORITY_JWT_AUDIENCE")
.unwrap_or_else(|| DEFAULT_AUTHORITY_AUDIENCE.into());
let invocation_audience = get("DURABLE_ACTORS_INVOKE_JWT_AUDIENCE")
.unwrap_or_else(|| DEFAULT_INVOCATION_AUDIENCE.into());
let jwt_max_lifetime = Duration::from_secs(
get("DURABLE_ACTORS_JWT_MAX_TTL_SECONDS")
.map(|value| value.parse())
.transpose()
.context("DURABLE_ACTORS_JWT_MAX_TTL_SECONDS must be an integer")?
.unwrap_or(DEFAULT_JWT_TTL_SECONDS),
);
ensure!(
!jwt_max_lifetime.is_zero(),
"DURABLE_ACTORS_JWT_MAX_TTL_SECONDS must be positive"
);
let api_key = get("DURABLE_ACTORS_SECRET");
if let Some(api_key) = &api_key {
ensure!(
!api_key.is_empty(),
"DURABLE_ACTORS_SECRET must not be empty"
);
ensure!(
api_key.trim() == api_key,
"DURABLE_ACTORS_SECRET has surrounding whitespace"
);
}
let bucket = required(&mut get, "DURABLE_ACTORS_BUCKET")?;
crate::storage::validate_bucket(&bucket)?;
let replica_regions = crate::replication::replica_regions(&mut get)?;
let region = get("DURABLE_ACTORS_REGION");
if let Some(region) = ®ion {
crate::placement::validate_region(region)?;
}
let storage = ControlPlaneStorageConfig {
replica_regions,
postgres_url: required(&mut get, "DURABLE_ACTORS_POSTGRES_URL")?,
bucket,
};
let sandbox_provider =
sandbox_provider_config(&mut get, &jwt_issuer, &invocation_audience)?;
let socket_event_sink = socket_event_sink_config(&mut get)?;
Ok(Self {
bind,
jwt_signing_key,
jwt_key_id,
jwt_issuer,
authority_audience,
invocation_audience,
jwt_max_lifetime,
api_key,
storage,
sandbox_provider,
socket_event_sink,
region,
})
}
}
fn socket_event_sink_config(
get: &mut impl FnMut(&str) -> Option<String>,
) -> Result<Option<SocketEventSinkConfig>> {
get("DURABLE_ACTORS_SOCKET_EVENT_URL")
.map(|url| {
Ok(SocketEventSinkConfig {
url: validated_http_url(&url, "DURABLE_ACTORS_SOCKET_EVENT_URL")?,
})
})
.transpose()
}
fn sandbox_provider_config(
get: &mut impl FnMut(&str) -> Option<String>,
jwt_issuer: &str,
invocation_audience: &str,
) -> Result<SandboxProviderConfig> {
let provider_name = required(get, "DURABLE_ACTORS_SANDBOX_PROVIDER")?;
ensure!(
provider_name == "modal",
"unsupported sandbox provider {provider_name:?}"
);
let mut environment = HashMap::from([
(
"MODAL_TOKEN_ID".into(),
provider_credential(get, "MODAL_TOKEN_ID")?,
),
(
"MODAL_TOKEN_SECRET".into(),
provider_credential(get, "MODAL_TOKEN_SECRET")?,
),
]);
if let Some(value) = get("DURABLE_ACTORS_MODAL_MUTABLE_NETWORK") {
let enabled: bool = value
.parse()
.context("DURABLE_ACTORS_MODAL_MUTABLE_NETWORK must be true or false")?;
environment.insert(
"DURABLE_ACTORS_MODAL_MUTABLE_NETWORK".into(),
enabled.to_string(),
);
}
let control_plane_url = validated_http_url(
&required(get, "DURABLE_ACTORS_CONTROL_PLANE_URL")?,
"DURABLE_ACTORS_CONTROL_PLANE_URL",
)?;
let idle = pool_number(get, "DURABLE_ACTORS_SPARE_IDLE", 5, 0, 32)?;
let regions = get("DURABLE_ACTORS_SPARE_REGIONS")
.unwrap_or_else(|| "north-america-east".into())
.split(',')
.map(|region| region.trim().to_owned())
.collect::<Vec<_>>();
for region in ®ions {
super::regions::storage_region(region)?;
}
Ok(SandboxProviderConfig {
runtime_image: {
let image = required(get, "DURABLE_ACTORS_RUNTIME_IMAGE")?;
ensure!(
image.starts_with("im-") && image.len() > 3 && image.len() <= 255,
"DURABLE_ACTORS_RUNTIME_IMAGE must be a published Modal runtime image ID"
);
image
},
pool: crate::sandbox::pool::PoolConfig {
kind: crate::sandbox::SpareKind::Actor,
idle,
idle_ttl_seconds: pool_number(get, "DURABLE_ACTORS_SPARE_TTL_SECONDS", 600, 30, 3600)?,
regions,
resources: crate::sandbox::ResourceLimits {
cpu_millis: pool_number(get, "DURABLE_ACTORS_HOST_CPU_MILLIS", 1000, 100, 64000)?,
memory_mib: pool_number(get, "DURABLE_ACTORS_HOST_MEMORY_MIB", 1024, 128, 262144)?,
},
},
provider_name,
command: get("DURABLE_ACTORS_SANDBOX_COMMAND")
.unwrap_or_else(|| "durable-actors-modal-go".into()),
environment,
runtime: HostSandboxRuntimeConfig {
control_plane_url,
jwt_issuer: jwt_issuer.into(),
invocation_jwt_audience: invocation_audience.into(),
actor_idle_timeout_seconds: actor_idle_timeout_seconds(get)?,
host_idle_timeout_ms: idle_timeout(
get,
"DURABLE_ACTORS_HOST_IDLE_TIMEOUT_MS",
DEFAULT_HOST_IDLE_TIMEOUT_MS,
MAX_IDLE_TIMEOUT_MS,
)?,
},
})
}
fn pool_number(
get: &mut impl FnMut(&str) -> Option<String>,
name: &str,
default: u32,
min: u32,
max: u32,
) -> Result<u32> {
let value = get(name)
.map(|value| value.parse::<u32>())
.transpose()
.with_context(|| format!("invalid {name}"))?
.unwrap_or(default);
ensure!(
(min..=max).contains(&value),
"{name} must be between {min} and {max}"
);
Ok(value)
}
fn provider_credential(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
let value = required(get, name)?;
ensure!(value.trim() == value, "{name} has surrounding whitespace");
Ok(value)
}
fn required(get: &mut impl FnMut(&str) -> Option<String>, name: &str) -> Result<String> {
let value = get(name).with_context(|| format!("{name} is required"))?;
ensure!(!value.is_empty(), "{name} must not be empty");
Ok(value)
}
fn validated_http_url(value: &str, name: &str) -> Result<String> {
let url = reqwest::Url::parse(value).with_context(|| format!("{name} must be a URL"))?;
ensure!(
matches!(url.scheme(), "http" | "https") && url.host_str().is_some(),
"{name} must be HTTP or HTTPS"
);
Ok(url.to_string())
}
pub(crate) fn actor_idle_timeout_seconds(
get: &mut impl FnMut(&str) -> Option<String>,
) -> Result<u64> {
idle_timeout(
get,
"DURABLE_ACTORS_ACTOR_IDLE_TIMEOUT_SECONDS",
DEFAULT_ACTOR_IDLE_TIMEOUT_SECONDS,
86_400,
)
}
fn idle_timeout(
get: &mut impl FnMut(&str) -> Option<String>,
name: &str,
default: u64,
maximum: u64,
) -> Result<u64> {
let value = get(name)
.map(|value| value.parse())
.transpose()
.with_context(|| format!("{name} must be an integer"))?
.unwrap_or(default);
ensure!(
(1..=maximum).contains(&value),
"{name} must be an integer between 1 and {maximum}"
);
Ok(value)
}
#[cfg(test)]
#[path = "../../tests/unit/control_plane/process.rs"]
mod tests;