Feat/s3 storage

Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "control-plane/failover-controller",
     "control-plane/sdn-controller",
     "control-plane/volume-manager",
+    "control-plane/object-storage",
     "control-plane/registry",
     "control-plane/shared/entity",
     "control-plane/shared/migration",

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.88-slim-bookworm AS base
+FROM rust:1.97-slim-bookworm AS base
 
 WORKDIR /app
 
@@ -26,6 +26,7 @@ COPY control-plane/scheduler/Cargo.toml ./control-plane/scheduler/
 COPY control-plane/failover-controller/Cargo.toml ./control-plane/failover-controller/
 COPY control-plane/sdn-controller/Cargo.toml ./control-plane/sdn-controller/
 COPY control-plane/volume-manager/Cargo.toml ./control-plane/volume-manager/
+COPY control-plane/object-storage/Cargo.toml ./control-plane/object-storage/
 COPY control-plane/registry/Cargo.toml ./control-plane/registry/
 COPY control-plane/shared/entity/Cargo.toml ./control-plane/shared/entity/
 COPY control-plane/shared/migration/Cargo.toml ./control-plane/shared/migration/
@@ -39,6 +40,7 @@ RUN mkdir -p agent/src \
     control-plane/failover-controller/src \
     control-plane/sdn-controller/src \
     control-plane/volume-manager/src \
+    control-plane/object-storage/src \
     control-plane/registry/src \
     control-plane/shared/entity/src \
     control-plane/shared/migration/src \
@@ -51,6 +53,7 @@ RUN mkdir -p agent/src \
     && echo "fn main() {}" > control-plane/failover-controller/src/main.rs \
     && echo "fn main() {}" > control-plane/sdn-controller/src/main.rs \
     && echo "fn main() {}" > control-plane/volume-manager/src/main.rs \
+    && echo "fn main() {}" > control-plane/object-storage/src/main.rs \
     && echo "fn main() {}" > control-plane/registry/src/main.rs \
     && echo "fn main() {}" > control-plane/csfx-migrate/src/main.rs \
     && echo "fn main() {}" > control-plane/csfx-updater/src/main.rs \

agent/src/firecracker/rootfs.rs

@@ -68,10 +68,7 @@ impl RootfsBuilder {
             (Some(tag), None) => format!(":{tag}"),
             (None, None) => String::new(),
         };
-        let mirrored = format!(
-            "{mirror}/{}{tag_or_digest}",
-            reference.repository()
-        );
+        let mirrored = format!("{mirror}/{}{tag_or_digest}", reference.repository());
         mirrored.parse().context("Invalid mirrored image reference")
     }
 

agent/src/firecracker/runtime.rs

@@ -3,6 +3,7 @@ use serde_json::json;
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::UnixStream;
 use tokio::process::Command;
