From 37e1c07791cc30eeb747602c5b2fdbb02f854104 Mon Sep 17 00:00:00 2001 From: Brett Chien Date: Mon, 7 Sep 2026 18:05:01 +0800 Subject: [PATCH] feat(studio-cp,oab-mcp): k8s roster observe + runtime_context dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit studio#146 slice 2 of 4. Builds the k8s counterpart to the ECS read-model (observe_services/observe_deployment/build_deployment), producing the same runtime-agnostic Deployment/InstancePhase shape via slice 1's K8sDriver instead of EcsDriver: - studio-cp: observe_k8s_services (list OAB-managed Deployments in a namespace, via the `oab/name` label) and observe_k8s_deployment (one Deployment's replica counters + per-Pod phase, mapping a live Pod's phase/readiness/CrashLoopBackOff onto the canonical 6-state). - oab-mcp: deploy_list/deploy_get/runtime_context now dispatch on a named fleet's runtime — k8s calls the new studio-cp functions (and the already-existing observe_k8s_identity for runtime_context), ecs is unchanged. `target()` itself and every other tool (scale/delete/apply/ provision/events) stay ecs-only on purpose — this only touches the three calls that are actually runtime-aware now. Fleet-membership matching note: a k8s Deployment's own on-cluster resource name is `oab-{slug(name)}` (k8s_driver.rs, no namespace embedded), but fleets.toml's `members` stores `oab-{namespace}-{name}` (deploy.ts writes this uniformly for both runtimes) — observe_k8s_services/ observe_k8s_deployment reconstruct the latter from the Deployment's `oab/name` label so FleetBinding::includes matches correctly. 🤖 Generated with Claude Code --- crates/oab-mcp/src/lib.rs | 88 +++++++++++++-- crates/studio-cp/src/lib.rs | 210 ++++++++++++++++++++++++++++++++++++ 2 files changed, 289 insertions(+), 9 deletions(-) diff --git a/crates/oab-mcp/src/lib.rs b/crates/oab-mcp/src/lib.rs index 3015965..132ff02 100644 --- a/crates/oab-mcp/src/lib.rs +++ b/crates/oab-mcp/src/lib.rs @@ -66,24 +66,24 @@ pub fn tools() -> Vec { vec![ Tool::new( "deploy_list", - "List all OAB deployments (ECS services) in the cluster with replica counts and status.", + "List all OAB deployments in the cluster (ECS) or namespace (k8s) with replica counts. For a k8s-runtime fleet, pass `fleet` — the fleet's own context/namespace govern the call, `cluster` is ignored.", as_map(json!({ "type": "object", "properties": { - "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg." }, - "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster)." } + "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster/context and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg. Required for a k8s-runtime fleet." }, + "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster). Ignored for a k8s-runtime fleet." } } })), ), Tool::new( "deploy_get", - "Get one deployment's read-model: replica counters plus each instance's canonical lifecycle phase (6-state).", + "Get one deployment's read-model: replica counters plus each instance's canonical lifecycle phase (6-state). Works for both ECS and k8s-runtime fleets (pass `fleet` for k8s).", as_map(json!({ "type": "object", "properties": { - "service": { "type": "string", "description": "ECS service name (or bare agent name)." }, - "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg." }, - "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster)." } + "service": { "type": "string", "description": "ECS service name, k8s reconstructed oab-{namespace}-{name}, or bare agent name." }, + "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster/context and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg. Required for a k8s-runtime fleet." }, + "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster). Ignored for a k8s-runtime fleet." } }, "required": ["service"] })), @@ -207,7 +207,7 @@ pub fn tools() -> Vec { ), Tool::new( "runtime_context", - "Show the effective runtime identity/context this control plane resolved for a cluster/fleet: the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), a best-effort credential-source hint, the fleet binding in effect (if any), and — when the binding declares an expected_principal — whether the resolved identity matches it (identity_matches; a non-blocking IdentityMismatch when false). Read-only; answers \"who am I acting as, against what account?\" and surfaces silent credential fallback.", + "Show the effective runtime identity/context this control plane resolved for a cluster/fleet: for ECS, the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), a best-effort credential-source hint; for a k8s-runtime fleet, the SelfSubjectReview-resolved principal against the fleet's kubeconfig context (cluster is null, context/namespace are set instead). Also returns the fleet binding in effect (if any) and — when the binding declares an expected_principal — whether the resolved identity matches it (identity_matches; a non-blocking IdentityMismatch when false). Read-only; answers \"who am I acting as, against what account?\" and surfaces silent credential fallback.", as_map(json!({ "type": "object", "properties": { @@ -469,6 +469,27 @@ impl OabMcp { }) } + /// Like [`Self::target`], but doesn't reject a k8s-runtime fleet — for + /// the calls this made runtime-aware (studio#146): roster list/get and + /// `runtime_context`. Returns the named fleet's binding, or `None` for + /// an unscoped call (back-compat, same as `target()`'s bare-cluster + /// path). An unknown fleet name is still an error, same as `target()`. + /// `target()` itself stays ecs-only — every other call (scale/delete/ + /// apply/provision/events) genuinely has no k8s-runtime dispatch yet, so + /// it should keep failing loudly rather than silently acting against an + /// empty cluster string. + fn named_fleet(&self, args: &Map) -> Result> { + let Some(name) = args.get("fleet").and_then(Value::as_str) else { + return Ok(None); + }; + let guard = self.bindings.read().unwrap(); + let binding = guard.get(name).cloned().ok_or_else(|| { + let known: Vec<&str> = guard.fleets.iter().map(|f| f.name.as_str()).collect(); + anyhow::anyhow!("unknown fleet {name:?}; configured fleets: [{}]", known.join(", ")) + })?; + Ok(Some(binding)) + } + /// The AWS config to act as for `cluster`: the fleet binding's credential /// when one governs it (resolved once, then memoized), else the default /// chain. This is where the per-fleet **switch** takes effect — a bound @@ -496,6 +517,18 @@ impl OabMcp { } async fn t_list(&self, args: &Map) -> Result { + if let Some(b) = self.named_fleet(args)? { + if b.runtime == scp::FleetRuntime::K8s { + let namespace = b.namespace.clone().unwrap_or_else(|| "default".to_string()); + let svcs = scp::observe_k8s_services(b.context.as_deref(), &namespace).await?; + let deployments: Vec = svcs + .iter() + .filter(|s| b.includes(&s.service_name, &s.name)) + .map(|s| json!({ "name": s.name, "namespace": s.namespace, "desired": s.desired })) + .collect(); + return Ok(json!({ "context": b.context, "namespace": namespace, "deployments": deployments })); + } + } let t = self.target(args)?; let cluster = t.cluster.clone(); let svcs = scp::observe_services(&self.aws_for(&cluster).await, &cluster).await?; @@ -519,11 +552,20 @@ impl OabMcp { } async fn t_get(&self, args: &Map) -> Result { - let cluster = self.target(args)?.cluster; let service = args .get("service") .and_then(Value::as_str) .ok_or_else(|| anyhow::anyhow!("missing required arg: service"))?; + if let Some(b) = self.named_fleet(args)? { + if b.runtime == scp::FleetRuntime::K8s { + let namespace = b.namespace.clone().unwrap_or_else(|| "default".to_string()); + return match scp::observe_k8s_deployment(b.context.as_deref(), &namespace, service).await? { + Some(d) => Ok(deployment_json(&d)), + None => Ok(json!({ "found": false, "service": service })), + }; + } + } + let cluster = self.target(args)?.cluster; match scp::observe_deployment(&self.aws_for(&cluster).await, &cluster, service).await? { Some(d) => Ok(deployment_json(&d)), None => Ok(json!({ "found": false, "service": service })), @@ -848,6 +890,34 @@ impl OabMcp { } async fn t_runtime_context(&self, args: &Map) -> Result { + if let Some(b) = self.named_fleet(args)? { + if b.runtime == scp::FleetRuntime::K8s { + let ctx = scp::observe_k8s_identity(b.context.as_deref()).await?; + let expected = b.expected_principal.clone(); + let identity_matches = expected + .as_ref() + .map(|e| scp::principal_matches(e, &ctx.principal)); + return Ok(json!({ + "cluster": null, + "context": b.context, + "namespace": b.namespace, + "principal": ctx.principal, + "principal_kind": ctx.principal_kind, + "scope": ctx.scope, + "location": ctx.location, + "source": ctx.source, + "caller_id": ctx.caller_id, + "binding": json!({ + "name": b.name, + "context": b.context, + "namespace": b.namespace, + "expected_principal": b.expected_principal, + }), + "expected_principal": expected, + "identity_matches": identity_matches, + })); + } + } let t = self.target(args)?; let cluster = t.cluster.clone(); let aws = self.aws_for(&cluster).await; diff --git a/crates/studio-cp/src/lib.rs b/crates/studio-cp/src/lib.rs index 4423b93..27be58a 100644 --- a/crates/studio-cp/src/lib.rs +++ b/crates/studio-cp/src/lib.rs @@ -170,6 +170,216 @@ pub async fn observe_events( oabctl::fetch_ecs_events(aws_config, log_group, Some(cluster), service, since_ms, limit).await } +// ---- k8s observe (studio#146) -------------------------------------------- +// +// The k8s counterpart to the ECS read-model above (`observe_services` / +// `observe_deployment` / `build_deployment`), producing the exact same +// runtime-agnostic `Deployment`/`InstancePhase` types via +// `agent_lifecycle::k8s::K8sDriver` instead of `EcsDriver`. This was +// deliberately deferred when the k8s `ProvisionDriver` write path landed +// (`oabctl::k8s_driver`'s module doc, studio#63 slice 3b) — this is that +// follow-up (studio#146). + +/// One k8s Deployment discovered while listing a namespace — the k8s +/// counterpart to ECS's [`oabctl::ServiceStatus`] enumeration step (only the +/// fields fleet-membership filtering and the follow-up per-Deployment fetch +/// need; full per-Pod detail comes from [`observe_k8s_deployment`], mirroring +/// the ECS `deploy_list` → `deploy_get` two-step the console roster already +/// does). +pub struct K8sServiceStatus { + /// The agent's original (un-slugified) name — read back from the + /// Deployment's `oab/name` label (`k8s_driver.rs::build_deployment` + /// sets it verbatim, never the k8s-safe slug `k8s_deployment_name` uses + /// for the resource's own name). + pub name: String, + pub namespace: String, + /// Reconstructed `oab-{namespace}-{name}` — **not** the on-cluster + /// Deployment resource name (`oab-{slug(name)}`, no namespace embedded; + /// see `k8s_driver.rs`'s module doc on why the two differ). This is the + /// form the New Fleet / Add Instance wizard actually writes into + /// `fleets.toml`'s `members` array (`console/src/deploy.ts`'s `service = + /// oab-${namespace}-${name}`, unconditional for both runtimes) — carried + /// so `FleetBinding::includes` matches the same way it does for ECS. + pub service_name: String, + pub desired: i32, +} + +fn k8s_oab_name(dep: &k8s_openapi::api::apps::v1::Deployment) -> Option { + dep.metadata.labels.as_ref()?.get("oab/name").cloned() +} + +/// List every OAB-managed Deployment (`oab/name` label present) in one +/// `(context, namespace)` — the k8s counterpart to [`observe_services`]. +pub async fn observe_k8s_services( + context: Option<&str>, + namespace: &str, +) -> anyhow::Result> { + use k8s_openapi::api::apps::v1::Deployment as K8sDeployment; + use kube::api::{Api, ListParams}; + + let client = k8s_client_for(context).await?; + let api: Api = Api::namespaced(client, namespace); + let list = api + .list(&ListParams::default().labels("oab/name")) + .await + .map_err(|e| anyhow::anyhow!("failed to list k8s deployments in '{namespace}': {e}"))?; + Ok(list + .items + .iter() + .filter_map(|dep| { + let name = k8s_oab_name(dep)?; + let desired = dep.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0); + Some(K8sServiceStatus { + service_name: format!("oab-{namespace}-{name}"), + name, + namespace: namespace.to_string(), + desired, + }) + }) + .collect()) +} + +fn k8s_pod_phase(pod: &k8s_openapi::api::core::v1::Pod) -> agent_lifecycle::k8s::PodPhase { + use agent_lifecycle::k8s::PodPhase; + match pod.status.as_ref().and_then(|s| s.phase.as_deref()) { + Some("Running") => PodPhase::Running, + Some("Succeeded") => PodPhase::Succeeded, + Some("Failed") => PodPhase::Failed, + Some("Unknown") => PodPhase::Unknown, + // "Pending" or not yet reported — not started, not a fault. + _ => PodPhase::Pending, + } +} + +fn k8s_pod_ready(pod: &k8s_openapi::api::core::v1::Pod) -> bool { + pod.status + .as_ref() + .and_then(|s| s.conditions.as_ref()) + .into_iter() + .flatten() + .any(|c| c.type_ == "Ready" && c.status == "True") +} + +fn k8s_pod_ready_check_defined(pod: &k8s_openapi::api::core::v1::Pod) -> bool { + pod.spec + .as_ref() + .into_iter() + .flat_map(|s| s.containers.iter()) + .any(|c| c.readiness_probe.is_some()) +} + +fn k8s_pod_crash_loop_back_off(pod: &k8s_openapi::api::core::v1::Pod) -> bool { + pod.status + .as_ref() + .and_then(|s| s.container_statuses.as_ref()) + .into_iter() + .flatten() + .any(|cs| { + cs.state + .as_ref() + .and_then(|st| st.waiting.as_ref()) + .is_some_and(|w| w.reason.as_deref() == Some("CrashLoopBackOff")) + }) +} + +/// One-shot approximation of the `identity_verified` latch (same caveat as +/// ECS's `latched_verified`: the real latch needs CP-persisted history) — a +/// Pod counts as verified once it's reached Running, or is already +/// Terminating (it must have run to get there). +fn k8s_latched_verified(pod: &k8s_openapi::api::core::v1::Pod) -> bool { + matches!(k8s_pod_phase(pod), agent_lifecycle::k8s::PodPhase::Running) + || pod.metadata.deletion_timestamp.is_some() +} + +/// Map a live k8s Pod onto the canonical [`AgentState`], via +/// [`agent_lifecycle::k8s::K8sDriver`] — the k8s counterpart to +/// `instance_phase`'s ECS derivation. `lease_valid`/`accepting_work` are +/// CP-level, not yet k8s-observable, so they default to valid/admitting — +/// same stance `instance_phase` takes for ECS today. +pub fn k8s_instance_phase(pod: &k8s_openapi::api::core::v1::Pod, verified_before: bool) -> AgentState { + use agent_lifecycle::k8s::{K8sDriver, K8sPod}; + use agent_lifecycle::RuntimeDriver; + + let native = K8sPod { + phase: k8s_pod_phase(pod), + deletion_timestamp_set: pod.metadata.deletion_timestamp.is_some(), + ready: k8s_pod_ready(pod), + ready_check_defined: k8s_pod_ready_check_defined(pod), + crash_loop_back_off: k8s_pod_crash_loop_back_off(pod), + lease_valid: true, + accepting_work: true, + }; + K8sDriver.project(&native, verified_before).classify() +} + +/// Observe one k8s Deployment end-to-end: replica counters + per-Pod phase — +/// the k8s counterpart to [`observe_deployment`]. `service` matches either +/// the reconstructed `oab-{namespace}-{name}` form or the bare agent name +/// (same dual-match spirit as ECS's `resolve_service`). +pub async fn observe_k8s_deployment( + context: Option<&str>, + namespace: &str, + service: &str, +) -> anyhow::Result> { + use k8s_openapi::api::apps::v1::Deployment as K8sDeployment; + use k8s_openapi::api::core::v1::Pod; + use kube::api::{Api, ListParams}; + + let client = k8s_client_for(context).await?; + let dep_api: Api = Api::namespaced(client.clone(), namespace); + let deployments = dep_api + .list(&ListParams::default().labels("oab/name")) + .await + .map_err(|e| anyhow::anyhow!("failed to list k8s deployments in '{namespace}': {e}"))?; + let Some(dep) = deployments.items.into_iter().find(|d| { + let Some(name) = k8s_oab_name(d) else { return false }; + service == format!("oab-{namespace}-{name}") || service == name + }) else { + return Ok(None); + }; + let name = k8s_oab_name(&dep).unwrap_or_default(); + let desired = dep.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0); + + let selector = dep + .spec + .as_ref() + .and_then(|s| s.selector.match_labels.as_ref()) + .map(|m| { + m.iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(",") + }) + .unwrap_or_default(); + let pod_api: Api = Api::namespaced(client, namespace); + let pods = pod_api + .list(&ListParams::default().labels(&selector)) + .await + .map_err(|e| anyhow::anyhow!("failed to list pods for k8s deployment '{name}': {e}"))?; + + let instances: Vec = pods + .items + .iter() + .map(|p| InstancePhase { + id: p.metadata.uid.clone().unwrap_or_default(), + phase: k8s_instance_phase(p, k8s_latched_verified(p)), + }) + .collect(); + let ready = instances + .iter() + .filter(|p| p.phase == AgentState::Running) + .count() as i32; + + Ok(Some(Deployment { + name, + namespace: namespace.to_string(), + desired, + current: instances.len() as i32, + ready, + instances, + })) +} + // ---- Effective runtime identity/context (ADR: Per-Fleet managing identity) -- // // Read-only observation of *who this control plane is actually acting as*. The