diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 25bf8dfc3d..409aaa734a 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -555,6 +555,7 @@ async fn lookup_volume( .send() .await .map_err(|e| format!("lookup request failed: {}", e))?; + let status = resp.status(); let result: LookupResult = resp .json() .await @@ -562,7 +563,18 @@ async fn lookup_volume( if let Some(err) = result.error && !err.is_empty() { - return Err(err); + // Mirror Go: a missing volume arrives as a per-volume error in an + // answered lookup (HTTP 404), meaning "no locations" — a real master + // failure (e.g. 503 while warming up) stays an error so proxy mode + // can tell missing data (404) from a failed lookup (500). + if status != StatusCode::NOT_FOUND { + return Err(err); + } + tracing::debug!("volume {} lookup: {}", volume_id, err); + } else if !status.is_success() && status != StatusCode::NOT_FOUND { + // A non-success answer without an error body (e.g. an intermediary's + // 503 as `{}`) is a failed lookup, not a missing volume. + return Err(format!("lookup failed: {}", status)); } Ok(result.locations.unwrap_or_default()) } @@ -812,6 +824,7 @@ async fn do_replicated_request( /// Extracted request info needed for proxy/redirect (avoids borrowing Request across await). struct ProxyRequestInfo { + method: Method, original_headers: HeaderMap, original_query: String, path: String, @@ -823,12 +836,14 @@ fn build_proxy_request_info( path: &str, headers: &HeaderMap, query_string: &str, + method: &Method, ) -> Option { // Go redirects to "vid,fid" built from parseURLPath, so the extension must // already be off the fid here (volume_server_handlers_read.go:128-137). let parsed = parse_needle_path(path)?; Some(ProxyRequestInfo { + method: method.clone(), original_headers: headers.clone(), original_query: query_string.to_string(), path: path.to_string(), @@ -856,7 +871,11 @@ async fn proxy_or_redirect_to_target( Ok(locs) => locs, Err(e) => { tracing::warn!("volume lookup failed for {}: {}", vid.0, e); - return StatusCode::NOT_FOUND.into_response(); + // Go: master lookup failure is 500 in proxy mode, 404 for redirect + return match state.read_mode { + ReadMode::Proxy => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + _ => StatusCode::NOT_FOUND.into_response(), + }; } }; @@ -893,24 +912,58 @@ async fn proxy_or_redirect_to_target( } } +/// Hop-by-hop field names that must not be relayed: the fixed set plus every +/// field the header's own Connection values nominate (RFC 9110 7.6.1). +fn hop_by_hop_names(headers: &HeaderMap, extra: &[&str]) -> std::collections::HashSet { + let mut names: std::collections::HashSet = headers + .get_all(header::CONNECTION) + .iter() + .filter_map(|v| v.to_str().ok()) + .flat_map(|v| v.split(',')) + .map(|token| token.trim().to_ascii_lowercase()) + .filter(|token| !token.is_empty()) + .collect(); + for fixed in [ + "connection", + "keep-alive", + "te", + "trailer", + "upgrade", + "proxy-connection", + "proxy-authenticate", + "proxy-authorization", + ] { + names.insert(fixed.to_string()); + } + for name in extra { + names.insert(name.to_string()); + } + names +} + /// Proxy the request to the target volume server. async fn proxy_request( state: &VolumeServerState, info: &ProxyRequestInfo, target: &VolumeLocation, ) -> Response { - // Build target URL, adding proxied=true query param + // Build target URL, adding proxied=true query param. Go's query.Set + // replaces any proxied params the client sent, so a stale proxied=false + // cannot outrank the marker this hop adds. let path = info.path.trim_start_matches('/'); // Defensively strip a `.grpcPort` suffix in case target.url carries it. let target_http = to_http_address(&target.url); - let raw_target = if info.original_query.is_empty() { + let forwarded_query = info + .original_query + .split('&') + .filter(|kv| kv.split('=').next() != Some("proxied")) + .collect::>() + .join("&"); + let raw_target = if forwarded_query.is_empty() { format!("{}/{}?proxied=true", target_http, path) } else { - format!( - "{}/{}?{}&proxied=true", - target_http, path, info.original_query - ) + format!("{}/{}?{}&proxied=true", target_http, path, forwarded_query) }; let target_url = match normalize_outgoing_http_url(&state.outgoing_http_scheme, &raw_target) { Ok(url) => url, @@ -920,11 +973,21 @@ async fn proxy_request( } }; - // Build the proxy request - let mut req_builder = state.http_client.get(&target_url); + // Build the proxy request, preserving the inbound method (Go parity: + // a client HEAD must not turn into a GET download at the replica). + let mut req_builder = state.http_client.request(info.method.clone(), &target_url); - // Forward all original headers as raw bytes, as Go does. + // Forward the original headers, as Go does — minus hop-by-hop fields. + // The proxied request carries no body, so this hop's Host and framing + // must not reach the replica either. + let hop_by_hop = hop_by_hop_names( + &info.original_headers, + &["host", "content-length", "transfer-encoding"], + ); for (name, value) in &info.original_headers { + if hop_by_hop.contains(name.as_str()) { + continue; + } req_builder = req_builder.header(name.clone(), value.clone()); } @@ -936,12 +999,15 @@ async fn proxy_request( } }; - // Build response, copying headers and body from remote + // Build response, copying headers and body from remote. Hop-by-hop + // response fields describe the replica hop, not ours — Content-Length + // stays because it still describes the relayed body. let status = StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + let resp_hop_by_hop = hop_by_hop_names(resp.headers(), &[]); let mut response_headers = HeaderMap::new(); for (name, value) in resp.headers() { - if name.as_str().eq_ignore_ascii_case("server") { + if name.as_str().eq_ignore_ascii_case("server") || resp_hop_by_hop.contains(name.as_str()) { continue; } response_headers.append(name.clone(), value.clone()); @@ -1088,14 +1154,30 @@ async fn get_or_head_handler_inner( let has_ec_volume = state.store.read().unwrap().has_ec_volume(vid); if !has_volume && !has_ec_volume { - return proxy_missing_volume(&state, request.uri(), request.headers(), &path, vid).await; + return proxy_missing_volume( + &state, + request.uri(), + request.headers(), + &path, + vid, + &method, + ) + .await; } - let track_download = - match check_download_limit(&state, request.uri(), request.headers(), &path, vid).await { - ControlFlow::Continue(track_download) => track_download, - ControlFlow::Break(resp) => return resp, - }; + let track_download = match check_download_limit( + &state, + request.uri(), + request.headers(), + &path, + vid, + &method, + ) + .await + { + ControlFlow::Continue(track_download) => track_download, + ControlFlow::Break(resp) => return resp, + }; let read_deleted = query.read_deleted.as_deref() == Some("true"); // Go reads Range once and ignores an empty value; a non-UTF-8 byte fails its parse. @@ -1277,6 +1359,7 @@ async fn proxy_missing_volume( request_headers: &HeaderMap, path: &str, vid: VolumeId, + method: &Method, ) -> Response { let query_string = uri.query().unwrap_or("").to_string(); if is_proxied_query(&query_string) @@ -1287,7 +1370,7 @@ async fn proxy_missing_volume( } // For redirect, fid must be stripped of extension (Go parity: parseURLPath returns raw fid). - let info = match build_proxy_request_info(path, request_headers, &query_string) { + let info = match build_proxy_request_info(path, request_headers, &query_string, method) { Some(info) => info, None => return StatusCode::NOT_FOUND.into_response(), }; @@ -1308,6 +1391,7 @@ async fn check_download_limit( request_headers: &HeaderMap, path: &str, vid: VolumeId, + method: &Method, ) -> ControlFlow { if state.concurrent_download_limit <= 0 { return ControlFlow::Continue(false); @@ -1332,7 +1416,8 @@ async fn check_download_limit( .is_some_and(|(_, vol)| vol.super_block().replica_placement.has_replication()) }; if should_try_replica - && let Some(info) = build_proxy_request_info(path, request_headers, &query_string) + && let Some(info) = + build_proxy_request_info(path, request_headers, &query_string, method) { return ControlFlow::Break(proxy_or_redirect_to_target(state, info, vid, true).await); } @@ -5181,6 +5266,7 @@ mod tests { #[test] fn test_redirect_request_uses_outgoing_http_scheme() { let info = ProxyRequestInfo { + method: Method::GET, original_headers: HeaderMap::new(), original_query: "collection=photos&readDeleted=true".to_string(), path: "/3,01637037d6".to_string(), @@ -5209,6 +5295,7 @@ mod tests { #[test] fn test_redirect_request_strips_grpc_port_from_target_url() { let info = ProxyRequestInfo { + method: Method::GET, original_headers: HeaderMap::new(), original_query: String::new(), path: "/3,01637037d6".to_string(), @@ -5240,7 +5327,8 @@ mod tests { #[test] fn test_redirect_location_drops_extension_for_slash_form() { let info = - build_proxy_request_info("/3/01637037d6.jpg", &HeaderMap::new(), "").expect("parses"); + build_proxy_request_info("/3/01637037d6.jpg", &HeaderMap::new(), "", &Method::GET) + .expect("parses"); assert_eq!(info.vid_str, "3"); assert_eq!(info.fid_str, "01637037d6"); @@ -5334,6 +5422,66 @@ mod tests { server.abort(); } + /// The master answers a missing volume with 404 and a per-volume `error` + /// in the lookup body. Go's LookupResult.Error never becomes the Go error + /// return — the volume server treats it as empty locations and replies + /// 404. Surfacing it as Err would make proxy-mode reads answer 500. + /// A lookup answered with any other failure status (e.g. 503 while the + /// master warms up) stays an error, like Go's transport-level failure. + #[tokio::test] + async fn test_lookup_volume_missing_volume_returns_empty() { + use axum::{Router, routing::get}; + + let app = Router::new().route( + "/dir/lookup", + get(|axum::extract::Query(params): axum::extract::Query< + std::collections::HashMap, + >| async move { + let (status, error) = match params.get("volumeId").map(String::as_str) { + Some("7") => (StatusCode::NOT_FOUND, Some("volume id 7 not found")), + Some("9") => (StatusCode::SERVICE_UNAVAILABLE, None), + _ => ( + StatusCode::SERVICE_UNAVAILABLE, + Some("service warming up, please retry"), + ), + }; + let mut body = serde_json::json!({ + "volumeOrFileId": params.get("volumeId") + }); + if let Some(error) = error { + body["error"] = serde_json::json!(error); + } + (status, axum::Json(body)) + }), + ); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let master_url = format!("127.0.0.1:{}", addr.port()); + let client = reqwest::Client::new(); + + let locations = super::lookup_volume(&client, "http", &master_url, 7) + .await + .expect("missing volume is not a lookup failure"); + assert!(locations.is_empty()); + + let err = super::lookup_volume(&client, "http", &master_url, 8) + .await + .expect_err("master failure is a lookup failure"); + assert!(err.contains("warming up")); + + let err = super::lookup_volume(&client, "http", &master_url, 9) + .await + .expect_err("a failed lookup without an error body is still a failure"); + assert!(err.contains("503")); + + server.abort(); + } + /// A body streaming `data_size` bytes of `file` from offset 0. fn file_streaming_body( file: &std::fs::File, @@ -5392,6 +5540,7 @@ mod tests { }); let info = ProxyRequestInfo { + method: Method::GET, original_headers: HeaderMap::new(), original_query: String::new(), path: "/3,01637037d6".to_string(), @@ -5424,6 +5573,260 @@ mod tests { server.abort(); } + /// Go now forwards the inbound method (volume_server_handlers_read.go), so + /// a client HEAD reaches the replica as HEAD, not GET. + #[tokio::test] + async fn test_proxy_request_preserves_method() { + use axum::{Router, routing::get}; + use std::sync::Arc as StdArc; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + + let head_count = StdArc::new(AtomicUsize::new(0)); + let get_count = StdArc::new(AtomicUsize::new(0)); + let app = { + let head_count = head_count.clone(); + let get_count = get_count.clone(); + Router::new().route( + "/{*path}", + get(move |request: axum::extract::Request| { + let head_count = head_count.clone(); + let get_count = get_count.clone(); + async move { + if *request.method() == Method::HEAD { + head_count.fetch_add(1, AtomicOrdering::SeqCst); + } else { + get_count.fetch_add(1, AtomicOrdering::SeqCst); + } + StatusCode::OK + } + }), + ) + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let target = VolumeLocation { + url: addr.to_string(), + public_url: String::new(), + grpc_port: 0, + read_only: false, + read_only_can_delete: false, + }; + let state = streaming_test_state(); + for method in [Method::HEAD, Method::GET] { + let info = ProxyRequestInfo { + method, + original_headers: HeaderMap::new(), + original_query: String::new(), + path: "/3,01637037d6".to_string(), + vid_str: "3".to_string(), + fid_str: "01637037d6".to_string(), + }; + let response = proxy_request(&state, &info, &target).await; + assert_eq!(response.status(), StatusCode::OK); + } + assert_eq!(head_count.load(AtomicOrdering::SeqCst), 1); + assert_eq!(get_count.load(AtomicOrdering::SeqCst), 1); + + server.abort(); + } + + /// Go's query.Set replaces any proxied params the client sent, so the + /// forwarded query carries a single proxied=true marker — a stale + /// proxied=false cannot outrank it on the replica's first-value lookup. + #[tokio::test] + async fn test_proxy_request_replaces_stale_proxied_param() { + use axum::{Router, routing::get}; + use std::sync::Arc as StdArc; + use std::sync::Mutex; + + let seen = StdArc::new(Mutex::new(String::new())); + let app = { + let seen = seen.clone(); + Router::new().route( + "/{*path}", + get(move |request: axum::extract::Request| { + let seen = seen.clone(); + async move { + *seen.lock().unwrap() = + request.uri().query().unwrap_or("").to_string(); + StatusCode::OK + } + }), + ) + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let target = VolumeLocation { + url: addr.to_string(), + public_url: String::new(), + grpc_port: 0, + read_only: false, + read_only_can_delete: false, + }; + let state = streaming_test_state(); + let info = ProxyRequestInfo { + method: Method::GET, + original_headers: HeaderMap::new(), + original_query: "proxied=false&collection=photos&proxied".to_string(), + path: "/3,01637037d6".to_string(), + vid_str: "3".to_string(), + fid_str: "01637037d6".to_string(), + }; + let response = proxy_request(&state, &info, &target).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + seen.lock().unwrap().as_str(), + "collection=photos&proxied=true" + ); + + server.abort(); + } + + /// End-to-end headers like Range reach the replica; hop-by-hop headers do + /// not — a forwarded Content-Length or Host describes this hop, not the + /// replica's, and a lying Content-Length stalls the peer's body read. + #[tokio::test] + async fn test_proxy_request_drops_hop_by_hop_headers() { + use axum::{Router, routing::get}; + use std::sync::Arc as StdArc; + use std::sync::Mutex; + + let seen = StdArc::new(Mutex::new(Vec::new())); + let app = { + let seen = seen.clone(); + Router::new().route( + "/{*path}", + get(move |request: axum::extract::Request| { + let seen = seen.clone(); + async move { + let headers: Vec<(String, String)> = request + .headers() + .iter() + .map(|(k, v)| { + (k.as_str().to_string(), v.to_str().unwrap_or("").to_string()) + }) + .collect(); + seen.lock().unwrap().push(headers); + ( + [ + (header::CONNECTION, "keep-alive, x-resp-marker"), + (header::HeaderName::from_static("keep-alive"), "timeout=5"), + (header::HeaderName::from_static("x-resp-marker"), "private"), + (header::ETAG, "\"peer-etag\""), + ], + request + .headers() + .get(header::RANGE) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(), + ) + } + }), + ) + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let target = VolumeLocation { + url: addr.to_string(), + public_url: String::new(), + grpc_port: 0, + read_only: false, + read_only_can_delete: false, + }; + let mut original_headers = HeaderMap::new(); + original_headers.insert(header::RANGE, header::HeaderValue::from_static("bytes=0-3")); + original_headers.insert( + header::HOST, + header::HeaderValue::from_static("client-facing:8080"), + ); + original_headers.insert( + header::CONTENT_LENGTH, + header::HeaderValue::from_static("999"), + ); + original_headers.insert( + header::TRANSFER_ENCODING, + header::HeaderValue::from_static("chunked"), + ); + original_headers.insert( + header::CONNECTION, + header::HeaderValue::from_static("keep-alive, x-hop-marker"), + ); + original_headers.insert( + "x-hop-marker", + header::HeaderValue::from_static("private"), + ); + original_headers.insert( + "proxy-authorization", + header::HeaderValue::from_static("Basic cHJveHk6c2VjcmV0"), + ); + let info = ProxyRequestInfo { + method: Method::GET, + original_headers, + original_query: String::new(), + path: "/3,01637037d6".to_string(), + vid_str: "3".to_string(), + fid_str: "01637037d6".to_string(), + }; + let response = proxy_request(&streaming_test_state(), &info, &target).await; + assert_eq!(response.status(), StatusCode::OK); + + let seen = seen.lock().unwrap(); + let headers = seen.first().expect("peer saw a request"); + assert!( + headers + .iter() + .any(|(n, v)| n == "range" && v == "bytes=0-3"), + "Range was not forwarded: {headers:?}" + ); + for dropped in [ + "content-length", + "transfer-encoding", + "connection", + "x-hop-marker", + "proxy-authorization", + ] { + assert!( + !headers.iter().any(|(n, _)| n == dropped), + "hop-by-hop header {dropped} reached the replica: {headers:?}" + ); + } + // The peer must see its own host, not the client's. + let host = headers + .iter() + .find(|(n, _)| n == "host") + .map(|(_, v)| v.as_str()) + .expect("replica request has a host header"); + assert_eq!(host, addr.to_string()); + + // End-to-end response fields reach the client; the replica's + // hop-by-hop fields and Connection-nominated markers do not. + assert_eq!( + response.headers().get(header::ETAG).unwrap(), + "\"peer-etag\"" + ); + for dropped in ["connection", "keep-alive", "x-resp-marker"] { + assert!( + response.headers().get(dropped).is_none(), + "hop-by-hop response header {dropped} reached the client" + ); + } + + server.abort(); + } + fn streaming_test_state() -> Arc { use crate::storage::needle_map::NeedleMapKind; use crate::storage::store::Store; @@ -7182,7 +7585,7 @@ mod tests { header::RANGE, header::HeaderValue::from_bytes(b"bytes=0-1\xff").unwrap(), ); - let info = build_proxy_request_info(&path, &headers, "").unwrap(); + let info = build_proxy_request_info(&path, &headers, "", &Method::GET).unwrap(); let resp = proxy_request(&proxy_state, &info, &target).await; assert_eq!(resp.status(), StatusCode::RANGE_NOT_SATISFIABLE); diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go index 6719f289b0..a82504277a 100644 --- a/weed/operation/lookup.go +++ b/weed/operation/lookup.go @@ -79,7 +79,11 @@ func LookupFileId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, fileId s } func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) { - results, err := LookupVolumeIds(masterFn, grpcDialOption, []string{vid}) + return LookupVolumeIdWithContext(context.Background(), masterFn, grpcDialOption, vid) +} + +func LookupVolumeIdWithContext(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid string) (*LookupResult, error) { + results, err := LookupVolumeIdsWithContext(ctx, masterFn, grpcDialOption, []string{vid}) return results[vid], err } @@ -90,6 +94,12 @@ func InvalidateVolumeIdLocationCache(vid string) { // LookupVolumeIds find volume locations by cache and actual lookup func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string, useCache ...bool) (map[string]*LookupResult, error) { + return LookupVolumeIdsWithContext(context.Background(), masterFn, grpcDialOption, vids, useCache...) +} + +// LookupVolumeIdsWithContext is LookupVolumeIds bound to ctx, so a caller's +// cancellation (e.g. a dropped inbound request) also stops the master query. +func LookupVolumeIdsWithContext(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string, useCache ...bool) (map[string]*LookupResult, error) { ret := make(map[string]*LookupResult) var unknown_vids []string cacheLocations := len(useCache) == 0 || useCache[0] @@ -112,12 +122,12 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids //only query unknown_vids - err := WithMasterServerClient(context.Background(), false, masterFn(context.Background()), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { + err := WithMasterServerClient(ctx, false, masterFn(ctx), grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.LookupVolumeRequest{ VolumeOrFileIds: unknown_vids, } - resp, grpcErr := masterClient.LookupVolume(context.Background(), req) + resp, grpcErr := masterClient.LookupVolume(ctx, req) if grpcErr != nil { return grpcErr } diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go index 16ab5bf96a..99c51cb851 100644 --- a/weed/server/volume_server_handlers_read.go +++ b/weed/server/volume_server_handlers_read.go @@ -12,6 +12,7 @@ import ( "net/http" "net/url" "path/filepath" + "slices" "strconv" "strings" "sync/atomic" @@ -56,6 +57,27 @@ func InternalError(w http.ResponseWriter, r *http.Request, err error) { writeJsonError(w, r, http.StatusInternalServerError, err) } +// hopByHopHeaders names the fields that describe a single hop and must not +// be relayed: the fixed hop-by-hop set plus every field the header's own +// Connection values nominate (RFC 9110 7.6.1). +func hopByHopHeaders(h http.Header, extra ...string) map[string]bool { + drop := map[string]bool{ + "Connection": true, "Keep-Alive": true, "Te": true, "Trailer": true, + "Upgrade": true, "Proxy-Connection": true, "Proxy-Authenticate": true, "Proxy-Authorization": true, + } + for _, v := range h.Values("Connection") { + for _, token := range strings.Split(v, ",") { + if name := http.CanonicalHeaderKey(strings.TrimSpace(token)); name != "" { + drop[name] = true + } + } + } + for _, name := range extra { + drop[http.CanonicalHeaderKey(name)] = true + } + return drop +} + func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Request) { vid, fid, _, _, _ := parseURLPath(r.URL.Path) volumeId, err := needle.NewVolumeId(vid) @@ -64,17 +86,28 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re w.WriteHeader(http.StatusBadRequest) return } - lookupResult, err := operation.LookupVolumeId(vs.GetMaster, vs.grpcDialOption, volumeId.String()) - if err != nil || len(lookupResult.Locations) <= 0 { + lookupResult, err := operation.LookupVolumeIdWithContext(r.Context(), vs.GetMaster, vs.grpcDialOption, volumeId.String()) + if err != nil && vs.ReadMode == "proxy" { + if r.Context().Err() != nil { + return + } + glog.V(0).InfofCtx(r.Context(), "lookup error: %v %s", err, r.URL.Path) + InternalError(w, r, nil) + return + } + if err != nil || lookupResult == nil || len(lookupResult.Locations) <= 0 { glog.V(0).Infoln("lookup error:", err, r.URL.Path) stats.VolumeServerFileReadInvalidNeedles.Inc() NotFound(w) return } if len(lookupResult.Locations) >= 2 { - rand.Shuffle(len(lookupResult.Locations), func(i, j int) { - lookupResult.Locations[i], lookupResult.Locations[j] = lookupResult.Locations[j], lookupResult.Locations[i] + // cached locations are shared between lookups, so shuffle a copy + locations := slices.Clone(lookupResult.Locations) + rand.Shuffle(len(locations), func(i, j int) { + locations[i], locations[j] = locations[j], locations[i] }) + lookupResult.Locations = locations } var targetUrl *url.URL location := fmt.Sprintf("%s:%d", vs.store.Ip, vs.store.Port) @@ -100,13 +133,19 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re query := r.URL.Query() query.Set(reqIsProxied, "true") r.URL.RawQuery = query.Encode() - request, err := http.NewRequest(http.MethodGet, r.URL.String(), nil) + request, err := http.NewRequestWithContext(r.Context(), r.Method, r.URL.String(), nil) if err != nil { glog.V(0).Infof("failed to instance http request of url %s: %v", redactedURL(r), err) InternalError(w, r, nil) return } + // the proxied request carries no body, so it must not forward this + // hop's framing either + hopByHop := hopByHopHeaders(r.Header, "Host", "Content-Length", "Transfer-Encoding") for k, vv := range r.Header { + if hopByHop[http.CanonicalHeaderKey(k)] { + continue + } for _, v := range vv { request.Header.Add(k, v) } @@ -114,6 +153,9 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re response, err := util_http.GetGlobalHttpClient().Do(request) if err != nil { + if r.Context().Err() != nil { + return + } stats.VolumeServerHandlerCounter.WithLabelValues(stats.FailedReadProxyReq).Inc() glog.V(0).Infof("request remote url %s: %v", redactedURL(r), err) InternalError(w, r, nil) @@ -121,8 +163,9 @@ func (vs *VolumeServer) proxyReqToTargetServer(w http.ResponseWriter, r *http.Re } defer util_http.CloseResponse(response) // proxy target response to client + respHopByHop := hopByHopHeaders(response.Header) for k, vv := range response.Header { - if k == "Server" { + if k == "Server" || respHopByHop[http.CanonicalHeaderKey(k)] { continue } for _, v := range vv { @@ -189,7 +232,7 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) hasVolume := vs.store.HasVolume(volumeId) _, hasEcVolume := vs.store.FindEcVolume(volumeId) if !hasVolume && !hasEcVolume { - if vs.ReadMode == "local" { + if vs.ReadMode == "local" || r.URL.Query().Get(reqIsProxied) == "true" { stats.VolumeServerFileReadInvalidNeedles.Inc() glog.V(0).Infoln("volume is not local:", err, r.URL.Path) NotFound(w) diff --git a/weed/server/volume_server_handlers_read_test.go b/weed/server/volume_server_handlers_read_test.go index 1402a9954d..769f870eb2 100644 --- a/weed/server/volume_server_handlers_read_test.go +++ b/weed/server/volume_server_handlers_read_test.go @@ -1,12 +1,26 @@ package weed_server import ( + "context" "errors" + "io" + "net" "net/http" "net/http/httptest" "strings" + "sync" "testing" "time" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) func TestInternalErrorWritesTheReadError(t *testing.T) { @@ -96,3 +110,323 @@ func TestNotModifiedChecksIfNoneMatchFirst(t *testing.T) { }) } } + +type proxyTestMaster struct { + master_pb.UnimplementedSeaweedServer + lookup func(context.Context, *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) +} + +func (m *proxyTestMaster) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + return m.lookup(ctx, req) +} + +func newProxyTestHTTPServer(t *testing.T, handler http.Handler) *httptest.Server { + t.Helper() + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := &httptest.Server{Listener: listener, Config: &http.Server{Handler: handler}} + server.Start() + t.Cleanup(server.Close) + return server +} + +type proxyTestResponseWriter struct { + http.ResponseWriter + statusWritten chan struct{} +} + +func (w proxyTestResponseWriter) WriteHeader(status int) { + w.ResponseWriter.WriteHeader(status) + w.statusWritten <- struct{}{} +} + +func newProxyTestVolumeServer(t *testing.T, volumeID string, locations []*master_pb.Location, lookupErr error) *VolumeServer { + t.Helper() + operation.InvalidateVolumeIdLocationCache(volumeID) + + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + masterAddress := pb.NewServerAddress("127.0.0.1", 9333, listener.Addr().(*net.TCPAddr).Port) + masterServer := grpc.NewServer() + master_pb.RegisterSeaweedServer(masterServer, &proxyTestMaster{lookup: func(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + if lookupErr != nil { + return nil, lookupErr + } + if len(req.VolumeOrFileIds) == 0 { + return nil, errors.New("lookup request contains no volume IDs") + } + return &master_pb.LookupVolumeResponse{VolumeIdLocations: []*master_pb.LookupVolumeResponse_VolumeIdLocation{{ + VolumeOrFileId: req.VolumeOrFileIds[0], + Locations: locations, + }}}, nil + }}) + go func() { + _ = masterServer.Serve(listener) + }() + + vs := &VolumeServer{ + store: &storage.Store{Ip: "192.0.2.10", Port: 8080}, + ReadMode: "proxy", + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + vs.setCurrentMaster(masterAddress) + t.Cleanup(func() { + operation.InvalidateVolumeIdLocationCache(volumeID) + pb.InvalidateGrpcConnection(masterAddress.ToGrpcAddress()) + masterServer.Stop() + }) + return vs +} + +func TestProxyReqToTargetServerPreservesMethodAndResponse(t *testing.T) { + const volumeID = "849300101" + type observedRequest struct { + method string + rangeHeader string + ifNoneMatch string + ifModifiedAt string + header http.Header + } + observed := make(chan observedRequest, 2) + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + observed <- observedRequest{ + method: r.Method, + rangeHeader: r.Header.Get("Range"), + ifNoneMatch: r.Header.Get("If-None-Match"), + ifModifiedAt: r.Header.Get("If-Modified-Since"), + header: r.Header.Clone(), + } + w.Header().Set("ETag", `"peer-etag"`) + w.Header().Set("Content-Range", "bytes 0-3/10") + w.Header().Set("X-Peer-Header", "preserved") + w.Header().Set("Keep-Alive", "timeout=5") + w.Header().Set("X-Resp-Marker", "private") + w.Header()["Connection"] = []string{"keep-alive, X-Resp-Marker"} + w.WriteHeader(http.StatusPartialContent) + _, _ = io.WriteString(w, "data") + })) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + r := httptest.NewRequest(method, "/"+volumeID+",abcdef", nil) + r.Header.Set("Range", "bytes=0-3") + r.Header.Set("If-None-Match", `"client-etag"`) + r.Header.Set("If-Modified-Since", "Wed, 21 Oct 2015 07:28:00 GMT") + r.Header.Set("Connection", "keep-alive, X-Hop-Marker") + r.Header.Set("Keep-Alive", "timeout=5") + r.Header.Set("Upgrade", "h2c") + r.Header.Set("Te", "trailers") + r.Header.Set("X-Hop-Marker", "private") + r.Header.Set("Proxy-Connection", "keep-alive") + r.Header.Set("Proxy-Authorization", "Basic cHJveHk6c2VjcmV0") + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, r) + + var got observedRequest + select { + case got = <-observed: + case <-time.After(2 * time.Second): + t.Fatalf("peer did not receive the proxied request; proxy status %d", w.Code) + } + if got.method != method { + t.Errorf("peer received method %q, want %q", got.method, method) + } + if got.rangeHeader != "bytes=0-3" || got.ifNoneMatch != `"client-etag"` || got.ifModifiedAt != "Wed, 21 Oct 2015 07:28:00 GMT" { + t.Errorf("peer received conditional/range headers %+v", got) + } + for _, dropped := range []string{"Connection", "Keep-Alive", "Upgrade", "Te", "X-Hop-Marker", "Proxy-Connection", "Proxy-Authorization"} { + if got.header.Get(dropped) != "" { + t.Errorf("hop-by-hop header %s reached the replica: %v", dropped, got.header) + } + } + if w.Code != http.StatusPartialContent { + t.Errorf("status %d, want %d", w.Code, http.StatusPartialContent) + } + if w.Header().Get("ETag") != `"peer-etag"` || w.Header().Get("Content-Range") != "bytes 0-3/10" || w.Header().Get("X-Peer-Header") != "preserved" { + t.Errorf("peer response headers were not preserved: %v", w.Header()) + } + for _, dropped := range []string{"Connection", "Keep-Alive", "X-Resp-Marker"} { + if w.Header().Get(dropped) != "" { + t.Errorf("hop-by-hop response header %s reached the client: %v", dropped, w.Header()) + } + } + wantBody := "data" + if method == http.MethodHead { + wantBody = "" + } + if gotBody := w.Body.String(); gotBody != wantBody { + t.Errorf("response body %q, want %q", gotBody, wantBody) + } + }) + } +} + +func TestProxyReqToTargetServerCancelsUpstreamWithInboundRequest(t *testing.T) { + const volumeID = "849300102" + peerStarted := make(chan struct{}) + peerCanceled := make(chan struct{}) + releasePeer := make(chan struct{}) + var releasePeerOnce sync.Once + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(peerStarted) + select { + case <-r.Context().Done(): + close(peerCanceled) + case <-releasePeer: + } + })) + defer releasePeerOnce.Do(func() { close(releasePeer) }) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r := httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil).WithContext(ctx) + w := httptest.NewRecorder() + proxyDone := make(chan struct{}) + go func() { + vs.proxyReqToTargetServer(w, r) + close(proxyDone) + }() + + select { + case <-peerStarted: + case <-time.After(2 * time.Second): + t.Fatal("peer did not receive the proxied request") + } + cancel() + select { + case <-peerCanceled: + case <-time.After(time.Second): + t.Error("canceling the inbound request did not cancel the peer request") + } + releasePeerOnce.Do(func() { close(releasePeer) }) + select { + case <-proxyDone: + case <-time.After(2 * time.Second): + t.Error("proxy handler did not finish after inbound cancellation") + } +} + +func TestProxyReqToTargetServerCancelsUpstreamBodyReadWithInboundRequest(t *testing.T) { + const volumeID = "849300106" + peerCanceled := make(chan struct{}) + releasePeer := make(chan struct{}) + var releasePeerOnce sync.Once + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + close(peerCanceled) + case <-releasePeer: + } + })) + defer releasePeerOnce.Do(func() { close(releasePeer) }) + + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + r := httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil) + ctx, cancel := context.WithCancel(r.Context()) + r = r.WithContext(ctx) + recorder := httptest.NewRecorder() + statusWritten := make(chan struct{}, 1) + w := proxyTestResponseWriter{ResponseWriter: recorder, statusWritten: statusWritten} + proxyDone := make(chan struct{}) + go func() { + vs.proxyReqToTargetServer(w, r) + close(proxyDone) + }() + + select { + case <-statusWritten: + case <-time.After(2 * time.Second): + cancel() + t.Fatal("proxy did not receive the peer response headers") + } + cancel() + select { + case <-peerCanceled: + case <-time.After(time.Second): + t.Error("canceling the inbound request did not cancel the peer response body") + } + releasePeerOnce.Do(func() { close(releasePeer) }) + select { + case <-proxyDone: + case <-time.After(2 * time.Second): + t.Error("proxy handler did not finish reading after inbound cancellation") + } +} + +func TestProxyReqToTargetServerDistinguishesLookupFailureFromNoLocation(t *testing.T) { + t.Run("cancelled request skips the lookup error response", func(t *testing.T) { + const volumeID = "849300107" + vs := newProxyTestVolumeServer(t, volumeID, nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil).WithContext(ctx)) + if w.Code != http.StatusOK { + t.Errorf("status %d, want no response written for a cancelled request", w.Code) + } + }) + + t.Run("master lookup failure is internal error", func(t *testing.T) { + const volumeID = "849300103" + vs := newProxyTestVolumeServer(t, volumeID, nil, status.Error(codes.Unavailable, "master unavailable")) + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil)) + if w.Code != http.StatusInternalServerError { + t.Errorf("status %d, want %d", w.Code, http.StatusInternalServerError) + } + }) + + for _, tc := range []struct { + name string + locations []*master_pb.Location + }{ + {name: "lookup returned no replica locations"}, + {name: "only local replica is not a usable proxy target", locations: []*master_pb.Location{{Url: "192.0.2.10:8080"}}}, + } { + t.Run(tc.name, func(t *testing.T) { + volumeID := "84930010" + if tc.name == "lookup returned no replica locations" { + volumeID += "4" + } else { + volumeID += "5" + } + vs := newProxyTestVolumeServer(t, volumeID, tc.locations, nil) + w := httptest.NewRecorder() + vs.proxyReqToTargetServer(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",abcdef", nil)) + if w.Code != http.StatusNotFound { + t.Errorf("status %d, want %d", w.Code, http.StatusNotFound) + } + }) + } +} + +// A request that already carries proxied=true must not be proxied again: a +// stale lookup pointing at servers that already answered would otherwise +// bounce the read between them until the client gives up. +func TestGetOrHeadHandlerDoesNotReproxyAProxiedRequest(t *testing.T) { + const volumeID = "849300106" + peer := newProxyTestHTTPServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("a proxied request must not be forwarded again") + })) + remoteAddress := strings.TrimPrefix(peer.URL, "http://") + vs := newProxyTestVolumeServer(t, volumeID, []*master_pb.Location{{Url: remoteAddress}}, nil) + vs.guard = security.NewGuard(nil, "", 0, "", 0) + + w := httptest.NewRecorder() + vs.GetOrHeadHandler(w, httptest.NewRequest(http.MethodGet, "/"+volumeID+",01637037d6?proxied=true", nil)) + if w.Code != http.StatusNotFound { + t.Errorf("status %d, want %d", w.Code, http.StatusNotFound) + } +}