@@ -153,16 +154,21 @@ struct GuestNetwork {
 pub struct FirecrackerRuntime {
     wg_private_key_b64: String,
     dns_supervisor: RgDnsProcessSupervisor,
+    rg_dns_registry: Arc<crate::rg_dns::RgDnsRegistry>,
     handles: Mutex<HashMap<String, VmHandle>>,
     next_cid: Mutex<u32>,
     reconciled: AtomicBool,
 }
 
 impl FirecrackerRuntime {
-    pub fn new(wg_private_key_b64: String) -> Self {
+    pub fn new(
+        wg_private_key_b64: String,
+        rg_dns_registry: Arc<crate::rg_dns::RgDnsRegistry>,
+    ) -> Self {
         Self {
             wg_private_key_b64,
             dns_supervisor: RgDnsProcessSupervisor::new(),
+            rg_dns_registry,
             handles: Mutex::new(HashMap::new()),
             next_cid: Mutex::new(next_free_cid_on_host()),
             reconciled: AtomicBool::new(false),
@@ -194,9 +200,13 @@ impl FirecrackerRuntime {
         resource_group_id: &str,
         resource_group_cidr: Option<&str>,
     ) -> Result<String> {
-        let iface = crate::rg_network::ensure_bridge(resource_group_id, resource_group_cidr)
-            .await
-            .context("Failed to ensure resource group bridge")?;
+        let iface = crate::rg_network::ensure_bridge(
+            resource_group_id,
+            resource_group_cidr,
+            &self.rg_dns_registry,
+        )
+        .await
+        .context("Failed to ensure resource group bridge")?;
 
         if let Some(cidr) = resource_group_cidr {
             self.dns_supervisor

agent/src/main.rs

@@ -137,8 +137,11 @@ async fn main() -> Result<()> {
         warn!(error = %e, "Failed to initialize nftables resource group isolation");
     }
 
+    let rg_dns_registry = Arc::new(rg_dns::RgDnsRegistry::new());
+
     let firecracker_runtime = Arc::new(firecracker::runtime::FirecrackerRuntime::new(
         wg_identity.private_key_b64.clone(),
+        Arc::clone(&rg_dns_registry),
     ));
 
     let running_containers: Arc<Mutex<HashMap<String, String>>> =
@@ -153,8 +156,6 @@ async fn main() -> Result<()> {
     let service_dns_registry: Arc<Mutex<HashMap<String, (String, String)>>> =
         Arc::new(Mutex::new(HashMap::new()));
 
-    let rg_dns_registry = rg_dns::RgDnsRegistry::new();
-
     if let Some(port) = std::env::var("CSFX_AGENT_PORT")
         .ok()
         .and_then(|v| v.parse::<u16>().ok())
@@ -363,7 +364,7 @@ async fn run_heartbeat_loop(
     mounted_volumes: Arc<Mutex<HashMap<String, String>>>,
     restart_counts: Arc<Mutex<HashMap<String, u32>>>,
     service_dns_registry: Arc<Mutex<HashMap<String, (String, String)>>>,
-    rg_dns_registry: rg_dns::RgDnsRegistry,
+    rg_dns_registry: Arc<rg_dns::RgDnsRegistry>,
     mut assignment_signal: tokio::sync::mpsc::UnboundedReceiver<()>,
 ) {
     let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));

agent/src/nftables.rs

@@ -130,6 +130,30 @@ pub async fn add_rg_port_dnat(
     .await
 }
 
+pub async fn dnat_bridge_port(bridge_name: &str, rg_gateway_ip: &str, port: u16) -> Result<()> {
+    run_nft(&[
+        "add",
+        "rule",
+        "ip",
+        NAT_TABLE_NAME,
+        NAT_CHAIN_NAME,
+        "iifname",
+        bridge_name,
+        "ip",
+        "daddr",
+        rg_gateway_ip,
+        "tcp",
+        "dport",
+        &port.to_string(),
+        "dnat",
+        "to",
+        &format!("127.0.0.1:{}", port),
+        "comment",
+        &format!("\"{}-s3\"", bridge_name),
+    ])
+    .await
+}
+
 pub async fn remove_node_port_rules(workload_id: &str) -> Result<()> {
     let output = Command::new("nft")
         .args(["-a", "list", "chain", "ip", NAT_TABLE_NAME, NAT_CHAIN_NAME])

agent/src/rg_network.rs

@@ -3,11 +3,18 @@ use std::path::{Path, PathBuf};
 use tokio::process::Command;
 use tracing::info;
 
+use crate::rg_dns::RgDnsRegistry;
 use crate::spec::{rg_bridge_iface_name, second_host_ip};
 
 const RG_REGISTRY_DIR: &str = "/var/lib/csfx-agent/rg-networks";
-
-pub async fn ensure_bridge(resource_group_id: &str, cidr: Option<&str>) -> Result<String> {
+const S3_SERVICE_NAME: &str = "s3";
+const S3_DNAT_PORT: u16 = 3900;
+
+pub async fn ensure_bridge(
+    resource_group_id: &str,
+    cidr: Option<&str>,
+    rg_dns_registry: &RgDnsRegistry,
+) -> Result<String> {
     let iface = rg_bridge_iface_name(resource_group_id);
 
     write_registry_entry(resource_group_id).await?;
@@ -29,6 +36,18 @@ pub async fn ensure_bridge(resource_group_id: &str, cidr: Option<&str>) -> Resul
                 &iface,
             ])
             .await?;
+
+            if let Err(e) = rg_dns_registry
+                .upsert(resource_group_id, S3_SERVICE_NAME, &gateway)
+                .await
+            {
+                info!(resource_group_id = %resource_group_id, error = %e, "Failed to register s3 dns record");
+            }
+
+            if let Err(e) = crate::nftables::dnat_bridge_port(&iface, &gateway, S3_DNAT_PORT).await
+            {
+                info!(resource_group_id = %resource_group_id, error = %e, "Failed to set up s3 dnat rule");
+            }
         }
     }
 

agent/src/server.rs

@@ -187,13 +187,15 @@ async fn logs_handler(
         })?;
 
     info!(workload_id = %workload_id, container_id = %container_id, "opening log stream to guest");
-    let stream = futures_util::stream::once(async {
-        Ok::<_, std::io::Error>(axum::body::Bytes::new())
-    })
-    .chain(state.firecracker.logs(&container_id));
+    let stream =
+        futures_util::stream::once(async { Ok::<_, std::io::Error>(axum::body::Bytes::new()) })
+            .chain(state.firecracker.logs(&container_id));
 
     Ok((
-        [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
+        [(
+            axum::http::header::CONTENT_TYPE,
+            "text/plain; charset=utf-8",
+        )],
         axum::body::Body::from_stream(stream),
     ))
 }