feat: send reqwest to the upstream server

batleforc · batleforc · commit 153392455b4b · 2025-11-04T00:22:35.000+01:00
diff --git a/.compose/compose.otel.yaml b/.compose/compose.otel.yaml
@@ -0,0 +1,20 @@
+services:
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:latest
+    command: ["--config=/etc/otel-collector-config.yaml"]
+    volumes:
+      - ./tracing/otel-collector-config.yaml:/etc/otel-collector-config.yaml
+    ports:
+      - "4317:4317" # OTLP gRPC receiver
+  jaeger:
+    image: jaegertracing/all-in-one:latest
+    ports:
+      - "6831:6831/udp" # UDP port for Jaeger agent
+      - "16686:16686" # Web UI
+      - "14268:14268" # HTTP port for spans
+  prometheus:
+    image: prom/prometheus:latest
+    volumes:
+      - ./tracing/prometheus.yml:/etc/prometheus/prometheus.yml
+    ports:
+      - "9090:9090"
diff --git a/.compose/tracing/otel-collector-config.yaml b/.compose/tracing/otel-collector-config.yaml
@@ -0,0 +1,23 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+processors:
+extensions:
+  health_check: {}
+exporters:
+  otlp:
+    endpoint: jaeger:4317
+    tls:
+      insecure: true
+  prometheus:
+    endpoint: "0.0.0.0:9090"
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [otlp]
+    metrics:
+      receivers: [otlp]
+      exporters: [prometheus]
diff --git a/.compose/tracing/prometheus.yml b/.compose/tracing/prometheus.yml
@@ -0,0 +1,7 @@
+global:
+  scrape_interval: 5s
+
+scrape_configs:
+  - job_name: 'otel-collector'
+    static_configs:
+      - targets: ['otel-collector:9090']
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -6,6 +6,7 @@
     "finalizer",
     "proxykubeapis",
     "proxys",
+    "reqwest",
     "schemars",
     "taskfile",
     "thiserror",
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -15,6 +15,7 @@ members = [
 [workspace.dependencies]
 actix-web = '4'
 tokio = { version = '1', features = ['macros', 'rt-multi-thread'] }
+tokio-stream = '0.1'
 anyhow = '1'
 utoipa = { version = '5', features = ['actix_extras'] }
 utoipa-actix-web = { version = '0.1' }
@@ -28,6 +29,7 @@ kube = { version = '2', features = ['derive', 'runtime', 'client'] }
 reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"] }
 schemars = { version = '1' }
 futures = '0.3'
+futures-util = "0.3"
 chrono = { version = '0.4', features = ['serde'] }
 deadpool-redis = { version = "0.22.0", features = ["serde"] }
 opentelemetry = { version = '0.30', features = ['trace', 'metrics', 'logs'] }
diff --git a/Taskfile.yaml b/Taskfile.yaml
@@ -18,8 +18,8 @@ tasks:
     desc: "Run the backend"
     env:
       KUBECONFIG: ./kubeconfig.local.yaml
-      OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:5081/api/default"
-      OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Basic cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM=,organization=default,stream-name=default"
+      #OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:5081/api/default"
+      #OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Basic cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM=,organization=default,stream-name=default"
       REDIS_URL: redis://:{{ .REDIS_PASSWORD }}@localhost:6379
     silent: true
     cmds:
diff --git a/apps/server/Cargo.toml b/apps/server/Cargo.toml
@@ -12,6 +12,7 @@ utoipa = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true }
 kube = { workspace = true }
+reqwest = { workspace = true }
 
 utoipa-actix-web = { workspace = true }
 
