lfg2025/archy
1
0
Fork 0
Code Issues 25 Pull Requests 4 Actions Packages Projects Releases 74 Wiki Activity

feat(container): ContainerOrchestrator trait, RpcHandler uses it in prod

Browse Source 
Step 4 of the rust-orchestrator migration. Unifies the container lifecycle
surface behind a single trait so the RPC layer stops caring whether it is
talking to the dev or prod orchestrator.

  * New trait core/archipelago/src/container/traits.rs: ContainerOrchestrator
    with install / start / stop / restart / remove / upgrade / status / list /
    logs / health, all keyed by app_id. Every method is async_trait-based.

  * ProdContainerOrchestrator: the lifecycle methods are moved from inherent
    impl into the trait impl (avoids name-shadowing recursion). Adoption and
    reconcile remain inherent since only main.rs / BootReconciler call them.

  * DevContainerOrchestrator: new trait impl that forwards to the existing
    Dev-named methods, applying the dev container-name + port-offset rules
    internally. New load_manifest_for() helper resolves app_id to
    <data_dir>/apps/<app_id>/manifest.yml so trait-level install(app_id)
    works in dev too. install_container(manifest, path) stays inherent for
    the manifest-path RPC shape.

  * RpcHandler now holds Option<Arc<dyn ContainerOrchestrator>> and, when in
    dev mode, a separate Option<Arc<DevContainerOrchestrator>> for the
    manifest_path install RPC. In prod mode RpcHandler::new() constructs a
    ProdContainerOrchestrator and calls load_manifests() at startup.

  * All seven container-* RPC guards no longer say dev mode required.
    container-install still requires dev mode because its manifest_path
    argument has no prod meaning; every other container RPC now works in both
    modes via the trait.

BOOT STILL DOES NOT USE THIS. main.rs wire-up (Step 6) and BootReconciler
(Step 5) come next. Until then the prod orchestrator is constructed but nothing
populates /opt/archipelago/apps so it has zero manifests to manage, matching
the pre-Step-4 behaviour.

