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) + } +}