From 5c92a54a1229f98c834d63245dd2ec75a95ae137 Mon Sep 17 00:00:00 2001 From: Javier de la Torre Date: Thu, 29 Oct 2026 08:57:37 +0100 Subject: [PATCH] remote: reuse one guarded remote storage client per configuration (#12189) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * remote: reuse one guarded remote storage client per configuration BuildGuardedRemoteStorageClient (the filer's lazy remote stat and list, remote deletes and renames, the volume server's cold reads, the S3 gateway's remote reads) built a new client, with its own http.Transport, on every call. Each transport kept its connection idle for IdleConnTimeout (60s) and was never reused or closed, so a steady stream of remote lookups held one open connection per request. An endpoint that caps connections per source then refuses every new connection. - remote_storage: GetRemoteStorage's per-name cache becomes a bounded LRU (WEED_REMOTE_CLIENT_CACHE_MAX, default 64) keyed by kind, name, a hash of the whole configuration and a credentials version, built outside the lock once per key (singleflight) from a copy of the configuration. A client leaving the cache has its idle connections closed. GetRemoteStorageWith exposes the cache to other builders. - BuildGuardedRemoteStorageClient caches the guarded client per configuration. The endpoint is checked against the deny-list when a client is built; every new connection still dials through the guarded dialer. A gcs credentials file, and for an S3-compatible remote using the default credential chain the shared credentials file and AWS_* variables, are part of the key by content. - The shared transport keeps 32 idle connections per host, caps connections per host (WEED_REMOTE_MAX_CONNS_PER_HOST, default 64) and has dial, TLS handshake and response header deadlines, so a hung remote frees its slots. - The filer drops the clients of removed remotes (and all but the newest client per name) on every /etc/remote reload, also when /etc/remote does not exist; reloads are serialized. * remote: drop clients whose build spans a reload or whose remote is gone * server: hash the shared AWS config file into the guarded credential version * server: bound a stalled body on the shared remote transport * remote: waive path-injection on the configured credentials file * remote: keep caching remotes the local filer does not name * remote: drop the inline codeql waiver the analyzer ignores * remote: version the plain gcs client's credentials file * remote: hash the environment's inline gcs credentials GOOGLE_APPLICATION_CREDENTIALS can hold the JSON itself, which the gcs client accepts; it is not in the configuration hash, so a changed value kept the old client. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * remote: version the well-known gcs credentials file With neither the configuration nor GOOGLE_APPLICATION_CREDENTIALS set, the SDK reads application_default_credentials.json; rotating it left the cache key unchanged and the old client behind. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * remote: share the gcs credentials version with the guarded client The guarded gcs client resolves the same environment and well-known credentials as the plain one; keying it only on a configured path left those rotations invisible to the cache. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * remote: version s3 keys resolved from the environment A remote whose access and secret keys both arrive through its type's environment fallback (aliyun, baidu, contabo, filebase, storj, tencent) never entered the default-chain branch of guardedCredentialsVersion, so an in-process environment change kept serving the stale cached client. The version now hashes the resolved keys for every S3-compatible remote. * volume: reuse one remote storage client per configuration (rust) fetch_and_write_needle built a fresh aws_sdk_s3::Client per request, so every cold read kept its connector's idle connections without ever reusing them — the same leak the Go side fixes. Cache clients keyed by the configuration's protobuf encoding in an LRU of WEED_REMOTE_CLIENT_CACHE_MAX entries (default 64), mirroring weed/remote_storage/client_cache.go. An evicted entry drops the client's last Arc and with it the connector's idle connections. The per-request endpoint validation is unchanged: Rust has no dial-time guard yet, so the up-front DNS check stays on every call. --------- Co-authored-by: user.email Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../src/remote_storage/client_cache.rs | 160 +++++++ seaweed-volume/src/remote_storage/mod.rs | 2 + seaweed-volume/src/server/grpc_server.rs | 6 +- weed/filer/remote_storage.go | 15 + weed/filer/remote_storage_test.go | 48 +++ weed/remote_storage/client_cache.go | 301 +++++++++++++ weed/remote_storage/remote_storage.go | 35 +- .../remote_storage_cache_test.go | 408 ++++++++++++++++++ weed/remote_storage/s3/default_chain_test.go | 29 ++ weed/remote_storage/s3/s3_storage_client.go | 18 + weed/server/volume_grpc_remote.go | 179 +++++++- weed/server/volume_grpc_remote_test.go | 206 +++++++-- .../server/volume_grpc_remote_timeout_test.go | 159 +++++++ 13 files changed, 1482 insertions(+), 84 deletions(-) create mode 100644 seaweed-volume/src/remote_storage/client_cache.rs create mode 100644 weed/remote_storage/client_cache.go create mode 100644 weed/remote_storage/remote_storage_cache_test.go create mode 100644 weed/remote_storage/s3/default_chain_test.go create mode 100644 weed/server/volume_grpc_remote_timeout_test.go diff --git a/seaweed-volume/src/remote_storage/client_cache.rs b/seaweed-volume/src/remote_storage/client_cache.rs new file mode 100644 index 0000000000..58da6f7b42 --- /dev/null +++ b/seaweed-volume/src/remote_storage/client_cache.rs @@ -0,0 +1,160 @@ +//! One remote storage client per configuration. Mirrors Go's +//! weed/remote_storage/client_cache.go. +//! +//! A client built per request keeps its connector's idle connections for the +//! pool's idle timeout and never reuses them, so a steady stream of cold reads +//! (fetch_and_write_needle) holds one open connection per request. Clients are +//! cached instead, keyed by the configuration's protobuf encoding (RemoteConf +//! holds no maps, so the encoding is deterministic), in an LRU of +//! WEED_REMOTE_CLIENT_CACHE_MAX entries (default 64). An evicted entry drops +//! the client's Arc; dropping its last Arc drops the connector and its idle +//! connections. The Go cache's retain hook and credentials version do not +//! apply here: the volume server never loads /etc/remote, and Rust +//! credentials come only from the configuration itself. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; + +use prost::Message; + +use crate::pb::remote_pb::RemoteConf; + +use super::{RemoteStorageClient, RemoteStorageError, make_remote_storage_client}; + +struct CachedClient { + client: Arc, + last_used: u64, +} + +struct CacheInner { + max: usize, + ticks: u64, + entries: HashMap, CachedClient>, +} + +impl CacheInner { + fn new(max: usize) -> Self { + CacheInner { + max, + ticks: 0, + entries: HashMap::new(), + } + } + + fn get( + &mut self, + conf: &RemoteConf, + ) -> Result, RemoteStorageError> { + self.ticks += 1; + let tick = self.ticks; + let key = conf.encode_to_vec(); + if let Some(entry) = self.entries.get_mut(&key) { + entry.last_used = tick; + return Ok(entry.client.clone()); + } + // The build is a pure allocation (the connector dials lazily), so it + // runs under the lock; the lock itself serializes concurrent builds of + // one key. + let client: Arc = Arc::from(make_remote_storage_client(conf)?); + self.entries.insert( + key, + CachedClient { + client: client.clone(), + last_used: tick, + }, + ); + while self.entries.len() > self.max { + if let Some(oldest) = self + .entries + .iter() + .min_by_key(|(_, e)| e.last_used) + .map(|(k, _)| k.clone()) + { + self.entries.remove(&oldest); + } else { + break; + } + } + Ok(client) + } +} + +fn lock_inner(cache: &Mutex) -> MutexGuard<'_, CacheInner> { + cache.lock().unwrap_or_else(|e| e.into_inner()) +} + +static CACHE: OnceLock> = OnceLock::new(); + +fn client_cache_max() -> usize { + match std::env::var("WEED_REMOTE_CLIENT_CACHE_MAX") { + Ok(v) => match v.parse::() { + Ok(n) if n > 0 => n, + _ => { + tracing::warn!( + "WEED_REMOTE_CLIENT_CACHE_MAX={:?} is not a positive number; using 64", + v + ); + 64 + } + }, + Err(_) => 64, + } +} + +/// Return the cached client of a configuration, building and caching it on a +/// miss. Mirrors Go's remote_storage.GetRemoteStorage. +pub fn get_remote_storage_client( + conf: &RemoteConf, +) -> Result, RemoteStorageError> { + lock_inner(CACHE.get_or_init(|| Mutex::new(CacheInner::new(client_cache_max())))).get(conf) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn s3_conf(name: &str, secret: &str) -> RemoteConf { + RemoteConf { + r#type: "s3".to_string(), + name: name.to_string(), + s3_endpoint: "https://s3.example".to_string(), + s3_access_key: "k".to_string(), + s3_secret_key: secret.to_string(), + s3_region: "us-east-1".to_string(), + ..Default::default() + } + } + + #[test] + fn same_configuration_one_client() { + let mut cache = CacheInner::new(64); + let first = cache.get(&s3_conf("a", "x")).unwrap(); + for _ in 0..5 { + let again = cache.get(&s3_conf("a", "x")).unwrap(); + assert!(Arc::ptr_eq(&first, &again)); + } + } + + #[test] + fn changed_configuration_new_client() { + let mut cache = CacheInner::new(64); + let first = cache.get(&s3_conf("a", "x")).unwrap(); + let next = cache.get(&s3_conf("a", "y")).unwrap(); + assert!(!Arc::ptr_eq(&first, &next)); + } + + #[test] + fn lru_bound_evicts_least_used() { + let mut cache = CacheInner::new(2); + let a = cache.get(&s3_conf("a", "x")).unwrap(); + let b = cache.get(&s3_conf("b", "x")).unwrap(); + // touch a so b is the least recently used + cache.get(&s3_conf("a", "x")).unwrap(); + cache.get(&s3_conf("c", "x")).unwrap(); + assert_eq!(cache.entries.len(), 2); + let a2 = cache.get(&s3_conf("a", "x")).unwrap(); + assert!(Arc::ptr_eq(&a, &a2)); + let b2 = cache.get(&s3_conf("b", "x")).unwrap(); + assert!(!Arc::ptr_eq(&b, &b2)); + } +} diff --git a/seaweed-volume/src/remote_storage/mod.rs b/seaweed-volume/src/remote_storage/mod.rs index b22393e506..d8fab18f47 100644 --- a/seaweed-volume/src/remote_storage/mod.rs +++ b/seaweed-volume/src/remote_storage/mod.rs @@ -3,10 +3,12 @@ //! Provides a trait-based abstraction over cloud storage providers (S3, GCS, Azure, etc.) //! and a registry to create clients from protobuf RemoteConf messages. +pub mod client_cache; pub mod endpoint_guard; pub mod s3; pub mod s3_tier; +pub use client_cache::get_remote_storage_client; pub use endpoint_guard::{guarded_tcp_connect, validate_remote_endpoint, validate_replica_target}; use crate::pb::remote_pb::{RemoteConf, RemoteStorageLocation}; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index e0b4507914..cd86ac1bf4 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -4988,9 +4988,11 @@ impl VolumeServer for VolumeGrpcService { .map_err(|e| Status::invalid_argument(format!("reject remote endpoint: {}", e)))?; } - // Create remote storage client + // One client per configuration: a client built per request kept its + // connector's idle connections and never reused them. Mirrors the Go + // volume server. let client = - crate::remote_storage::make_remote_storage_client(remote_conf).map_err(|e| { + crate::remote_storage::get_remote_storage_client(remote_conf).map_err(|e| { Status::internal(format!( "get remote client: make remote storage client {}: {}", remote_conf.name, e, diff --git a/weed/filer/remote_storage.go b/weed/filer/remote_storage.go index e1733ce777..1ea416ab41 100644 --- a/weed/filer/remote_storage.go +++ b/weed/filer/remote_storage.go @@ -35,6 +35,7 @@ type FilerRemoteStorage struct { // guards rules and storageNameToConf, which are replaced wholesale // whenever /etc/remote changes mu sync.RWMutex + reloadMu sync.Mutex // serializes reloads, so an older listing never overwrites a newer one rules ptrie.Trie[*remote_pb.RemoteStorageLocation] mountDirs []util.FullPath storageNameToConf map[string]*remote_pb.RemoteConf @@ -74,11 +75,18 @@ func (rs *FilerRemoteStorage) SetConfValidator(v RemoteStorageConfValidator) { func (rs *FilerRemoteStorage) LoadRemoteStorageConfigurationsAndMapping(filer *Filer) (err error) { // execute this on filer + // reloads run one at a time, so the listing a reload applies (and the cached + // clients it retains) is never older than one already applied + rs.reloadMu.Lock() + defer rs.reloadMu.Unlock() + limit := int64(math.MaxInt32) entries, _, err := filer.ListDirectoryEntries(context.Background(), DirectoryEtcRemote, "", false, limit, "", "", "") if err != nil { if err == filer_pb.ErrNotFound { + // no /etc/remote: no remote is configured, so none keeps a client + remote_storage.RetainRemoteStorages(nil) return nil } glog.Errorf("read remote storage %s: %v", DirectoryEtcRemote, err) @@ -123,6 +131,13 @@ func (rs *FilerRemoteStorage) LoadRemoteStorageConfigurationsAndMapping(filer *F rs.rules, rs.mountDirs, rs.storageNameToConf = rules, mountDirs, storageNameToConf rs.mu.Unlock() + // a remote removed from /etc/remote keeps no cached client and no open connections + names := make(map[string]bool, len(storageNameToConf)) + for name := range storageNameToConf { + names[name] = true + } + remote_storage.RetainRemoteStorages(names) + return nil } diff --git a/weed/filer/remote_storage_test.go b/weed/filer/remote_storage_test.go index 9ef2dd5e83..5cb64124f0 100644 --- a/weed/filer/remote_storage_test.go +++ b/weed/filer/remote_storage_test.go @@ -4,7 +4,9 @@ import ( "context" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -129,3 +131,49 @@ func TestLoadRemoteStorageConfigurationsAndMapping_RejectsBlockedEndpoint(t *tes type assertError string func (e assertError) Error() string { return string(e) } + +// Every reload drops the cached clients of remotes that are no longer +// configured, also when /etc/remote is gone entirely, so a removed remote keeps +// no client and no open connections. +func TestLoadRemoteStorageConfigurationsAndMapping_DropsClientsOfRemovedRemotes(t *testing.T) { + remote_storage.ResetRemoteStorageClients() + t.Cleanup(remote_storage.ResetRemoteStorageClients) + closed := map[string]int{} + cache := func(name string) { + conf := &remote_pb.RemoteConf{Name: name, Type: "s3", S3Endpoint: "https://s3.example.com", S3Region: "us-east-1"} + _, err := remote_storage.GetRemoteStorageWith("guarded", conf, "", func(*remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, func(), error) { + return &stubRemoteClient{}, func() { closed[name]++ }, nil + }) + require.NoError(t, err) + } + + // /etc/remote holds "good" only: the client of "gone" is dropped and closed, "good"'s kept + goodBytes, _ := proto.Marshal(&remote_pb.RemoteConf{Name: "good", Type: "s3", S3Endpoint: "https://s3.example.com", S3Region: "us-east-1"}) + store := newStubFilerStore() + store.entries[string(DirectoryEtcRemote)+"/good.conf"] = &Entry{ + FullPath: util.FullPath(string(DirectoryEtcRemote) + "/good.conf"), Attr: Attr{Mode: 0644}, Content: goodBytes, + } + cache("good") + cache("gone") + f := newTestFiler(t, store, NewFilerRemoteStorage()) + require.NoError(t, f.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(f)) + assert.Zero(t, closed["good"], "a configured remote keeps its client") + assert.Positive(t, closed["gone"], "a removed remote's client is closed") + + // no /etc/remote at all (the listing answers ErrNotFound): every remote is gone + empty := newTestFiler(t, newStubFilerStore(), NewFilerRemoteStorage()) + empty.Store = NewFilerStoreWrapper(etcRemoteNotFoundStore{newStubFilerStore()}) + require.NoError(t, empty.RemoteStorage.LoadRemoteStorageConfigurationsAndMapping(empty)) + assert.Positive(t, closed["good"], "with no /etc/remote, no remote keeps a client") +} + +// a store in which /etc/remote does not exist: its listing answers filer_pb.ErrNotFound +type etcRemoteNotFoundStore struct{ *stubFilerStore } + +func (s etcRemoteNotFoundStore) ListDirectoryEntries(context.Context, util.FullPath, string, bool, int64, ListEachEntryFunc) (string, error) { + return "", filer_pb.ErrNotFound +} + +func (s etcRemoteNotFoundStore) ListDirectoryPrefixedEntries(context.Context, util.FullPath, string, bool, int64, string, ListEachEntryFunc) (string, error) { + return "", filer_pb.ErrNotFound +} diff --git a/weed/remote_storage/client_cache.go b/weed/remote_storage/client_cache.go new file mode 100644 index 0000000000..5fed1db67c --- /dev/null +++ b/weed/remote_storage/client_cache.go @@ -0,0 +1,301 @@ +package remote_storage + +import ( + "container/list" + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "os/user" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "time" + + "golang.org/x/sync/singleflight" + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// Remote storage clients are cached, one per (kind, configuration, +// credentials version). +// +// Each client carries its own HTTP transport. A client built per request keeps +// its connection open for the transport's idle timeout and never reuses it, so +// a steady stream of lazy remote stats, cold reads or remote deletes holds one +// open connection per request; a storage endpoint that caps connections per +// source then refuses every new one. So clients are cached: +// - the key is the kind ("" for the plain client of GetRemoteStorage, +// "guarded" for the servers' SSRF-guarded one; the two are never shared), +// the configuration's name, a hash of the whole configuration and a +// credentials version (the content hash of a credentials file the +// configuration points at, which can change while the configuration does +// not). Two configurations under one name are two entries; neither evicts +// the other. +// - the cache is an LRU of WEED_REMOTE_CLIENT_CACHE_MAX entries (default 64). +// - a client is built outside the cache's lock, once per key at a time +// (singleflight): a slow build (a DNS check, a credentials file) never +// stalls the lookups of other keys. +// - a client leaving the cache (evicted, or dropped by RetainRemoteStorages) +// has its idle connections closed at once and again after a grace period, +// for the connections its in-flight requests return to the pool afterwards. +// - RetainRemoteStorages (every filer reload) also retires, per kind and +// name, every entry but the newest: a key rotation, a new endpoint or a new +// credentials file leaves no client of the old configuration behind. A +// build finishing after a reload never enters the cache either: it is +// retired instead when a reload passed while it was running. +// - the cached client is built from a copy of the configuration, so a caller +// changing its value later changes nothing. + +// ClientBuild builds the client of a configuration and, optionally, a function that closes its idle connections. +type ClientBuild func(conf *remote_pb.RemoteConf) (client RemoteStorageClient, closeIdle func(), err error) + +type cachedClient struct { + key string + kind string + name string + born uint64 // insertion order: retain keeps only the newest per kind and name + generation uint64 // the retain generation the build started under + client RemoteStorageClient + closeIdle func() +} + +type clientCache struct { + mu sync.Mutex + max int + order *list.List // front: most recently used + entries map[string]*list.Element + building singleflight.Group + births uint64 + generation uint64 // bumped by every retain: a build started under an older one is stale + grace time.Duration + afterFunc func(time.Duration, func()) // tests +} + +var remoteClients = newClientCache(clientCacheMaxFromEnv(), time.Minute) + +func newClientCache(max int, grace time.Duration) *clientCache { + return &clientCache{max: max, order: list.New(), entries: map[string]*list.Element{}, grace: grace, + afterFunc: func(d time.Duration, f func()) { time.AfterFunc(d, f) }} +} + +func clientCacheMaxFromEnv() int { + if v := os.Getenv("WEED_REMOTE_CLIENT_CACHE_MAX"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + glog.Warningf("WEED_REMOTE_CLIENT_CACHE_MAX=%q is not a positive number; using 64", v) + } + return 64 +} + +// clientKey names a configuration exactly: kind, name, a hash of its deterministic encoding, and the credentials version. +func clientKey(kind string, conf *remote_pb.RemoteConf, version string) string { + encoded, _ := proto.MarshalOptions{Deterministic: true}.Marshal(conf) + sum := sha256.Sum256(encoded) + return kind + "\x00" + conf.GetName() + "\x00" + hex.EncodeToString(sum[:16]) + "\x00" + version +} + +func (c *clientCache) get(kind string, conf *remote_pb.RemoteConf, version string, build ClientBuild) (RemoteStorageClient, error) { + key := clientKey(kind, conf, version) + if client := c.lookup(key); client != nil { + return client, nil + } + own := proto.Clone(conf).(*remote_pb.RemoteConf) + v, err, _ := c.building.Do(key, func() (interface{}, error) { + if client := c.lookup(key); client != nil { // another caller's build finished between the lookup and Do + return client, nil + } + generation := c.generationSnapshot() + client, closeIdle, err := build(own) + if err != nil { + return nil, err + } + c.insert(&cachedClient{key: key, kind: kind, name: own.GetName(), generation: generation, client: client, closeIdle: closeIdle}) + return client, nil + }) + if err != nil { + return nil, err + } + return v.(RemoteStorageClient), nil +} + +func (c *clientCache) lookup(key string) RemoteStorageClient { + c.mu.Lock() + defer c.mu.Unlock() + if el, ok := c.entries[key]; ok { + c.order.MoveToFront(el) + return el.Value.(*cachedClient).client + } + return nil +} + +func (c *clientCache) generationSnapshot() uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.generation +} + +func (c *clientCache) insert(e *cachedClient) { + var evicted []*cachedClient + c.mu.Lock() + if el, ok := c.entries[e.key]; ok { // cannot happen under singleflight; keep the first, retire the newcomer + c.order.MoveToFront(el) + evicted = append(evicted, e) + } else if e.generation < c.generation { + // a reload ran while the build was in flight: the caller keeps the + // client but it never enters the cache + evicted = append(evicted, e) + } else { + c.births++ + e.born = c.births + c.entries[e.key] = c.order.PushFront(e) + for c.order.Len() > c.max { + back := c.order.Back() + old := back.Value.(*cachedClient) + c.order.Remove(back) + delete(c.entries, old.key) + evicted = append(evicted, old) + } + } + c.mu.Unlock() + for _, old := range evicted { + c.retire(old) + } +} + +// retire closes a client's idle connections now and again after the grace period (requests in flight when it left the +// cache return their connections to its pool afterwards). +func (c *clientCache) retire(e *cachedClient) { + if e.closeIdle == nil { + return + } + e.closeIdle() + c.afterFunc(c.grace, e.closeIdle) +} + +func (c *clientCache) retain(names map[string]bool) { + var dropped []*cachedClient + c.mu.Lock() + c.generation++ + newest := map[string]uint64{} // kind and name -> the newest entry's birth + for _, el := range c.entries { + e := el.Value.(*cachedClient) + if k := e.kind + "\x00" + e.name; e.born > newest[k] { + newest[k] = e.born + } + } + for key, el := range c.entries { + e := el.Value.(*cachedClient) + if names[e.name] && e.born == newest[e.kind+"\x00"+e.name] { + continue + } + c.order.Remove(el) + delete(c.entries, key) + dropped = append(dropped, e) + } + c.mu.Unlock() + for _, e := range dropped { + c.retire(e) + } +} + +func (c *clientCache) len() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.order.Len() +} + +// GetRemoteStorage returns the plain client of a configuration (one per configuration). +func GetRemoteStorage(remoteConf *remote_pb.RemoteConf) (RemoteStorageClient, error) { + return GetRemoteStorageWith("", remoteConf, CredentialsVersion(remoteConf), func(conf *remote_pb.RemoteConf) (RemoteStorageClient, func(), error) { + client, err := makeRemoteStorageClient(conf) + if err != nil { + return nil, nil, fmt.Errorf("make remote storage client %s: %w", conf.GetName(), err) + } + return client, nil, nil + }) +} + +// GetRemoteStorageWith returns the cached client of one kind for a configuration and credentials version, building it with +// build (given a copy of the configuration) when there is none. version names whatever the client reads at build time +// that the configuration does not hold (a credentials file's content hash); "" when nothing. +func GetRemoteStorageWith(kind string, remoteConf *remote_pb.RemoteConf, version string, build ClientBuild) (RemoteStorageClient, error) { + return remoteClients.get(kind, remoteConf, version, build) +} + +// CredentialsVersion names what a gcs client, plain or guarded, reads at +// build time that its configuration does not hold: a credentials file's +// content, inline JSON from GOOGLE_APPLICATION_CREDENTIALS, or the well-known +// application default credentials file when neither names one. The +// configuration's own inline JSON needs none; it is already inside the +// configuration's hash. "" for a non-gcs remote. +func CredentialsVersion(remoteConf *remote_pb.RemoteConf) string { + if remoteConf.GetType() != "gcs" { + return "" + } + path := remoteConf.GetGcsGoogleApplicationCredentials() + fromEnv := false + if path == "" { + path, fromEnv = os.Getenv("GOOGLE_APPLICATION_CREDENTIALS"), true + } + if strings.HasPrefix(path, "{") { + if !fromEnv { + return "" + } + sum := sha256.Sum256([]byte(path)) + return hex.EncodeToString(sum[:16]) + } + if path == "" { + path = wellKnownCredentialsFile() + } + return FileVersion(path) +} + +// wellKnownCredentialsFile is the application default credentials file the +// gcs SDK reads when neither the configuration nor +// GOOGLE_APPLICATION_CREDENTIALS names one. +func wellKnownCredentialsFile() string { + const f = "application_default_credentials.json" + if runtime.GOOS == "windows" { + return filepath.Join(os.Getenv("APPDATA"), "gcloud", f) + } + home := os.Getenv("HOME") + if home == "" { + if u, err := user.Current(); err == nil { + home = u.HomeDir + } + } + return filepath.Join(home, ".config", "gcloud", f) +} + +// RetainRemoteStorages drops, and closes the idle connections of, every cached client whose configuration name is not in +// names, and every client of a name but its newest: a remote removed from /etc/remote, or the old configuration of a +// changed one, keeps no client and no pool. The filer calls it on every reload. +func RetainRemoteStorages(names map[string]bool) { + remoteClients.retain(names) +} + +// FileVersion is the credentials version of a file: a hash of its content, "" when it cannot be read (the build then +// reads it, or fails, as it would without the cache). +func FileVersion(path string) string { + if path == "" { + return "" + } + data, err := os.ReadFile(util.ResolvePath(path)) // "~/…" as the gcs client reads it + if err != nil { + return "" + } + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:16]) +} + +// ResetRemoteStorageClients empties the cache, retiring every client. For tests. +func ResetRemoteStorageClients() { + remoteClients.retain(nil) +} diff --git a/weed/remote_storage/remote_storage.go b/weed/remote_storage/remote_storage.go index 73f568f5a8..1bf2be90f4 100644 --- a/weed/remote_storage/remote_storage.go +++ b/weed/remote_storage/remote_storage.go @@ -7,12 +7,10 @@ import ( "io" "sort" "strings" - "sync" "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" - "google.golang.org/protobuf/proto" ) const slash = "/" @@ -176,16 +174,7 @@ type RemoteStorageClientMaker interface { HasBucket() bool } -type CachedRemoteStorageClient struct { - *remote_pb.RemoteConf - RemoteStorageClient -} - -var ( - RemoteStorageClientMakers = make(map[string]RemoteStorageClientMaker) - remoteStorageClients = make(map[string]CachedRemoteStorageClient) - remoteStorageClientsLock sync.Mutex -) +var RemoteStorageClientMakers = make(map[string]RemoteStorageClientMaker) func GetAllRemoteStorageNames() string { var storageNames []string @@ -215,25 +204,3 @@ func makeRemoteStorageClient(remoteConf *remote_pb.RemoteConf) (RemoteStorageCli } return maker.Make(remoteConf) } - -func GetRemoteStorage(remoteConf *remote_pb.RemoteConf) (RemoteStorageClient, error) { - remoteStorageClientsLock.Lock() - defer remoteStorageClientsLock.Unlock() - - existingRemoteStorageClient, found := remoteStorageClients[remoteConf.Name] - if found && proto.Equal(existingRemoteStorageClient.RemoteConf, remoteConf) { - return existingRemoteStorageClient.RemoteStorageClient, nil - } - - newRemoteStorageClient, err := makeRemoteStorageClient(remoteConf) - if err != nil { - return nil, fmt.Errorf("make remote storage client %s: %v", remoteConf.Name, err) - } - - remoteStorageClients[remoteConf.Name] = CachedRemoteStorageClient{ - RemoteConf: remoteConf, - RemoteStorageClient: newRemoteStorageClient, - } - - return newRemoteStorageClient, nil -} diff --git a/weed/remote_storage/remote_storage_cache_test.go b/weed/remote_storage/remote_storage_cache_test.go new file mode 100644 index 0000000000..9fbc33217a --- /dev/null +++ b/weed/remote_storage/remote_storage_cache_test.go @@ -0,0 +1,408 @@ +package remote_storage + +import ( + "fmt" + "os" + "os/user" + "path/filepath" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" +) + +// a client that is only an identity: the cache never calls it +type cacheTestClient struct { + RemoteStorageClient + id int64 +} + +// a fresh cache for one test, swapped in for the process-wide one and restored after +func useTestCache(t *testing.T, max int) (*clientCache, *[]time.Duration) { + t.Helper() + c := newClientCache(max, time.Minute) + var delayed []time.Duration + var mu sync.Mutex + c.afterFunc = func(d time.Duration, f func()) { mu.Lock(); delayed = append(delayed, d); mu.Unlock(); f() } + prev := remoteClients + remoteClients = c + t.Cleanup(func() { remoteClients = prev }) + return c, &delayed +} + +type counter struct { + builds atomic.Int64 + mu sync.Mutex + closed map[int64]int + confs []*remote_pb.RemoteConf +} + +func (k *counter) build(conf *remote_pb.RemoteConf) (RemoteStorageClient, func(), error) { + id := k.builds.Add(1) + k.mu.Lock() + k.confs = append(k.confs, conf) + k.mu.Unlock() + return &cacheTestClient{id: id}, func() { + k.mu.Lock() + defer k.mu.Unlock() + if k.closed == nil { + k.closed = map[int64]int{} + } + k.closed[id]++ + }, nil +} + +func (k *counter) closes(id int64) int { + k.mu.Lock() + defer k.mu.Unlock() + return k.closed[id] +} + +func s3conf(name, secret string) *remote_pb.RemoteConf { + return &remote_pb.RemoteConf{Name: name, Type: "s3", S3Endpoint: "https://s3.example", S3AccessKey: "k", S3SecretKey: secret} +} + +// One client per configuration: the same configuration (another copy of it) gets the same client, built once. +func TestRemoteClientCache_SameConfigurationOneClient(t *testing.T) { + useTestCache(t, 64) + var k counter + first, err := GetRemoteStorageWith("guarded", s3conf("a", "x"), "", k.build) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 5; i++ { + again, _ := GetRemoteStorageWith("guarded", s3conf("a", "x"), "", k.build) + if again != first { + t.Fatalf("call %d built a new client for the same configuration", i) + } + } + if k.builds.Load() != 1 { + t.Fatalf("builds = %d, want 1", k.builds.Load()) + } + plain, _ := GetRemoteStorageWith("", s3conf("a", "x"), "", k.build) + if plain == first { + t.Fatal("the plain and the guarded client of one configuration must not be shared") + } +} + +// Two configurations under one name (a key being rotated while the old one is still in use) coexist: alternating between +// them builds each once, never thrashing. +func TestRemoteClientCache_TwoConfigurationsUnderOneNameCoexist(t *testing.T) { + useTestCache(t, 64) + var k counter + for i := 0; i < 10; i++ { + secret := "old" + if i%2 == 1 { + secret = "new" + } + if _, err := GetRemoteStorageWith("guarded", s3conf("a", secret), "", k.build); err != nil { + t.Fatal(err) + } + } + if k.builds.Load() != 2 { + t.Fatalf("builds = %d for two alternating configurations under one name, want 2", k.builds.Load()) + } + if k.closes(1) != 0 || k.closes(2) != 0 { + t.Fatal("neither client may be closed while both are in use") + } +} + +// The credentials version is part of the key: a changed credentials file is a new client. +func TestRemoteClientCache_VersionIsPartOfTheKey(t *testing.T) { + useTestCache(t, 64) + var k counter + a, _ := GetRemoteStorageWith("guarded", s3conf("a", ""), "v1", k.build) + b, _ := GetRemoteStorageWith("guarded", s3conf("a", ""), "v1", k.build) + c, _ := GetRemoteStorageWith("guarded", s3conf("a", ""), "v2", k.build) + if a != b || a == c || k.builds.Load() != 2 { + t.Fatalf("same version must share, another version must not (builds = %d)", k.builds.Load()) + } +} + +// The cache is bounded: past its size the least recently used client is evicted and its idle connections closed, at once +// and again after the grace period. +func TestRemoteClientCache_BoundedLRU(t *testing.T) { + c, delayed := useTestCache(t, 4) + var k counter + for i := 0; i < 4; i++ { + _, _ = GetRemoteStorageWith("guarded", s3conf(fmt.Sprintf("r%d", i), "x"), "", k.build) + } + _, _ = GetRemoteStorageWith("guarded", s3conf("r0", "x"), "", k.build) // r0 is now the most recently used + _, _ = GetRemoteStorageWith("guarded", s3conf("r4", "x"), "", k.build) // evicts r1 (client 2) + if n := c.len(); n != 4 { + t.Fatalf("cache holds %d clients, want 4", n) + } + if k.closes(2) != 2 { + t.Fatalf("the evicted client must be closed now and after the grace period: closed %d time(s)", k.closes(2)) + } + if len(*delayed) != 1 || (*delayed)[0] != time.Minute { + t.Fatalf("one delayed close after the grace period, got %v", *delayed) + } + if k.closes(1) != 0 { + t.Fatal("the recently used client was evicted instead of the least recently used one") + } + again, _ := GetRemoteStorageWith("guarded", s3conf("r0", "x"), "", k.build) + if again.(*cacheTestClient).id != 1 { + t.Fatal("r0 must still be cached") + } +} + +// RetainRemoteStorages drops and closes the clients of every configuration not named, of every kind. +func TestRemoteClientCache_RetainDropsWhatIsGone(t *testing.T) { + useTestCache(t, 64) + var k counter + _, _ = GetRemoteStorageWith("guarded", s3conf("keep", "x"), "", k.build) + _, _ = GetRemoteStorageWith("guarded", s3conf("gone", "x"), "", k.build) + _, _ = GetRemoteStorageWith("", s3conf("gone", "x"), "", k.build) + RetainRemoteStorages(map[string]bool{"keep": true}) + if k.closes(1) != 0 || k.closes(2) != 2 || k.closes(3) != 2 { + t.Fatalf("closes: keep %d, gone %d and %d", k.closes(1), k.closes(2), k.closes(3)) + } + _, _ = GetRemoteStorageWith("guarded", s3conf("gone", "x"), "", k.build) + if k.builds.Load() != 4 { + t.Fatal("a dropped configuration is built again on its next use") + } +} + +// Builds run outside the cache's lock, once per key: a slow build does not stall another key, and concurrent callers of +// one key share one build. +func TestRemoteClientCache_BuildsOutsideTheLockOncePerKey(t *testing.T) { + useTestCache(t, 64) + var k counter + release := make(chan struct{}) + slow := func(conf *remote_pb.RemoteConf) (RemoteStorageClient, func(), error) { + <-release + return k.build(conf) + } + var wg sync.WaitGroup + results := make([]RemoteStorageClient, 20) + for i := range results { + wg.Add(1) + go func(i int) { + defer wg.Done() + results[i], _ = GetRemoteStorageWith("guarded", s3conf("slow", "x"), "", slow) + }(i) + } + time.Sleep(50 * time.Millisecond) // the slow build is in progress + done := make(chan struct{}) + go func() { + _, _ = GetRemoteStorageWith("guarded", s3conf("other", "x"), "", k.build) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("another key waited for a slow build: the build holds the cache's lock") + } + close(release) + wg.Wait() + for i, r := range results { + if r != results[0] || r == nil { + t.Fatalf("caller %d got another client: concurrent callers of one key must share one build", i) + } + } + if k.builds.Load() != 2 { + t.Fatalf("builds = %d, want 2 (one per key)", k.builds.Load()) + } +} + +// The client is built from a copy of the configuration: the caller changing its value afterwards changes nothing. +func TestRemoteClientCache_BuildsFromACopy(t *testing.T) { + useTestCache(t, 64) + var k counter + conf := s3conf("a", "x") + _, _ = GetRemoteStorageWith("guarded", conf, "", k.build) + if k.confs[0] == conf { + t.Fatal("the build was given the caller's own configuration") + } + conf.S3SecretKey = "changed" + if k.confs[0].S3SecretKey != "x" || !proto.Equal(k.confs[0], s3conf("a", "x")) { + t.Fatal("the cached client's configuration changed with the caller's value") + } +} + +// The process-wide cache can be emptied (tests in other packages use it). +func TestRemoteClientCache_Reset(t *testing.T) { + useTestCache(t, 64) + var k counter + _, _ = GetRemoteStorageWith("guarded", s3conf("a", "x"), "", k.build) + ResetRemoteStorageClients() + if remoteClients.len() != 0 || k.closes(1) != 2 { + t.Fatalf("after a reset: %d cached, closed %d time(s)", remoteClients.len(), k.closes(1)) + } +} + +// A reload (RetainRemoteStorages) keeps, per kind and name, only the newest client: the old configuration of a rotated +// key, or the old version of a credentials file, leaves no client and no pool behind; another kind keeps its own. +func TestRemoteClientCache_RetainKeepsOnlyTheNewestPerName(t *testing.T) { + useTestCache(t, 64) + var k counter + _, _ = GetRemoteStorageWith("guarded", s3conf("a", "old"), "", k.build) // 1 + _, _ = GetRemoteStorageWith("guarded", s3conf("a", "new"), "", k.build) // 2: the rotated key + _, _ = GetRemoteStorageWith("guarded", s3conf("a", ""), "v1", k.build) // 3 + _, _ = GetRemoteStorageWith("guarded", s3conf("a", ""), "v2", k.build) // 4: the credentials file changed + _, _ = GetRemoteStorageWith("", s3conf("a", "old"), "", k.build) // 5: the plain kind's own + RetainRemoteStorages(map[string]bool{"a": true}) + if k.closes(1) != 2 || k.closes(2) != 2 || k.closes(3) != 2 { + t.Fatalf("superseded clients must be closed: %d %d %d", k.closes(1), k.closes(2), k.closes(3)) + } + if k.closes(4) != 0 || k.closes(5) != 0 { + t.Fatalf("the newest of each kind must stay: guarded %d, plain %d", k.closes(4), k.closes(5)) + } + again, _ := GetRemoteStorageWith("guarded", s3conf("a", ""), "v2", k.build) + if again.(*cacheTestClient).id != 4 { + t.Fatal("the newest client must still be cached") + } +} + +// A build that finishes after a reload removed its remote must not put the client back: the caller still gets it, but +// it never enters the cache, and the next use builds again. +func TestRemoteClientCache_BuildSpanningRetainIsDropped(t *testing.T) { + useTestCache(t, 64) + var k counter + started := make(chan struct{}) + release := make(chan struct{}) + slow := func(conf *remote_pb.RemoteConf) (RemoteStorageClient, func(), error) { + close(started) + <-release + return k.build(conf) + } + built := make(chan RemoteStorageClient, 1) + go func() { + client, _ := GetRemoteStorageWith("guarded", s3conf("gone", "x"), "", slow) + built <- client + }() + <-started + RetainRemoteStorages(map[string]bool{"keep": true}) + close(release) + if client := <-built; client == nil { + t.Fatal("the caller still gets its built client") + } + if k.closes(1) != 2 { + t.Fatalf("the dropped client must be retired: closes %d", k.closes(1)) + } + _, _ = GetRemoteStorageWith("guarded", s3conf("gone", "x"), "", k.build) + if k.builds.Load() != 2 { + t.Fatalf("builds = %d, want 2: the dropped client was not cached", k.builds.Load()) + } +} + +// A remote the local filer never named — another filer's, proxied through this volume server — keeps connection reuse: +// retention only ever drops what was built, never what a later get is for. +func TestRemoteClientCache_UnknownNameStillCaches(t *testing.T) { + useTestCache(t, 64) + var k counter + RetainRemoteStorages(map[string]bool{"keep": true}) + first, _ := GetRemoteStorageWith("guarded", s3conf("peer", "x"), "", k.build) + again, _ := GetRemoteStorageWith("guarded", s3conf("peer", "x"), "", k.build) + if first == nil || first != again || k.builds.Load() != 1 { + t.Fatal("a remote the local filer does not name must still be cached") + } +} + +// The plain gcs client's version follows its credentials file: a conf path or GOOGLE_APPLICATION_CREDENTIALS, while +// inline JSON needs none (it is inside the configuration's hash already). +func TestPlainClientGcsCredentialsVersion(t *testing.T) { + dir := t.TempDir() + creds := filepath.Join(dir, "creds.json") + if err := os.WriteFile(creds, []byte("one"), 0o600); err != nil { + t.Fatal(err) + } + conf := &remote_pb.RemoteConf{Name: "g", Type: "gcs", GcsGoogleApplicationCredentials: creds} + v1 := CredentialsVersion(conf) + if v1 == "" { + t.Fatal("a credentials path must produce a version") + } + if err := os.WriteFile(creds, []byte("two"), 0o600); err != nil { + t.Fatal(err) + } + if CredentialsVersion(conf) == v1 { + t.Fatal("a rotated credentials file must change the version") + } + if CredentialsVersion(&remote_pb.RemoteConf{Type: "gcs", GcsGoogleApplicationCredentials: "{}"}) != "" { + t.Fatal("inline JSON credentials need no version") + } + t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", creds) + if CredentialsVersion(&remote_pb.RemoteConf{Type: "gcs"}) == "" { + t.Fatal("the environment's credentials path must produce a version") + } + t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", `{"type":"one"}`) + vEnvJSON := CredentialsVersion(&remote_pb.RemoteConf{Type: "gcs"}) + if vEnvJSON == "" { + t.Fatal("the environment's inline JSON must produce a version") + } + t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", `{"type":"two"}`) + if CredentialsVersion(&remote_pb.RemoteConf{Type: "gcs"}) == vEnvJSON { + t.Fatal("rotated inline JSON credentials must change the version") + } + if CredentialsVersion(&remote_pb.RemoteConf{Type: "s3"}) != "" { + t.Fatal("a non-gcs remote has no credentials file version") + } +} + +// With neither the configuration nor GOOGLE_APPLICATION_CREDENTIALS naming credentials, the version follows the +// well-known application default credentials file the SDK reads. +func TestPlainClientGcsWellKnownCredentialsVersion(t *testing.T) { + dir := t.TempDir() + if runtime.GOOS == "windows" { + t.Setenv("APPDATA", dir) + } else { + t.Setenv("HOME", dir) + } + t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", "") + adc := wellKnownCredentialsFile() + if err := os.MkdirAll(filepath.Dir(adc), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(adc, []byte("one"), 0o600); err != nil { + t.Fatal(err) + } + conf := &remote_pb.RemoteConf{Type: "gcs"} + v1 := CredentialsVersion(conf) + if v1 == "" { + t.Fatal("the well-known credentials file must produce a version") + } + if err := os.WriteFile(adc, []byte("two"), 0o600); err != nil { + t.Fatal(err) + } + if CredentialsVersion(conf) == v1 { + t.Fatal("a rotated well-known credentials file must change the version") + } +} + +// FileVersion reads "~/…" as the gcs client does (util.ResolvePath), so a rotated file under the home directory is seen. +func TestFileVersionResolvesHome(t *testing.T) { + u, err := user.Current() + if err != nil { + t.Skip("no current user") + } + name := fmt.Sprintf(".seaweedfs-fileversion-test-%d.json", os.Getpid()) + path := filepath.Join(u.HomeDir, name) + if err := os.WriteFile(path, []byte("one"), 0o600); err != nil { + t.Skip("home directory not writable") + } + t.Cleanup(func() { os.Remove(path) }) + v1 := FileVersion("~/" + name) + if v1 == "" { + t.Fatal(`FileVersion("~/...") is empty: "~" was not resolved`) + } + _ = os.WriteFile(path, []byte("two"), 0o600) + if v2 := FileVersion("~/" + name); v2 == v1 { + t.Fatal("a rotated file must change the version") + } +} + +// A failed plain build names the remote configuration in its error. +func TestGetRemoteStorageErrorNamesTheRemote(t *testing.T) { + useTestCache(t, 64) + _, err := GetRemoteStorage(&remote_pb.RemoteConf{Name: "mybucketremote", Type: "no-such-type"}) + if err == nil || !strings.Contains(err.Error(), "make remote storage client mybucketremote") { + t.Fatalf("error %v must name the remote", err) + } +} diff --git a/weed/remote_storage/s3/default_chain_test.go b/weed/remote_storage/s3/default_chain_test.go new file mode 100644 index 0000000000..d646f4777c --- /dev/null +++ b/weed/remote_storage/s3/default_chain_test.go @@ -0,0 +1,29 @@ +package s3 + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" +) + +// The default credential chain applies when an access key or a secret is +// missing, for every S3-compatible type, and only then: the servers hash the +// shared credentials file into the client cache key in that case alone. +func TestUsesDefaultCredentialChain(t *testing.T) { + for _, tc := range []struct { + name string + conf *remote_pb.RemoteConf + want bool + }{ + {"s3 with both keys", &remote_pb.RemoteConf{Type: "s3", S3AccessKey: "a", S3SecretKey: "b"}, false}, + {"s3 with a key, no secret", &remote_pb.RemoteConf{Type: "s3", S3AccessKey: "a"}, true}, + {"s3 with no keys", &remote_pb.RemoteConf{Type: "s3"}, true}, + {"b2 with both keys (S3AccessKey unset)", &remote_pb.RemoteConf{Type: "b2", BackblazeKeyId: "a", BackblazeApplicationKey: "b", BackblazeEndpoint: "https://s3.example"}, false}, + {"b2 with no secret", &remote_pb.RemoteConf{Type: "b2", BackblazeKeyId: "a", BackblazeEndpoint: "https://s3.example"}, true}, + {"not S3-compatible", &remote_pb.RemoteConf{Type: "azure"}, false}, + } { + if got := UsesDefaultCredentialChain(tc.conf); got != tc.want { + t.Errorf("%s: %v, want %v", tc.name, got, tc.want) + } + } +} diff --git a/weed/remote_storage/s3/s3_storage_client.go b/weed/remote_storage/s3/s3_storage_client.go index 101aff6a26..df9e02eddf 100644 --- a/weed/remote_storage/s3/s3_storage_client.go +++ b/weed/remote_storage/s3/s3_storage_client.go @@ -170,6 +170,24 @@ func S3CompatibleEndpoint(conf *remote_pb.RemoteConf) (string, bool) { return opt.endpoint, true } +// UsesDefaultCredentialChain reports whether a client of conf, of any +// S3-compatible type, may take its credentials from the SDK's default chain +// (an access key or a secret missing, after the type's own environment +// fallbacks): the only case where they live outside the configuration. +func UsesDefaultCredentialChain(conf *remote_pb.RemoteConf) bool { + opt, ok := s3CompatibleClientOptions(conf) + return ok && !(opt.accessKey != "" && opt.secretKey != "") +} + +// ResolvedCredentials returns the static credentials a client of conf carries +// — the configuration's own keys, or the type's environment fallback for an +// unset field — and whether conf is S3-compatible at all. An environment +// fallback's change shows up here where the configuration's hash would not. +func ResolvedCredentials(conf *remote_pb.RemoteConf) (accessKey, secretKey string, ok bool) { + opt, ok := s3CompatibleClientOptions(conf) + return opt.accessKey, opt.secretKey, ok +} + // awsCredentialSourceConfigured reports whether the environment names a source // for the AWS SDK default credential chain, such as static keys, a profile, // IRSA web identity or the ECS/EKS container credentials endpoint. diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index adeb59e5de..3943886899 100644 --- a/weed/server/volume_grpc_remote.go +++ b/weed/server/volume_grpc_remote.go @@ -2,13 +2,17 @@ package weed_server import ( "context" + "crypto/sha256" + "encoding/hex" "errors" "fmt" "net" "net/http" "net/url" "os" + "path/filepath" "slices" + "strconv" "strings" "sync" "time" @@ -17,6 +21,7 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -284,10 +289,130 @@ func (vs *VolumeServer) guardedGrpcDialOption(endpoint string) grpc.DialOption { } // newGuardedHTTPClient returns an *http.Client whose transport refuses to -// dial addresses that fail checkBlockedIP at connect time. It is meant for -// per-request use; do not share across remote configs. +// dial addresses that fail checkBlockedIP at connect time. One client serves +// one remote configuration (BuildGuardedRemoteStorageClient caches it per +// configuration); never share it across remote configs. func newGuardedHTTPClient(endpoint string) *http.Client { - return newGuardedHTTPClientPolicy(endpoint, false) + client := newGuardedHTTPClientPolicy(endpoint, false) + if t, ok := client.Transport.(*http.Transport); ok { + tuneSharedTransport(t) + } + return client +} + +// The shared transport's timeouts: with one client per remote and a cap per +// host, a request that never finishes would hold a slot forever, and enough of +// them stall every later lazy stat, cold read and remote delete of that remote. +// A dial, a TLS handshake and a response's headers each have a deadline, and a +// stalled body an idle one. +const ( + guardedDialTimeout = 10 * time.Second + guardedTLSHandshakeTimeout = 10 * time.Second + guardedResponseHeaderTimeout = 60 * time.Second +) + +// guardedIOIdleTimeout bounds the stall of one exchange: a body that stops +// mid-transfer frees its connection slot while a slow but progressing transfer +// is never cut off. A var so tests can shorten it. +var guardedIOIdleTimeout = 2 * time.Minute + +// idleTimeoutConn refreshes a deadline on every read and write: a peer that +// stops sending or receiving is cut off after idle has passed without +// progress. +type idleTimeoutConn struct { + net.Conn + idle time.Duration +} + +func (c *idleTimeoutConn) Read(b []byte) (int, error) { + _ = c.SetReadDeadline(time.Now().Add(c.idle)) + return c.Conn.Read(b) +} + +func (c *idleTimeoutConn) Write(b []byte) (int, error) { + _ = c.SetWriteDeadline(time.Now().Add(c.idle)) + return c.Conn.Write(b) +} + +// tuneSharedTransport sets what a transport shared by every request to one +// remote needs: enough idle connections for a burst to reuse (Go's default is +// 2 per host), a cap on connections per host (a burst of cold reads queues +// instead of opening hundreds, which an endpoint limiting connections per +// source would refuse), and deadlines, so a hung request frees its slot. +func tuneSharedTransport(t *http.Transport) { + t.MaxIdleConns = 64 + t.MaxIdleConnsPerHost = 32 + t.MaxConnsPerHost = guardedMaxConnsPerHost() + t.TLSHandshakeTimeout = guardedTLSHandshakeTimeout + t.ResponseHeaderTimeout = guardedResponseHeaderTimeout + if dial := t.DialContext; dial != nil { + t.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + ctx, cancel := context.WithTimeout(ctx, guardedDialTimeout) + defer cancel() // an established connection outlives the dial's context + conn, err := dial(ctx, network, addr) + if err != nil { + return nil, err + } + return &idleTimeoutConn{Conn: conn, idle: guardedIOIdleTimeout}, nil + } + } +} + +// guardedMaxConnsPerHost is WEED_REMOTE_MAX_CONNS_PER_HOST, default 64. +func guardedMaxConnsPerHost() int { + if v := os.Getenv("WEED_REMOTE_MAX_CONNS_PER_HOST"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + glog.Warningf("WEED_REMOTE_MAX_CONNS_PER_HOST=%q is not a positive number; using 64", v) + } + return 64 +} + +// guardedCredentialsVersion names the credentials a guarded client reads when +// it is built that its configuration does not hold, so that a change of them +// builds a new client: the gcs credentials sources of CredentialsVersion, or, +// for an s3-compatible remote, the resolved keys (which a type's environment +// fallback may supply), plus — with a key or secret still missing (the SDK's +// default chain) — the shared credentials and config files and the AWS_* +// variables the chain reads. "" for a remote of another kind. +func guardedCredentialsVersion(conf *remote_pb.RemoteConf) string { + if conf.GetType() == "gcs" { + return remote_storage.CredentialsVersion(conf) + } + accessKey, secretKey, ok := s3remote.ResolvedCredentials(conf) + if !ok { + return "" + } + h := sha256.New() + h.Write([]byte(accessKey)) + h.Write([]byte{0}) + h.Write([]byte(secretKey)) + if s3remote.UsesDefaultCredentialChain(conf) { + shared := awsSharedFile("AWS_SHARED_CREDENTIALS_FILE", "credentials") + config := awsSharedFile("AWS_CONFIG_FILE", "config") + h.Write([]byte{0}) + h.Write([]byte(remote_storage.FileVersion(shared))) + h.Write([]byte{0}) + h.Write([]byte(remote_storage.FileVersion(config))) + for _, k := range []string{"AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "AWS_PROFILE", "AWS_DEFAULT_PROFILE"} { + h.Write([]byte{0}) + h.Write([]byte(os.Getenv(k))) + } + } + return hex.EncodeToString(h.Sum(nil)[:16]) +} + +// awsSharedFile is the SDK's effective path of a shared file: the environment +// override or the default under ~/.aws. +func awsSharedFile(env, name string) string { + if path := os.Getenv(env); path != "" { + return path + } + if home, err := os.UserHomeDir(); err == nil { + return filepath.Join(home, ".aws", name) + } + return "" } // newGuardedHTTPClientPolicy is newGuardedHTTPClient with the allowPrivate knob @@ -363,12 +488,6 @@ func guardedRemoteClient(remoteConf *remote_pb.RemoteConf) (endpoint string, mak return "", nil, false } -// gcsCredentialsArePath reports whether a gcs credentials value is a filesystem -// path rather than inline JSON, matching the gcs client's own inline detection. -func gcsCredentialsArePath(creds string) bool { - return creds != "" && !strings.HasPrefix(creds, "{") -} - var errGcsCredentialsUnreadable = errors.New("gcs credentials file is not readable or does not contain valid credentials") // loadGcsCredentialsContent returns the credential JSON for a gcs credentials @@ -424,18 +543,38 @@ func BuildGuardedRemoteStorageClient(ctx context.Context, remoteConf *remote_pb. return nil, fmt.Errorf("reject remote credentials: %w", credsErr) } } - if endpoint, makeClient, ok := guardedRemoteClient(remoteConf); ok { - if validateErr := validateRemoteEndpoint(ctx, endpoint); validateErr != nil { - return nil, fmt.Errorf("reject remote endpoint: %w", validateErr) - } - // Build a one-shot client whose dial path re-validates the resolved - // IP every time. This pins the validated endpoint against DNS - // rebinding (a hostname that resolves to a public IP for - // validateRemoteEndpoint and then flips to 127.0.0.1 / 169.254.x.x - // when the SDK dials). - client, err := makeClient(newGuardedHTTPClient(endpoint)) + if _, _, ok := guardedRemoteClient(remoteConf); ok { + // One client per configuration and credentials version: + // a client built per call kept its transport's connection open for + // IdleConnTimeout and never reused it, one open connection per lazy + // stat, cold read or remote delete. The endpoint is resolved and + // checked against the deny-list when a client is built (a cache + // miss); every new connection of a cached client still dials through + // the guarded dialer, which checks the address it actually connects to. + buildCtx := context.WithoutCancel(ctx) + client, err := remote_storage.GetRemoteStorageWith("guarded", remoteConf, guardedCredentialsVersion(remoteConf), + func(conf *remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, func(), error) { + endpoint, makeClient, ok := guardedRemoteClient(conf) + if !ok { + return nil, nil, fmt.Errorf("remote %s: no guarded client", conf.GetName()) + } + if validateErr := validateRemoteEndpoint(buildCtx, endpoint); validateErr != nil { + return nil, nil, fmt.Errorf("reject remote endpoint: %w", validateErr) + } + // A client whose dial path re-validates the resolved IP every + // time. This pins the validated endpoint against DNS rebinding + // (a hostname that resolves to a public IP for + // validateRemoteEndpoint and then flips to 127.0.0.1 / + // 169.254.x.x when the SDK dials). + httpClient := newGuardedHTTPClient(endpoint) + client, err := makeClient(httpClient) + if err != nil { + return nil, nil, fmt.Errorf("get remote client: %w", err) + } + return client, httpClient.CloseIdleConnections, nil + }) if err != nil { - return nil, fmt.Errorf("get remote client: %w", err) + return nil, err } return client, nil } diff --git a/weed/server/volume_grpc_remote_test.go b/weed/server/volume_grpc_remote_test.go index cdee1f8db1..50957a0991 100644 --- a/weed/server/volume_grpc_remote_test.go +++ b/weed/server/volume_grpc_remote_test.go @@ -21,8 +21,11 @@ import ( "testing" "time" + "google.golang.org/protobuf/proto" + "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" + "github.com/seaweedfs/seaweedfs/weed/remote_storage" s3remote "github.com/seaweedfs/seaweedfs/weed/remote_storage/s3" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) @@ -490,34 +493,6 @@ func TestGuardedRemoteClientAzureBuildsGuardedClient(t *testing.T) { } } -// TestGcsCredentialsArePath confirms a caller-supplied gcs credentials value is -// only accepted as inline JSON. A filesystem path would otherwise be read from -// disk by the SDK when handling the request. -func TestGcsCredentialsArePath(t *testing.T) { - paths := []string{ - "/etc/hostname", - "/etc/shadow", - "/nope/nothere", - "~/creds.json", - "relative/creds.json", - } - for _, p := range paths { - if !gcsCredentialsArePath(p) { - t.Errorf("expected %q to be treated as a path", p) - } - } - inlineOrEmpty := []string{ - "", - `{"type":"service_account"}`, - `{}`, - } - for _, c := range inlineOrEmpty { - if gcsCredentialsArePath(c) { - t.Errorf("expected %q to be accepted (inline or empty)", c) - } - } -} - // TestGuardedRemoteClientGuardsGcsTokenURL confirms the token endpoint named by // inline gcs credentials is the endpoint the guard validates, so a loopback // token_uri is refused while the Google default passes. @@ -970,3 +945,178 @@ func TestGuardedPeerHTTPClientRejectsHTTPSDowngrade(t *testing.T) { t.Error("HTTPS peer never saw the original upload") } } + +// guardedCacheTest empties the process-wide remote client cache before and +// after a test and stubs DNS with a lookup counter. +func guardedCacheTest(t *testing.T, ips map[string][]net.IP) *atomic.Int64 { + t.Helper() + remote_storage.ResetRemoteStorageClients() + t.Cleanup(remote_storage.ResetRemoteStorageClients) + original := lookupIPAddrFunc + t.Cleanup(func() { lookupIPAddrFunc = original }) + var lookups atomic.Int64 + stub := stubLookup(t, ips) + lookupIPAddrFunc = func(ctx context.Context, host string) ([]net.IPAddr, error) { + lookups.Add(1) + return stub(ctx, host) + } + return &lookups +} + +// The same configuration gets the same client; a rotated key a new one; a loopback endpoint under a cached name is +// still refused (it is another configuration: a miss, checked when it is built). +func TestBuildGuardedRemoteStorageClientReusesOneClientPerConf(t *testing.T) { + guardedCacheTest(t, nil) + conf := &remote_pb.RemoteConf{ + Name: "reuse", + Type: "s3", + S3Endpoint: "https://93.184.216.34", // a public address literal: validated without DNS + S3AccessKey: "k", + S3SecretKey: "s", + S3Region: "us-east-1", + } + first, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if err != nil { + t.Fatalf("build: %v", err) + } + for i := 0; i < 5; i++ { + again, err := BuildGuardedRemoteStorageClient(context.Background(), proto.Clone(conf).(*remote_pb.RemoteConf), false) + if err != nil { + t.Fatalf("build again: %v", err) + } + if again != first { + t.Fatalf("call %d built a new client for the same configuration", i) + } + } + rotated := proto.Clone(conf).(*remote_pb.RemoteConf) + rotated.S3SecretKey = "s2" + if next, err := BuildGuardedRemoteStorageClient(context.Background(), rotated, false); err != nil || next == first { + t.Errorf("a changed configuration (a rotated key) must get a new client (err %v)", err) + } + loop := proto.Clone(conf).(*remote_pb.RemoteConf) + loop.S3Endpoint = "http://127.0.0.1:8000" + if _, err := BuildGuardedRemoteStorageClient(context.Background(), loop, false); err == nil || + !strings.Contains(err.Error(), "reject remote endpoint") { + t.Errorf("a loopback endpoint under a cached name must still be rejected, got %v", err) + } +} + +// The endpoint's DNS is checked when a client is built (a cache miss), not on every call; the guarded dialer still +// checks every new connection. +func TestBuildGuardedRemoteStorageClientResolvesOnlyOnAMiss(t *testing.T) { + lookups := guardedCacheTest(t, map[string][]net.IP{"s3.example.test": {net.ParseIP("52.216.10.10")}}) + conf := &remote_pb.RemoteConf{Name: "dns", Type: "s3", S3Endpoint: "https://s3.example.test", + S3AccessKey: "k", S3SecretKey: "s", S3Region: "us-east-1"} + for i := 0; i < 10; i++ { + if _, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false); err != nil { + t.Fatalf("build: %v", err) + } + } + if n := lookups.Load(); n != 1 { + t.Fatalf("%d DNS lookups for 10 calls with one configuration, want 1", n) + } +} + +// The shared transport keeps a burst's connections for reuse and caps them per host (WEED_REMOTE_MAX_CONNS_PER_HOST). +func TestGuardedHTTPClientBoundsConnectionsPerHost(t *testing.T) { + tr := newGuardedHTTPClient("https://93.184.216.34").Transport.(*http.Transport) + if tr.MaxIdleConnsPerHost < 32 || tr.MaxConnsPerHost != 64 { + t.Fatalf("MaxIdleConnsPerHost %d (want >= 32), MaxConnsPerHost %d (want 64)", tr.MaxIdleConnsPerHost, tr.MaxConnsPerHost) + } + t.Setenv("WEED_REMOTE_MAX_CONNS_PER_HOST", "8") + if got := newGuardedHTTPClient("https://93.184.216.34").Transport.(*http.Transport).MaxConnsPerHost; got != 8 { + t.Fatalf("MaxConnsPerHost %d with WEED_REMOTE_MAX_CONNS_PER_HOST=8", got) + } +} + +// A gcs credentials file is part of the key by its content: the same file shares one client, a changed file builds anew. +func TestBuildGuardedRemoteStorageClientGcsFileCredentialsByContent(t *testing.T) { + guardedCacheTest(t, map[string][]net.IP{"oauth2.googleapis.com": {net.ParseIP("142.250.0.95")}}) + credsFile := filepath.Join(t.TempDir(), "service-account.json") + write := func(email string) { + creds := `{"type":"service_account","token_uri":"https://oauth2.googleapis.com/token","client_email":"` + email + `","private_key":"-----BEGIN PRIVATE KEY-----\nMIIBVwIBADANBgkqhkiG9w0BAQEFAASCAUEwggE9AgEAAkEAxY\n-----END PRIVATE KEY-----\n","private_key_id":"key1"}` + if err := os.WriteFile(credsFile, []byte(creds), 0600); err != nil { + t.Fatal(err) + } + } + write("sa@example.iam.gserviceaccount.com") + conf := &remote_pb.RemoteConf{Name: "gcsfile", Type: "gcs", GcsGoogleApplicationCredentials: credsFile} + first, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if err != nil { + t.Fatalf("build: %v", err) + } + again, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if again != first { + t.Fatal("the same credentials file must share one client") + } + write("other@example.iam.gserviceaccount.com") + if next, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false); err != nil || next == first { + t.Fatalf("a changed credentials file must build a new client (err %v)", err) + } +} + +// An s3-compatible remote without keys (the SDK's default chain) is keyed by the shared credentials file and the AWS_* +// variables: unchanged, one client; changed, a new one. +func TestBuildGuardedRemoteStorageClientS3DefaultChainByCredentials(t *testing.T) { + guardedCacheTest(t, nil) + shared := filepath.Join(t.TempDir(), "credentials") + if err := os.WriteFile(shared, []byte("[default]\naws_access_key_id = a\naws_secret_access_key = b\n"), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("AWS_SHARED_CREDENTIALS_FILE", shared) + conf := &remote_pb.RemoteConf{Name: "chain", Type: "s3", S3Endpoint: "https://93.184.216.34", S3Region: "us-east-1"} + first, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if err != nil { + t.Fatalf("build: %v", err) + } + if again, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false); again != first { + t.Fatal("unchanged credentials must share one client") + } + if err := os.WriteFile(shared, []byte("[default]\naws_access_key_id = c\naws_secret_access_key = d\n"), 0600); err != nil { + t.Fatal(err) + } + if next, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false); err != nil || next == first { + t.Fatalf("a changed shared credentials file must build a new client (err %v)", err) + } + awsConfig := filepath.Join(t.TempDir(), "config") + if err := os.WriteFile(awsConfig, []byte("[default]\nregion = us-east-1\n"), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("AWS_CONFIG_FILE", awsConfig) + withConfig, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if err != nil { + t.Fatalf("build: %v", err) + } + if err := os.WriteFile(awsConfig, []byte("[default]\nregion = us-west-2\n"), 0600); err != nil { + t.Fatal(err) + } + if next, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false); next == withConfig { + t.Fatal("a changed shared config file must build a new client") + } + t.Setenv("AWS_PROFILE", "other") + prev, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + t.Setenv("AWS_PROFILE", "third") + if next, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false); next == prev { + t.Fatal("a changed AWS_PROFILE must build a new client") + } +} + +// An s3-compatible remote whose keys come from its type's environment fallback is keyed by them too: a changed variable +// builds a new client even though nothing in the configuration changed. +func TestBuildGuardedRemoteStorageClientS3EnvironmentCredentials(t *testing.T) { + guardedCacheTest(t, nil) + t.Setenv("ALICLOUD_ACCESS_KEY_ID", "a") + t.Setenv("ALICLOUD_ACCESS_KEY_SECRET", "b") + conf := &remote_pb.RemoteConf{Name: "env", Type: "aliyun", AliyunEndpoint: "https://93.184.216.34"} + first, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false) + if err != nil { + t.Fatalf("build: %v", err) + } + if again, _ := BuildGuardedRemoteStorageClient(context.Background(), conf, false); again != first { + t.Fatal("unchanged environment credentials must share one client") + } + t.Setenv("ALICLOUD_ACCESS_KEY_SECRET", "c") + if next, err := BuildGuardedRemoteStorageClient(context.Background(), conf, false); err != nil || next == first { + t.Fatalf("a changed environment credential must build a new client (err %v)", err) + } +} diff --git a/weed/server/volume_grpc_remote_timeout_test.go b/weed/server/volume_grpc_remote_timeout_test.go new file mode 100644 index 0000000000..249eb21ef8 --- /dev/null +++ b/weed/server/volume_grpc_remote_timeout_test.go @@ -0,0 +1,159 @@ +package weed_server + +import ( + "context" + "io" + "net" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" +) + +// The guarded client of a remote carries the shared transport's cap and deadlines. +func TestNewGuardedHTTPClientHasDeadlines(t *testing.T) { + tr, ok := newGuardedHTTPClient("https://storage.example").Transport.(*http.Transport) + if !ok { + t.Fatal("not an *http.Transport") + } + if tr.ResponseHeaderTimeout != guardedResponseHeaderTimeout || tr.TLSHandshakeTimeout != guardedTLSHandshakeTimeout { + t.Fatalf("deadlines: header %v, tls %v", tr.ResponseHeaderTimeout, tr.TLSHandshakeTimeout) + } + if tr.MaxConnsPerHost != guardedMaxConnsPerHost() || tr.MaxIdleConnsPerHost != 32 { + t.Fatalf("caps: per host %d, idle per host %d", tr.MaxConnsPerHost, tr.MaxIdleConnsPerHost) + } +} + +// A remote that accepts and never answers: each request fails at the header deadline instead of hanging, and once the +// hung requests have freed the capped slots a later request to the same remote gets through. +func TestSharedTransportHungRequestsFreeTheirSlots(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/hang" { + select { + case <-release: + case <-r.Context().Done(): + } + return + } + io.WriteString(w, "ok") + })) + defer srv.Close() + defer close(release) + + tr := &http.Transport{DialContext: (&net.Dialer{}).DialContext} + tuneSharedTransport(tr) + tr.ResponseHeaderTimeout = 300 * time.Millisecond // the production value is a minute + tr.MaxConnsPerHost = 2 + client := &http.Client{Transport: tr} + + var wg sync.WaitGroup + errs := make(chan error, 2) + start := time.Now() + for i := 0; i < 2; i++ { // fill both slots + wg.Add(1) + go func() { + defer wg.Done() + resp, err := client.Get(srv.URL + "/hang") + if err == nil { + resp.Body.Close() + } + errs <- err + }() + } + time.Sleep(50 * time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/ok", nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("a request after the hung ones: %v (the slots were never freed)", err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if string(body) != "ok" { + t.Fatalf("body %q", body) + } + wg.Wait() + close(errs) + for err := range errs { + if err == nil { + t.Fatal("a hung request succeeded; expected the header deadline") + } + } + if d := time.Since(start); d > 3*time.Second { + t.Fatalf("took %v; the hung requests should fail at the 300 ms header deadline", d) + } +} + +// A remote that sends headers and then stalls mid-body: the connection's idle deadline cuts the stalled read, freeing +// its slot so a later request to the same remote gets through. +func TestSharedTransportStalledBodyFreesItsSlot(t *testing.T) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/stall" { + w.Header().Set("Content-Length", "100") + w.WriteHeader(http.StatusOK) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + select { + case <-release: + case <-r.Context().Done(): + } + return + } + io.WriteString(w, "ok") + })) + defer srv.Close() + defer close(release) + + old := guardedIOIdleTimeout + guardedIOIdleTimeout = 300 * time.Millisecond // the production value is minutes + defer func() { guardedIOIdleTimeout = old }() + + tr := &http.Transport{DialContext: (&net.Dialer{}).DialContext} + tuneSharedTransport(tr) + tr.MaxConnsPerHost = 2 + client := &http.Client{Transport: tr} + + var wg sync.WaitGroup + errs := make(chan error, 2) + start := time.Now() + for i := 0; i < 2; i++ { // fill both slots + wg.Add(1) + go func() { + defer wg.Done() + resp, err := client.Get(srv.URL + "/stall") + if err == nil { + _, err = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + } + errs <- err + }() + } + time.Sleep(50 * time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/ok", nil) + resp, err := client.Do(req) + if err != nil { + t.Fatalf("a request after the stalled ones: %v (the slots were never freed)", err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if string(body) != "ok" { + t.Fatalf("body %q", body) + } + wg.Wait() + close(errs) + for err := range errs { + if err == nil { + t.Fatal("a stalled body read succeeded; expected the idle deadline") + } + } + if d := time.Since(start); d > 3*time.Second { + t.Fatalf("took %v; the stalled reads should fail at the 300 ms idle deadline", d) + } +}