diff --git a/apps/server/src/main.rs b/apps/server/src/main.rs
@@ -20,6 +20,7 @@ async fn main() -> anyhow::Result<()> {
     let controller = controller::run(state.clone());
     let mut api_doc = ApiDoc::openapi();
     api_doc.info.version = env!("CARGO_PKG_VERSION").to_string();
+    let reqwest_client = reqwest::Client::default();
     let server = HttpServer::new(move || {
         let cors = Cors::default()
             .allow_any_origin()
@@ -62,6 +63,7 @@ async fn main() -> anyhow::Result<()> {
             .map(|app| app.wrap(Compress::default()))
             .map(|app| app.wrap(cors))
             .app_data(Data::new(state.clone()))
+            .app_data(Data::new(reqwest_client.clone()))
             .service(scope("/api/v1").configure(init_api()))
             .service(scope("/clusters").configure(init_cluster_api()))
             .split_for_parts();
diff --git a/compose.yaml b/compose.yaml
@@ -1,30 +1,6 @@
 include:
-  - .compose/compose.openobserve.yaml
+  #- .compose/compose.openobserve.yaml
+  - .compose/compose.otel.yaml
   - .compose/compose.redis.yaml
   - .compose/compose.traefik.yaml
-  - .compose/compose.authelia.yaml
-
-services:
-  secure:
-    image: "traefik/whoami"
-    container_name: "secure"
-    labels:
-      traefik.enable: "true"
-      traefik.http.routers.secure.rule: "Host(`secure.k8s.localhost`)"
-      traefik.http.routers.secure.entrypoints: "https"
-      traefik.http.routers.secure.tls: "true"
-      traefik.http.routers.secure.tls.options: "default"
-      traefik.http.routers.secure.middlewares: "authelia@docker"
-    restart: "unless-stopped"
-
-  public:
-    image: "traefik/whoami"
-    container_name: "public"
-    labels:
-      traefik.enable: "true"
-      traefik.http.routers.public.rule: "Host(`public.k8s.localhost`)"
-      traefik.http.routers.public.entrypoints: "https"
-      traefik.http.routers.public.tls: "true"
-      traefik.http.routers.public.tls.options: "default"
-      traefik.http.routers.public.middlewares: "authelia@docker"
-    restart: "unless-stopped"
+  #- .compose/compose.authelia.yaml
diff --git a/libs/api/Cargo.toml b/libs/api/Cargo.toml
@@ -11,6 +11,9 @@ tracing = { workspace = true }
 deadpool-redis = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+futures-util = { workspace = true }
 
 common = { path = "../common" }
 crd = { path = "../crd" }
diff --git a/libs/api/src/cluster/redirect/mod.rs b/libs/api/src/cluster/redirect/mod.rs
@@ -1,4 +1,4 @@
-use actix_web::{delete, get, patch, post, put, web, HttpRequest, Responder};
+use actix_web::{delete, dev::PeerAddr, get, http, patch, post, put, web, HttpRequest, Responder};
 use common::State;
 use redirect::redirect;
 use tracing::instrument;
@@ -18,9 +18,15 @@ pub mod redirect;
     )
 )]
 #[get("/{ns}/{cluster}/{path:.*}")]
-#[instrument(name = "get_redirect", skip(data))]
-pub async fn get_redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
-    redirect(req, data).await
+#[instrument(name = "get_redirect", skip(data, payload))]
+pub async fn get_redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
+    redirect(req, data, payload, method, peer_addr).await
 }
 
 /// Cluster redirect
@@ -34,9 +40,15 @@ pub async fn get_redirect(req: HttpRequest, data: web::Data<State>) -> impl Resp
     )
 )]
 #[post("/{ns}/{cluster}/{path:.*}")]
-#[instrument(name = "post_redirect", skip(data))]
-pub async fn post_redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
-    redirect(req, data).await
+#[instrument(name = "post_redirect", skip(data, payload))]
+pub async fn post_redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
+    redirect(req, data, payload, method, peer_addr).await
 }
 
 /// Cluster redirect
@@ -50,9 +62,15 @@ pub async fn post_redirect(req: HttpRequest, data: web::Data<State>) -> impl Res
     )
 )]
 #[put("/{ns}/{cluster}/{path:.*}")]
-#[instrument(name = "put_redirect", skip(data))]
-pub async fn put_redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
-    redirect(req, data).await
+#[instrument(name = "put_redirect", skip(data, payload))]
+pub async fn put_redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
+    redirect(req, data, payload, method, peer_addr).await
 }
 
 /// Cluster redirect
@@ -66,9 +84,15 @@ pub async fn put_redirect(req: HttpRequest, data: web::Data<State>) -> impl Resp
     )
 )]
 #[patch("/{ns}/{cluster}/{path:.*}")]
-#[instrument(name = "patch_redirect", skip(data))]
-pub async fn patch_redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
-    redirect(req, data).await
+#[instrument(name = "patch_redirect", skip(data, payload))]
+pub async fn patch_redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
+    redirect(req, data, payload, method, peer_addr).await
 }
 
 /// Cluster redirect
@@ -82,7 +106,13 @@ pub async fn patch_redirect(req: HttpRequest, data: web::Data<State>) -> impl Re
     )
 )]
 #[delete("/{ns}/{cluster}/{path:.*}")]
-#[instrument(name = "delete_redirect", skip(data))]
-pub async fn delete_redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
-    redirect(req, data).await
+#[instrument(name = "delete_redirect", skip(data, payload))]
+pub async fn delete_redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
+    redirect(req, data, payload, method, peer_addr).await
 }
