diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 43e17789a..1b24446e9 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -1,6 +1,6 @@ # Azure Storage Fuse (cloudfuse) -Cloudfuse is a FUSE filesystem driver that provides virtual filesystem backed by S3 or Azure Blob Storage. It uses libfuse (fuse3) to communicate with the Linux FUSE kernel module and implements filesystem operations using the AWS S3 or Azure Storage REST APIs. +Cloudfuse is a FUSE filesystem driver that provides virtual filesystem backed by S3 or Azure Blob Storage. It uses libfuse (fuse3) to communicate with the Linux FUSE kernel module and implements filesystem operations using the AWS S3 or Azure Storage REST APIs. Always reference these instructions first and fallback to search or bash commands only when you encounter unexpected information that does not match the info here. @@ -11,26 +11,32 @@ Always reference these instructions first and fallback to search or bash command **CRITICAL**: All build and test commands include specific timeout warnings. NEVER CANCEL long-running operations. - Install required dependencies: + ```bash sudo apt update sudo apt install -y libfuse3-dev fuse3 gcc ``` - Install Go 1.27.0+ (already available in most environments; version is pinned in [go.mod](../go.mod)): + ```bash go version # Should show 1.27.0 or higher ``` - Build cloudfuse binary: + ```bash ./build.sh ``` + **Timing**: ~30 seconds. NEVER CANCEL. Use timeout 120+ seconds. - Build health monitor binary: + ```bash ./build.sh health ``` + **Timing**: ~5 seconds. Use timeout 60+ seconds. - Verify binary functionality: @@ -42,18 +48,23 @@ Always reference these instructions first and fallback to search or bash command ### Testing - Run unit tests (core components only): + ```bash go test -v -timeout=10m ./internal/... ./common/... --tags=unittest,fuse3 ``` + **Timing**: ~2 minutes. NEVER CANCEL. Use timeout 15+ minutes. - Run full unit tests (some may fail without Azure credentials): + ```bash go test -v -timeout=45m ./... --tags=unittest,fuse3 ``` + **WARNING**: Expected network/credential test failures. **Timing**: ~5-10 minutes. NEVER CANCEL. Use timeout 60+ minutes. - Run linting: + ```bash # Install golangci-lint if not available curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin @@ -61,6 +72,7 @@ Always reference these instructions first and fallback to search or bash command # Run linting $(go env GOPATH)/bin/golangci-lint run --tests=false --build-tags fuse3 --max-issues-per-linter=0 ``` + **Timing**: ~10 seconds. Use timeout 60+ seconds. - Check code formatting: @@ -74,6 +86,7 @@ Always reference these instructions first and fallback to search or bash command **ALWAYS test these scenarios after making changes**: 1. **Binary Creation and Basic Commands**: + ```bash ./build.sh ./cloudfuse --version @@ -82,6 +95,7 @@ Always reference these instructions first and fallback to search or bash command ``` 2. **Config Generation**: + ```bash mkdir -p /tmp/cloudfuse-test ./cloudfuse gen-config --tmp-path=/tmp/cloudfuse-test --o /tmp/cloudfuse-test/config.yaml @@ -89,6 +103,7 @@ Always reference these instructions first and fallback to search or bash command ``` 3. **Health Monitor**: + ```bash ./build.sh health ./cfusemon --help @@ -197,8 +212,9 @@ $(go env GOPATH)/bin/golangci-lint run --tests=false --build-tags fuse3 --max-is ## Key Files to Monitor When making changes, always check these files for consistency: + - `go.mod` - dependency versions - `main.go` - entry point - `build.sh` - build configuration - `cmd/mount.go` - core mount functionality -- Configuration templates in `setup/` and root directory \ No newline at end of file +- Configuration templates in `setup/` and root directory diff --git a/.gitignore b/.gitignore index 6bf9e7432..b5b21334c 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,9 @@ config*.yaml testcnf.yaml cloudfuse.log cloudfuse.test +blobfuse2.log +blobfuse2.log.* +blobfuse2.test azuretest.json cloudfuse cloudfuse.exe @@ -43,7 +46,15 @@ component/azstorage/logfile.txt logfile.txt **/logfile.txt **/logfile.txt.* +# Test trace files +*/_tmp_*.trace +_tmp_*.trace # Emacs backup and auto-save files *~ \#*\# -.\#* \ No newline at end of file +.\#* +# Python bytecode +__pycache__/ +*.pyc +*.pyo +.agents \ No newline at end of file diff --git a/cmd/mount.go b/cmd/mount.go index 3522ad277..389f1e267 100644 --- a/cmd/mount.go +++ b/cmd/mount.go @@ -499,6 +499,12 @@ var mountCmd = &cobra.Command{ return fmt.Errorf("failed to initialize logger [%s]", err.Error()) } + // It's best to destroy the logger before we return to the caller, as caller may abruptly exit on error and we + // might lose some logs in the channel which are not yet flushed to the file in case of not destroying the logger + defer func() { + _ = log.Destroy() + }() + if !disableVersionCheck { err := VersionCheck() if err != nil { @@ -527,6 +533,13 @@ var mountCmd = &cobra.Command{ log.Crit("Log options: %+v", options.Logging) log.Debug("Mount allowed on nonempty path : %v", options.NonEmpty) + // Mirror runtime panic/fatal stack traces to a log file (in addition to stderr, which the daemon library + // redirects to the per-mount .trace file). Also re-attaches the fd after in-process rotation and on SIGHUP + // from external rotators (logrotate, AKS Blob CSI driver, ...). Captures panics in any goroutine, including + // those spawned by libfuse callbacks. Called after the first log.Crit so that in syslog mode rsyslog has + // already created /var/log/Cloudfuse.log. + log.SetupCrashOutput(options.Logging.Type, options.Logging.LogFilePath) + if directIO { // Direct IO is enabled, so remove the attr-cache from the pipeline for i, name := range options.Components { diff --git a/cmd/root.go b/cmd/root.go index c28cc1999..cd04208a9 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -340,6 +340,28 @@ func parseArgs(cmdArgs []string) []string { return args } +func getTransport() *http.Transport { + // Prefer cloning the default transport so we inherit standard + // proxy/TLS settings (including HTTP(S)_PROXY and NO_PROXY). + if dt, ok := http.DefaultTransport.(*http.Transport); ok { + cp := dt.Clone() + cp.MaxIdleConns = 10 + cp.IdleConnTimeout = 30 * time.Second + cp.DisableCompression = true // GitHub API responses are small + cp.DisableKeepAlives = false // Connections are reused + return cp + } + + // Fallback: construct a transport that at least respects proxy env vars. + return &http.Transport{ + Proxy: http.ProxyFromEnvironment, + MaxIdleConns: 10, + IdleConnTimeout: 30 * time.Second, + DisableCompression: true, // GitHub API responses are small + DisableKeepAlives: false, // Connections are reused + } +} + // Execute : Actual command execution starts from here func Execute() error { parsedArgs := parseArgs(os.Args) diff --git a/cmd/root_test.go b/cmd/root_test.go index 841814abf..3acdaae10 100644 --- a/cmd/root_test.go +++ b/cmd/root_test.go @@ -97,7 +97,6 @@ func (suite *rootCmdSuite) SetupTest() { if err != nil { panic(fmt.Sprintf("Unable to set silent logger as default: %v", err)) } - // suite.testExecute() } func (suite *rootCmdSuite) cleanupTest() { diff --git a/common/cache_policy/lru_cache.go b/common/cache_policy/lru_cache.go new file mode 100644 index 000000000..9db0f9278 --- /dev/null +++ b/common/cache_policy/lru_cache.go @@ -0,0 +1,281 @@ +/* + Licensed under the MIT License . + + Copyright © 2023-2026 Seagate Technology LLC and/or its Affiliates + Copyright © 2020-2026 Microsoft Corporation. All rights reserved. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE +*/ + +package cache_policy + +import ( + "container/list" + "sync" + "unsafe" +) + +// mapBucketOverhead estimates the per-entry overhead from Go's internal hash map structure. +// Go's map uses a bucket-based design; 64 B covers the amortised cost of key/value copies +// and tophash bookkeeping per entry. +const mapBucketOverhead = 64 + +// lruItem is the value stored inside each list.Element in an LRU[K,V]. +type lruItem[K comparable, V any] struct { + key K + val V + userSize int64 // sizeOf(key, val) frozen at insertion time +} + +// LRU is a memory-bounded, generic least-recently-used cache. +// +// It is safe for concurrent use by multiple goroutines. +// +// Memory accounting works in two layers: +// - perEntryOverhead: fixed structural bytes per entry (list.Element + lruItem + map +// bucket), computed once in NewLRU via unsafe.Sizeof for the concrete K/V types. +// - sizeOf(k, v): caller-provided estimate of the heap bytes for one key-value pair, +// covering any heap allocations reachable from k and v (e.g. string data, pointed-to +// structs). It should NOT include the structural overhead above. +type LRU[K comparable, V any] struct { + mu sync.RWMutex + items map[K]*list.Element + list *list.List + maxSize int64 // 0 = no limit; immutable after construction + currSize int64 + perEntryOverhead int64 + sizeOf func(K, V) int64 +} + +// NewLRU creates an LRU with the given memory limit (bytes) and size estimator. +// maxSize == 0 disables memory-based eviction. +func NewLRU[K comparable, V any](maxSize int64, sizeOf func(K, V) int64) *LRU[K, V] { + var zeroItem lruItem[K, V] + overhead := int64(unsafe.Sizeof(zeroItem)) + + int64(unsafe.Sizeof(list.Element{})) + + mapBucketOverhead + l := &LRU[K, V]{ + items: make(map[K]*list.Element), + list: list.New(), + sizeOf: sizeOf, + perEntryOverhead: overhead, + maxSize: maxSize, + } + return l +} + +// Put inserts or updates a key-value pair and promotes it to MRU. +// After insertion, LRU entries are evicted until currSize <= maxSize. +// Returns false (and leaves the cache unchanged) if the entry's size alone +// exceeds maxSize — oversized entries are rejected upfront so existing entries +// are never displaced by a key that can never fit. +func (l *LRU[K, V]) Put(key K, val V) bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.put(key, val) +} + +// put is the lock-free inner implementation, called with l.mu write-locked. +func (l *LRU[K, V]) put(key K, val V) bool { + userSize := l.sizeOf(key, val) + // Reject entries that can never fit, so existing LRU contents are not displaced. + if l.maxSize > 0 && l.perEntryOverhead+userSize > l.maxSize { + return false + } + if elem, ok := l.items[key]; ok { + item := elem.Value.(*lruItem[K, V]) + l.currSize -= item.userSize + item.val = val + item.userSize = userSize + l.currSize += userSize + l.list.MoveToFront(elem) + l.evictIfNeeded() + return true + } + item := &lruItem[K, V]{key: key, val: val, userSize: userSize} + elem := l.list.PushFront(item) + l.items[key] = elem + l.currSize += l.perEntryOverhead + userSize + l.evictIfNeeded() + return true +} + +// Get retrieves a value and promotes it to MRU. Returns (zero, false) on miss. +func (l *LRU[K, V]) Get(key K) (V, bool) { + l.mu.Lock() + defer l.mu.Unlock() + if elem, ok := l.items[key]; ok { + l.list.MoveToFront(elem) + return elem.Value.(*lruItem[K, V]).val, true + } + var zero V + return zero, false +} + +// Peek retrieves a value without changing LRU order. Returns (zero, false) on miss. +func (l *LRU[K, V]) Peek(key K) (V, bool) { + l.mu.RLock() + defer l.mu.RUnlock() + if elem, ok := l.items[key]; ok { + return elem.Value.(*lruItem[K, V]).val, true + } + var zero V + return zero, false +} + +// Has reports whether key is present without changing LRU order. +func (l *LRU[K, V]) Has(key K) bool { + l.mu.RLock() + defer l.mu.RUnlock() + _, ok := l.items[key] + return ok +} + +// Delete removes an entry from the cache. No-op if the key is absent. +func (l *LRU[K, V]) Delete(key K) { + l.mu.Lock() + defer l.mu.Unlock() + l.delete(key) +} + +// delete is the lock-free inner implementation, called with l.mu write-locked. +func (l *LRU[K, V]) delete(key K) { + if elem, ok := l.items[key]; ok { + item := elem.Value.(*lruItem[K, V]) + l.currSize -= l.perEntryOverhead + item.userSize + delete(l.items, key) + l.list.Remove(elem) + } +} + +// Len returns the number of entries currently in the cache. +func (l *LRU[K, V]) Len() int { + l.mu.RLock() + defer l.mu.RUnlock() + return len(l.items) +} + +// Size returns the current estimated memory usage in bytes. +func (l *LRU[K, V]) Size() int64 { + l.mu.RLock() + defer l.mu.RUnlock() + return l.currSize +} + +// MaxSize returns the configured memory limit (0 = no limit). +// This value is immutable after construction and requires no locking. +func (l *LRU[K, V]) MaxSize() int64 { + return l.maxSize +} + +// PerEntryOverhead returns the structural bytes added per entry (useful for testing). +// This value is immutable after construction and requires no locking. +func (l *LRU[K, V]) PerEntryOverhead() int64 { + return l.perEntryOverhead +} + +// Range calls fn for every entry in MRU→LRU order. +// +// The cache is read-locked for the duration of the call. fn must not call any +// LRU methods (Put, Get, Delete, etc.) — doing so will deadlock. Mutating a +// value pointed to by V is safe as long as the caller manages value-level +// concurrency separately. +func (l *LRU[K, V]) Range(fn func(K, V) bool) { + l.mu.RLock() + defer l.mu.RUnlock() + for elem := l.list.Front(); elem != nil; elem = elem.Next() { + item := elem.Value.(*lruItem[K, V]) + if !fn(item.key, item.val) { + return + } + } +} + +// Purge removes all entries. +func (l *LRU[K, V]) Purge() { + l.mu.Lock() + defer l.mu.Unlock() + l.items = make(map[K]*list.Element) + l.list.Init() + l.currSize = 0 +} + +// DeleteIf removes all entries for which pred returns true. +// It holds the write lock for the entire scan — single pass, no intermediate allocation. +func (l *LRU[K, V]) DeleteIf(pred func(K, V) bool) { + l.mu.Lock() + defer l.mu.Unlock() + for elem := l.list.Front(); elem != nil; { + item := elem.Value.(*lruItem[K, V]) + next := elem.Next() + if pred(item.key, item.val) { + l.currSize -= l.perEntryOverhead + item.userSize + delete(l.items, item.key) + l.list.Remove(elem) + } + elem = next + } +} + +// ReplaceIf replaces the value of every entry for which pred returns true with newVal(key). +// The factory is called once per matching entry so each gets its own value. +// It holds the write lock for the entire scan — no entry matching the predicate can be +// inserted or removed by another goroutine between the scan and the replacements. +func (l *LRU[K, V]) ReplaceIf(pred func(K, V) bool, newVal func(K) V) { + l.mu.Lock() + defer l.mu.Unlock() + for elem := l.list.Front(); elem != nil; { + next := elem.Next() + item := elem.Value.(*lruItem[K, V]) + if pred(item.key, item.val) { + v := newVal(item.key) + newSize := l.sizeOf(item.key, v) + l.currSize += newSize - item.userSize + item.val = v + item.userSize = newSize + l.list.MoveToFront(elem) + } + elem = next + } + l.evictIfNeeded() +} + +// evictIfNeeded removes the least-recently-used entries until currSize <= maxSize. +// Must be called with l.mu write-locked. +func (l *LRU[K, V]) evictIfNeeded() { + if l.maxSize <= 0 { + return + } + for l.currSize > l.maxSize && l.list.Len() > 0 { + l.evictLast() + } +} + +// evictLast removes the least-recently-used entry. +// Must be called with l.mu write-locked. +func (l *LRU[K, V]) evictLast() { + elem := l.list.Back() + if elem == nil { + return + } + item := elem.Value.(*lruItem[K, V]) + l.currSize -= l.perEntryOverhead + item.userSize + delete(l.items, item.key) + l.list.Remove(elem) +} diff --git a/common/cache_policy/lru_cache_test.go b/common/cache_policy/lru_cache_test.go new file mode 100644 index 000000000..1429c4cd8 --- /dev/null +++ b/common/cache_policy/lru_cache_test.go @@ -0,0 +1,591 @@ +/* + Licensed under the MIT License . + + Copyright © 2023-2026 Seagate Technology LLC and/or its Affiliates + Copyright © 2020-2026 Microsoft Corporation. All rights reserved. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE +*/ + +package cache_policy + +import ( + "container/list" + "sync" + "testing" + "unsafe" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/suite" +) + +type lruCacheTestSuite struct { + suite.Suite + assert *assert.Assertions +} + +func (s *lruCacheTestSuite) SetupTest() { + s.assert = assert.New(s.T()) +} + +// identity sizeOf: returns len(key) + len(val) for string caches. +func strSizeOf(k, v string) int64 { + return int64(len(k) + len(v)) +} + +// zeroSizeOf ignores key and value; useful when we want to reason about overhead only. +func zeroSizeOf[K comparable, V any](K, V) int64 { return 0 } + +// ---- basic operations ---- + +func (s *lruCacheTestSuite) TestBasicPutGet() { + lru := NewLRU[string, string](0, strSizeOf) + lru.Put("a", "1") + lru.Put("b", "2") + + v, ok := lru.Get("a") + s.assert.True(ok) + s.assert.Equal("1", v) + + v, ok = lru.Get("b") + s.assert.True(ok) + s.assert.Equal("2", v) + + _, ok = lru.Get("missing") + s.assert.False(ok) +} + +func (s *lruCacheTestSuite) TestPeekDoesNotPromote() { + lru := NewLRU[string, string](0, strSizeOf) + lru.Put("a", "1") + lru.Put("b", "2") // b is now MRU + + // Peek at "a" — should not promote it + v, ok := lru.Peek("a") + s.assert.True(ok) + s.assert.Equal("1", v) + + // Without any limit the order won't matter for eviction, but verify via Range (MRU first) + var order []string + lru.Range(func(k, _ string) bool { + order = append(order, k) + return true + }) + s.assert.Equal([]string{"b", "a"}, order) // b is still MRU +} + +func (s *lruCacheTestSuite) TestHas() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + lru.Put("x", 42) + s.assert.True(lru.Has("x")) + s.assert.False(lru.Has("y")) +} + +func (s *lruCacheTestSuite) TestDelete() { + lru := NewLRU[string, string](0, strSizeOf) + lru.Put("a", "1") + lru.Put("b", "2") + lru.Delete("a") + + s.assert.False(lru.Has("a")) + s.assert.True(lru.Has("b")) + s.assert.Equal(1, lru.Len()) +} + +func (s *lruCacheTestSuite) TestDeleteAbsent() { + lru := NewLRU[string, string](0, strSizeOf) + // Should not panic + lru.Delete("nonexistent") + s.assert.Equal(0, lru.Len()) +} + +func (s *lruCacheTestSuite) TestLen() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + s.assert.Equal(0, lru.Len()) + lru.Put("a", 1) + s.assert.Equal(1, lru.Len()) + lru.Put("b", 2) + s.assert.Equal(2, lru.Len()) + lru.Delete("a") + s.assert.Equal(1, lru.Len()) +} + +func (s *lruCacheTestSuite) TestPurge() { + lru := NewLRU[string, string](0, strSizeOf) + lru.Put("a", "1") + lru.Put("b", "2") + lru.Purge() + s.assert.Equal(0, lru.Len()) + s.assert.Equal(int64(0), lru.Size()) + s.assert.False(lru.Has("a")) +} + +// ---- size accounting ---- + +func (s *lruCacheTestSuite) TestSizeAccounting() { + lru := NewLRU[string, string](0, strSizeOf) + overhead := lru.PerEntryOverhead() + + lru.Put("ab", "cd") // userSize = 2+2 = 4 + s.assert.Equal(overhead+4, lru.Size()) + + lru.Put("ab", "cdef") // update: userSize changes from 4 to 6 + s.assert.Equal(overhead+6, lru.Size()) + + lru.Delete("ab") + s.assert.Equal(int64(0), lru.Size()) +} + +func (s *lruCacheTestSuite) TestPerEntryOverheadUsesUnsafeSizeof() { + lru := NewLRU[string, string](0, strSizeOf) + overhead := lru.PerEntryOverhead() + + // Overhead must cover at least: lruItem struct + list.Element struct + var item lruItem[string, string] + minExpected := int64(unsafe.Sizeof(item)) + int64(unsafe.Sizeof(list.Element{})) + s.assert.GreaterOrEqual(overhead, minExpected, + "perEntryOverhead should be >= sizeof(lruItem) + sizeof(list.Element)") + // And must include the map bucket overhead constant + s.assert.GreaterOrEqual(overhead, minExpected+mapBucketOverhead) +} + +// ---- eviction ---- + +func (s *lruCacheTestSuite) TestNoLimitNoEviction() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) // maxSize=0 means no limit + for i := 0; i < 10000; i++ { + k := string(rune('a' + i%26)) + lru.Put(k, i) + } + // All distinct keys are "a".."z" (26 entries), none evicted + s.assert.Equal(26, lru.Len()) +} + +func (s *lruCacheTestSuite) TestEvictionAtCapacity() { + // Allow room for exactly 2 entries (overhead + 0 user bytes each) + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + lru = NewLRU[string, int](overhead*2, zeroSizeOf[string, int]) + + lru.Put("a", 1) + lru.Put("b", 2) + s.assert.Equal(2, lru.Len()) + + // Adding "c" should evict "a" (LRU) + lru.Put("c", 3) + s.assert.Equal(2, lru.Len()) + s.assert.False(lru.Has("a"), "LRU entry 'a' should have been evicted") + s.assert.True(lru.Has("b")) + s.assert.True(lru.Has("c")) +} + +func (s *lruCacheTestSuite) TestEvictionPromotion() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + lru = NewLRU[string, int](overhead*2, zeroSizeOf[string, int]) + + lru.Put("a", 1) + lru.Put("b", 2) // order: b(MRU), a(LRU) + + // Promote "a" via Get + lru.Get("a") // order: a(MRU), b(LRU) + + // Adding "c" should evict "b" (now LRU), not "a" + lru.Put("c", 3) + s.assert.Equal(2, lru.Len()) + s.assert.True(lru.Has("a"), "'a' was recently accessed and should survive") + s.assert.False(lru.Has("b"), "'b' should be evicted (LRU)") + s.assert.True(lru.Has("c")) +} + +func (s *lruCacheTestSuite) TestEvictionOrder() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + lru = NewLRU[string, int](overhead*3, zeroSizeOf[string, int]) + + lru.Put("a", 1) // LRU order: a + lru.Put("b", 2) // LRU order: b, a + lru.Put("c", 3) // LRU order: c, b, a — full + + lru.Put("d", 4) // evicts "a" + s.assert.False(lru.Has("a")) + lru.Put("e", 5) // evicts "b" + s.assert.False(lru.Has("b")) + + s.assert.True(lru.Has("c")) + s.assert.True(lru.Has("d")) + s.assert.True(lru.Has("e")) +} + +// ---- MaxSize accessor ---- + +func (s *lruCacheTestSuite) TestMaxSizeAccessor() { + lru := NewLRU[string, int](1024, zeroSizeOf[string, int]) + s.assert.Equal(int64(1024), lru.MaxSize()) +} + +// ---- oversized entry rejection ---- + +func (s *lruCacheTestSuite) TestOversizedEntryRejected() { + lru := NewLRU[string, string](0, strSizeOf) + overhead := lru.PerEntryOverhead() + lru = NewLRU[string, string](overhead+4, strSizeOf) // room for 1 entry with 4 user bytes + + lru.Put("ab", "cd") // userSize=4, fits exactly + s.assert.Equal(1, lru.Len()) + + // Update with a value too large to ever fit — rejected upfront, original entry survives. + retained := lru.Put("ab", "cdefgh") // userSize=8, exceeds limit + s.assert.False(retained, "entry whose size exceeds maxSize should be rejected") + s.assert.Equal(1, lru.Len(), "existing entry must not be displaced") + v, ok := lru.Get("ab") + s.assert.True(ok) + s.assert.Equal("cd", v, "original value must be unchanged") +} + +// ---- Range ---- + +func (s *lruCacheTestSuite) TestRange() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + lru.Put("a", 1) + lru.Put("b", 2) + lru.Put("c", 3) // MRU order: c, b, a + + var keys []string + lru.Range(func(k string, _ int) bool { + keys = append(keys, k) + return true + }) + s.assert.Equal([]string{"c", "b", "a"}, keys) +} + +func (s *lruCacheTestSuite) TestRangeEarlyStop() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + for _, k := range []string{"a", "b", "c", "d"} { + lru.Put(k, 1) + } + + count := 0 + lru.Range(func(_ string, _ int) bool { + count++ + return count < 2 // stop after 2 + }) + s.assert.Equal(2, count) +} + +// ---- integration: integer keys ---- + +func (s *lruCacheTestSuite) TestIntKeys() { + sizeOf := func(k int, v []byte) int64 { return int64(len(v)) } + lru := NewLRU[int, []byte](0, sizeOf) + lru.Put(1, []byte("hello")) + lru.Put(2, []byte("world")) + + v, ok := lru.Get(1) + s.assert.True(ok) + s.assert.Equal([]byte("hello"), v) + s.assert.Equal(2, lru.Len()) +} + +// ---- frozen userSize ---- + +func (s *lruCacheTestSuite) TestFrozenUserSizeOnMutation() { + // sizeOf reports the length of the slice the pointer points to. + // After insertion, we grow the slice — sizeOf would return a larger value, + // but the LRU must still use the size frozen at insertion time. + type buf struct{ data []byte } + sizeOf := func(_ string, v *buf) int64 { return int64(len(v.data)) } + + lru := NewLRU[string, *buf](0, sizeOf) + overhead := lru.PerEntryOverhead() + + b := &buf{data: []byte("hi")} // userSize = 2 at insertion + lru.Put("k", b) + s.assert.Equal(overhead+2, lru.Size()) + + // Mutate the pointed-to value — sizeOf would now return 100 + b.data = make([]byte, 100) + // Size must remain overhead+2 (frozen at insertion) + s.assert.Equal(overhead+2, lru.Size()) + + // Delete must subtract the frozen size, not the current size + lru.Delete("k") + s.assert.Equal(int64(0), lru.Size()) +} + +func (s *lruCacheTestSuite) TestFrozenUserSizeOnUpdate() { + // When Put is called again for an existing key the new userSize replaces the old one. + type buf struct{ data []byte } + sizeOf := func(_ string, v *buf) int64 { return int64(len(v.data)) } + + lru := NewLRU[string, *buf](0, sizeOf) + overhead := lru.PerEntryOverhead() + + b := &buf{data: []byte("hi")} + lru.Put("k", b) + + // Update with a value that reports a different size at Put time + b2 := &buf{data: make([]byte, 50)} + lru.Put("k", b2) + s.assert.Equal(overhead+50, lru.Size()) + + // Now mutate b2 — size must stay at 50 + b2.data = make([]byte, 200) + s.assert.Equal(overhead+50, lru.Size()) + + lru.Delete("k") + s.assert.Equal(int64(0), lru.Size()) +} + +// ---- delete size accounting ---- + +func (s *lruCacheTestSuite) TestDeleteSizeAccounting() { + lru := NewLRU[string, string](0, strSizeOf) + overhead := lru.PerEntryOverhead() + + lru.Put("ab", "cd") // userSize = 4 + s.assert.Equal(overhead+4, lru.Size()) + + lru.Delete("ab") + s.assert.Equal(int64(0), lru.Size()) + s.assert.Equal(0, lru.Len()) +} + +// ---- DeleteIf ---- + +func (s *lruCacheTestSuite) TestDeleteIf() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + + for _, k := range []string{"a", "b", "c", "d"} { + lru.Put(k, 1) + } + s.assert.Equal(int64(4)*overhead, lru.Size()) + + // Remove entries whose key >= "c" + lru.DeleteIf(func(k string, _ int) bool { return k >= "c" }) + + s.assert.Equal(2, lru.Len()) + s.assert.True(lru.Has("a")) + s.assert.True(lru.Has("b")) + s.assert.False(lru.Has("c")) + s.assert.False(lru.Has("d")) + s.assert.Equal(int64(2)*overhead, lru.Size()) +} + +func (s *lruCacheTestSuite) TestDeleteIfAll() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + lru.Put("a", 1) + lru.Put("b", 2) + + lru.DeleteIf(func(_ string, _ int) bool { return true }) + + s.assert.Equal(0, lru.Len()) + s.assert.Equal(int64(0), lru.Size()) +} + +func (s *lruCacheTestSuite) TestDeleteIfNone() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + lru.Put("a", 1) + lru.Put("b", 2) + + lru.DeleteIf(func(_ string, _ int) bool { return false }) + + s.assert.Equal(2, lru.Len()) + s.assert.Equal(int64(2)*overhead, lru.Size()) +} + +func (s *lruCacheTestSuite) TestDeleteIfEmpty() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + // Should not panic on empty cache + lru.DeleteIf(func(_ string, _ int) bool { return true }) + s.assert.Equal(0, lru.Len()) +} + +func (s *lruCacheTestSuite) TestReplaceIf() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + for _, k := range []string{"a", "b", "c", "d"} { + lru.Put(k, 1) + } + + lru.ReplaceIf(func(k string, _ int) bool { return k >= "c" }, func(_ string) int { return 99 }) + + s.assert.Equal(4, lru.Len()) + s.assert.Equal(int64(4)*overhead, lru.Size()) + v, ok := lru.Peek("a") + s.assert.True(ok) + s.assert.Equal(1, v) + v, ok = lru.Peek("c") + s.assert.True(ok) + s.assert.Equal(99, v) + v, ok = lru.Peek("d") + s.assert.True(ok) + s.assert.Equal(99, v) +} + +func (s *lruCacheTestSuite) TestReplaceIfAll() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + lru.Put("a", 1) + lru.Put("b", 2) + + lru.ReplaceIf(func(_ string, _ int) bool { return true }, func(_ string) int { return 0 }) + + s.assert.Equal(2, lru.Len()) + v, _ := lru.Peek("a") + s.assert.Equal(0, v) + v, _ = lru.Peek("b") + s.assert.Equal(0, v) +} + +func (s *lruCacheTestSuite) TestReplaceIfNone() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + overhead := lru.PerEntryOverhead() + lru.Put("a", 1) + lru.Put("b", 2) + + lru.ReplaceIf(func(_ string, _ int) bool { return false }, func(_ string) int { return 99 }) + + s.assert.Equal(2, lru.Len()) + s.assert.Equal(int64(2)*overhead, lru.Size()) + v, _ := lru.Peek("a") + s.assert.Equal(1, v) + v, _ = lru.Peek("b") + s.assert.Equal(2, v) +} + +func (s *lruCacheTestSuite) TestReplaceIfEmpty() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + // Should not panic on empty cache + lru.ReplaceIf(func(_ string, _ int) bool { return true }, func(_ string) int { return 0 }) + s.assert.Equal(0, lru.Len()) +} + +func (s *lruCacheTestSuite) TestReplaceIfPromotesToMRU() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + lru.Put("a", 1) + lru.Put("b", 2) // b is MRU + + lru.ReplaceIf(func(k string, _ int) bool { return k == "a" }, func(_ string) int { return 99 }) + + // "a" should have been promoted to MRU after replacement + var first string + lru.Range(func(k string, _ int) bool { first = k; return false }) + s.assert.Equal("a", first) +} + +// ---- purge then reuse ---- + +func (s *lruCacheTestSuite) TestPurgeAndReuse() { + lru := NewLRU[string, string](0, strSizeOf) + lru.Put("a", "1") + lru.Put("b", "2") + lru.Purge() + + // Cache must be fully functional after Purge + lru.Put("c", "3") + v, ok := lru.Get("c") + s.assert.True(ok) + s.assert.Equal("3", v) + s.assert.Equal(1, lru.Len()) + s.assert.False(lru.Has("a")) + s.assert.False(lru.Has("b")) +} + +// ---- Range on empty cache ---- + +func (s *lruCacheTestSuite) TestRangeOnEmptyCache() { + lru := NewLRU[string, int](0, zeroSizeOf[string, int]) + calls := 0 + lru.Range(func(_ string, _ int) bool { + calls++ + return true + }) + s.assert.Equal(0, calls) +} + +// ---- Peek on absent key ---- + +func (s *lruCacheTestSuite) TestPeekAbsent() { + lru := NewLRU[string, string](0, strSizeOf) + v, ok := lru.Peek("missing") + s.assert.False(ok) + s.assert.Empty(v) +} + +// ---- concurrent access ---- + +func (s *lruCacheTestSuite) TestConcurrentAccess() { + // Run under `go test -race` to detect data races. + lru := NewLRU[int, int](0, zeroSizeOf[int, int]) + const goroutines = 8 + const ops = 500 + + var wg sync.WaitGroup + wg.Add(goroutines) + for g := 0; g < goroutines; g++ { + go func(id int) { + defer wg.Done() + for i := 0; i < ops; i++ { + key := (id*ops + i) % 64 + lru.Put(key, id*ops+i) + lru.Get(key) + lru.Peek(key) + lru.Has(key) + if i%10 == 0 { + lru.Delete(key) + } + } + }(g) + } + wg.Wait() + // Just verify the LRU is in a consistent state + s.assert.GreaterOrEqual(lru.Len(), 0) + s.assert.GreaterOrEqual(lru.Size(), int64(0)) +} + +func (s *lruCacheTestSuite) TestConcurrentAccessWithLimit() { + lru := NewLRU[int, int](0, zeroSizeOf[int, int]) + overhead := lru.PerEntryOverhead() + lru = NewLRU[int, int](overhead*32, zeroSizeOf[int, int]) + + const goroutines = 8 + const ops = 500 + + var wg sync.WaitGroup + wg.Add(goroutines) + for g := 0; g < goroutines; g++ { + go func(id int) { + defer wg.Done() + for i := 0; i < ops; i++ { + key := (id*ops + i) % 128 + lru.Put(key, id*ops+i) + lru.Get(key) + } + }(g) + } + wg.Wait() + + s.assert.LessOrEqual(lru.Size(), overhead*32) +} + +func TestLRUCacheTestSuite(t *testing.T) { + suite.Run(t, new(lruCacheTestSuite)) +} diff --git a/common/log/base_logger.go b/common/log/base_logger.go index 90d7e4d92..a6823359b 100644 --- a/common/log/base_logger.go +++ b/common/log/base_logger.go @@ -315,5 +315,8 @@ func (l *BaseLogger) LogRotate() error { l.logger.SetOutput(l.logFileHandle) l.fileConfig.currentLogSize = 0 + // Notify post-rotation hooks (e.g. runtime crash output) that the underlying file has changed. + invokeRotateHooks() + return err } diff --git a/common/log/logger.go b/common/log/logger.go index 561a427da..a790d67ff 100644 --- a/common/log/logger.go +++ b/common/log/logger.go @@ -28,6 +28,12 @@ package log import ( "fmt" "log" + "os" + "os/signal" + "runtime/debug" + "sync" + "sync/atomic" + "syscall" "time" "github.com/Seagate/cloudfuse/common" @@ -182,6 +188,203 @@ func LogRotate() error { return logObj.LogRotate() } +// ------------------ Runtime crash-output mirroring + rotate hooks ------------------ +// +// This section wires Go runtime crash dumps (panics, fatal errors from any goroutine) to the +// Cloudfuse log file in addition to the per-mount .trace file, and keeps that crash-output fd +// attached to the live log file across rotations. Entry point: SetupCrashOutput. + +var ( + // rotateHooksMu guards rotateHooks against concurrent registration and invocation. + rotateHooksMu sync.Mutex + + // rotateHooks holds the callbacks invoked after each successful log rotation. + rotateHooks []func() + + // setupCrashOutputOnce makes SetupCrashOutput idempotent: multiple calls (from tests, or if + // the mount entry point ever runs more than once) will not accumulate duplicate rotate hooks. + // The first successful call wins; subsequent calls with different arguments are ignored. + setupCrashOutputOnce sync.Once + + // sighupOnce ensures the SIGHUP listener goroutine is only started once per process. + sighupOnce sync.Once + + // sighupCh is the channel signal.Notify writes to; retained at package scope so tests can + // shut the listener goroutine down cleanly (production never reads it back). + sighupCh chan os.Signal + + // sighupInstalled reports whether the SIGHUP listener has been installed. Read from tests + // via sync/atomic; production code has no reason to read it. + sighupInstalled atomic.Bool + + // sighupHandled counts the number of SIGHUPs the listener goroutine has processed. Used by + // tests to verify that the crash-output handler actually ran (as opposed to merely observing + // that the signal was delivered to the process). + sighupHandled atomic.Uint64 +) + +// registerLogRotateHook registers fn to be invoked after each successful log rotation performed by the +// underlying logger. Useful for callers that hold a file descriptor to the log file (e.g. runtime +// crash output) and need to re-open it after rotation. Hooks are invoked sequentially and must be +// cheap and non-blocking; they run inline on the rotation path. +func registerLogRotateHook(fn func()) { + if fn == nil { + return + } + rotateHooksMu.Lock() + rotateHooks = append(rotateHooks, fn) + rotateHooksMu.Unlock() +} + +// invokeRotateHooks runs all registered post-rotation hooks. Called by logger implementations +// after a successful rotation. Each hook runs under its own defer/recover so a panicking hook +// cannot take down the log-dumper goroutine or the rotation path -- rotation is a best-effort +// notification, not a critical correctness path. +func invokeRotateHooks() { + rotateHooksMu.Lock() + hooks := append([]func(){}, rotateHooks...) + rotateHooksMu.Unlock() + for _, fn := range hooks { + func() { + defer func() { + if r := recover(); r != nil { + fmt.Fprintf(os.Stderr, "log: rotate hook panicked [%v]\n", r) + } + }() + fn() + }() + } +} + +// SetupCrashOutput wires the Go runtime crash output (panics, fatal errors from any goroutine) to the +// Cloudfuse log file in addition to stderr (which the daemon library redirects to the per-mount +// .trace file). Supported logger modes: +// +// - "base" -> writes to logFilePath (the configured log file). Skipped when logFilePath is empty +// or "stdout". +// - "" / "default" / "syslog" -> writes to common.SyslogFilePath (the rsyslog sink for cloudfuse +// messages, declared in setup/11-cloudfuse.conf). +// +// A no-op for the "silent" logger, unknown logger types, and base-with-stdout -- in those cases no +// rotate hook is registered and no SIGHUP handler is installed. +// +// The crash output fd is kept attached to the live log file across rotations via two mechanisms +// selected per logger mode: +// 1. "base" -> BaseLogger's in-process size-based rotation invokes the registered rotate hook. +// No SIGHUP handler is installed because BaseLogger owns its file and does not participate in +// external rotation. +// 2. syslog family -> external rotators (logrotate's postrotate, the AKS Blob CSI driver, ...) +// signal the process via SIGHUP after rotating /var/log/cloudfuse.log aside. SysLogger has no +// in-process rotation, so SIGHUP is the only trigger. +func SetupCrashOutput(loggerType, logFilePath string) { + // Skip the whole setup (crash-output fd, rotate hook, SIGHUP goroutine) when the logger + // configuration has no meaningful file target to mirror crash dumps to. + if crashOutputTarget(loggerType, logFilePath) == "" { + return + } + + setupCrashOutputOnce.Do(func() { + setCrashOutput(loggerType, logFilePath) + + // Re-attach the crash output fd after BaseLogger's in-process size-based rotation. The rename + // moves the original file aside while our dup'd fd stays pinned to the old inode, so we must + // re-open the live file. Registered unconditionally: it is a no-op for logger types whose + // LogRotate() implementations never invoke the hook (SysLogger, SilentLogger). + registerLogRotateHook(func() { + setCrashOutput(loggerType, logFilePath) + }) + + // SIGHUP is only meaningful for the syslog family: SysLogger has no in-process rotation, so + // external rotators (logrotate + postrotate, AKS Blob CSI driver, ...) are the only way to + // know that /var/log/cloudfuse.log has been rotated. For "base" mode the in-process hook above + // already covers rotation and we don't want to hijack SIGHUP for other consumers. + if isSyslogFamily(loggerType) { + installCrashSighupHandler(loggerType, logFilePath) + } + }) +} + +// isSyslogFamily reports whether loggerType routes crash output to common.SyslogFilePath and +// therefore relies on external rotators + SIGHUP for rotation notification. +func isSyslogFamily(loggerType string) bool { + switch loggerType { + case "", "default", "syslog": + return true + default: + return false + } +} + +// crashOutputTarget returns the filesystem path where runtime crash dumps should be mirrored for +// the given logger configuration, or "" if the configuration does not support mirroring crash +// dumps to a log file (silent logger, stdout base logger, unknown types, ...). +func crashOutputTarget(loggerType, logFilePath string) string { + switch loggerType { + case "base": + // BaseLogger may be configured with "stdout" or no file at all; nothing useful to also write to. + if logFilePath == "" || logFilePath == "stdout" { + return "" + } + return logFilePath + case "", "default", "syslog": + // syslog can't be redirected to via a file descriptor, so target the rsyslog sink for cloudfuse messages. + return common.SyslogFilePath + default: + // "silent" or unknown logger. + return "" + } +} + +func setCrashOutput(loggerType, logFilePath string) { + // Panic-safe: this can be invoked from the SIGHUP handler goroutine that outlives log.Destroy(), + // which closes BaseLogger's channel. Swallow any panic so a signal during teardown cannot crash + // the process. The .trace file still captures runtime panics via stderr redirection. + defer func() { _ = recover() }() + + crashFilePath := crashOutputTarget(loggerType, logFilePath) + if crashFilePath == "" { + return + } + + // Open without O_CREATE: in base mode BaseLogger has already created the file; in syslog mode rsyslog owns + // the file and must have created it first. If the file is missing (e.g. rsyslog not restarted yet on a fresh + // install), let it fail -- the per-mount .trace file still captures the panic via stderr redirection. + f, err := os.OpenFile(crashFilePath, os.O_WRONLY|os.O_APPEND, 0) + if err != nil { + // Write diagnostics directly to stderr rather than through Warn(). This function runs on the + // log-rotation callback path (invokeRotateHooks) and on the SIGHUP handler goroutine; a Warn() + // here would go through BaseLogger's buffered channel, which can block if full or panic if the + // logger has been destroyed. Stderr is non-blocking and always safe. + fmt.Fprintf(os.Stderr, "log: failed to open %s for crash output [%s]\n", crashFilePath, err.Error()) + return + } + // SetCrashOutput dups the fd, so the file handle can be closed immediately. + defer f.Close() + + if err := debug.SetCrashOutput(f, debug.CrashOptions{}); err != nil { + fmt.Fprintf(os.Stderr, "log: failed to set crash output to %s [%s]\n", crashFilePath, err.Error()) + } +} + +// installCrashSighupHandler arranges for the crash output fd to be re-attached when the process receives +// SIGHUP. Intended for the syslog family only: SysLogger has no in-process rotation, so external +// rotators (logrotate's postrotate, the AKS Blob CSI driver's rotation hook, etc.) are the only +// way to be notified that /var/log/cloudfuse.log has been rotated aside. Guarded by sync.Once so the +// listener is installed at most once per process. +func installCrashSighupHandler(loggerType, logFilePath string) { + sighupOnce.Do(func() { + sighupCh = make(chan os.Signal, 1) + signal.Notify(sighupCh, syscall.SIGHUP) + sighupInstalled.Store(true) + go func() { + for range sighupCh { + setCrashOutput(loggerType, logFilePath) + sighupHandled.Add(1) + } + }() + }) +} + func init() { logObj, _ = NewLogger("base", common.LogConfig{ Level: common.ELogLevel.LOG_DEBUG(), diff --git a/common/log/logger_test.go b/common/log/logger_test.go index 312f6fe33..716152fd7 100644 --- a/common/log/logger_test.go +++ b/common/log/logger_test.go @@ -26,7 +26,15 @@ package log import ( + "os" + "os/signal" + "path/filepath" + "runtime/debug" + "sync" + "sync/atomic" + "syscall" "testing" + "time" "github.com/Seagate/cloudfuse/common" @@ -136,6 +144,288 @@ func (lts *LoggerTestSuite) TestNegative() { assert.Error(err, "Negative : did not get logger object") } +// resetCrashOutputState clears process-global state used by the crash-output / rotation-hook +// machinery so each test starts from a known baseline. logObj is swapped for a fresh silent logger +// because earlier tests may have called Destroy() on a base logger, which closes its channel and +// would cause subsequent Warn/Info calls (e.g. from setCrashOutput) to panic. +// +// The previously-installed SIGHUP listener goroutine (if any) is stopped: signal.Stop unregisters +// the channel from the runtime signal dispatch, and closing the channel lets the "for range" +// goroutine drain the buffer and exit. Without this, tests that call SetupCrashOutput repeatedly +// would leak goroutines still registered for SIGHUP, and future SIGHUPs would fan out to all of +// them, making later tests flaky. +func resetCrashOutputState() { + rotateHooksMu.Lock() + rotateHooks = nil + rotateHooksMu.Unlock() + + if sighupInstalled.Load() && sighupCh != nil { + signal.Stop(sighupCh) + close(sighupCh) + sighupCh = nil + } + sighupOnce = sync.Once{} + sighupInstalled.Store(false) + sighupHandled.Store(0) + setupCrashOutputOnce = sync.Once{} + + _ = debug.SetCrashOutput(nil, debug.CrashOptions{}) + logObj = &SilentLogger{} +} + +func hookCount() int { + rotateHooksMu.Lock() + defer rotateHooksMu.Unlock() + return len(rotateHooks) +} + +func (lts *LoggerTestSuite) TestRegisterLogRotateHook() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + // nil hook must be a no-op. + registerLogRotateHook(nil) + assert.Equal(0, hookCount()) + + var order []int + registerLogRotateHook(func() { order = append(order, 1) }) + registerLogRotateHook(func() { order = append(order, 2) }) + registerLogRotateHook(func() { order = append(order, 3) }) + assert.Equal(3, hookCount()) + + invokeRotateHooks() + assert.Equal([]int{1, 2, 3}, order, "hooks must fire in registration order") + + // Re-invocation must re-run all hooks (they are not one-shot). + invokeRotateHooks() + assert.Equal([]int{1, 2, 3, 1, 2, 3}, order) +} + +// TestInvokeRotateHooksIsolatesPanics validates that a panicking hook does not prevent later +// hooks from running and does not propagate out of invokeRotateHooks. This matters because +// invokeRotateHooks runs inline on BaseLogger's log-rotation path; a bare panic there would +// take down the log-dumper goroutine and effectively break logging. +func (lts *LoggerTestSuite) TestInvokeRotateHooksIsolatesPanics() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + var ran []int + registerLogRotateHook(func() { ran = append(ran, 1) }) + registerLogRotateHook(func() { panic("hook 2 exploded") }) + registerLogRotateHook(func() { ran = append(ran, 3) }) + + assert.NotPanics(invokeRotateHooks, "invokeRotateHooks must swallow per-hook panics") + assert.Equal([]int{1, 3}, ran, "hooks after a panicking hook must still run") +} + +func (lts *LoggerTestSuite) TestBaseLoggerRotateInvokesHook() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + tmpDir := lts.T().TempDir() + cfg := common.LogConfig{ + FilePath: filepath.Join(tmpDir, "rotate.log"), + MaxFileSize: 1, + FileCount: 3, + Level: common.ELogLevel.LOG_DEBUG(), + } + err := SetDefaultLogger("base", cfg) + assert.NoError(err) + defer func() { _ = Destroy() }() + + var fired int32 + registerLogRotateHook(func() { atomic.AddInt32(&fired, 1) }) + + assert.NoError(LogRotate()) + assert.Equal(int32(1), atomic.LoadInt32(&fired)) + + assert.NoError(LogRotate()) + assert.Equal(int32(2), atomic.LoadInt32(&fired)) +} + +func (lts *LoggerTestSuite) TestSetCrashOutput() { + assert := assert.New(lts.T()) + + // "base" with a real, writable file -- success path; runtime crash output is updated. + resetCrashOutputState() + tmp, err := os.CreateTemp("", "cloudfuse-crash-base-*.log") + assert.NoError(err) + defer os.Remove(tmp.Name()) + assert.NoError(tmp.Close()) + assert.NotPanics(func() { setCrashOutput("base", tmp.Name()) }) + + // "base" with empty path or "stdout" -- early no-op, no panic. + assert.NotPanics(func() { setCrashOutput("base", "") }) + assert.NotPanics(func() { setCrashOutput("base", "stdout") }) + + // "silent" and unknown logger types -- early no-op, no panic. + assert.NotPanics(func() { setCrashOutput("silent", "ignored") }) + assert.NotPanics(func() { setCrashOutput("not-a-real-type", "ignored") }) + + // "base" pointing at a non-existent path -- must Warn and return (no panic, no crash). + // O_CREATE is intentionally not used, so missing files are tolerated. + assert.NotPanics(func() { + setCrashOutput("base", filepath.Join(lts.T().TempDir(), "does-not-exist.log")) + }) + + // "syslog"/""/"default" branches all target common.SyslogFilePath. In a test environment that + // file is usually not writable; the call must still not panic and must not return an error. + assert.NotPanics(func() { setCrashOutput("", "") }) + assert.NotPanics(func() { setCrashOutput("default", "") }) + assert.NotPanics(func() { setCrashOutput("syslog", "") }) +} + +func (lts *LoggerTestSuite) TestCrashOutputTarget() { + assert := assert.New(lts.T()) + + // "base" with a real path returns that path. + assert.Equal("/tmp/cloudfuse.log", crashOutputTarget("base", "/tmp/cloudfuse.log")) + + // "base" with empty / stdout returns "" (no file to mirror to). + assert.Empty(crashOutputTarget("base", "")) + assert.Empty(crashOutputTarget("base", "stdout")) + + // syslog family always routes to common.SyslogFilePath regardless of logFilePath. + assert.Equal(common.SyslogFilePath, crashOutputTarget("", "")) + assert.Equal(common.SyslogFilePath, crashOutputTarget("default", "")) + assert.Equal(common.SyslogFilePath, crashOutputTarget("syslog", "")) + assert.Equal(common.SyslogFilePath, crashOutputTarget("syslog", "/ignored")) + + // silent and unknown types -- no target. + assert.Empty(crashOutputTarget("silent", "ignored")) + assert.Empty(crashOutputTarget("not-a-real-type", "ignored")) +} + +// TestSetCrashOutputPanicSafeAfterDestroy validates the defer/recover guard in setCrashOutput: +// the SIGHUP handler goroutine outlives Destroy(), which closes BaseLogger's channel. Any Warn() +// from setCrashOutput after Destroy would panic on send-to-closed-channel; the recover swallows +// it so a SIGHUP during teardown cannot crash the process. +func (lts *LoggerTestSuite) TestSetCrashOutputPanicSafeAfterDestroy() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + // Install a base logger and immediately destroy it -- the channel is now closed. + cfg := common.LogConfig{ + FilePath: filepath.Join(lts.T().TempDir(), "destroyed.log"), + MaxFileSize: 10, + FileCount: 3, + Level: common.ELogLevel.LOG_DEBUG(), + } + assert.NoError(SetDefaultLogger("base", cfg)) + assert.NoError(Destroy()) + + // setCrashOutput's error path calls Warn() -- which would attempt to send on the closed + // channel. Point it at a non-existent file so the error path is exercised. The deferred + // recover inside setCrashOutput must absorb the panic. + assert.NotPanics(func() { + setCrashOutput("base", filepath.Join(lts.T().TempDir(), "does-not-exist.log")) + }) + + // Restore a healthy logger so subsequent tests are not affected. + logObj = &SilentLogger{} +} + +// TestSetupCrashOutputSkipsUnsupportedConfigs validates that SetupCrashOutput does not register +// a rotate hook and does not install the SIGHUP handler for logger configurations that have no +// meaningful file target to mirror crash dumps to. +func (lts *LoggerTestSuite) TestSetupCrashOutputSkipsUnsupportedConfigs() { + assert := assert.New(lts.T()) + + cases := []struct { + loggerType, logFilePath string + }{ + {"silent", "ignored"}, + {"base", ""}, + {"base", "stdout"}, + {"unknown-logger", "/tmp/x"}, + } + for _, c := range cases { + resetCrashOutputState() + before := hookCount() + SetupCrashOutput(c.loggerType, c.logFilePath) + assert.Equalf(before, hookCount(), + "SetupCrashOutput must not register a rotate hook for (%q, %q)", c.loggerType, c.logFilePath) + assert.Falsef(sighupInstalled.Load(), + "SetupCrashOutput must not install SIGHUP handler for (%q, %q)", c.loggerType, c.logFilePath) + } +} + +// TestSetupCrashOutputBaseSkipsSighupHandler validates that "base" mode registers the in-process +// rotate hook but does not install the SIGHUP handler: BaseLogger owns its file and rotates in +// process, so hijacking SIGHUP would be inappropriate. +func (lts *LoggerTestSuite) TestSetupCrashOutputBaseSkipsSighupHandler() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + tmp, err := os.CreateTemp("", "cloudfuse-crash-base-nosighup-*.log") + assert.NoError(err) + defer os.Remove(tmp.Name()) + assert.NoError(tmp.Close()) + + before := hookCount() + SetupCrashOutput("base", tmp.Name()) + assert.Equal(before+1, hookCount(), "base mode must register exactly one rotate hook") + assert.False(sighupInstalled.Load(), "base mode must not install SIGHUP handler") + + _ = debug.SetCrashOutput(nil, debug.CrashOptions{}) +} + +func (lts *LoggerTestSuite) TestSetupCrashOutputRegistersHookAndHandler() { + assert := assert.New(lts.T()) + resetCrashOutputState() + + tmp, err := os.CreateTemp("", "cloudfuse-crash-setup-*.log") + assert.NoError(err) + defer os.Remove(tmp.Name()) + assert.NoError(tmp.Close()) + + // Pre-arm a sentinel SIGHUP listener so the test can confirm signal delivery without risk of + // the default SIGHUP action (process termination) firing if installation somehow failed. + sentinel := make(chan os.Signal, 1) + signal.Notify(sentinel, syscall.SIGHUP) + defer signal.Stop(sentinel) + + before := hookCount() + // Use "syslog" so SetupCrashOutput actually installs the SIGHUP handler (base mode skips it). + SetupCrashOutput("syslog", tmp.Name()) + assert.Equal(before+1, hookCount(), "SetupCrashOutput must register exactly one rotate hook") + assert.True(sighupInstalled.Load(), "syslog mode must install SIGHUP handler") + + // Invoking rotate hooks must not panic (the registered closure re-runs setCrashOutput). + assert.NotPanics(invokeRotateHooks) + + // Idempotence: a second call must not register another rotate hook and must not panic. The + // outer sync.Once in SetupCrashOutput ensures repeated invocations are cheap no-ops. + hooksAfterFirst := hookCount() + assert.NotPanics(func() { SetupCrashOutput("syslog", tmp.Name()) }) + assert.Equal(hooksAfterFirst, hookCount(), "second SetupCrashOutput must not register another rotate hook") + + // Signal delivery sanity check: SIGHUP reaches the process (proving signal.Notify was wired). + handledBefore := sighupHandled.Load() + assert.NoError(syscall.Kill(syscall.Getpid(), syscall.SIGHUP)) + select { + case <-sentinel: + // delivered to the sentinel; our listener may or may not have run yet + case <-time.After(2 * time.Second): + lts.T().Fatal("SIGHUP was not delivered to the process within 2s") + } + + // Also verify that the installed crash-output handler goroutine actually ran, not just that + // the signal reached the process. Poll the counter since the handler runs asynchronously. + deadline := time.Now().Add(2 * time.Second) + for sighupHandled.Load() == handledBefore { + if time.Now().After(deadline) { + lts.T().Fatalf("crash-output SIGHUP handler did not execute within 2s (before=%d, after=%d)", + handledBefore, sighupHandled.Load()) + } + time.Sleep(10 * time.Millisecond) + } + assert.Greater(sighupHandled.Load(), handledBefore, "SIGHUP handler goroutine must run at least once") + + // Clean up runtime crash output so it doesn't leak to other tests. + _ = debug.SetCrashOutput(nil, debug.CrashOptions{}) +} + func TestLoggerTestSuite(t *testing.T) { suite.Run(t, new(LoggerTestSuite)) } diff --git a/common/types.go b/common/types.go index 7d9ac3c51..3c27e9bb4 100644 --- a/common/types.go +++ b/common/types.go @@ -149,6 +149,10 @@ var DefaultWorkDir string var DefaultLogFilePath string var StatsConfigFilePath string +// SyslogFilePath is where rsyslog routes cloudfuse-tagged messages (see setup/11-cloudfuse.conf). +// Used as a fallback target for runtime crash dumps when the syslog logger is in use. +var SyslogFilePath = "/var/log/cloudfuse.log" + var EnableMonitoring = false var CfsDisabled = false diff --git a/common/util.go b/common/util.go index e951f307d..6487a42a0 100644 --- a/common/util.go +++ b/common/util.go @@ -51,6 +51,7 @@ import ( "github.com/awnumar/memguard" "github.com/petermattis/goid" + "github.com/prometheus/procfs" "gopkg.in/ini.v1" ) @@ -711,3 +712,31 @@ func PrettyOpenFlags(f int) string { func GetGoroutineID() uint64 { return (uint64)(goid.Get()) } + +// GetAvailableMemoryBytes returns the available system memory in bytes. +// It uses MemAvailable from /proc/meminfo and falls back to MemFree when MemAvailable is unavailable. +func GetAvailableMemoryBytes() (uint64, error) { + fs, err := procfs.NewDefaultFS() + if err != nil { + return 0, err + } + + memInfo, err := fs.Meminfo() + if err != nil { + return 0, err + } + + return getAvailableMemoryBytesFromMeminfo(memInfo) +} + +func getAvailableMemoryBytesFromMeminfo(memInfo procfs.Meminfo) (uint64, error) { + if memInfo.MemAvailableBytes != nil && *memInfo.MemAvailableBytes > 0 { + return *memInfo.MemAvailableBytes, nil + } + + if memInfo.MemFreeBytes != nil && *memInfo.MemFreeBytes > 0 { + return *memInfo.MemFreeBytes, nil + } + + return 0, fmt.Errorf("neither MemAvailable nor MemFree found in /proc/meminfo") +} diff --git a/common/util_linux_test.go b/common/util_linux_test.go new file mode 100644 index 000000000..32045853a --- /dev/null +++ b/common/util_linux_test.go @@ -0,0 +1,96 @@ +//go:build linux + +/* + Licensed under the MIT License . + + Copyright © 2023-2026 Seagate Technology LLC and/or its Affiliates + Copyright © 2020-2026 Microsoft Corporation. All rights reserved. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE +*/ + +package common + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +// func (suite *utilTestSuite) TestSetFrsize() { +// st := &unix.Statfs_t{} +// var val uint64 = 4096 +// SetFrsize(st, val) +// suite.assert.Equal(int64(val), st.Frsize) +// } + +func (suite *utilTestSuite) TestGetAvailableMemoryBytes() { + availableMemory, err := GetAvailableMemoryBytes() + suite.assert.NoError(err) + + meminfoMemory, err := readAvailableMemoryBytesFromProcMeminfo() + suite.assert.NoError(err) + + var difference uint64 + if availableMemory > meminfoMemory { + difference = availableMemory - meminfoMemory + } else { + difference = meminfoMemory - availableMemory + } + + tolerance := meminfoMemory / 100 + suite.assert.LessOrEqual(difference, tolerance) +} + +func readAvailableMemoryBytesFromProcMeminfo() (uint64, error) { + data, err := os.ReadFile("/proc/meminfo") + if err != nil { + return 0, err + } + + var memFree uint64 + for _, line := range strings.Split(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) < 2 { + continue + } + + value, err := strconv.ParseUint(fields[1], 10, 64) + if err != nil { + return 0, err + } + + value *= 1024 + switch fields[0] { + case "MemAvailable:": + if value > 0 { + return value, nil + } + case "MemFree:": + memFree = value + } + } + + if memFree > 0 { + return memFree, nil + } + + return 0, fmt.Errorf("neither MemAvailable nor MemFree found in /proc/meminfo") +} diff --git a/common/util_test.go b/common/util_test.go index 7d2529f9f..c6e9b214b 100644 --- a/common/util_test.go +++ b/common/util_test.go @@ -38,6 +38,7 @@ import ( "testing" "github.com/awnumar/memguard" + "github.com/prometheus/procfs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/suite" ) @@ -783,3 +784,36 @@ func (suite *utilTestSuite) TestGetGoroutineIDParallel() { suite.Len(idMap, workers, "expected unique goroutine ids equal to workers") } + +func (suite *utilTestSuite) TestGetAvailableMemoryBytesFromMeminfoFallback() { + availableMemory := uint64(1024) + freeMemory := uint64(512) + + actual, err := getAvailableMemoryBytesFromMeminfo(procfs.Meminfo{ + MemAvailableBytes: &availableMemory, + MemFreeBytes: &freeMemory, + }) + suite.assert.NoError(err) + suite.assert.Equal(availableMemory, actual) + + actual, err = getAvailableMemoryBytesFromMeminfo(procfs.Meminfo{ + MemFreeBytes: &freeMemory, + }) + suite.assert.NoError(err) + suite.assert.Equal(freeMemory, actual) + + zeroMemory := uint64(0) + actual, err = getAvailableMemoryBytesFromMeminfo(procfs.Meminfo{ + MemAvailableBytes: &zeroMemory, + MemFreeBytes: &freeMemory, + }) + suite.assert.NoError(err) + suite.assert.Equal(freeMemory, actual) + + actual, err = getAvailableMemoryBytesFromMeminfo(procfs.Meminfo{ + MemAvailableBytes: &zeroMemory, + MemFreeBytes: &zeroMemory, + }) + suite.assert.Error(err) + suite.assert.Equal(uint64(0), actual) +} diff --git a/common/version.go b/common/version.go index d8f34aa9e..b7f6419fc 100644 --- a/common/version.go +++ b/common/version.go @@ -125,3 +125,8 @@ func (v Version) OlderThan(v2 Version) bool { func (v Version) NewerThan(v2 Version) bool { return v.compare(v2) == 1 } + +// return the original version string +func (v Version) String() string { + return v.original +} diff --git a/component/azstorage/block_blob.go b/component/azstorage/block_blob.go index a266c53fc..90260037e 100644 --- a/component/azstorage/block_blob.go +++ b/component/azstorage/block_blob.go @@ -99,7 +99,12 @@ func (bb *BlockBlob) Configure(cfg AzStorageConfig) error { } bb.listDetails = container.ListBlobsInclude{ - Metadata: true, + Metadata: true, + // Include.Tags is only honored on Block Blob (flat namespace) accounts. + // On HNS (ADLS Gen2), the list endpoint does not populate BlobTags in + // the response; tags must be fetched per-blob via GetTags — handled in + // processBlobItems. + Tags: bb.Config.filterHasTag && !bb.Config.isHNS, Deleted: false, Snapshots: false, Permissions: false, //Added to get permissions, acl, group, owner for HNS accounts @@ -653,17 +658,34 @@ func (bb *BlockBlob) GetAttr(ctx context.Context, name string) (attr *internal.O // To support virtual directories with no marker blob, we call list instead of get properties since list will not return a 404 if bb.Config.virtualDirectory { - attr, err = bb.getAttrUsingList(ctx, name) - } else { - attr, err = bb.getAttrUsingRest(ctx, name) + // getAttrUsingList -> List -> processBlobItems already evaluates the + // filter (including tag filters), so a successful return means the blob + // passed the filter. + return bb.getAttrUsingList(ctx, name) } + attr, err = bb.getAttrUsingRest(ctx, name) + if err != nil { + return attr, err + } if bb.Config.filter != nil && attr != nil { - if !bb.Config.filter.IsAcceptable(&blobfilter.BlobAttr{ + filterAttr := blobfilter.BlobAttr{ Name: attr.Name, Mtime: attr.Mtime, Size: attr.Size, - }) { + } + // GetProperties does not return blob index tags. When the configured + // filter references tags, fetch them explicitly before deciding. + if bb.Config.filterHasTag && !attr.IsDir() { + tags, terr := bb.getBlobTags(name) + if terr != nil { + log.Err("BlockBlob::GetAttr : Failed to get tags for %s [%s]", name, terr.Error()) + return attr, syscall.EACCES + } + filterAttr.Tags = tags + } + + if !bb.Config.filter.IsAcceptable(&filterAttr) { log.Debug("BlockBlob::GetAttr : Filtered out %s", name) return nil, syscall.ENOENT } @@ -672,6 +694,18 @@ func (bb *BlockBlob) GetAttr(ctx context.Context, name string) (attr *internal.O return attr, err } +// getBlobTags fetches blob index tags for the given blob name and returns them +// as a flat map. Returns nil if the blob has no tags. +func (bb *BlockBlob) getBlobTags(name string) (map[string]string, error) { + blobClient := bb.Container.NewBlockBlobClient(filepath.Join(bb.Config.prefixPath, name)) + resp, err := blobClient.GetTags(context.Background(), nil) + if err != nil { + log.Err("BlockBlob::getBlobTags : Failed to get tags for %s [%s]", name, err.Error()) + return nil, err + } + return parseBlobTags(&resp.BlobTags), nil +} + // List : Get a list of blobs matching the given prefix // This fetches the list using a marker so the caller code should handle marker logic // If count=0 - fetch max entries @@ -711,7 +745,7 @@ func (bb *BlockBlob) List( // APIs that may be affected include IsDirEmpty, ReadDir and StreamDir if err != nil { - log.Err("BlockBlob::List : Failed to list the container with the prefix [%s]", err.Error()) + log.Err("BlockBlob::List : Failed to list the container with the prefix %s: %v", listPath, err) return nil, nil, err } @@ -770,6 +804,23 @@ func (bb *BlockBlob) processBlobItems( filterAttr.Name = blobAttr.Name filterAttr.Mtime = blobAttr.Mtime filterAttr.Size = blobAttr.Size + filterAttr.Tags = nil + if bb.Config.filterHasTag { + // HNS (ADLS Gen2) does not return blob index tags in the list response + // (BlobTags is nil and BlobTagCount is not populated on list items even + // when Include.Tags is requested). Fetch tags per-blob via GetTags so + // the tag filter can be evaluated. This N+1 cost only applies when a + // tag= filter is configured on an HNS account. + if bb.Config.isHNS { + tags, err := bb.getBlobTags(blobAttr.Path) + if err != nil { + return nil, nil, syscall.EACCES + } + filterAttr.Tags = tags + } else { + filterAttr.Tags = parseBlobTags(blobInfo.BlobTags) + } + } if bb.Config.filter.IsAcceptable(&filterAttr) { blobList = append(blobList, blobAttr) @@ -788,12 +839,9 @@ func (bb *BlockBlob) getBlobAttr( ctx context.Context, blobInfo *container.BlobItem, ) (*internal.ObjAttr, error) { - if blobInfo.Properties.CustomerProvidedKeySHA256 != nil && - *blobInfo.Properties.CustomerProvidedKeySHA256 != "" { - log.Trace( - "BlockBlob::List : blob is encrypted with customer provided key so fetching metadata explicitly using REST", - ) - return bb.getAttrUsingRest(ctx, *blobInfo.Name) + if blobInfo.Properties.CustomerProvidedKeySHA256 != nil && *blobInfo.Properties.CustomerProvidedKeySHA256 != "" { + log.Trace("BlockBlob::List : blob is encrypted with customer provided key so fetching metadata explicitly using REST") + return bb.getAttrUsingRest(ctx, removePrefixPath(bb.Config.prefixPath, *blobInfo.Name)) } mode, err := bb.getFileMode(blobInfo.Properties.Permissions) if err != nil { @@ -1380,11 +1428,7 @@ func (bb *BlockBlob) GetFileBlockOffsets( ) if err != nil { - log.Err( - "BlockBlob::GetFileBlockOffsets : Failed to get block list %s [%s]", - name, - err.Error(), - ) + log.Err("BlockBlob::GetFileBlockOffsets : Failed to get block list %s [%s]", name, err.Error()) return &common.BlockOffsetList{}, err } @@ -1922,11 +1966,7 @@ func (bb *BlockBlob) Write(ctx context.Context, options *internal.WriteFileOptio if exceedsFileBlocks { newBufferSize, err = bb.createNewBlocks(fileOffsets, offset, length) if err != nil { - log.Err( - "BlockBlob::Write : Failed to create new blocks for file %s [%s]", - name, - err.Error(), - ) + log.Err("BlockBlob::Write : Failed to create new blocks for file %s [%s]", name, err.Error()) return err } } @@ -2152,11 +2192,7 @@ func (bb *BlockBlob) GetCommittedBlockList( ) if err != nil { - log.Err( - "BlockBlob::GetFileBlockOffsets : Failed to get block list %s [%s]", - name, - err.Error(), - ) + log.Err("BlockBlob::GetFileBlockOffsets : Failed to get block list %s [%s]", name, err.Error()) return nil, err } @@ -2283,9 +2319,18 @@ func (bb *BlockBlob) getFormattedPath(name string) string { func (bb *BlockBlob) SetFilter(filter string) error { if filter == "" { bb.Config.filter = nil + bb.Config.filterHasTag = false + bb.listDetails.Tags = false return nil } bb.Config.filter = &blobfilter.BlobFilter{} - return bb.Config.filter.Configure(filter) + if err := bb.Config.filter.Configure(filter); err != nil { + return err + } + bb.Config.filterHasTag = filterReferencesTag(filter) + // HNS accounts do not populate tags in the list response; leave Include.Tags off + // and let processBlobItems fetch tags per-blob when needed. + bb.listDetails.Tags = bb.Config.filterHasTag && !bb.Config.isHNS + return nil } diff --git a/component/azstorage/block_blob_test.go b/component/azstorage/block_blob_test.go index 2ed5f70dd..26edac05e 100644 --- a/component/azstorage/block_blob_test.go +++ b/component/azstorage/block_blob_test.go @@ -41,6 +41,7 @@ import ( "math" "net/http" "os" + "path/filepath" "runtime" "strings" "syscall" @@ -2652,6 +2653,90 @@ func (s *blockBlobTestSuite) TestGetAttrDirWithCPKEnabled() { s.assert.True(checkMetadata(props.Metadata, folderKey, "true")) } +func (s *blockBlobTestSuite) TestGetAttrFileWithCPKEnabledAndPrefixPath() { + defer s.cleanupTest() + CPKEncryptionKey, CPKEncryptionKeySHA256 := generateCPKInfo() + config := fmt.Sprintf( + "azstorage:\n account-name: %s\n endpoint: https://%s.blob.core.windows.net/\n type: block\n cpk-enabled: true\n cpk-encryption-key: %s\n cpk-encryption-key-sha256: %s\n account-key: %s\n mode: key\n container: %s\n", + storageTestConfigurationParameters.BlockAccount, + storageTestConfigurationParameters.BlockAccount, + CPKEncryptionKey, + CPKEncryptionKeySHA256, + storageTestConfigurationParameters.BlockKey, + s.container, + ) + + s.tearDownTestHelper(false) + s.setupTestHelper(config, s.container, false) + + // Create a subdirectory to use as a prefix path and a file inside it + prefix := generateDirectoryName() + err := s.az.CreateDir(internal.CreateDirOptions{Name: prefix}) + s.assert.NoError(err) + + fileName := generateFileName() + filePath := prefix + "/" + fileName + _, err = s.az.CreateFile(internal.CreateFileOptions{Name: filePath}) + s.assert.NoError(err) + + // Set the prefix path to the subdirectory + _ = s.az.storage.SetPrefixPath(prefix) + + // GetAttr should resolve the file correctly without duplicating the prefix path + props, err := s.az.GetAttr(internal.GetAttrOptions{Name: fileName}) + s.assert.NoError(err) + s.assert.NotNil(props) + s.assert.Equal(fileName, props.Path) + s.assert.False(props.IsDir()) +} + +func (s *blockBlobTestSuite) TestStreamDirWithCPKEnabledAndPrefixPath() { + defer s.cleanupTest() + CPKEncryptionKey, CPKEncryptionKeySHA256 := generateCPKInfo() + config := fmt.Sprintf( + "azstorage:\n account-name: %s\n endpoint: https://%s.blob.core.windows.net/\n type: block\n cpk-enabled: true\n cpk-encryption-key: %s\n cpk-encryption-key-sha256: %s\n account-key: %s\n mode: key\n container: %s\n", + storageTestConfigurationParameters.BlockAccount, + storageTestConfigurationParameters.BlockAccount, + CPKEncryptionKey, + CPKEncryptionKeySHA256, + storageTestConfigurationParameters.BlockKey, + s.container, + ) + + s.tearDownTestHelper(false) + s.setupTestHelper(config, s.container, false) + + // Create a subdirectory to use as a prefix path with files inside it + prefix := generateDirectoryName() + err := s.az.CreateDir(internal.CreateDirOptions{Name: prefix}) + s.assert.NoError(err) + + fileName := generateFileName() + filePath := prefix + "/" + fileName + _, err = s.az.CreateFile(internal.CreateFileOptions{Name: filePath}) + s.assert.NoError(err) + + // Set the prefix path to the subdirectory + _ = s.az.storage.SetPrefixPath(prefix) + + // StreamDir should list the file correctly without duplicating the prefix path + entries, _, err := s.az.StreamDir(internal.StreamDirOptions{Name: "/"}) + s.assert.NoError(err) + s.assert.NotEmpty(entries) + + // Verify the file path does not have a duplicated prefix + found := false + for _, entry := range entries { + if entry.Name == fileName { + found = true + s.assert.Equal(fileName, entry.Path) + s.assert.False(entry.IsDir()) + break + } + } + s.assert.True(found, "Expected file not found in StreamDir results") +} + func (s *blockBlobTestSuite) TestGetAttrFile() { defer s.cleanupTest() vdConfig := fmt.Sprintf( @@ -4498,12 +4583,92 @@ func (s *blockBlobTestSuite) TestBlobFilters() { s.assert.NoError(err) } +func (s *blockBlobTestSuite) TestBlobTagFilter() { + defer s.cleanupTest() + // Setup: create files and stamp blob index tags on a subset + bb := s.az.storage.(*BlockBlob) + + name := generateDirectoryName() + err := s.az.CreateDir(internal.CreateDirOptions{Name: name}) + s.assert.NoError(err) + + type blobSpec struct { + path string + tags map[string]string + } + specs := []blobSpec{ + {name + "/a.txt", map[string]string{"domain": "optical"}}, + {name + "/b.txt", map[string]string{"domain": "optical", "owner": "team-a"}}, + {name + "/c.txt", map[string]string{"domain": "radar"}}, + {name + "/d.txt", nil}, + } + for _, sp := range specs { + _, err = s.az.CreateFile(internal.CreateFileOptions{Name: sp.path}) + s.assert.NoError(err) + if sp.tags != nil { + client := bb.Container.NewBlockBlobClient(sp.path) + _, err = client.SetTags(ctx, sp.tags, nil) + s.assert.NoError(err) + } + } + + // list should return all entries when no filter is applied + listAll := func() []*internal.ObjAttr { + out := make([]*internal.ObjAttr, 0) + marker := "" + for { + page, next, err := s.az.StreamDir( + internal.StreamDirOptions{Name: name + "/", Token: marker, Count: 50}, + ) + s.assert.NoError(err) + out = append(out, page...) + marker = next + if marker == "" { + return out + } + } + } + + s.assert.Len(listAll(), 4) + + // Filter by an existing tag - should match the two blobs tagged optical + err = bb.SetFilter("tag=domain:optical") + s.assert.NoError(err) + s.assert.True(bb.Config.filterHasTag) + + blobs := listAll() + s.assert.Len(blobs, 2) + + got := map[string]bool{} + for _, b := range blobs { + got[filepath.Base(b.Path)] = true + } + s.assert.True(got["a.txt"]) + s.assert.True(got["b.txt"]) + + // GetAttr on a non-matching blob should report ENOENT via the filter, + // while a matching one should succeed. + _, err = bb.GetAttr(ctx, name + "/c.txt") + s.assert.Equal(syscall.ENOENT, err) + + attr, err := bb.GetAttr(ctx, name + "/a.txt") + s.assert.NoError(err) + s.assert.NotNil(attr) + + // Filter that does not reference tags should not flip filterHasTag. + err = bb.SetFilter("name=^a.*") + s.assert.NoError(err) + s.assert.False(bb.Config.filterHasTag) + + err = bb.SetFilter("") + s.assert.NoError(err) + s.assert.False(bb.Config.filterHasTag) +} + func (s *blockBlobTestSuite) UtilityFunctionTestTruncateFileToSmaller( size int, truncatedLength int, ) { - s.T().Helper() - defer s.cleanupTest() // Setup vdConfig := fmt.Sprintf( diff --git a/component/azstorage/config.go b/component/azstorage/config.go index fee0f1cf7..d1147a949 100644 --- a/component/azstorage/config.go +++ b/component/azstorage/config.go @@ -227,6 +227,8 @@ type AzStorageOptions struct { PreserveACL bool `config:"preserve-acl" yaml:"preserve-acl"` Filter string `config:"filter" yaml:"filter"` UserAssertion string `config:"user-assertion" yaml:"user-assertions"` + CapMbpsRead int64 `config:"cap-mbps-read" yaml:"cap-mbps-read"` + CapIOps int64 `config:"cap-iops" yaml:"cap-iops"` } // RegisterEnvVariables : Register environment variables @@ -587,44 +589,15 @@ func ParseAndValidateConfig(az *AzStorage, opt AzStorageOptions) error { } } - log.Crit( - "ParseAndValidateConfig : account %s, container %s, account-type %s, auth %s, prefix %s, endpoint %s, MD5 %v %v, virtual-directory %v, disable-compression %v, CPK %v, restricted-characters-windows %v", - az.stConfig.authConfig.AccountName, - az.stConfig.container, - az.stConfig.authConfig.AccountType, - az.stConfig.authConfig.AuthMode, - az.stConfig.prefixPath, - az.stConfig.authConfig.Endpoint, - az.stConfig.validateMD5, - az.stConfig.updateMD5, - az.stConfig.virtualDirectory, - az.stConfig.disableCompression, - az.stConfig.cpkEnabled, - az.stConfig.restrictedCharsWin, - ) - log.Crit( - "ParseAndValidateConfig : use-HTTP %t, block-size %d, max-concurrency %d, default-tier %v, fail-unsupported-op %t, mount-all-containers %t", - az.stConfig.authConfig.UseHTTP, - az.stConfig.blockSize, - az.stConfig.maxConcurrency, - az.stConfig.defaultTier, - az.stConfig.ignoreAccessModifiers, - az.stConfig.mountAllContainers, - ) - log.Crit( - "ParseAndValidateConfig : Retry Config: retry-count %d, max-timeout %d, backoff-time %d, max-delay %d, preserve-acl: %v", - az.stConfig.maxRetries, - az.stConfig.maxTimeout, - az.stConfig.backoffTime, - az.stConfig.maxRetryDelay, - az.stConfig.preserveACL, - ) + log.Crit("ParseAndValidateConfig : account %s, container %s, account-type %s, auth %s, prefix %s, endpoint %s, MD5 %v %v, virtual-directory %v, disable-compression %v, CPK %v", + az.stConfig.authConfig.AccountName, az.stConfig.container, az.stConfig.authConfig.AccountType, az.stConfig.authConfig.AuthMode, + az.stConfig.prefixPath, az.stConfig.authConfig.Endpoint, az.stConfig.validateMD5, az.stConfig.updateMD5, az.stConfig.virtualDirectory, az.stConfig.disableCompression, az.stConfig.cpkEnabled) + log.Crit("ParseAndValidateConfig : use-HTTP %t, block-size %d, max-concurrency %d, default-tier %v, fail-unsupported-op %t, mount-all-containers %t", az.stConfig.authConfig.UseHTTP, az.stConfig.blockSize, az.stConfig.maxConcurrency, az.stConfig.defaultTier, az.stConfig.ignoreAccessModifiers, az.stConfig.mountAllContainers) + log.Crit("ParseAndValidateConfig : Retry Config: retry-count %d, max-timeout %d, backoff-time %d, max-delay %d, preserve-acl: %v", + az.stConfig.maxRetries, az.stConfig.maxTimeout, az.stConfig.backoffTime, az.stConfig.maxRetryDelay, az.stConfig.preserveACL) - log.Crit( - "ParseAndValidateConfig : Telemetry : %s, honour-ACL %v", - az.stConfig.telemetry, - az.stConfig.honourACL, - ) + log.Crit("ParseAndValidateConfig : Telemetry : %s, honour-ACL %v, cap-mbps-read %d, cap-iops %d", + az.stConfig.telemetry, az.stConfig.honourACL, az.stConfig.capMbpsRead, az.stConfig.capIOps) return nil } @@ -643,11 +616,34 @@ func configureBlobFilter(azStorage *AzStorage, opt AzStorageOptions) error { log.Err("configureBlobFilter : Failed to configure blob filter %s", err.Error()) return errors.New("failed to configure blob filter") } + azStorage.stConfig.filterHasTag = filterReferencesTag(opt.Filter) log.Crit("configureBlobFilter : Blob filter configured %s", opt.Filter) return nil } +// filterReferencesTag returns true when the raw filter expression includes a +// `tag=` clause. The blobfilter package does not expose its parsed filter set, +// so we inspect the input string ourselves to know whether GetAttr paths must +// fetch blob index tags before evaluating the filter. +func filterReferencesTag(filter string) bool { + for clause := range strings.SplitSeq(filter, "||") { + for sub := range strings.SplitSeq(clause, "&&") { + token := strings.TrimSpace(sub) + lowered := strings.ToLower(strings.Map(func(r rune) rune { + if r == ' ' || r == '\t' { + return -1 + } + return r + }, token)) + if strings.HasPrefix(lowered, "tag=") || strings.HasPrefix(lowered, "tag!=") { + return true + } + } + } + return false +} + // ParseAndReadDynamicConfig : On config change read only the required config func ParseAndReadDynamicConfig(az *AzStorage, opt AzStorageOptions, reload bool) error { log.Trace("ParseAndReadDynamicConfig : Reparsing config") @@ -740,5 +736,17 @@ func ParseAndReadDynamicConfig(az *AzStorage, opt AzStorageOptions, reload bool) } } + // Rate limiting, default is no limit + az.stConfig.capMbpsRead = -1 + az.stConfig.capIOps = -1 + + if opt.CapMbpsRead > 0 { + az.stConfig.capMbpsRead = opt.CapMbpsRead + } + + if opt.CapIOps > 0 { + az.stConfig.capIOps = opt.CapIOps + } + return nil } diff --git a/component/azstorage/config_test.go b/component/azstorage/config_test.go index ce02ebe21..26920414a 100644 --- a/component/azstorage/config_test.go +++ b/component/azstorage/config_test.go @@ -453,6 +453,40 @@ func (s *configTestSuite) TestSASRefresh() { assert.NoError(err) } +func (s *configTestSuite) TestRateLimitConfig() { + defer config.ResetConfig() + assert := assert.New(s.T()) + az := &AzStorage{} + opt := AzStorageOptions{} + opt.AccountName = "abcd" + opt.Container = "abcd" + opt.AuthMode = "key" + opt.AccountKey = "abcd" + + // Test default values (no limit) + err := ParseAndReadDynamicConfig(az, opt, false) + assert.NoError(err) + assert.Equal(int64(-1), az.stConfig.capMbpsRead) + assert.Equal(int64(-1), az.stConfig.capIOps) + + // Test setting limits + opt.CapMbpsRead = 100 + opt.CapIOps = 10 + err = ParseAndReadDynamicConfig(az, opt, false) + assert.NoError(err) + assert.Equal(int64(100), az.stConfig.capMbpsRead) + assert.Equal(int64(10), az.stConfig.capIOps) + + // Test setting only one limit + opt.CapMbpsRead = 200 + opt.CapIOps = 0 // reset to no limit + + err = ParseAndReadDynamicConfig(az, opt, false) + assert.NoError(err) + assert.Equal(int64(200), az.stConfig.capMbpsRead) + assert.Equal(int64(-1), az.stConfig.capIOps) +} + func TestConfigTestSuite(t *testing.T) { suite.Run(t, new(configTestSuite)) } diff --git a/component/azstorage/connection.go b/component/azstorage/connection.go index b17b56300..e3509234e 100644 --- a/component/azstorage/connection.go +++ b/component/azstorage/connection.go @@ -80,7 +80,13 @@ type AzStorageConfig struct { cpkEncryptionKeySha256 string // Blob filters - filter *blobfilter.BlobFilter + filter *blobfilter.BlobFilter + filterHasTag bool // true when the configured filter references blob tags + isHNS bool // true when the underlying account is ADLS Gen2 (hierarchical namespace) + + // Rate limiting + capMbpsRead int64 + capIOps int64 } type AzStorageConnection struct { diff --git a/component/azstorage/datalake.go b/component/azstorage/datalake.go index 87b657c35..8a5b02981 100644 --- a/component/azstorage/datalake.go +++ b/component/azstorage/datalake.go @@ -86,6 +86,7 @@ func transformConfig(dlConfig AzStorageConfig) AzStorageConfig { bbConfig := dlConfig bbConfig.authConfig.AccountType = EAccountType.BLOCK() bbConfig.authConfig.Endpoint = transformAccountEndpoint(dlConfig.authConfig.Endpoint) + bbConfig.isHNS = true return bbConfig } @@ -299,6 +300,9 @@ func (dl *Datalake) CreateDirectory(ctx context.Context, name string) error { err.Error(), ) return syscall.EEXIST + case ErrPathTooDeep: + log.Err("Datalake::CreateDirectory : Path is too deep for %s [%s]", name, err.Error()) + return syscall.ENAMETOOLONG default: log.Err( "Datalake::CreateDirectory : Failed to create directory %s [%s]", @@ -426,10 +430,19 @@ func (dl *Datalake) RenameDirectory(ctx context.Context, source string, target s ) if err != nil { serr := storeDatalakeErrToErr(err) - if serr == ErrFileNotFound { + switch serr { + case ErrFileNotFound: log.Err("Datalake::RenameDirectory : %s does not exist", source) return syscall.ENOENT - } else { + case ErrPathTooDeep: + log.Err( + "Datalake::RenameDirectory : Path is too deep for %s -> %s [%s]", + source, + target, + err.Error(), + ) + return syscall.ENAMETOOLONG + default: log.Err( "Datalake::RenameDirectory : Failed to rename directory %s to %s [%s]", source, @@ -517,11 +530,20 @@ func (dl *Datalake) GetAttr( } if dl.Config.filter != nil { - if !dl.Config.filter.IsAcceptable(&blobfilter.BlobAttr{ + filterAttr := &blobfilter.BlobAttr{ Name: blobAttr.Name, Mtime: blobAttr.Mtime, Size: blobAttr.Size, - }) { + } + if dl.Config.filterHasTag && !blobAttr.IsDir() { + tagResp, err := fileClient.GetTags(context.Background(), nil) + if err != nil { + log.Err("Datalake::GetAttr : Failed to get tags for %s [%s]", name, err.Error()) + return blobAttr, syscall.EACCES + } + filterAttr.Tags = parseBlobTags(&tagResp.BlobTags) + } + if !dl.Config.filter.IsAcceptable(filterAttr) { log.Debug("Datalake::GetAttr : Filtered out %s", name) return nil, syscall.ENOENT } @@ -748,12 +770,14 @@ func (dl *Datalake) CommitBlocks( func (dl *Datalake) SetFilter(filter string) error { if filter == "" { dl.Config.filter = nil + dl.Config.filterHasTag = false } else { dl.Config.filter = &blobfilter.BlobFilter{} err := dl.Config.filter.Configure(filter) if err != nil { return err } + dl.Config.filterHasTag = filterReferencesTag(filter) } return dl.BlockBlob.SetFilter(filter) diff --git a/component/azstorage/datalake_test.go b/component/azstorage/datalake_test.go index 47fe35248..a0e23cc22 100644 --- a/component/azstorage/datalake_test.go +++ b/component/azstorage/datalake_test.go @@ -3501,6 +3501,81 @@ func (s *datalakeTestSuite) TestBlobFilters() { s.assert.NoError(err) } +func (s *datalakeTestSuite) TestBlobTagFilter() { + defer s.cleanupTest() + dl := s.az.storage.(*Datalake) + + name := generateDirectoryName() + err := s.az.CreateDir(internal.CreateDirOptions{Name: name}) + s.assert.NoError(err) + + type blobSpec struct { + path string + tags map[string]string + } + specs := []blobSpec{ + {name + "/a.txt", map[string]string{"domain": "optical"}}, + {name + "/b.txt", map[string]string{"domain": "optical", "owner": "team-a"}}, + {name + "/c.txt", map[string]string{"domain": "radar"}}, + {name + "/d.txt", nil}, + } + for _, sp := range specs { + _, err = s.az.CreateFile(internal.CreateFileOptions{Name: sp.path}) + s.assert.NoError(err) + if sp.tags != nil { + _, err = dl.Filesystem.NewFileClient(sp.path).SetTags(ctx, sp.tags, nil) + s.assert.NoError(err) + } + } + + listAll := func() []*internal.ObjAttr { + out := make([]*internal.ObjAttr, 0) + marker := "" + for { + page, next, err := s.az.StreamDir(internal.StreamDirOptions{Name: name + "/", Token: marker, Count: 50}) + s.assert.NoError(err) + out = append(out, page...) + marker = next + if marker == "" { + return out + } + } + } + + s.assert.Len(listAll(), 4) + + err = dl.SetFilter("tag=domain:optical") + s.assert.NoError(err) + s.assert.True(dl.Config.filterHasTag) + s.assert.True(dl.BlockBlob.Config.filterHasTag) + + blobs := listAll() + s.assert.Len(blobs, 2) + + got := map[string]bool{} + for _, b := range blobs { + got[filepath.Base(b.Path)] = true + } + s.assert.True(got["a.txt"]) + s.assert.True(got["b.txt"]) + + _, err = dl.GetAttr(ctx, name + "/c.txt") + s.assert.Equal(syscall.ENOENT, err) + + attr, err := dl.GetAttr(ctx, name + "/a.txt") + s.assert.NoError(err) + s.assert.NotNil(attr) + + err = dl.SetFilter("name=^a.*") + s.assert.NoError(err) + s.assert.False(dl.Config.filterHasTag) + s.assert.False(dl.BlockBlob.Config.filterHasTag) + + err = dl.SetFilter("") + s.assert.NoError(err) + s.assert.False(dl.Config.filterHasTag) +} + func (s *datalakeTestSuite) TestList() { defer s.cleanupTest() // Setup diff --git a/component/azstorage/policies.go b/component/azstorage/policies.go index 6714730ef..0d2a26d98 100644 --- a/component/azstorage/policies.go +++ b/component/azstorage/policies.go @@ -27,10 +27,15 @@ package azstorage import ( "fmt" + "math" "net/http" + "strings" + + "golang.org/x/time/rate" "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" "github.com/Seagate/cloudfuse/common" + "github.com/Seagate/cloudfuse/common/log" ) // cloudfuseTelemetryPolicy is a custom pipeline policy to prepend the cloudfuse user agent string to the one coming from SDK. @@ -71,3 +76,108 @@ func (r *serviceVersionPolicy) Do(req *policy.Request) (*http.Response, error) { req.Raw().Header["x-ms-version"] = []string{r.serviceApiVersion} return req.Next() } + +// --------------------------------------------------------------------------------------------------------------------------------------------------- +// Policy to limit the rate of requests +type rateLimitingPolicy struct { + ingressBandwidthLimiter *rate.Limiter + opsLimiter *rate.Limiter +} + +func newRateLimitingPolicy(readBytesPerSec int64, opsPerSec int64) policy.Policy { + p := &rateLimitingPolicy{} + + // Use 10 second window for burst size calculation for rate limiter. + // This allows for short bursts while still enforcing the average rate over a reasonable time period. + // A larger window size would allow larger bursts but would be less effective at limiting short term spikes. + // A smaller window size would limit bursts more but could lead to underutilization of available bandwidth/ops. + // 10 seconds is a reasonable compromise between these factors. + // Note: The burst size is the maximum number of tokens that can be accumulated in the bucket. + // Therefore, a larger burst size allows for larger bursts of traffic, + // but does not affect the average rate over time. + // Users can adjust the bytesPerSec and opsPerSec values to fine-tune the rate limiting behavior as needed. + // For example, setting a higher bytesPerSec value will allow for higher average bandwidth, + // while setting a lower opsPerSec value will limit the number of operations per second. + const windowSize = 10 + + if readBytesPerSec > 0 { + ingressBandwidthBurstSize := readBytesPerSec * int64(windowSize) + burst := int(ingressBandwidthBurstSize) + // On 32-bit systems, int is 32-bit. If bandwidthBurstSize > MaxInt, we need to clamp it. + // math.MaxInt is platform dependent. + if ingressBandwidthBurstSize > int64(math.MaxInt) { + burst = math.MaxInt + } + + p.ingressBandwidthLimiter = rate.NewLimiter(rate.Limit(readBytesPerSec), burst) + log.Info( + "RateLimitingPolicy : Bandwidth limit set to %d bytes/sec with burst size of %d bytes", + readBytesPerSec, + burst, + ) + } + + if opsPerSec > 0 { + opsBurstSize := opsPerSec * int64(windowSize) + burst := int(opsBurstSize) + if opsBurstSize > int64(math.MaxInt) { + burst = math.MaxInt + } + + p.opsLimiter = rate.NewLimiter(rate.Limit(opsPerSec), burst) + log.Info("RateLimitingPolicy : Ops limit set to %d ops/sec with burst size of %d ops", + opsPerSec, burst) + } + + return p +} + +func (p *rateLimitingPolicy) Do(req *policy.Request) (*http.Response, error) { + ctx := req.Raw().Context() + + // Limit operations per second + if p.opsLimiter != nil { + // Wait for 1 token + err := p.opsLimiter.Wait(ctx) + if err != nil { + log.Err("RateLimitingPolicy : Ops limit wait failed [%s]", err.Error()) + return nil, err + } + } + + // Limit ingress bandwidth for blob downloads (Azure egress: data leaving Azure Storage). + // This policy intentionally applies only to GET requests, which represent download operations. + if p.ingressBandwidthLimiter != nil && req.Raw().Method == http.MethodGet { + // Check for x-ms-range header + // We are not using req.Raw().Header.Get() as it canonicalizes the header name. + // Whereas SDK stores the header in the request is stored in lower case. + // So we directly access the header map with lower case key. + // NOTE: using strings.ToLower to ignore the lint error regarding canonicalized headers. + rangeHeader := req.Raw().Header[strings.ToLower(X_Ms_Range)] + if len(rangeHeader) == 0 { + rangeHeader = req.Raw().Header[RangeHeader] + } + + if len(rangeHeader) > 0 { + size, err := parseRangeHeader(rangeHeader[0]) + if err == nil && size > 0 { + // Wait for tokens equal to size. + // NOTE: range size is guaranteed to be within int range by parseRangeHeader. + err := p.ingressBandwidthLimiter.WaitN(ctx, int(size)) + if err != nil { + log.Err("RateLimitingPolicy : Bandwidth limit wait failed [%s]", err.Error()) + return nil, err + } + } else if err != nil { + log.Err( + "RateLimitingPolicy : Failed to parse Range header %s: [%s]", + rangeHeader[0], + err.Error(), + ) + return nil, err + } + } + } + + return req.Next() +} diff --git a/component/azstorage/policies_test.go b/component/azstorage/policies_test.go new file mode 100644 index 000000000..f2d0bb302 --- /dev/null +++ b/component/azstorage/policies_test.go @@ -0,0 +1,204 @@ +/* + Licensed under the MIT License . + + Copyright © 2023-2026 Seagate Technology LLC and/or its Affiliates + Copyright © 2020-2026 Microsoft Corporation. All rights reserved. + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in all + copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + SOFTWARE +*/ + +package azstorage + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" + "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime" + "github.com/Seagate/cloudfuse/common" + "github.com/Seagate/cloudfuse/common/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/suite" +) + +type mockTransport struct{} + +func (m *mockTransport) Do(req *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200}, nil +} + +type policiesTestSuite struct { + suite.Suite +} + +func (s *policiesTestSuite) SetupTest() { + // Initialize the logger + err := log.SetDefaultLogger("silent", common.LogConfig{Level: common.ELogLevel.LOG_DEBUG()}) + if err != nil { + panic("Unable to set silent logger as default.") + } +} + +func (s *policiesTestSuite) TearDownTest() { + _ = log.Destroy() +} + +func (s *policiesTestSuite) TestRateLimitingPolicy_OpsLimit() { + assert := assert.New(s.T()) + + // Limit to 1 op/sec. Burst will be 1 * 10 = 10 ops. + p := newRateLimitingPolicy(-1, 1) + pipeline := runtime.NewPipeline("test", "v1", runtime.PipelineOptions{ + PerRetry: []policy.Policy{p}, + }, &policy.ClientOptions{Transport: &mockTransport{}}) + + req, err := runtime.NewRequest(context.Background(), http.MethodGet, "http://localhost") + assert.NoError(err) + + // Consume burst + for i := 0; i < 10; i++ { + _, err := pipeline.Do(req) + assert.NoError(err) + } + + // Next request should be delayed by ~1 sec + start := time.Now() + _, err = pipeline.Do(req) + assert.NoError(err) + duration := time.Since(start) + + // It should take at least 1 second (minus some tolerance) + assert.GreaterOrEqual(duration, 900*time.Millisecond, "Expected delay of ~1s, got %v", duration) +} + +func (s *policiesTestSuite) TestRateLimitingPolicy_BandwidthLimit() { + assert := assert.New(s.T()) + + // Limit 100 bytes/sec. Burst will be 100 * 10 = 1000 bytes. + p := newRateLimitingPolicy(100, -1) + pipeline := runtime.NewPipeline("test", "v1", runtime.PipelineOptions{ + PerRetry: []policy.Policy{p}, + }, &policy.ClientOptions{Transport: &mockTransport{}}) + + req, err := runtime.NewRequest(context.Background(), http.MethodGet, "http://localhost") + assert.NoError(err) + + req.Raw().Header["Range"] = []string{"bytes=0-99"} // 100 bytes + + // Consume burst (10 requests of 100 bytes = 1000 bytes) + for i := 0; i < 10; i++ { + _, err := pipeline.Do(req) + assert.NoError(err) + } + + // Next request of 100 bytes should be delayed by ~1 sec + start := time.Now() + _, err = pipeline.Do(req) + assert.NoError(err) + duration := time.Since(start) + + assert.GreaterOrEqual(duration, 900*time.Millisecond, "Expected delay of ~1s, got %v", duration) +} + +func (s *policiesTestSuite) TestRateLimitingPolicy_NoLimit() { + assert := assert.New(s.T()) + + p := newRateLimitingPolicy(-1, -1) + pipeline := runtime.NewPipeline("test", "v1", runtime.PipelineOptions{ + PerRetry: []policy.Policy{p}, + }, &policy.ClientOptions{Transport: &mockTransport{}}) + + req, _ := runtime.NewRequest(context.Background(), http.MethodGet, "http://localhost") + req.Raw().Header["Range"] = []string{"bytes=0-99"} + + start := time.Now() + for i := 0; i < 20; i++ { + _, err := pipeline.Do(req) + assert.NoError(err) + } + duration := time.Since(start) + + // Should be very fast + assert.Less(duration, 100*time.Millisecond, "Expected fast execution, got %v", duration) +} + +func (s *policiesTestSuite) TestRateLimitingPolicy_BandwidthLimit_XMsRange() { + assert := assert.New(s.T()) + + // Limit 100 bytes/sec. Burst 1000 bytes. + p := newRateLimitingPolicy(100, -1) + pipeline := runtime.NewPipeline("test", "v1", runtime.PipelineOptions{ + PerRetry: []policy.Policy{p}, + }, &policy.ClientOptions{Transport: &mockTransport{}}) + + req, _ := runtime.NewRequest(context.Background(), http.MethodGet, "http://localhost") + req.Raw().Header["x-ms-range"] = []string{"bytes=0-99"} // 100 bytes + + // Consume burst + for i := 0; i < 10; i++ { + _, err := pipeline.Do(req) + assert.NoError(err) + } + + // Next request should be delayed + start := time.Now() + _, err := pipeline.Do(req) + assert.NoError(err) + duration := time.Since(start) + + assert.GreaterOrEqual(duration, 900*time.Millisecond, "Expected delay of ~1s, got %v", duration) +} + +func (s *policiesTestSuite) TestRateLimitingPolicy_BandwidthLimit_SkipNonGet() { + assert := assert.New(s.T()) + + // Limit 100 bytes/sec. burst 1000. + p := newRateLimitingPolicy(100, -1) + pipeline := runtime.NewPipeline("test", "v1", runtime.PipelineOptions{ + PerRetry: []policy.Policy{p}, + }, &policy.ClientOptions{Transport: &mockTransport{}}) + + // Create requests for checkable methods + methods := []string{http.MethodPut, http.MethodPost, http.MethodDelete, http.MethodHead} + + for _, method := range methods { + req, _ := runtime.NewRequest(context.Background(), method, "http://localhost") + // Even if we have a range header that implies a large payload + // the policy should ignore it because it's not GET. + req.Raw().Header["Range"] = []string{"bytes=0-999"} // 1000 bytes + + start := time.Now() + // Execute multiple times - if limited, this would take ~10 seconds (1000 bytes * 10 / 100 bytes/sec) + // But since we are skipping non-GET, it should be instant. + for i := 0; i < 11; i++ { + _, err := pipeline.Do(req) + assert.NoError(err) + } + duration := time.Since(start) + + // Each request should be effectively instant, so total should be very fast + assert.Less(duration, 100*time.Millisecond, "Expected fast execution for method %s, got %v", method, duration) + } +} + +func TestPoliciesSuite(t *testing.T) { + suite.Run(t, new(policiesTestSuite)) +} diff --git a/component/azstorage/utils.go b/component/azstorage/utils.go index ace474f22..86962b1d6 100644 --- a/component/azstorage/utils.go +++ b/component/azstorage/utils.go @@ -35,6 +35,7 @@ import ( "net/url" "os" "path/filepath" + "strconv" "strings" "time" @@ -49,6 +50,7 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/azcore/policy" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service" serviceBfs "github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/service" @@ -74,6 +76,9 @@ const ( DisableKeepAlives bool = false DisableCompression bool = false MaxResponseHeaderBytes int64 = 0 + + X_Ms_Range string = "x-ms-range" + RangeHeader string = "Range" ) // getAzStorageClientOptions : Create client options based on the config @@ -114,11 +119,19 @@ func getAzStorageClientOptions(conf *AzStorageConfig) (azcore.ClientOptions, err perCallPolicies = append(perCallPolicies, newServiceVersionPolicy(serviceApiVersion)) } + perRetryPolicies := []policy.Policy{} + if conf.capMbpsRead > 0 || conf.capIOps > 0 { + // Convert Mbps to Bytes/sec: 1 Mbps = (1024* 1024) / 8 = 131072 Bytes/sec + bytesPerSec := conf.capMbpsRead * 131072 + perRetryPolicies = append(perRetryPolicies, newRateLimitingPolicy(bytesPerSec, conf.capIOps)) + } + return azcore.ClientOptions{ - Retry: retryOptions, - Logging: logOptions, - PerCallPolicies: perCallPolicies, - Transport: transportOptions, + Retry: retryOptions, + Logging: logOptions, + PerCallPolicies: perCallPolicies, + PerRetryPolicies: perRetryPolicies, + Transport: transportOptions, }, err } @@ -240,6 +253,7 @@ const ( InvalidRange BlobIsUnderLease InvalidPermission + ErrPathTooDeep ) // For detailed error list refer below link, @@ -285,6 +299,8 @@ func storeDatalakeErrToErr(err error) uint16 { return InvalidPermission case datalakeerror.BlobNotFound: return ErrFileNotFound + case datalakeerror.PathIsTooDeep: + return ErrPathTooDeep default: return ErrUnknown } @@ -608,20 +624,65 @@ func sanitizeEtag(ETag *azcore.ETag) string { return "" } -// func parseBlobTags(tags *container.BlobTags) map[string]string { +// parseRangeHeader parses the x-ms-range header and returns the size of the range requested. +// Examples of x-ms-range header: +// +// bytes=0-499 --> returns 500 +// bytes=500-999 --> returns 500 +// bytes=500- --> returns error (open ended range) +// bytes=-500 --> returns error +// bytes=1000-500 --> returns error (invalid range) +func parseRangeHeader(rangeHeader string) (int64, error) { + if rangeHeader == "" { + return 0, fmt.Errorf("empty x-ms-range header") + } + + if !strings.HasPrefix(rangeHeader, "bytes=") { + return 0, fmt.Errorf("invalid x-ms-range header format %s", rangeHeader) + } + + parts := strings.Split(strings.TrimPrefix(rangeHeader, "bytes="), "-") + if len(parts) != 2 { + return 0, fmt.Errorf("invalid x-ms-range header format %s", rangeHeader) + } + + start, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return 0, err + } + + // Open ended range + if parts[1] == "" { + return 0, fmt.Errorf("invalid x-ms-range header format %s, open ended range not supported", rangeHeader) + } -// if tags == nil { -// return nil -// } + end, err := strconv.ParseInt(parts[1], 10, 64) + if err != nil { + return 0, err + } -// blobtags := make(map[string]string) -// for _, tag := range tags.BlobTagSet { -// if tag != nil { -// if tag.Key != nil { -// blobtags[*tag.Key] = *tag.Value -// } -// } -// } + // Invalid range + if end < start { + return 0, fmt.Errorf("invalid range %s", rangeHeader) + } -// return blobtags -// } + return end - start + 1, nil +} + +// parseBlobTags converts the SDK BlobTags into a flat map suitable for the +// blobfilter package. Returns nil if no tag set is present. +func parseBlobTags(tags *container.BlobTags) map[string]string { + if tags == nil || len(tags.BlobTagSet) == 0 { + return nil + } + + blobtags := make(map[string]string, len(tags.BlobTagSet)) + for _, tag := range tags.BlobTagSet { + if tag == nil || tag.Key == nil || tag.Value == nil { + continue + } + blobtags[*tag.Key] = *tag.Value + } + + return blobtags +} diff --git a/component/azstorage/utils_test.go b/component/azstorage/utils_test.go index 5330d7fde..2311456c9 100644 --- a/component/azstorage/utils_test.go +++ b/component/azstorage/utils_test.go @@ -39,7 +39,8 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/azcore" "github.com/Azure/azure-sdk-for-go/sdk/azcore/to" "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob" - + "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container" + "github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/datalakeerror" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/suite" ) @@ -340,7 +341,6 @@ func (s *utilsTestSuite) TestBlockNonProxyOptions() { opt, err := getAzBlobServiceClientOptions(&AzStorageConfig{}) assert.NoError(err) assert.EqualValues(0, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) } func (s *utilsTestSuite) TestBlockProxyOptions() { @@ -350,21 +350,18 @@ func (s *utilsTestSuite) TestBlockProxyOptions() { ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) opt, err = getAzBlobServiceClientOptions( &AzStorageConfig{proxyAddress: "http://127.0.0.1:8080", maxRetries: 3}, ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) opt, err = getAzBlobServiceClientOptions( &AzStorageConfig{proxyAddress: "https://128.0.0.1:8080", maxRetries: 3}, ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) } func (s *utilsTestSuite) TestBfsNonProxyOptions() { @@ -372,7 +369,6 @@ func (s *utilsTestSuite) TestBfsNonProxyOptions() { opt, err := getAzDatalakeServiceClientOptions(&AzStorageConfig{}) assert.NoError(err) assert.EqualValues(0, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) } func (s *utilsTestSuite) TestBfsProxyOptions() { @@ -382,21 +378,18 @@ func (s *utilsTestSuite) TestBfsProxyOptions() { ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) opt, err = getAzDatalakeServiceClientOptions( &AzStorageConfig{proxyAddress: "http://127.0.0.1:8080", maxRetries: 3}, ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) opt, err = getAzDatalakeServiceClientOptions( &AzStorageConfig{proxyAddress: "https://128.0.0.1:8080", maxRetries: 3}, ) assert.NoError(err) assert.EqualValues(3, opt.Retry.MaxRetries) - assert.GreaterOrEqual(len(opt.Logging.AllowedHeaders), 1) } type endpointAccountType struct { @@ -760,6 +753,38 @@ func (s *utilsTestSuite) TestRemoveLeadingSlashes() { } } +func (s *utilsTestSuite) TestStoreDatalakeErrToErr() { + assert := assert.New(s.T()) + + type testCase struct { + name string + code datalakeerror.StorageErrorCode + expected uint16 + } + + inputs := []testCase{ + {name: "PathAlreadyExists", code: datalakeerror.PathAlreadyExists, expected: ErrFileAlreadyExists}, + {name: "PathNotFound", code: datalakeerror.PathNotFound, expected: ErrFileNotFound}, + {name: "SourcePathNotFound", code: datalakeerror.SourcePathNotFound, expected: ErrFileNotFound}, + {name: "LeaseIDMissing", code: datalakeerror.LeaseIDMissing, expected: BlobIsUnderLease}, + {name: "AuthorizationPermissionMismatch", code: datalakeerror.AuthorizationPermissionMismatch, expected: InvalidPermission}, + {name: "PathIsTooDeep", code: datalakeerror.PathIsTooDeep, expected: ErrPathTooDeep}, + {name: "Unknown", code: datalakeerror.StorageErrorCode("UnknownCode"), expected: ErrUnknown}, + } + + for _, i := range inputs { + s.Run(i.name, func() { + respErr := &azcore.ResponseError{ErrorCode: string(i.code)} + result := storeDatalakeErrToErr(respErr) + assert.Equal(i.expected, result) + }) + } + + // nil error returns ErrNoErr + result := storeDatalakeErrToErr(nil) + assert.Equal(uint16(ErrNoErr), result) +} + func (s *utilsTestSuite) TestRemovePrefixPath() { assert := assert.New(s.T()) @@ -787,6 +812,89 @@ func (s *utilsTestSuite) TestRemovePrefixPath() { } } +func (s *utilsTestSuite) TestParseRangeHeader() { + assert := assert.New(s.T()) + + tests := []struct { + header string + expected int64 + hasError bool + }{ + {"bytes=0-100", 101, false}, + {"bytes=100-200", 101, false}, + {"bytes=0-0", 1, false}, + {"bytes=0-", 0, true}, // open ended range not supported + {"", 0, true}, + {"invalid", 0, true}, + {"bytes=abc-def", 0, true}, + {"bytes=100-50", 0, true}, // invalid range + } + + for _, test := range tests { + size, err := parseRangeHeader(test.header) + + if test.hasError { + assert.Error(err) + } else { + assert.NoError(err) + assert.Equal(test.expected, size) + } + } +} + +func (s *utilsTestSuite) TestParseBlobTags() { + assert := assert.New(s.T()) + + // nil tags + assert.Nil(parseBlobTags(nil)) + + // empty tag set + assert.Nil(parseBlobTags(&container.BlobTags{})) + + // nil key/value entries are skipped, valid entries preserved + k1, v1 := "domain", "optical" + k2, v2 := "owner", "team-a" + tags := &container.BlobTags{ + BlobTagSet: []*container.BlobTag{ + {Key: &k1, Value: &v1}, + nil, + {Key: nil, Value: &v1}, + {Key: &k2, Value: nil}, + {Key: &k2, Value: &v2}, + }, + } + got := parseBlobTags(tags) + assert.Equal(map[string]string{"domain": "optical", "owner": "team-a"}, got) +} + +func (s *utilsTestSuite) TestFilterReferencesTag() { + assert := assert.New(s.T()) + + cases := []struct { + in string + want bool + }{ + {"", false}, + {"name=^foo.*", false}, + {"size>100", false}, + {"tag=domain:optical", true}, + {"TAG=domain:optical", true}, + {" Tag = domain:optical", true}, + {"tag!=domain:optical", true}, + {"TAG!=domain:optical", true}, + {" Tag != domain:optical", true}, + {"name=^foo.* && tag=key:value", true}, + {"name=^foo.* || tag=key:value", true}, + {"name=^foo.* && tag!=key:value", true}, + {"name=^foo.* || tag!=key:value", true}, + {"size>100 && name=^foo.*", false}, + } + + for _, c := range cases { + assert.Equal(c.want, filterReferencesTag(c.in), "input %q", c.in) + } +} + func TestUtilsTestSuite(t *testing.T) { suite.Run(t, new(utilsTestSuite)) } diff --git a/component/block_cache/block_cache.go b/component/block_cache/block_cache.go index 071c1cb44..ca49966dc 100644 --- a/component/block_cache/block_cache.go +++ b/component/block_cache/block_cache.go @@ -390,20 +390,13 @@ func (bc *BlockCache) getDefaultDiskSize(path string) uint64 { } func (bc *BlockCache) getDefaultMemSize() uint64 { - var memSize uint64 - freeRam, err := common.GetFreeRam() + availableMemory, err := common.GetAvailableMemoryBytes() if err != nil { - log.Err( - "BlockCache::Configure : config error %s [%s]. Assigning a pre-defined value of 4GB.", - bc.Name(), - err.Error(), - ) - memSize = uint64(4192) * _1MB - } else { - memSize = uint64(0.8 * (float64)(freeRam)) + log.Info("BlockCache::getDefaultMemSize : config error %s [%s]. Assigning a pre-defined value of 4192MB (~4GB).", bc.Name(), err.Error()) + return uint64(4192) * _1MB } - return memSize + return uint64(0.6 * float64(availableMemory)) } // CreateFile: Create a new file diff --git a/component/block_cache/block_cache_test.go b/component/block_cache/block_cache_test.go index 5ccbbb1b1..721f2c08b 100644 --- a/component/block_cache/block_cache_test.go +++ b/component/block_cache/block_cache_test.go @@ -218,14 +218,9 @@ func (suite *blockCacheTestSuite) TestMemory() { suite.assert.NoError(err) suite.assert.Equal("block_cache", tobj.blockCache.Name()) - cmd := exec.Command("bash", "-c", "free -b | grep Mem | awk '{print $4}'") - var out bytes.Buffer - cmd.Stdout = &out - err = cmd.Run() - suite.assert.NoError(err) - free, err := strconv.Atoi(strings.TrimSpace(out.String())) + availableMemory, err := common.GetAvailableMemoryBytes() suite.assert.NoError(err) - expected := uint64(0.8 * float64(free)) + expected := uint64(0.6 * float64(availableMemory)) actual := tobj.blockCache.memSize difference := math.Abs(float64(actual) - float64(expected)) tolerance := 0.10 * float64(math.Max(float64(actual), float64(expected))) @@ -276,14 +271,9 @@ func (suite *blockCacheTestSuite) TestStatfsMemory() { suite.assert.NoError(err) suite.assert.Equal("block_cache", tobj.blockCache.Name()) - cmd := exec.Command("bash", "-c", "free -b | grep Mem | awk '{print $4}'") - var out bytes.Buffer - cmd.Stdout = &out - err = cmd.Run() - suite.assert.NoError(err) - free, err := strconv.Atoi(strings.TrimSpace(out.String())) + availableMemory, err := common.GetAvailableMemoryBytes() suite.assert.NoError(err) - expected := uint64(0.8 * float64(free)) + expected := uint64(0.6 * float64(availableMemory)) stat, ret, err := tobj.blockCache.StatFs() suite.assert.True(ret) suite.assert.NoError(err) diff --git a/component/file_cache/file_cache_test.go b/component/file_cache/file_cache_test.go index 6977502ba..219582791 100644 --- a/component/file_cache/file_cache_test.go +++ b/component/file_cache/file_cache_test.go @@ -40,6 +40,8 @@ import ( "runtime" "strconv" "strings" + "sync" + "sync/atomic" "syscall" "testing" "time" @@ -65,6 +67,7 @@ type fileCacheTestSuite struct { loopback internal.Component cache_path string // uses os.Separator (filepath.Join) fake_storage_path string // uses os.Separator (filepath.Join) + configString string useMock bool mockCtrl *gomock.Controller mock *internal.MockComponent @@ -136,6 +139,7 @@ func (suite *fileCacheTestSuite) SetupTest() { func (suite *fileCacheTestSuite) setupTestHelper(configuration string) { suite.assert = assert.New(suite.T()) + suite.configString = configuration err := config.ReadConfigFromReader(strings.NewReader(configuration)) suite.assert.NoError(err) @@ -3243,6 +3247,282 @@ loopbackfs: suite.assert.Equal(expectedData, cloudData, "File should be updated immediately in the second window") } +// setupMockFileCacheForFlush creates a file cache backed by a gomock NextComponent +// and returns (fileCache, mockComponent, cachePath, cleanup). +// The caller must defer cleanup() to stop the independent file cache and remove its temp dir. +// The suite-level loopback and fileCache are left running throughout and are not affected. +func (suite *fileCacheTestSuite) setupMockFileCacheForFlush(mockCtrl *gomock.Controller) (*FileCache, *internal.MockComponent, string, func()) { + mockComponent := internal.NewMockComponent(mockCtrl) + + randStr := randomString(8) + cachePath := filepath.Join(home_dir, "file_cache"+randStr) + cfg := fmt.Sprintf("file_cache:\n path: %s\n offload-io: true\n timeout-sec: 0", cachePath) + + // Overwrite global config to configure the independent FileCache, then immediately + // restore the original config so the suite-level components are unaffected. + err := config.ReadConfigFromReader(strings.NewReader(cfg)) + suite.assert.NoError(err) + + fcComp := NewFileCacheComponent() + fcComp.SetNextComponent(mockComponent) + err = fcComp.Configure(true) + suite.assert.NoError(err) + + // Restore the suite's original config so subsequent suite-level operations work correctly. + err = config.ReadConfigFromReader(strings.NewReader(suite.configString)) + suite.assert.NoError(err) + + mockComponent.EXPECT().Start(gomock.Any()).Return(nil).Times(1) + mockComponent.EXPECT().Stop().Return(nil).Times(1) + + err = mockComponent.Start(context.Background()) + suite.assert.NoError(err) + err = fcComp.Start(context.Background()) + suite.assert.NoError(err) + + fc := fcComp.(*FileCache) + + cleanup := func() { + _ = fc.Stop() + _ = mockComponent.Stop() + os.RemoveAll(cachePath) + } + + return fc, mockComponent, cachePath, cleanup +} + +// TestFlushFileConcurrent verifies that concurrent FlushFile calls for the same file +// are serialized by the per-file lock. +// +// The mock NextComponent's CopyFromFile tracks the number of concurrent in-flight +// uploads via an atomic counter and sleeps briefly to widen the race window. +// If two uploads ever overlap (counter > 1), the mock returns an error that mimics +// the Azure Storage InvalidBlockList failure. +// With the per-file lock this can never happen; without it, the overlapping calls +// would cause the test to fail. +func (suite *fileCacheTestSuite) TestFlushFileConcurrent() { + defer suite.cleanupTest() + mockCtrl := gomock.NewController(suite.T()) + defer mockCtrl.Finish() + + fc, mockComponent, cachePath, cleanup := suite.setupMockFileCacheForFlush(mockCtrl) + defer cleanup() + + // Create a real local file in the cache (createEmptyFile=false means no NextComponent call). + path := "concurrent_flush.txt" + localPath := filepath.Join(cachePath, path) + err := os.MkdirAll(filepath.Dir(localPath), 0755) + suite.assert.NoError(err) + f, err := os.Create(localPath) + suite.assert.NoError(err) + _, err = f.WriteString("concurrent flush test data") + suite.assert.NoError(err) + + handle := handlemap.NewHandle(path) + handle.UnixFD = uint64(f.Fd()) + handle.SetFileObject(f) + handle.Flags.Set(handlemap.HandleFlagDirty) + fc.fileLocks.Get(path).Inc() // simulate an open handle + + // Track concurrent in-flight CopyFromFile calls. + var inFlight atomic.Int32 + var maxInFlight atomic.Int32 + + // The mock CopyFromFile blocks briefly so that without serialization, + // multiple goroutines would be inside it at the same time. + mockComponent.EXPECT(). + CopyFromFile(gomock.Any()). + DoAndReturn(func(opts internal.CopyFromFileOptions) error { + cur := inFlight.Add(1) + // Record the high-water mark of concurrent callers. + for { + old := maxInFlight.Load() + if cur <= old || maxInFlight.CompareAndSwap(old, cur) { + break + } + } + + // Hold the "upload" open long enough for other goroutines to arrive. + time.Sleep(50 * time.Millisecond) + + inFlight.Add(-1) + + if cur > 1 { + // This is what Azure Storage would return when two PutBlockList calls race. + return fmt.Errorf("InvalidBlockList: concurrent upload detected (%d in-flight)", cur) + } + return nil + }). + AnyTimes() + + // Launch concurrent flushes (simulating multiple libfuse_flush calls from dup'd fds). + concurrency := 5 + var wg sync.WaitGroup + errs := make([]error, concurrency) + wg.Add(concurrency) + for i := 0; i < concurrency; i++ { + go func(idx int) { + defer wg.Done() + // Re-dirty the handle before each flush so that the fast-path dirty check + // doesn't skip the upload. In the real world each fd's close sets dirty via + // libfuse_flush (fileHandle.dirty != 0). + handle.Flags.Set(handlemap.HandleFlagDirty) + errs[idx] = fc.FlushFile(internal.FlushFileOptions{Handle: handle}) + }(i) + } + wg.Wait() + + for i, e := range errs { + suite.assert.NoError(e, "FlushFile goroutine %d returned error: %v", i, e) + } + + // The per-file lock must have serialized all uploads; at no point should more + // than one CopyFromFile have been in-flight simultaneously. + suite.assert.EqualValues(1, maxInFlight.Load(), + "expected max 1 concurrent CopyFromFile, got %d — uploads were not serialized", maxInFlight.Load()) + + f.Close() +} + +// TestFlushFileConcurrentWithRelease verifies that a concurrent FlushFile call +// (from libfuse_flush) and ReleaseFile (close) are properly serialized so that +// CopyFromFile never runs in parallel. +func (suite *fileCacheTestSuite) TestFlushFileConcurrentWithRelease() { + defer suite.cleanupTest() + mockCtrl := gomock.NewController(suite.T()) + defer mockCtrl.Finish() + + fc, mockComponent, cachePath, cleanup := suite.setupMockFileCacheForFlush(mockCtrl) + defer cleanup() + + path := "concurrent_flush_release.txt" + localPath := filepath.Join(cachePath, path) + err := os.MkdirAll(filepath.Dir(localPath), 0755) + suite.assert.NoError(err) + f, err := os.Create(localPath) + suite.assert.NoError(err) + _, err = f.WriteString("concurrent flush release test data") + suite.assert.NoError(err) + + handle := handlemap.NewHandle(path) + handle.UnixFD = uint64(f.Fd()) + handle.SetFileObject(f) + handle.Flags.Set(handlemap.HandleFlagDirty) + fc.fileLocks.Get(path).Inc() // simulate an open handle + + var inFlight atomic.Int32 + var maxInFlight atomic.Int32 + + mockComponent.EXPECT(). + CopyFromFile(gomock.Any()). + DoAndReturn(func(opts internal.CopyFromFileOptions) error { + cur := inFlight.Add(1) + for { + old := maxInFlight.Load() + if cur <= old || maxInFlight.CompareAndSwap(old, cur) { + break + } + } + time.Sleep(50 * time.Millisecond) + inFlight.Add(-1) + if cur > 1 { + return fmt.Errorf("InvalidBlockList: concurrent upload detected (%d in-flight)", cur) + } + return nil + }). + AnyTimes() + + // Start a concurrent flush to simulate libfuse_flush from a dup'd fd. + var wg sync.WaitGroup + var flushErr error + wg.Add(1) + go func() { + defer wg.Done() + flushErr = fc.FlushFile(internal.FlushFileOptions{Handle: handle}) + }() + + // ReleaseFile on the same handle races with the flush above. + // releaseFileInternal calls FlushFile internally, so both paths compete. + releaseErr := fc.ReleaseFile(internal.ReleaseFileOptions{Handle: handle}) + + wg.Wait() + suite.assert.NoError(flushErr, "concurrent FlushFile should not fail") + suite.assert.NoError(releaseErr, "ReleaseFile should not fail") + suite.assert.EqualValues(1, maxInFlight.Load(), + "expected max 1 concurrent CopyFromFile, got %d — uploads were not serialized", maxInFlight.Load()) +} + +// TestFlushFileSyncFileConcurrent verifies that SyncFile (which calls FlushFile internally +// when syncToFlush is enabled) and a direct FlushFile call are serialized so that +// CopyFromFile never runs concurrently. +func (suite *fileCacheTestSuite) TestFlushFileSyncFileConcurrent() { + defer suite.cleanupTest() + mockCtrl := gomock.NewController(suite.T()) + defer mockCtrl.Finish() + + fc, mockComponent, cachePath, cleanup := suite.setupMockFileCacheForFlush(mockCtrl) + defer cleanup() + fc.syncToFlush = true + + path := "sync_flush_concurrent.txt" + localPath := filepath.Join(cachePath, path) + err := os.MkdirAll(filepath.Dir(localPath), 0755) + suite.assert.NoError(err) + f, err := os.Create(localPath) + suite.assert.NoError(err) + _, err = f.WriteString("sync flush concurrent test data") + suite.assert.NoError(err) + + handle := handlemap.NewHandle(path) + handle.UnixFD = uint64(f.Fd()) + handle.SetFileObject(f) + handle.Flags.Set(handlemap.HandleFlagDirty) + fc.fileLocks.Get(path).Inc() // simulate an open handle + + var inFlight atomic.Int32 + var maxInFlight atomic.Int32 + + mockComponent.EXPECT(). + CopyFromFile(gomock.Any()). + DoAndReturn(func(opts internal.CopyFromFileOptions) error { + cur := inFlight.Add(1) + for { + old := maxInFlight.Load() + if cur <= old || maxInFlight.CompareAndSwap(old, cur) { + break + } + } + time.Sleep(50 * time.Millisecond) + inFlight.Add(-1) + if cur > 1 { + return fmt.Errorf("InvalidBlockList: concurrent upload detected (%d in-flight)", cur) + } + return nil + }). + AnyTimes() + + // Launch concurrent SyncFile (calls FlushFile internally) and direct FlushFile. + var wg sync.WaitGroup + var syncErr, flushErr error + + wg.Add(2) + go func() { + defer wg.Done() + syncErr = fc.SyncFile(internal.SyncFileOptions{Handle: handle}) + }() + go func() { + defer wg.Done() + flushErr = fc.FlushFile(internal.FlushFileOptions{Handle: handle}) + }() + wg.Wait() + + suite.assert.NoError(syncErr, "SyncFile should not fail") + suite.assert.NoError(flushErr, "FlushFile should not fail") + suite.assert.EqualValues(1, maxInFlight.Load(), + "expected max 1 concurrent CopyFromFile, got %d — uploads were not serialized", maxInFlight.Load()) + + f.Close() +} func (suite *fileCacheTestSuite) TestGetAttrCase1() { defer suite.cleanupTest() @@ -4473,6 +4753,98 @@ func (suite *fileCacheTestSuite) createRemoteDirectoryStructure() { suite.assert.NoError(err) } +// TestOpenFileDownloadFailure tests that when CopyToFile fails during OpenFile, +// the correct download error is returned (not the cleanup error) and the +// partially downloaded file is removed from the cache. +func (suite *fileCacheTestSuite) TestOpenFileDownloadFailure() { + // Create a mock controller and component + mockCtrl := gomock.NewController(suite.T()) + defer mockCtrl.Finish() + + mockComponent := internal.NewMockComponent(mockCtrl) + + // Create a new file cache with the mock component + // First, stop the default file cache + err := suite.fileCache.Stop() + suite.assert.NoError(err) + err = suite.loopback.Stop() + suite.assert.NoError(err) + + randStr := randomString(8) + cachePath := filepath.Join(home_dir, "file_cache"+randStr) + defaultConfig := fmt.Sprintf("file_cache:\n path: %s\n offload-io: true\n timeout-sec: 0", cachePath) + + err = config.ReadConfigFromReader(strings.NewReader(defaultConfig)) + suite.assert.NoError(err) + + fileCache := NewFileCacheComponent() + fileCache.SetNextComponent(mockComponent) + err = fileCache.Configure(true) + suite.assert.NoError(err) + + // Expect Start and Stop calls + mockComponent.EXPECT().Start(gomock.Any()).Return(nil).Times(1) + mockComponent.EXPECT().Stop().Return(nil).Times(1) + + err = mockComponent.Start(context.Background()) + suite.assert.NoError(err) + + err = fileCache.Start(context.Background()) + suite.assert.NoError(err) + + fc := fileCache.(*FileCache) + + // Test file path + path := "test_download_failure.txt" + localPath := filepath.Join(cachePath, path) + + // Set up expectations for GetAttr to return a valid file that exists in storage + mockComponent.EXPECT(). + GetAttr(gomock.Any()). + Return(&internal.ObjAttr{ + Path: path, + Name: filepath.Base(path), + Size: 1024, + Mode: 0644, + Flags: internal.NewFileBitMap(), + }, nil). + Times(1) + + // Set up expectation for CopyToFile to fail with a download error + downloadErr := fmt.Errorf("simulated download failure") + mockComponent.EXPECT(). + CopyToFile(gomock.Any()). + Return(downloadErr). + Times(1) + + // Attempt to open the file - this should fail + handle, err := fc.OpenFile(internal.OpenFileOptions{Name: path, Mode: 0644}) + + // Assert that the error returned is the download error, not a cleanup error + suite.assert.Error(err) + suite.assert.Equal(downloadErr, err) + suite.assert.Nil(handle) + + // Verify that the partially downloaded file was cleaned up from the cache + _, statErr := os.Stat(localPath) + suite.assert.True(os.IsNotExist(statErr), "Partially downloaded file should be cleaned up from cache") + + // Clean up + err = fc.Stop() + suite.assert.NoError(err) + err = mockComponent.Stop() + suite.assert.NoError(err) + os.RemoveAll(cachePath) + + // Restart the default file cache for other tests + suite.loopback = newLoopbackFS() + suite.fileCache = newTestFileCache(suite.loopback) + err = suite.loopback.Start(context.Background()) + suite.assert.NoError(err) + err = suite.fileCache.Start(context.Background()) + suite.assert.NoError(err) +} + // In order for 'go test' to run this suite, we need to create // a normal test function and pass our suite to suite.Run func TestFileCacheTestSuite(t *testing.T) { diff --git a/component/libfuse/libfuse2_handler_test_wrapper.go b/component/libfuse/libfuse2_handler_test_wrapper.go index 3f1328158..b55f13bfd 100644 --- a/component/libfuse/libfuse2_handler_test_wrapper.go +++ b/component/libfuse/libfuse2_handler_test_wrapper.go @@ -1510,6 +1510,22 @@ func testUnlinkError(suite *libfuseTestSuite) { // Rename +func testRenameDirEnametoolong(suite *libfuseTestSuite) { + defer suite.cleanupTest() + src := "src" + dst := "dst" + + srcAttr := &internal.ObjAttr{Name: src} + srcAttr.Flags.Set(internal.PropFlagIsDir) + suite.mock.EXPECT().GetAttr(internal.GetAttrOptions{Name: src}).Return(srcAttr, nil) + suite.mock.EXPECT().GetAttr(internal.GetAttrOptions{Name: dst}).Return(nil, syscall.ENOENT) + options := internal.RenameDirOptions{Src: src, Dst: dst} + suite.mock.EXPECT().RenameDir(options).Return(syscall.ENAMETOOLONG) + + err := cfuseFS.Rename("/"+src, "/"+dst) + suite.assert.Equal(-fuse.ENAMETOOLONG, err) +} + func testSymlink(suite *libfuseTestSuite) { defer suite.cleanupTest() name := "path" diff --git a/component/libfuse/libfuse_handler_test.go b/component/libfuse/libfuse_handler_test.go index 3e1678e51..ad9ccc409 100644 --- a/component/libfuse/libfuse_handler_test.go +++ b/component/libfuse/libfuse_handler_test.go @@ -608,6 +608,10 @@ func (suite *libfuseTestSuite) TestRenameSrcGetAttrError() { testRenameSrcGetAttrError(suite) } +func (suite *libfuseTestSuite) TestRenameDirEnametoolong() { + testRenameDirEnametoolong(suite) +} + func (suite *libfuseTestSuite) TestSymlink() { testSymlink(suite) } diff --git a/docker/dockerinstall.sh b/docker/dockerinstall.sh index b3c194e43..f57e0ee32 100755 --- a/docker/dockerinstall.sh +++ b/docker/dockerinstall.sh @@ -1,8 +1,8 @@ #!/bin/bash # Cleanup old installation sudo apt remove docker-desktop -rm-r $HOME/.docker/desktop -sudo rm/usr/local/bin/com.docker.cli +rm -rf "$HOME/.docker/desktop" +sudo rm -f /usr/local/bin/com.docker.cli sudo apt purge docker-desktop sudo apt-get update diff --git a/go.mod b/go.mod index a36ee2b2b..91e0b25eb 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/montanaflynn/stats v0.12.7 github.com/netresearch/go-cron v0.16.1 github.com/petermattis/goid v0.0.0-20260819104326-d9896a8858b2 + github.com/prometheus/procfs v0.22.0 github.com/radovskyb/watcher v1.0.7 github.com/rivo/tview v0.42.0 github.com/sevlyar/go-daemon v0.1.7 @@ -38,6 +39,7 @@ require ( go.yaml.in/yaml/v3 v3.0.5 golang.org/x/crypto v0.57.0 golang.org/x/sys v0.48.0 + golang.org/x/time v0.16.0 gopkg.in/ini.v1 v1.67.3 ) diff --git a/go.sum b/go.sum index 255f3df29..bb733d9c5 100644 --- a/go.sum +++ b/go.sum @@ -130,6 +130,8 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjL github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6 h1:jL3a8soXdzuTCcRnKhOmtcsVOObdDTFf4O2B403HPRU= github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/prometheus/procfs v0.22.0 h1:6q9+/JL9IKAPbCmBrv9n5O5Ty3NKnciV5X7YGw0oics= +github.com/prometheus/procfs v0.22.0/go.mod h1:CvmFr/GVhIjIvWJZW3tgkODBQMRIf0EyWMQLHCHab58= github.com/radovskyb/watcher v1.0.7 h1:AYePLih6dpmS32vlHfhCeli8127LzkIgwJGcwwe8tUE= github.com/radovskyb/watcher v1.0.7/go.mod h1:78okwvY5wPdzcb1UYnip1pvrZNIVEIh/Cm+ZuvsUYIg= github.com/rivo/tview v0.42.0 h1:b/ftp+RxtDsHSaynXTbJb+/n/BxDEi+W3UfF5jILK6c= @@ -236,6 +238,8 @@ golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= diff --git a/main.go b/main.go index 28d672f3d..77c5ade7d 100644 --- a/main.go +++ b/main.go @@ -27,23 +27,11 @@ package main import ( "github.com/Seagate/cloudfuse/cmd" - "github.com/Seagate/cloudfuse/common/log" ) //go:generate ./cmd/componentGenerator.sh $NAME // To use go:generate run command "NAME="component" go generate" func main() { - defer log.Destroy() // nolint:errcheck - // This recovers the panics only for the functions that run within this context. all the go-routine - // spawned by this function need to handle their panics separately if required. Also the FUSE callbacks - // wouldn't run in this context, so the panics originated from the callbacks can't get recovered here. - defer func() { - if panicErr := recover(); panicErr != nil { - log.Crit("PANIC: %v", panicErr) - panic(panicErr) - } - }() - _ = cmd.Execute() } diff --git a/perf_testing/scripts/fio_bench.sh b/perf_testing/scripts/fio_bench.sh index e98b17ae1..7019a3675 100755 --- a/perf_testing/scripts/fio_bench.sh +++ b/perf_testing/scripts/fio_bench.sh @@ -183,6 +183,118 @@ run_test_suite() { done } +# -------------------------------------------------------------------------------------------------- +# Helper: Run custom filecache read test +run_filecache_read_test() { + echo "Running custom filecache read test..." + + local TEST_FILE="filecache_read_test_100gb.data" + local BLOCK_SIZE="1M" + + # Step 1: Mount and create the test file if it doesn't exist + mount_blobfuse + rm -rf "${MOUNT_DIR}/"* + + if [ ! -f "${MOUNT_DIR}/${TEST_FILE}" ]; then + echo "Creating 100GB test file: ${TEST_FILE}..." + echo "This may take some time..." + + # Create the file using dd (this will upload to blob storage in file_cache mode) + dd if=/dev/urandom of="${MOUNT_DIR}/${TEST_FILE}" bs=1M count=102400 status=progress + + echo "Test file created successfully." + + # Sync to ensure file is uploaded + sync + sleep 5 + fi + + # Verify file exists and get its size + if [ ! -f "${MOUNT_DIR}/${TEST_FILE}" ]; then + echo "Error: Test file does not exist" + exit 1 + fi + + local FILE_SIZE=$(stat -c%s "${MOUNT_DIR}/${TEST_FILE}") + echo "Test file size: $FILE_SIZE bytes" + + # Step 2: Unmount to clear any cached data + echo "Unmounting to clear cache..." + cleanup_mount + sleep 5 + + # Step 3: Clear kernel page cache + echo "Dropping kernel caches..." + sudo sh -c "echo 3 > /proc/sys/vm/drop_caches" + sleep 2 + + # Step 4: Mount again for the read test + echo "Remounting for read test..." + mount_blobfuse + + # Step 5: Run the read test with direct I/O + echo "Starting sequential read test with direct I/O (block size: ${BLOCK_SIZE})..." + + # Get initial network stats + local start_rx=$(cat /sys/class/net/$INTERFACE/statistics/rx_bytes) + local start_tx=$(cat /sys/class/net/$INTERFACE/statistics/tx_bytes) + local start_time=$(date +%s.%N) + + # Perform the read test using dd with direct I/O + # O_DIRECT flag ensures we bypass the OS page cache + # This includes the open time (when filecache downloads the file) + dd if="${MOUNT_DIR}/${TEST_FILE}" of=/dev/null bs=${BLOCK_SIZE} iflag=direct 2>&1 | tee "${OUTPUT_DIR}/dd_output.txt" + + # Get final stats + local end_time=$(date +%s.%N) + local end_rx=$(cat /sys/class/net/$INTERFACE/statistics/rx_bytes) + local end_tx=$(cat /sys/class/net/$INTERFACE/statistics/tx_bytes) + + # Calculate metrics + local duration=$(echo "$end_time - $start_time" | bc) + local rx_bytes=$((end_rx - start_rx)) + local tx_bytes=$((end_tx - start_tx)) + + # Calculate bandwidth in Mbps + local rx_mbps=$(echo "scale=4; ($rx_bytes * 8) / ($duration * 1000000)" | bc) + local tx_mbps=$(echo "scale=4; ($tx_bytes * 8) / ($duration * 1000000)" | bc) + + # Calculate throughput in MiB/s (includes open time) + local throughput_mibs=$(echo "scale=4; $FILE_SIZE / ($duration * 1024 * 1024)" | bc) + + echo "-------------------------------------" + echo "Test completed!" + echo "Duration: ${duration} seconds" + echo "Throughput: ${throughput_mibs} MiB/s (including file open time)" + echo + echo "Network Statistics:" + echo "Interface: $INTERFACE" + echo "Received (RX): $rx_bytes bytes (${rx_mbps} Mbps)" + echo "Transmitted (TX): $tx_bytes bytes (${tx_mbps} Mbps)" + echo "-------------------------------------" + + # Generate JSON output for bandwidth results + cat > "${OUTPUT_DIR}/filecache_sequential_read_bandwidth_summary.json" < "${OUTPUT_DIR}/filecache_sequential_read_latency_summary.json" <// @@ -22,6 +26,10 @@ file_cache: attr_cache: timeout-sec: 7200 + # max-size-mb controls how much memory the attribute cache may use. + # Default: auto-tuned to 1% of total RAM, clamped to [64 MB, 1 GB]. + # Set to 0 or omit to use the auto-tuned value. + # max-size-mb: 256 azstorage: type: block diff --git a/sample_configs/sampleFileCacheWithSASConfigAzure.yaml b/sample_configs/sampleFileCacheWithSASConfigAzure.yaml index 926d94f7c..232a66761 100644 --- a/sample_configs/sampleFileCacheWithSASConfigAzure.yaml +++ b/sample_configs/sampleFileCacheWithSASConfigAzure.yaml @@ -14,6 +14,10 @@ components: libfuse: network-share: true + attribute-expiration-sec: 120 + entry-expiration-sec: 120 + negative-entry-expiration-sec: 240 + kernel-list-cache-expiration-sec: 120 file_cache: path: /// @@ -22,6 +26,10 @@ file_cache: attr_cache: timeout-sec: 7200 + # max-size-mb controls how much memory the attribute cache may use. + # Default: auto-tuned to 1% of total RAM, clamped to [64 MB, 1 GB]. + # Set to 0 or omit to use the auto-tuned value. + # max-size-mb: 256 azstorage: type: block diff --git a/scripts/update_seagate_copyright.sh b/scripts/update_seagate_copyright.sh index 6b936df03..db90c30ed 100755 --- a/scripts/update_seagate_copyright.sh +++ b/scripts/update_seagate_copyright.sh @@ -2,8 +2,8 @@ # URL of blobfuse2 repo blobfuse2_repo="https://github.com/Azure/azure-storage-fuse.git" -tags="tags/blobfuse2-2.5.0" -year="2025" +tags="tags/blobfuse2-2.5.4" +year="2026" # Create temporary directory to clone into tmp_dir=$(mktemp -d) diff --git a/setup/advancedConfig.yaml b/setup/advancedConfig.yaml index 66aae9390..e5e91d323 100644 --- a/setup/advancedConfig.yaml +++ b/setup/advancedConfig.yaml @@ -194,6 +194,8 @@ azstorage: cpk-encryption-key: cpk-encryption-key-sha256: preserve-acl: true|false + cap-mbps-read: + cap-iops: # S3 storage configuration s3storage: diff --git a/setup/baseConfig.yaml b/setup/baseConfig.yaml index 95790bbde..e3bde85cf 100644 --- a/setup/baseConfig.yaml +++ b/setup/baseConfig.yaml @@ -105,6 +105,11 @@ block_cache: cleanup-on-start: true|false prefetch-on-open: true|false +# Attribute cache related configuration +attr_cache: + timeout-sec: