use anyhow::Result;
use bollard::Docker;
use std::collections::HashMap;
use tracing::{debug, info, warn};
use super::*;
pub async fn find_staging(
docker: &Docker,
service_id: &str,
service_name: &str,
dep_short: &str,
idx: u32,
) -> Result<Option<String>> {
let atual = replica_staging_name(service_id, service_name, dep_short, idx);
if let Some(id) = find_owned_by_name(docker, &atual, service_id).await? {
return Ok(Some(id));
}
let legado = legacy_replica_staging_name(service_name, dep_short, idx);
find_owned_by_name(docker, &legado, service_id).await
}
pub(super) fn replica_index(labels: Option<&HashMap<String, String>>, name: &str) -> u32 {
if let Some(i) = labels
.and_then(|l| l.get("rustploy.replica"))
.and_then(|v| v.parse().ok())
{
return i;
}
name.rsplit_once("_r")
.and_then(|(_, n)| n.parse().ok())
.unwrap_or(0)
}
#[derive(Debug, Clone)]
pub struct LiveReplica {
pub id: String,
pub name: String,
pub replica: u32,
pub running: bool,
}
pub async fn find_live_replicas(docker: &Docker, service_id: &str) -> Result<Vec<LiveReplica>> {
use bollard::container::ListContainersOptions;
let mut filters = HashMap::new();
filters.insert("label".to_string(), vec!["rustploy.service_id".to_string()]);
let opts = ListContainersOptions {
all: true,
filters,
..Default::default()
};
let mut out: Vec<LiveReplica> = docker
.list_containers(Some(opts))
.await?
.into_iter()
.filter(|c| {
c.labels
.as_ref()
.and_then(|l| l.get("rustploy.service_id"))
.is_some_and(|o| label_owned_by(o, service_id))
})
.filter_map(|c| {
let name = c
.names
.as_ref()
.and_then(|n| n.first())
.map(|n| n.trim_start_matches('/').to_string())?;
if is_staging_name(&name) {
return None;
}
Some(LiveReplica {
id: c.id?,
replica: replica_index(c.labels.as_ref(), &name),
running: c.state.as_deref() == Some("running"),
name,
})
})
.collect();
out.sort_by_key(|r| (r.replica, !r.running));
debug!(service_id = %service_id, count = out.len(), "réplicas live encontradas");
Ok(out)
}
pub async fn live_replica_ips(
docker: &Docker,
service_id: &str,
network: &str,
replicas: u32,
) -> Vec<String> {
let lives = match find_live_replicas(docker, service_id).await {
Ok(v) => v,
Err(e) => {
warn!(service_id = %service_id, error = %e, "falha ao listar réplicas live");
return vec![];
}
};
let mut ips = Vec::new();
for i in 0..replicas {
for live in lives.iter().filter(|r| r.replica == i) {
if let Ok(ip) = get_container_ip(docker, &live.id, network).await {
ips.push(ip);
break;
}
}
}
ips
}
pub async fn find_all_by_service_id(docker: &Docker, service_id: &str) -> Result<Vec<String>> {
use bollard::container::ListContainersOptions;
debug!(
service_id = %service_id,
"buscando containers por service_id"
);
let mut filters = HashMap::new();
filters.insert(
"label".to_string(),
vec![format!("rustploy.service_id={service_id}")],
);
let opts = ListContainersOptions {
all: true,
filters,
..Default::default()
};
let list = docker.list_containers(Some(opts)).await?;
let ids: Vec<String> = list.into_iter().filter_map(|c| c.id).collect();
debug!(
service_id = %service_id,
count = ids.len(),
"containers encontrados"
);
Ok(ids)
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct ManagedContainer {
pub id: String,
pub name: String,
pub state: String,
}
#[derive(Default)]
pub struct ContainerIndex {
by_service_id: HashMap<String, Vec<ManagedContainer>>,
by_compose_project: HashMap<String, Vec<ManagedContainer>>,
}
impl ContainerIndex {
pub fn for_service(&self, service: &shared::Service) -> Vec<ManagedContainer> {
if let Some(v) = self.by_service_id.get(&service.id) {
return v.clone();
}
self.by_compose_project
.get(&service.compose_project_name())
.cloned()
.unwrap_or_default()
}
}
pub async fn list_compose_projects(
docker: &Docker,
) -> Result<std::collections::BTreeMap<String, std::collections::BTreeSet<String>>> {
use bollard::container::ListContainersOptions;
let list = docker
.list_containers(Some(ListContainersOptions::<String> {
all: true,
..Default::default()
}))
.await?;
let mut out: std::collections::BTreeMap<String, std::collections::BTreeSet<String>> =
Default::default();
for c in list {
let labels = c.labels.unwrap_or_default();
if let Some(project) = labels.get("com.docker.compose.project") {
let keys = out.entry(project.clone()).or_default();
if let Some(svc) = labels.get("com.docker.compose.service") {
keys.insert(svc.clone());
}
}
}
Ok(out)
}
pub async fn index_containers(docker: &Docker) -> ContainerIndex {
use bollard::container::ListContainersOptions;
let opts = ListContainersOptions::<String> {
all: true,
..Default::default()
};
let list = match docker.list_containers(Some(opts)).await {
Ok(l) => l,
Err(e) => {
warn!(
error = %e,
"containers::index_containers: falha ao listar containers"
);
return ContainerIndex::default();
}
};
let mut idx = ContainerIndex::default();
for c in list {
let labels = c.labels.clone().unwrap_or_default();
let name = c
.names
.as_ref()
.and_then(|n| n.first())
.map(|n| n.trim_start_matches('/').to_string())
.unwrap_or_default();
let mc = ManagedContainer {
id: c.id.unwrap_or_default(),
name,
state: c.state.unwrap_or_default(),
};
if let Some(service_id) = labels.get("rustploy.service_id") {
idx.by_service_id
.entry(service_id.clone())
.or_default()
.push(mc);
} else if let Some(project) = labels.get("com.docker.compose.project") {
idx.by_compose_project
.entry(project.clone())
.or_default()
.push(mc);
}
}
idx
}
pub async fn find_old_containers(
docker: &Docker,
service_id: &str,
exclude_deployment_id: &str,
) -> Result<Vec<String>> {
use bollard::container::ListContainersOptions;
debug!(
service_id = %service_id,
"containers::find_old_containers: buscando"
);
let mut filters = HashMap::new();
filters.insert(
"label".to_string(),
vec![format!("rustploy.service_id={service_id}")],
);
let opts = ListContainersOptions {
all: true,
filters,
..Default::default()
};
let list = docker.list_containers(Some(opts)).await?;
let ids: Vec<String> = list
.into_iter()
.filter(|c| {
let dep = c
.labels
.as_ref()
.and_then(|l| l.get("rustploy.deployment_id"))
.map(|s| s.as_str())
.unwrap_or("");
dep != exclude_deployment_id
})
.filter_map(|c| c.id)
.collect();
debug!(
service_id = %service_id,
count = ids.len(),
"containers antigos encontrados"
);
Ok(ids)
}
#[allow(dead_code)]
pub async fn find_by_name(docker: &Docker, name: &str) -> Result<Option<String>> {
use bollard::container::ListContainersOptions;
debug!(name = %name, "buscando container por nome");
let mut filters = HashMap::new();
filters.insert("name".to_string(), vec![format!("^/{name}$")]);
let opts = ListContainersOptions {
all: true,
filters,
..Default::default()
};
let containers = docker.list_containers(Some(opts)).await?;
let found = containers.into_iter().next().and_then(|c| c.id);
match &found {
Some(id) => {
debug!(
name = %name,
container_id = %id,
"container encontrado por nome"
)
}
None => debug!(
name = %name,
"container não encontrado"
),
}
Ok(found)
}
pub async fn name_owner(docker: &Docker, name: &str) -> Result<Option<(String, Option<String>)>> {
use bollard::container::ListContainersOptions;
let mut filters = HashMap::new();
filters.insert("name".to_string(), vec![format!("^/{name}$")]);
let opts = ListContainersOptions {
all: true,
filters,
..Default::default()
};
let found = docker.list_containers(Some(opts)).await?.into_iter().next();
Ok(found.and_then(|c| {
let owner = c
.labels
.as_ref()
.and_then(|l| l.get("rustploy.service_id"))
.cloned();
c.id.map(|id| (id, owner))
}))
}
pub async fn find_owned_by_name(
docker: &Docker,
name: &str,
service_id: &str,
) -> Result<Option<String>> {
match name_owner(docker, name).await? {
Some((id, Some(owner))) if label_owned_by(&owner, service_id) => Ok(Some(id)),
Some((id, owner)) => {
warn!(
name = %name,
container_id = %id,
owner = ?owner,
service_id = %service_id,
"container com o nome esperado pertence a outro serviço (ou a nenhum) — ignorado"
);
Ok(None)
}
None => Ok(None),
}
}
pub async fn list_running_by_prefix(
docker: &Docker,
prefix: &str,
) -> Result<Vec<(String, String)>> {
use bollard::container::ListContainersOptions;
let mut filters = HashMap::new();
filters.insert("name".to_string(), vec![format!("^/{prefix}")]);
let opts = ListContainersOptions {
all: false, filters,
..Default::default()
};
let mut out: Vec<(String, String)> = docker
.list_containers(Some(opts))
.await?
.into_iter()
.filter_map(|c| {
let id = c.id?;
let name = c
.names
.unwrap_or_default()
.into_iter()
.next()?
.trim_start_matches('/')
.to_string();
Some((id, name))
})
.collect();
out.sort_by(|a, b| a.1.cmp(&b.1));
Ok(out)
}
pub(super) async fn exposed_tcp_ports(docker: &Docker, container_id: &str) -> Vec<u16> {
let Ok(info) = inspect(docker, container_id).await else {
return Vec::new();
};
let mut out: Vec<u16> = info
.config
.as_ref()
.and_then(|c| c.exposed_ports.as_ref())
.map(|ports| {
ports
.keys()
.filter_map(|k| {
let (num, proto) = k.split_once('/').unwrap_or((k.as_str(), "tcp"));
(proto == "tcp").then(|| num.parse::<u16>().ok()).flatten()
})
.collect()
})
.unwrap_or_default();
out.sort_unstable();
out
}
pub(super) fn match_ingress_service(
candidates: &[(String, String)],
prefix: &str,
svc: &str,
) -> Option<usize> {
let exact = format!("{prefix}{svc}");
let replica = format!("{prefix}{svc}-");
candidates
.iter()
.position(|(_, n)| *n == exact || n.starts_with(&replica))
}
pub(super) fn match_exposed_port(exposed: &[Vec<u16>], want_ports: &[u16]) -> Option<usize> {
want_ports
.iter()
.find_map(|want| exposed.iter().position(|ports| ports.contains(want)))
}
pub async fn find_compose_ingress_container(
docker: &Docker,
project_name: &str,
ingress_service: Option<&str>,
want_ports: &[u16],
) -> Result<Option<String>> {
let prefix = format!("{project_name}-");
let candidates = list_running_by_prefix(docker, &prefix).await?;
if candidates.is_empty() {
return Ok(None);
}
if let Some(svc) = ingress_service.map(str::trim).filter(|s| !s.is_empty()) {
if let Some(i) = match_ingress_service(&candidates, &prefix, svc) {
let (id, name) = &candidates[i];
info!(container = %name, ingress_service = %svc, "compose ingress: alvo declarado");
return Ok(Some(id.clone()));
}
warn!(
ingress_service = %svc,
project = %project_name,
candidatos = ?candidates.iter().map(|(_, n)| n).collect::<Vec<_>>(),
"compose ingress: `ingress_service` não existe na stack — caindo para a descoberta por porta"
);
}
if !want_ports.is_empty() {
let mut portas_por_container: Vec<(String, String, Vec<u16>)> =
Vec::with_capacity(candidates.len());
for (id, name) in &candidates {
let ports = exposed_tcp_ports(docker, id).await;
portas_por_container.push((id.clone(), name.clone(), ports));
}
let so_portas: Vec<Vec<u16>> = portas_por_container
.iter()
.map(|(_, _, p)| p.clone())
.collect();
if let Some(i) = match_exposed_port(&so_portas, want_ports) {
let (id, name, _) = &portas_por_container[i];
info!(container = %name, "compose ingress: alvo por porta exposta");
return Ok(Some(id.clone()));
}
warn!(
project = %project_name,
?want_ports,
candidatos = ?portas_por_container
.iter()
.map(|(_, n, p)| format!("{n}{p:?}"))
.collect::<Vec<_>>(),
"compose ingress: nenhum container expõe a porta pedida — usando o primeiro por nome"
);
}
Ok(candidates.into_iter().next().map(|(id, _)| id))
}
#[cfg(test)]
mod tests_ingress {
use super::*;
fn cands(nomes: &[&str]) -> Vec<(String, String)> {
nomes
.iter()
.map(|n| (format!("id_{n}"), n.to_string()))
.collect()
}
const STACK: &[&str] = &[
"rp_01m11z0g_db-auth-1",
"rp_01m11z0g_db-db-1",
"rp_01m11z0g_db-functions-1",
"rp_01m11z0g_db-imgproxy-1",
"rp_01m11z0g_db-kong-1",
"rp_01m11z0g_db-meta-1",
"rp_01m11z0g_db-rest-1",
"rp_01m11z0g_db-storage-1",
"rp_01m11z0g_db-supavisor-1",
];
#[test]
fn alvo_declarado_acha_o_container_da_replica() {
let c = cands(STACK);
let i = match_ingress_service(&c, "rp_01m11z0g_db-", "kong").unwrap();
assert_eq!(c[i].1, "rp_01m11z0g_db-kong-1");
}
#[test]
fn alvo_declarado_aceita_nome_sem_sufixo_de_replica() {
let c = cands(&["rp_x-proxy", "rp_x-app-1"]);
let i = match_ingress_service(&c, "rp_x-", "proxy").unwrap();
assert_eq!(c[i].1, "rp_x-proxy");
}
#[test]
fn alvo_declarado_inexistente_nao_casa() {
let c = cands(&["rp_x-kong-1", "rp_x-storage-1"]);
assert!(match_ingress_service(&c, "rp_x-", "nginx").is_none());
}
#[test]
fn porta_do_dominio_escolhe_o_gateway_e_nao_o_primeiro() {
let exposed = vec![
vec![9999], vec![5432], vec![9000], vec![8080], vec![8000, 8443], vec![8080], vec![3000], vec![5000], vec![4000, 5452], ];
assert_eq!(match_exposed_port(&exposed, &[8000]), Some(4));
}
#[test]
fn sem_ninguem_expondo_a_porta_nao_ha_escolha_por_porta() {
let exposed = vec![vec![5432], vec![5000]];
assert_eq!(match_exposed_port(&exposed, &[8000]), None);
}
#[test]
fn a_primeira_porta_pedida_decide() {
let exposed = vec![vec![443], vec![8000]];
assert_eq!(match_exposed_port(&exposed, &[8000, 443]), Some(1));
assert_eq!(match_exposed_port(&exposed, &[443, 8000]), Some(0));
}
}
#[cfg(test)]
mod tests_owner {
use super::*;
use bollard::container::{Config, CreateContainerOptions};
const SVC: &str = "svc_01JABCDEFGHJKMNPQRSTVWXYZ0";
#[test]
fn nomes_atuais_tem_o_id() {
assert_eq!(replica_live_name(SVC, "api", 0), "rp_stvwxyz0_api");
assert_eq!(replica_live_name(SVC, "api", 2), "rp_stvwxyz0_api_r2");
assert_eq!(
replica_staging_name(SVC, "api", "01JQQQQQ", 1),
"rp_stvwxyz0_api_staging_01JQQQQQ_r1"
);
assert_eq!(
legacy_replica_staging_name("api", "01JQQQQQ", 0),
"rp_api_staging_01JQQQQQ"
);
}
#[test]
fn staging_reconhecido_nos_dois_formatos() {
assert!(is_staging_name("rp_stvwxyz0_api_staging_01JQQQQQ"));
assert!(is_staging_name("rp_stvwxyz0_api_staging_01JQQQQQ_r3"));
assert!(is_staging_name("rp_api_staging_01JQQQQQ"));
assert!(!is_staging_name("rp_stvwxyz0_api"));
assert!(!is_staging_name("rp_api_r1"));
assert!(!is_staging_name("rp_stvwxyz0_staging"));
assert!(!is_staging_name("rp_stvwxyz0_app_staging_area_de_testes"));
assert!(!is_staging_name("rp_stvwxyz0_app_staging_prod"));
assert!(!is_staging_name("rp_stvwxyz0_app_staging_prod_r1"));
}
#[test]
fn indice_de_replica_pelo_label_ou_pelo_nome() {
let mut l = HashMap::new();
l.insert("rustploy.replica".to_string(), "3".to_string());
assert_eq!(replica_index(Some(&l), "rp_stvwxyz0_api"), 3);
assert_eq!(replica_index(None, "rp_api_r2"), 2);
assert_eq!(replica_index(None, "rp_api"), 0);
assert_eq!(replica_index(None, "rp_x_redis"), 0);
}
#[test]
fn label_do_proprio_servico() {
assert!(label_owned_by("svc_01JABC", "svc_01JABC"));
}
#[test]
fn label_de_antes_dos_prefixos_de_id() {
assert!(label_owned_by("01JABC", "svc_01JABC"));
}
#[test]
fn label_de_outro_servico() {
assert!(!label_owned_by("svc_01JXYZ", "svc_01JABC"));
assert!(!label_owned_by("01JXYZ", "svc_01JABC"));
assert!(!label_owned_by("", "svc_01JABC"));
}
#[tokio::test]
#[ignore]
async fn nome_ocupado_por_outro_servico_nao_e_achado() {
let docker = Docker::connect_with_local_defaults().expect("docker");
let image = std::env::var("RP_TEST_IMAGE").unwrap_or_else(|_| "redis:6.2-alpine".into());
let name = format!("rp_teste_dono_{}", std::process::id());
let mut labels = HashMap::new();
labels.insert("rustploy.service_id".to_string(), "svc_OUTRO".to_string());
let id = docker
.create_container(
Some(CreateContainerOptions {
name: name.clone(),
platform: None,
}),
Config {
image: Some(image),
labels: Some(labels),
..Default::default()
},
)
.await
.expect("create")
.id;
let meu = find_owned_by_name(&docker, &name, "svc_MEU").await;
let dele = find_owned_by_name(&docker, &name, "svc_OUTRO").await;
let dono = name_owner(&docker, &name).await;
let _ = remove(&docker, &id).await;
assert_eq!(
meu.unwrap(),
None,
"container de outro serviço não pode ser o nosso"
);
assert_eq!(dele.unwrap(), Some(id.clone()));
assert_eq!(dono.unwrap(), Some((id, Some("svc_OUTRO".to_string()))));
assert_eq!(name_owner(&docker, &name).await.unwrap(), None);
}
}