Verification: cargo build -p archipelago clean (11 expected unused method
warnings for methods not yet wired from main.rs). cargo test -p archipelago:
all 21 container::* tests pass (16 prod_orchestrator + 5 others). 24 other
test failures are pre-existing and unrelated (identity_manager / session /
wallet / mesh / credentials — all independently flaky on file-backed state).
This commit is contained in:
archipelago
2026-04-22 18:56:52 -04:00
parent e103925a4e
commit 40a6eaca72
7 changed files with 262 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/rpc/container.rs
+70 -50
View File
@@ -7,8 +7,13 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
// The `container-install { manifest_path }` RPC is a dev-mode convenience
// that points at an arbitrary YAML on disk. Production install happens via
// the reconciler (BootReconciler, Step 5) and via the unified
// ContainerOrchestrator::install(app_id) trait call, which can be exposed
// through a separate `container-install-by-id` RPC when needed.
let dev = self.dev_orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("container-install with manifest_path is only available in dev mode")
})?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
@@ -45,7 +50,7 @@ impl RpcHandler {
let manifest: archipelago_container::AppManifest =
serde_yaml::from_str(&manifest_content).context("Failed to parse manifest")?;
let container_name = orchestrator
let container_name = dev
.install_container(&manifest, manifest_path)
.await
.context("Failed to install container")?;
@@ -57,9 +62,10 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
let app_id = params
@@ -69,7 +75,7 @@ impl RpcHandler {
validate_app_id(app_id)?;
orchestrator
.start_container(app_id)
.start(app_id)
.await
.context("Failed to start container")?;
@@ -80,9 +86,10 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
let app_id = params
@@ -92,7 +99,7 @@ impl RpcHandler {
validate_app_id(app_id)?;
orchestrator
.stop_container(app_id)
.stop(app_id)
.await
.context("Failed to stop container")?;
@@ -103,9 +110,10 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
let app_id = params
@@ -119,7 +127,7 @@ impl RpcHandler {
.unwrap_or(false);
orchestrator
.remove_container(app_id, preserve_data)
.remove(app_id, preserve_data)
.await
.context("Failed to remove container")?;
@@ -163,9 +171,9 @@ impl RpcHandler {
return Ok(serde_json::json!(containers));
}
// Fallback: scanner hasn't run yet, query podman directly
// Fallback: scanner hasn't run yet, query the orchestrator directly.
if let Some(orchestrator) = &self.orchestrator {
if let Ok(containers) = orchestrator.list_containers().await {
if let Ok(containers) = orchestrator.list().await {
if !containers.is_empty() {
return Ok(serde_json::to_value(containers)?);
}
@@ -242,9 +250,10 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
let app_id = params
@@ -254,7 +263,7 @@ impl RpcHandler {
validate_app_id(app_id)?;
let status = orchestrator
.get_container_status(app_id)
.status(app_id)
.await
.context("Failed to get container status")?;
@@ -265,9 +274,10 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let params = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
let app_id = params
@@ -278,7 +288,7 @@ impl RpcHandler {
let lines = params.get("lines").and_then(|v| v.as_u64()).unwrap_or(100) as u32;
let logs = orchestrator
.get_container_logs(app_id, lines)
.logs(app_id, lines)
.await
.context("Failed to get container logs")?;
@@ -291,12 +301,13 @@ impl RpcHandler {
app_id: &str,
lines: u32,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
let logs = orchestrator
.get_container_logs(app_id, lines)
.logs(app_id, lines)
.await
.context("Failed to get container logs")?;
@@ -307,43 +318,52 @@ impl RpcHandler {
&self,
params: Option<serde_json::Value>,
) -> Result<serde_json::Value> {
let orchestrator = self.orchestrator.as_ref().ok_or_else(|| {
anyhow::anyhow!("Container orchestrator not available (dev mode required)")
})?;
let orchestrator = self
.orchestrator
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Container orchestrator not available"))?;
// If app_id is provided, get health for that app
// If app_id is provided, get health for that app.
if let Some(params) = params {
if let Some(app_id) = params.get("app_id").and_then(|v| v.as_str()) {
let health = orchestrator
.get_health_status(app_id)
.health(app_id)
.await
.context("Failed to get container health")?;
return Ok(serde_json::json!({ app_id: health }));
}
}
// Otherwise, get health for all containers
// Otherwise, get health for all containers.
let containers = orchestrator
.list_containers()
.list()
.await
.context("Failed to list containers")?;
let mut health_map = serde_json::Map::new();
for container in containers {
if let Some(app_id) = container.name.strip_prefix("archipelago-") {
if let Some(app_id) = app_id.strip_suffix("-dev") {
match orchestrator.get_health_status(app_id).await {
Ok(health) => {
health_map
.insert(app_id.to_string(), serde_json::Value::String(health));
}
Err(_) => {
health_map.insert(
app_id.to_string(),
serde_json::Value::String("unknown".to_string()),
);
}
}
// Map the runtime container name back to the app_id the orchestrator
// knows about. Dev orchestrator uses `archipelago-<id>-dev`; Prod
// uses bare `<id>` (or `archy-<id>` for UIs — health() accepts the
// app_id either way since UI_APP_IDS is centralised).
let app_id_candidate = container
.name
.strip_prefix("archipelago-")
.and_then(|s| s.strip_suffix("-dev"))
.or_else(|| container.name.strip_prefix("archy-"))
.unwrap_or(container.name.as_str());
match orchestrator.health(app_id_candidate).await {
Ok(health) => {
health_map.insert(
app_id_candidate.to_string(),
serde_json::Value::String(health),
);
}
Err(_) => {
health_map.insert(
app_id_candidate.to_string(),
serde_json::Value::String("unknown".to_string()),
);
}
}
}
core/archipelago/src/api/rpc/mod.rs
+25 -7
View File
@@ -39,7 +39,7 @@ mod webhooks;
use crate::auth::AuthManager;
use crate::config::Config;
use crate::container::DevContainerOrchestrator;
use crate::container::{ContainerOrchestrator, DevContainerOrchestrator, ProdContainerOrchestrator};
use crate::monitoring::MetricsStore;
use crate::port_allocator::PortAllocator;
use crate::rate_limit::{EndpointRateLimiter, LoginRateLimiter};
@@ -62,7 +62,14 @@ pub(crate) const DEV_DEFAULT_PASSWORD: &str = "password123";
pub struct RpcHandler {
config: Config,
auth_manager: AuthManager,
orchestrator: Option<Arc<DevContainerOrchestrator>>,
/// Shared lifecycle orchestrator (Dev or Prod). Always `Some` in a normal
/// build — the only reason it is `Option` is so tests that don't exercise
/// container RPCs can skip constructing one.
orchestrator: Option<Arc<dyn ContainerOrchestrator>>,
/// Concrete handle to the dev orchestrator, when we're in dev mode. Used by
/// `container-install { manifest_path }` which takes an ad-hoc manifest
/// path and is not part of the shared trait.
dev_orchestrator: Option<Arc<DevContainerOrchestrator>>,
state_manager: Arc<StateManager>,
pub(crate) metrics_store: Arc<MetricsStore>,
port_allocator: Arc<tokio::sync::Mutex<PortAllocator>>,
@@ -89,12 +96,22 @@ impl RpcHandler {
session_store: SessionStore,
) -> Result<Self> {
let auth_manager = AuthManager::new(config.data_dir.clone());
let orchestrator = if config.dev_mode {
Some(Arc::new(
DevContainerOrchestrator::new(config.clone()).await?,
))
let (orchestrator, dev_orchestrator): (
Option<Arc<dyn ContainerOrchestrator>>,
Option<Arc<DevContainerOrchestrator>>,
) = if config.dev_mode {
let dev = Arc::new(DevContainerOrchestrator::new(config.clone()).await?);
let trait_obj: Arc<dyn ContainerOrchestrator> = dev.clone();
(Some(trait_obj), Some(dev))
} else {
None
let prod = Arc::new(ProdContainerOrchestrator::new(config.clone()).await?);
// Best-effort manifest load; a missing /opt/archipelago/apps is
// logged by load_manifests and not fatal.
if let Err(e) = prod.load_manifests().await {
tracing::error!(error = %e, "prod orchestrator: load_manifests failed at startup");
}
let trait_obj: Arc<dyn ContainerOrchestrator> = prod;
(Some(trait_obj), None)
};
let port_allocator = Arc::new(tokio::sync::Mutex::new(
PortAllocator::new(&config.data_dir).await?,
@@ -129,6 +146,7 @@ impl RpcHandler {
config,
auth_manager,
orchestrator,
dev_orchestrator,
state_manager,
metrics_store,
port_allocator,