diff --git a/libs/api/src/cluster/redirect/redirect.rs b/libs/api/src/cluster/redirect/redirect.rs
@@ -1,14 +1,22 @@
-use actix_web::{web, HttpRequest, HttpResponse, Responder};
+use actix_web::{dev::PeerAddr, http, web, HttpRequest, HttpResponse, Responder};
 use common::State;
 use crd::ProxyKubeApi;
 use deadpool_redis::redis::AsyncTypedCommands;
+use futures_util::stream::StreamExt;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::{error, instrument};
 
-#[instrument(name = "main_redirect", skip(data))]
-pub async fn redirect(req: HttpRequest, data: web::Data<State>) -> impl Responder {
+#[instrument(name = "main_redirect", skip(data, payload))]
+pub async fn redirect(
+    req: HttpRequest,
+    data: web::Data<State>,
+    mut payload: web::Payload,
+    method: http::Method,
+    peer_addr: Option<PeerAddr>,
+) -> impl Responder {
     let ns: String = req.match_info().get("ns").unwrap().parse().unwrap();
     let cluster: String = req.match_info().get("cluster").unwrap().parse().unwrap();
-    let path: String = req.match_info().get("path").unwrap().parse().unwrap();
     let mut conn = match data.get_redis_conn().await {
         Ok(conn) => conn,
         Err(e) => return HttpResponse::ServiceUnavailable().body(e.to_string()),
@@ -18,7 +26,7 @@ pub async fn redirect(req: HttpRequest, data: web::Data<State>) -> impl Responde
         Ok(None) => return HttpResponse::NotFound().finish(),
         Err(e) => return HttpResponse::ServiceUnavailable().body(e.to_string()),
     };
-
+    // TODO: MTLS ? https://github.com/actix/examples/blob/master/https-tls/rustls-client-cert/src/main.rs
     let proxy = match ProxyKubeApi::from_json(&proxy_json) {
         Some(proxy) => proxy,
         None => {
@@ -33,12 +41,61 @@ pub async fn redirect(req: HttpRequest, data: web::Data<State>) -> impl Responde
         .url_to_call(data.client.clone(), "default".to_string())
         .await
     {
-        Ok(url) => url,
+        Ok(url) => req
+            .uri()
+            .to_string()
+            .replace(&format!("/clusters/{}/{}", ns, cluster), &url),
         Err(err) => {
             error!(err);
             return HttpResponse::NotFound().finish();
         }
     };
+
+    let (tx, rx) = mpsc::unbounded_channel();
+    actix_web::rt::spawn(async move {
+        while let Some(chunk) = payload.next().await {
+            tx.send(chunk).unwrap();
+        }
+    });
+    let client = match proxy.get_client(data.into_inner()).await {
+        Ok(c) => c,
+        Err(err) => {
+            error!(err);
+            return HttpResponse::ServiceUnavailable().body(err);
+        }
+    };
+    let mut forwarded_req = client
+        .request(
+            reqwest::Method::from_bytes(method.as_str().as_bytes()).unwrap(),
+            url_to_call.clone(),
+        )
+        .body(reqwest::Body::wrap_stream(UnboundedReceiverStream::new(rx)));
+    for (h, v) in req.headers().iter() {
+        forwarded_req = forwarded_req.header(h.as_str(), v.clone().to_str().unwrap());
+        println!("Header: {}: {}", h.as_str(), v.clone().to_str().unwrap());
+    }
+    if let Some(PeerAddr(addr)) = peer_addr {
+        forwarded_req = forwarded_req.header("x-forwarded-for", addr.ip().to_string());
+    }
+
+    let res = match forwarded_req.send().await {
+        Ok(res) => res,
+        Err(e) => {
+            tracing::error!(error = %e);
+            return HttpResponse::ServiceUnavailable().body(e.to_string());
+        }
+    };
+
+    let mut client_resp =
+        HttpResponse::build(actix_web::http::StatusCode::from_u16(res.status().as_u16()).unwrap());
+
+    for (header_name, header_value) in res.headers().iter().filter(|(h, _)| *h != "connection") {
+        client_resp.insert_header((
+            actix_web::http::header::HeaderName::from_bytes(header_name.as_ref()).unwrap(),
+            actix_web::http::header::HeaderValue::from_bytes(header_value.as_ref()).unwrap(),
+        ));
+    }
+
+    client_resp.streaming(res.bytes_stream())
     // https://github.com/actix/examples/blob/master/http-proxy/src/main.rs#L56
-    HttpResponse::Ok().body(format!("{}/{}", url_to_call, path))
 }
diff --git a/libs/crd/src/lib.rs b/libs/crd/src/lib.rs
