feat: add X-Dragonfly-Use-P2P header for proxy to use P2P to distribute the content (#733)

Signed-off-by: Gaius <gaius.qi@gmail.com>

Cargo.lock

@@ -861,7 +861,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "anyhow",
  "blake3",
@@ -932,7 +932,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-backend"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "dragonfly-api",
  "dragonfly-client-core",
@@ -955,7 +955,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-config"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "bytesize",
  "bytesize-serde",
@@ -976,7 +976,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-core"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "hyper 1.4.1",
  "hyper-util",
@@ -991,7 +991,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-init"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "anyhow",
  "clap",
@@ -1007,7 +1007,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-storage"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "base16ct",
  "bincode",
@@ -1032,7 +1032,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-client-util"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "base16ct",
  "blake3",
@@ -1386,7 +1386,7 @@ dependencies = [
 
 [[package]]
 name = "hdfs"
-version = "0.1.107"
+version = "0.1.108"
 dependencies = [
  "dragonfly-client-backend",
  "dragonfly-client-core",

Cargo.toml

@@ -12,7 +12,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.1.107"
+version = "0.1.108"
 authors = ["The Dragonfly Developers"]
 homepage = "https://d7y.io/"
 repository = "https://github.com/dragonflyoss/client.git"
@@ -22,13 +22,13 @@ readme = "README.md"
 edition = "2021"
 
 [workspace.dependencies]
-dragonfly-client = { path = "dragonfly-client", version = "0.1.107" }
+dragonfly-client = { path = "dragonfly-client", version = "0.1.108" }
-dragonfly-client-core = { path = "dragonfly-client-core", version = "0.1.107" }
+dragonfly-client-core = { path = "dragonfly-client-core", version = "0.1.108" }
-dragonfly-client-config = { path = "dragonfly-client-config", version = "0.1.107" }
+dragonfly-client-config = { path = "dragonfly-client-config", version = "0.1.108" }
-dragonfly-client-storage = { path = "dragonfly-client-storage", version = "0.1.107" }
+dragonfly-client-storage = { path = "dragonfly-client-storage", version = "0.1.108" }
-dragonfly-client-backend = { path = "dragonfly-client-backend", version = "0.1.107" }
+dragonfly-client-backend = { path = "dragonfly-client-backend", version = "0.1.108" }
-dragonfly-client-util = { path = "dragonfly-client-util", version = "0.1.107" }
+dragonfly-client-util = { path = "dragonfly-client-util", version = "0.1.108" }
-dragonfly-client-init = { path = "dragonfly-client-init", version = "0.1.107" }
+dragonfly-client-init = { path = "dragonfly-client-init", version = "0.1.108" }
 thiserror = "1.0"
 dragonfly-api = "=2.0.154"
 reqwest = { version = "0.12.4", features = ["stream", "native-tls", "default-tls", "rustls-tls"] }

dragonfly-client/src/proxy/header.rs

@@ -39,6 +39,12 @@ pub const DRAGONFLY_REGISTRY_HEADER: &str = "X-Dragonfly-Registry";
 // Default value includes the filtered query params of s3, gcs, oss, obs, cos.
 pub const DRAGONFLY_FILTERED_QUERY_PARAMS_HEADER: &str = "X-Dragonfly-Filtered-Query-Params";
 
+// DRAGONFLY_USE_P2P_HEADER is the header key of use p2p in http request.
+// If the value is "true", the request will use P2P technology to distribute
+// the content. If the value is "false", but url matches the regular expression in proxy config.
+// The request will also use P2P technology to distribute the content.
+pub const DRAGONFLY_USE_P2P_HEADER: &str = "X-Dragonfly-Use-P2P";
+
 // get_tag gets the tag from http header.
 #[instrument(skip_all)]
 pub fn get_tag(header: &HeaderMap) -> Option<String> {
@@ -123,3 +129,18 @@ pub fn get_filtered_query_params(
         None => default_filtered_query_params,
     }
 }
+
+// get_use_p2p gets the use p2p from http header.
+#[instrument(skip_all)]
+pub fn get_use_p2p(header: &HeaderMap) -> bool {
+    match header.get(DRAGONFLY_USE_P2P_HEADER) {
+        Some(value) => match value.to_str() {
+            Ok(value) => value.eq_ignore_ascii_case("true"),
+            Err(err) => {
+                error!("get use p2p from header failed: {}", err);
+                false
+            }
+        },
+        None => false,
+    }
+}

dragonfly-client/src/proxy/mod.rs

@@ -348,7 +348,7 @@ pub async fn http_handler(
         find_matching_rule(config.proxy.rules.clone(), request_uri.to_string().as_str())
     {
         info!(
-            "proxy HTTP request via dfdaemon for method: {}, uri: {}",
+            "proxy HTTP request via dfdaemon by rule config for method: {}, uri: {}",
             request.method(),
             request_uri
         );
@@ -362,6 +362,24 @@ pub async fn http_handler(
         .await;
     }
 
+    // If the request header contains the X-Dragonfly-Use-P2P header, proxy the request via the
+    // dfdaemon.
+    if header::get_use_p2p(request.headers()) {
+        info!(
+            "proxy HTTP request via dfdaemon by X-Dragonfly-Use-P2P header for method: {}, uri: {}",
+            request.method(),
+            request_uri
+        );
+        return proxy_by_dfdaemon(
+            config,
+            task,
+            Rule::default(),
+            request,
+            dfdaemon_download_client,
+        )
+        .await;
+    }
+
     if request.uri().scheme().cloned() == Some(http::uri::Scheme::HTTPS) {
         info!(
             "proxy HTTPS request directly to remote server for method: {}, uri: {}",
@@ -512,7 +530,7 @@ pub async fn upgraded_handler(
         find_matching_rule(config.proxy.rules.clone(), request_uri.to_string().as_str())
     {
         info!(
-            "proxy HTTPS request via dfdaemon for method: {}, uri: {}",
+            "proxy HTTPS request via dfdaemon by rule config for method: {}, uri: {}",
             request.method(),
             request_uri
         );
@@ -526,6 +544,24 @@ pub async fn upgraded_handler(
         .await;
     }
 
+    // If the request header contains the X-Dragonfly-Use-P2P header, proxy the request via the
+    // dfdaemon.
+    if header::get_use_p2p(request.headers()) {
+        info!(
+            "proxy HTTP request via dfdaemon by X-Dragonfly-Use-P2P header for method: {}, uri: {}",
+            request.method(),
+            request_uri
+        );
+        return proxy_by_dfdaemon(
+            config,
+            task,
+            Rule::default(),
+            request,
+            dfdaemon_download_client,
+        )
+        .await;
+    }
+
     if request.uri().scheme().cloned() == Some(http::uri::Scheme::HTTPS) {
         info!(
             "proxy HTTPS request directly to remote server for method: {}, uri: {}",