mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-16 03:25:50 +00:00
Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e6cf190596 | ||
|
|
ee2d98f664 | ||
|
|
18b235da0d | ||
|
|
4c81015bd3 | ||
|
|
a5ccefb661 | ||
|
|
103328e70f | ||
|
|
1562621c5e | ||
|
|
3659c369e0 | ||
|
|
e12ec94747 | ||
|
|
9e56651808 | ||
|
|
d167521270 | ||
|
|
8608a4e75d | ||
|
|
896e4dd192 | ||
|
|
84512927ab | ||
|
|
590a75b9d6 |
@@ -27,7 +27,7 @@ jobs:
|
||||
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: Initialize CodeQL
|
||||
uses: github/codeql-action/init@v4.38.2
|
||||
uses: github/codeql-action/init@v4.38.3
|
||||
# Override language selection by uncommenting this and choosing your languages
|
||||
with:
|
||||
languages: go
|
||||
@@ -35,7 +35,7 @@ jobs:
|
||||
# Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
|
||||
# If this step fails, then you should remove it and run the build manually (see below).
|
||||
- name: Autobuild
|
||||
uses: github/codeql-action/autobuild@v4.38.2
|
||||
uses: github/codeql-action/autobuild@v4.38.3
|
||||
|
||||
# ℹ️ Command-line programs to run using the OS shell.
|
||||
# 📚 See https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idstepsrun
|
||||
@@ -49,4 +49,4 @@ jobs:
|
||||
# make release
|
||||
|
||||
- name: Perform CodeQL Analysis
|
||||
uses: github/codeql-action/analyze@v4.38.2
|
||||
uses: github/codeql-action/analyze@v4.38.3
|
||||
|
||||
@@ -405,7 +405,7 @@ jobs:
|
||||
output: trivy-results.sarif
|
||||
exit-code: '0'
|
||||
- name: Upload Trivy scan results to GitHub Security
|
||||
uses: github/codeql-action/upload-sarif@v4.38.2
|
||||
uses: github/codeql-action/upload-sarif@v4.38.3
|
||||
if: always()
|
||||
with:
|
||||
sarif_file: trivy-results.sarif
|
||||
|
||||
@@ -460,7 +460,7 @@ jobs:
|
||||
|
||||
- name: Upload Trivy scan results to GitHub Security
|
||||
if: always()
|
||||
uses: github/codeql-action/upload-sarif@v4.38.2
|
||||
uses: github/codeql-action/upload-sarif@v4.38.3
|
||||
with:
|
||||
sarif_file: trivy-results.sarif
|
||||
category: trivy-${{ matrix.variant }}
|
||||
|
||||
@@ -39,7 +39,7 @@ jobs:
|
||||
check-latest: true
|
||||
|
||||
- name: Set up chart-testing
|
||||
uses: helm/chart-testing-action@v2.8.0
|
||||
uses: helm/chart-testing-action@v2.9.0
|
||||
|
||||
- name: Run chart-testing (list-changed)
|
||||
id: list-changed
|
||||
@@ -1869,7 +1869,7 @@ jobs:
|
||||
echo "IMAGE_TAG=$TAG" >> $GITHUB_ENV
|
||||
|
||||
- name: Create kind cluster
|
||||
uses: helm/kind-action@v1.15.1
|
||||
uses: helm/kind-action@v1.15.0
|
||||
|
||||
- name: Verify volume resize hook security contexts
|
||||
run: |
|
||||
|
||||
@@ -24,7 +24,7 @@ require (
|
||||
github.com/facebookgo/stats v0.0.0-20151006221625-1b76add642e4
|
||||
github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4 // indirect
|
||||
github.com/fsnotify/fsnotify v1.9.0 // indirect
|
||||
github.com/go-redsync/redsync/v4 v4.17.0
|
||||
github.com/go-redsync/redsync/v4 v4.18.0
|
||||
github.com/go-sql-driver/mysql v1.10.1
|
||||
github.com/go-zookeeper/zk v1.0.4 // indirect
|
||||
github.com/golang/protobuf v1.5.4
|
||||
@@ -157,7 +157,7 @@ require (
|
||||
github.com/twmb/avro v1.9.0
|
||||
github.com/xeipuuv/gojsonschema v1.2.0
|
||||
github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.3
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.151.1
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.155.1
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.7.1
|
||||
go.uber.org/atomic v1.12.0
|
||||
golang.org/x/sync v0.24.0
|
||||
|
||||
@@ -527,8 +527,8 @@ github.com/go-redis/redis/v7 v7.4.1 h1:PASvf36gyUpr2zdOUS/9Zqc80GbM+9BDyiJSJDDOr
|
||||
github.com/go-redis/redis/v7 v7.4.1/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
|
||||
github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
|
||||
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
|
||||
github.com/go-redsync/redsync/v4 v4.17.0 h1:FFJ+uxZs44y4Sq10//IFKic9T94AYl+u3Sog98AHzBo=
|
||||
github.com/go-redsync/redsync/v4 v4.17.0/go.mod h1:CKVA6qwT07S/916i+Yd9h1/8YFQhCCpPYTQhvvYytJo=
|
||||
github.com/go-redsync/redsync/v4 v4.18.0 h1:DDM09PEhmNTf/vziolf2x6391y1hHxF5vQ4ytHl9xQc=
|
||||
github.com/go-redsync/redsync/v4 v4.18.0/go.mod h1:yuoqcQ55FS1VypSeVTMqFsJqTdWpEH5eA4MQGKJ9c4M=
|
||||
github.com/go-resty/resty/v2 v2.17.2 h1:FQW5oHYcIlkCNrMD2lloGScxcHJ0gkjshV3qcQAyHQk=
|
||||
github.com/go-resty/resty/v2 v2.17.2/go.mod h1:kCKZ3wWmwJaNc7S29BRtUhJwy7iqmn+2mLtQrOyQlVA=
|
||||
github.com/go-sql-driver/mysql v1.10.1 h1:arlSnNLq6a5yxGxV7qg9lF4j0C+KwD6NbQyKr9QL6ME=
|
||||
@@ -1063,10 +1063,10 @@ github.com/rdleal/intervalst v1.5.0 h1:SEB9bCFz5IqD1yhfH1Wv8IBnY/JQxDplwkxHjT6ha
|
||||
github.com/rdleal/intervalst v1.5.0/go.mod h1:xO89Z6BC+LQDH+IPQQw/OESt5UADgFD41tYMUINGpxQ=
|
||||
github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0=
|
||||
github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4=
|
||||
github.com/redis/rueidis v1.0.76 h1:RdDWuvlYBSp+bTrBvaXqJnNEL3VVzsnjo+0psPFgLc4=
|
||||
github.com/redis/rueidis v1.0.76/go.mod h1:UsfHPSbomB6QAVMk4iiFkzRy0nh9o7scDGa+SitvBY4=
|
||||
github.com/redis/rueidis/rueidiscompat v1.0.76 h1:7LikbiqCQqCsZXeZ+akgZMnjIV/J0VHih9PIX4gGZC4=
|
||||
github.com/redis/rueidis/rueidiscompat v1.0.76/go.mod h1:UatQQLVj4QMIsZtpvRWY28qm6r2d72idhcS+C/RM+Zg=
|
||||
github.com/redis/rueidis v1.0.77 h1:ZR41bgJcm7oRFb3aSDPrRhC0eonDSrPzjvvZvHIlNjM=
|
||||
github.com/redis/rueidis v1.0.77/go.mod h1:L8mnCQJJaSNL6I4pIR6Rz732HTGS9vmuXm0yT9dRvjo=
|
||||
github.com/redis/rueidis/rueidiscompat v1.0.77 h1:S1xiLQsFv8XQ46e0LelC8ihGqbDzVRocXXZLzVAeh64=
|
||||
github.com/redis/rueidis/rueidiscompat v1.0.77/go.mod h1:aXCvRVUTvTHZY0owHuz9JjveO1hyuzyDWrpJh8/E3Ds=
|
||||
github.com/rekby/fixenv v0.6.1 h1:jUFiSPpajT4WY2cYuc++7Y1zWrnCxnovGCIX72PZniM=
|
||||
github.com/rekby/fixenv v0.6.1/go.mod h1:/b5LRc06BYJtslRtHKxsPWFT/ySpHV+rWvzTg+XWk4c=
|
||||
github.com/relvacode/iso8601 v1.7.0 h1:BXy+V60stMP6cpswc+a93Mq3e65PfXCgDFfhvNNGrdo=
|
||||
@@ -1317,8 +1317,8 @@ github.com/ydb-platform/ydb-go-genproto v0.0.0-20260810122915-65bfd5c4b705 h1:7V
|
||||
github.com/ydb-platform/ydb-go-genproto v0.0.0-20260810122915-65bfd5c4b705/go.mod h1:Er+FePu1dNUieD+XTMDduGpQuCPssK5Q4BjF+IIXJ3I=
|
||||
github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.3 h1:A48PcmIETWwbEMIGwndv52uME6J1eJ28MBgF7HVR74o=
|
||||
github.com/ydb-platform/ydb-go-sdk-auth-environ v0.5.3/go.mod h1:oRGF+67GWLbtNzqGygsPIUsjXmsRXD1KS5m3o5ppm2k=
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.151.1 h1:T+fB2ZDHpYIGC7DWjK+rzZHLoexBF0zG/n3Q9DGNskY=
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.151.1/go.mod h1:dJXJ1u00IqO8Vsph8fWmYj8b1E7jyphnvJPqA69emBY=
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.155.1 h1:ZMByYe53zZ9R1o23fxJsbNhjzKFQkzpXm6uGj8gWHXQ=
|
||||
github.com/ydb-platform/ydb-go-sdk/v3 v3.155.1/go.mod h1:dJXJ1u00IqO8Vsph8fWmYj8b1E7jyphnvJPqA69emBY=
|
||||
github.com/ydb-platform/ydb-go-yc v0.12.5 h1:qYIeU1N6a92FaDA0+/vR55l5IzRF50MeLZwUWwvy2xQ=
|
||||
github.com/ydb-platform/ydb-go-yc v0.12.5/go.mod h1:DlSltVQKEdXJotaAYwoHzMGDF9EJw5S3hg3LXe/5yYw=
|
||||
github.com/ydb-platform/ydb-go-yc-metadata v0.6.2 h1:CZFawwHjLcjsxcl/g0dGIXbI2r41okPGLCydF4qTHcc=
|
||||
|
||||
+1080
-1050
File diff suppressed because it is too large
Load Diff
|
Before Width: | Height: | Size: 54 KiB After Width: | Height: | Size: 54 KiB |
+157
-54
@@ -12,6 +12,7 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -64,8 +65,9 @@ var (
|
||||
miniS3Options S3Options
|
||||
miniWebDavOptions WebDavOption
|
||||
miniAdminOptions AdminOptions
|
||||
// Track which port flags were explicitly passed on CLI before config file is applied
|
||||
explicitPortFlags map[string]bool
|
||||
// Flags explicitly provided on the command line or via WEED_* env vars,
|
||||
// captured before the saved options file is applied
|
||||
explicitFlags map[string]bool
|
||||
miniEnableWebDAV *bool
|
||||
miniEnableS3 *bool
|
||||
miniEnableAdminUI *bool
|
||||
@@ -376,6 +378,19 @@ The S3 endpoint is available at http://localhost:8333. For client
|
||||
configuration and IAM setup, see the project documentation or use the
|
||||
Admin UI (http://localhost:23646) to manage users and policies.
|
||||
|
||||
Notes:
|
||||
- Values mini computes itself (currently master.volumeSizeLimitMB) are
|
||||
pinned in <dir>/mini.options and applied again on the next start to
|
||||
keep them stable; applied values are printed at startup. Command-line
|
||||
flags are never saved and always win over saved values. Edit or
|
||||
delete mini.options to reset a computed value.
|
||||
- The official Docker image ships /etc/seaweedfs/filer.toml, which keeps
|
||||
the filer metadata in /data/filerldb2 regardless of -dir; mini warns
|
||||
at startup when a configured store dir is outside -dir. If you set
|
||||
-dir to another path, keep /data on a persistent volume as well.
|
||||
Otherwise the metadata is lost when the container is recreated, and
|
||||
buckets list no objects although the volume files are still in -dir.
|
||||
|
||||
`,
|
||||
}
|
||||
|
||||
@@ -823,8 +838,8 @@ func findAvailablePortOnIP(ip string, startPort int, maxAttempts int, reservedPo
|
||||
// If the port is not available, it finds the next available port and updates the pointer
|
||||
// The reservedPorts map contains ports that should not be allocated (for gRPC collision avoidance)
|
||||
func ensurePortAvailableOnIP(portPtr *int, serviceName string, ip string, reservedPorts map[int]bool, flagName string) error {
|
||||
// Check if this port was explicitly specified by the user (from CLI, before config file was applied)
|
||||
isExplicitPort := explicitPortFlags[flagName]
|
||||
// Check if this port was explicitly specified by the user (from CLI or env, before the config file was applied)
|
||||
isExplicitPort := explicitFlags[flagName]
|
||||
|
||||
if *portPtr == 0 {
|
||||
return nil
|
||||
@@ -1072,10 +1087,16 @@ func initializeGrpcPortsOnIP(bindIp string) {
|
||||
}
|
||||
}
|
||||
|
||||
// miniOptionsFilePath returns the path of the mini.options file under the
|
||||
// first -dir folder.
|
||||
func miniOptionsFilePath(dataFolder string) string {
|
||||
return filepath.Join(util.ResolvePath(util.StringSplit(dataFolder, ",")[0]), "mini.options")
|
||||
}
|
||||
|
||||
// loadMiniConfigurationFile reads the mini.options file and returns parsed options
|
||||
// File format: one option per line, without leading dash (e.g., "ip=127.0.0.1")
|
||||
func loadMiniConfigurationFile(dataFolder string) (map[string]string, error) {
|
||||
configFile := filepath.Join(util.ResolvePath(util.StringSplit(dataFolder, ",")[0]), "mini.options")
|
||||
configFile := miniOptionsFilePath(dataFolder)
|
||||
|
||||
options := make(map[string]string)
|
||||
|
||||
@@ -1123,31 +1144,112 @@ func loadMiniConfigurationFile(dataFolder string) (map[string]string, error) {
|
||||
return options, nil
|
||||
}
|
||||
|
||||
// applyConfigFileOptions sets command-line flags from loaded configuration file
|
||||
func applyConfigFileOptions(options map[string]string) {
|
||||
for key, value := range options {
|
||||
// Skip port flags that were explicitly passed on CLI
|
||||
if explicitPortFlags[key] {
|
||||
glog.V(2).Infof("Skipping config file option %s=%s (explicitly specified on command line)", key, value)
|
||||
// applyConfigFileOptions sets command-line flags from loaded configuration
|
||||
// file. Options for flags the user provided explicitly (command line or
|
||||
// WEED_* env var) are skipped, so passing a flag always wins over the saved
|
||||
// value — including passing it at its default value. Returns a map of the
|
||||
// applied keys to their display values (secrets redacted).
|
||||
func applyConfigFileOptions(options map[string]string) map[string]string {
|
||||
keys := make([]string, 0, len(options))
|
||||
for key := range options {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
|
||||
applied := make(map[string]string)
|
||||
for _, key := range keys {
|
||||
value := options[key]
|
||||
// Redact secrets (admin.password, admin.readOnlyPassword, ...) in
|
||||
// logs and the startup message; only flag.Value.Set sees the value.
|
||||
display := value
|
||||
if isSensitiveOption(key) {
|
||||
display = "*****"
|
||||
}
|
||||
if explicitFlags[key] {
|
||||
glog.V(2).Infof("Skipping config file option %s=%s (explicitly specified)", key, display)
|
||||
continue
|
||||
}
|
||||
// Set the flag value if it hasn't been explicitly set on command line
|
||||
flag := cmdMini.Flag.Lookup(key)
|
||||
if flag != nil {
|
||||
// Only set if not already set (by command line)
|
||||
if flag.Value.String() == flag.DefValue {
|
||||
if err := flag.Value.Set(value); err != nil {
|
||||
glog.Warningf("Failed to apply config file option: %s=%s: %v", key, value, err)
|
||||
} else {
|
||||
glog.V(2).Infof("Applied config file option: %s=%s", key, value)
|
||||
if flag == nil {
|
||||
continue
|
||||
}
|
||||
if err := flag.Value.Set(value); err != nil {
|
||||
glog.Warningf("Failed to apply config file option: %s=%s: %v", key, display, err)
|
||||
continue
|
||||
}
|
||||
glog.V(2).Infof("Applied config file option: %s=%s", key, display)
|
||||
applied[key] = display
|
||||
}
|
||||
return applied
|
||||
}
|
||||
|
||||
// isSensitiveOption reports whether a saved option holds a secret that must
|
||||
// not appear in logs or the startup message.
|
||||
func isSensitiveOption(key string) bool {
|
||||
k := strings.ToLower(key)
|
||||
return strings.Contains(k, "password") || strings.Contains(k, "secret") || strings.Contains(k, "token")
|
||||
}
|
||||
|
||||
// warnIfFilerStoreOutsideDir warns when a filer.toml configures a local
|
||||
// filer store outside the -dir folders and the master meta folder. When a
|
||||
// filer.toml is found the filer skips defaultLevelDbDirectory, so metadata
|
||||
// can silently land on a path the volume data doesn't use — in a container
|
||||
// that path is lost on recreate unless it is persisted separately.
|
||||
func warnIfFilerStoreOutsideDir(dataFolders string, metaFolder string) {
|
||||
if !util.LoadConfiguration("filer", false) {
|
||||
return
|
||||
}
|
||||
v := util.GetViper()
|
||||
folders := append(util.StringSplit(dataFolders, ","), metaFolder)
|
||||
inside := func(storeDir string) bool {
|
||||
// The filer stores open the configured dir verbatim, so absolutize it
|
||||
// the same way (relative to the working directory) rather than
|
||||
// expanding "~", which the store never expands.
|
||||
resolved, err := filepath.Abs(storeDir)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, folder := range folders {
|
||||
base, err := filepath.Abs(util.ResolvePath(folder))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
// Resolve symlinks only when both sides exist, so a not-yet-
|
||||
// created store dir under a symlinked -dir still compares
|
||||
// lexically against the same spelling.
|
||||
r, b := resolved, base
|
||||
if er, err := filepath.EvalSymlinks(resolved); err == nil {
|
||||
if eb, err := filepath.EvalSymlinks(base); err == nil {
|
||||
r, b = er, eb
|
||||
}
|
||||
}
|
||||
if rel, err := filepath.Rel(b, r); err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
// Local filer stores keep metadata under a "dir" key; check both the
|
||||
// default store (leveldb2.enabled) and path-specific stores
|
||||
// (leveldb2.<id>.enabled, keyed under <section>.location).
|
||||
localStores := map[string]bool{"leveldb": true, "leveldb2": true, "leveldb3": true}
|
||||
for _, key := range v.AllKeys() {
|
||||
if !strings.HasSuffix(key, ".enabled") || !v.GetBool(key) {
|
||||
continue
|
||||
}
|
||||
section := strings.TrimSuffix(key, ".enabled")
|
||||
if !localStores[strings.Split(section, ".")[0]] {
|
||||
continue
|
||||
}
|
||||
if storeDir := v.GetString(section + ".dir"); storeDir != "" && !inside(storeDir) {
|
||||
glog.Warningf("filer store %s uses dir %s from filer.toml, outside -dir %s; persist that path or filer metadata will be lost when the container is recreated", section, storeDir, dataFolders)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// saveMiniConfiguration saves the current mini configuration to a file
|
||||
// The file format uses option=value format without leading dashes
|
||||
// saveMiniConfiguration writes values mini computed itself to mini.options.
|
||||
// Command-line flags are never saved, so a dropped flag cannot resurrect on
|
||||
// the next start; the file exists to pin environment-derived values.
|
||||
func saveMiniConfiguration(dataFolder string) error {
|
||||
configDir := util.ResolvePath(util.StringSplit(dataFolder, ",")[0])
|
||||
if err := os.MkdirAll(configDir, 0755); err != nil {
|
||||
@@ -1159,29 +1261,15 @@ func saveMiniConfiguration(dataFolder string) error {
|
||||
|
||||
var sb strings.Builder
|
||||
sb.WriteString("#!/bin/bash\n")
|
||||
sb.WriteString("# Mini server configuration\n")
|
||||
sb.WriteString("# Mini server auto-computed options\n")
|
||||
sb.WriteString("# Format: option=value (no leading dash)\n")
|
||||
sb.WriteString("# This file is loaded on startup if it exists\n\n")
|
||||
sb.WriteString("# This file is loaded on startup if it exists\n")
|
||||
sb.WriteString("# Command-line flags are not saved; only values mini computed itself.\n\n")
|
||||
|
||||
// Collect all flags that were explicitly passed (except "dir")
|
||||
cmdMini.Flag.Visit(func(f *flag.Flag) {
|
||||
// Skip the "dir" option - it's environment-specific
|
||||
if f.Name == "dir" {
|
||||
return
|
||||
}
|
||||
value := f.Value.String()
|
||||
// Quote the value if it contains spaces
|
||||
if strings.Contains(value, " ") {
|
||||
sb.WriteString(fmt.Sprintf("%s=\"%s\"\n", f.Name, value))
|
||||
} else {
|
||||
sb.WriteString(fmt.Sprintf("%s=%s\n", f.Name, value))
|
||||
}
|
||||
})
|
||||
|
||||
// Add auto-calculated volume size if it was computed
|
||||
// Pin the auto-calculated volume size so it stays stable across restarts
|
||||
// even as free disk space changes. Delete this line to force
|
||||
// recalculation on the next startup.
|
||||
if !isFlagPassed("master.volumeSizeLimitMB") && miniMasterOptions.volumeSizeLimitMB != nil {
|
||||
sb.WriteString(fmt.Sprintf("\n# Auto-calculated volume size based on total disk capacity\n"))
|
||||
sb.WriteString(fmt.Sprintf("# Delete this line to force recalculation on next startup\n"))
|
||||
sb.WriteString(fmt.Sprintf("master.volumeSizeLimitMB=%d\n", *miniMasterOptions.volumeSizeLimitMB))
|
||||
}
|
||||
|
||||
@@ -1203,21 +1291,32 @@ func runMini(cmd *Command, args []string) bool {
|
||||
// can opt back into verbose output.
|
||||
quietMiniLogs()
|
||||
|
||||
// Capture which port flags were explicitly passed on CLI BEFORE config file is applied
|
||||
// This is necessary to distinguish user-specified ports from defaults or config file options
|
||||
explicitPortFlags = make(map[string]bool)
|
||||
portFlagNames := []string{"master.port", "filer.port", "volume.port", "s3.port", "s3.port.iceberg", "webdav.port", "admin.port", "s3.iam.readOnly"}
|
||||
for _, flagName := range portFlagNames {
|
||||
explicitPortFlags[flagName] = isFlagPassed(flagName)
|
||||
}
|
||||
// Capture which flags were explicitly provided on the command line or via
|
||||
// WEED_* env vars BEFORE the saved options file is applied. Saved options
|
||||
// only fill in flags the user did not provide this run, so passing a flag
|
||||
// at its default value also resets a saved value.
|
||||
explicitFlags = make(map[string]bool)
|
||||
cmdMini.Flag.Visit(func(f *flag.Flag) {
|
||||
explicitFlags[f.Name] = true
|
||||
})
|
||||
|
||||
// Load configuration from file if it exists
|
||||
// Load configuration from file if it exists. Capture the path before
|
||||
// applying: a saved "dir" option can change miniDataFolders.
|
||||
optionsFile := miniOptionsFilePath(*miniDataFolders)
|
||||
configOptions, err := loadMiniConfigurationFile(*miniDataFolders)
|
||||
if err != nil {
|
||||
glog.Warningf("Error loading configuration file: %v", err)
|
||||
}
|
||||
// Apply loaded options to flags (CLI flags will override these)
|
||||
applyConfigFileOptions(configOptions)
|
||||
// Apply loaded options to flags (explicitly provided flags win)
|
||||
appliedOptions := applyConfigFileOptions(configOptions)
|
||||
if len(appliedOptions) > 0 {
|
||||
pairs := make([]string, 0, len(appliedOptions))
|
||||
for k, v := range appliedOptions {
|
||||
pairs = append(pairs, k+"="+v)
|
||||
}
|
||||
sort.Strings(pairs)
|
||||
fmt.Printf(" Applying %d saved option(s) from %s: %s\n", len(appliedOptions), optionsFile, strings.Join(pairs, " "))
|
||||
}
|
||||
|
||||
if *miniOptions.debug {
|
||||
grace.StartDebugServer(*miniOptions.debugPort)
|
||||
@@ -1346,10 +1445,14 @@ func runMini(cmd *Command, args []string) bool {
|
||||
glog.Fatalf("Check Meta Folder (-dir=\"%s\") Writable: %s", *miniMasterOptions.metaFolder, err)
|
||||
}
|
||||
miniFilerOptions.defaultLevelDbDirectory = miniMasterOptions.metaFolder
|
||||
warnIfFilerStoreOutsideDir(*miniDataFolders, *miniMasterOptions.metaFolder)
|
||||
|
||||
// Calculate and set optimal volume size limit based on available disk space
|
||||
// Only auto-calculate if user didn't explicitly specify a value via -master.volumeSizeLimitMB
|
||||
if !isFlagPassed("master.volumeSizeLimitMB") {
|
||||
// Calculate and set optimal volume size limit based on available disk space.
|
||||
// Skip when the user passed -master.volumeSizeLimitMB explicitly, and when
|
||||
// the value was loaded from mini.options — file-applied flags are not in
|
||||
// Visit, and the saved pin must survive restarts.
|
||||
_, sizeLimitFromFile := appliedOptions["master.volumeSizeLimitMB"]
|
||||
if !isFlagPassed("master.volumeSizeLimitMB") && !sizeLimitFromFile {
|
||||
// User didn't override, use auto-calculated value
|
||||
// The -dir flag can accept comma-separated directories; use the first one for disk space calculation.
|
||||
// miniDataFolders was already tilde-resolved at the top of runMini.
|
||||
|
||||
@@ -0,0 +1,359 @@
|
||||
package ec
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
)
|
||||
|
||||
// ecBalanceFakeServer answers the EC-shard RPC subset a balance run drives:
|
||||
// copy (rejected when full), mount, inventory, unmount, delete. The disk state
|
||||
// a real server tracks lives in shards; what the planner saw is rebuilt from it
|
||||
// through FetchTopology, so a landed move is visible to the next plan.
|
||||
type ecBalanceFakeServer struct {
|
||||
volume_server_pb.UnimplementedVolumeServerServer
|
||||
|
||||
mu sync.Mutex
|
||||
shards map[uint32]map[uint32]bool // vid -> shardId -> present
|
||||
full bool // VolumeEcShardsCopy answers "no space left"
|
||||
copyCalls int
|
||||
grpcAddr string
|
||||
stop func()
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.copyCalls++
|
||||
if f.full {
|
||||
return nil, fmt.Errorf("no space left")
|
||||
}
|
||||
return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
for _, sid := range req.ShardIds {
|
||||
f.shards[req.VolumeId][sid] = true
|
||||
}
|
||||
return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
resp := &volume_server_pb.VolumeEcShardsInfoResponse{}
|
||||
for sid := range f.shards[req.VolumeId] {
|
||||
resp.EcShardInfos = append(resp.EcShardInfos, &volume_server_pb.EcShardInfo{
|
||||
VolumeId: req.VolumeId,
|
||||
ShardId: sid,
|
||||
Size: 1,
|
||||
})
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) {
|
||||
return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
for _, sid := range req.ShardIds {
|
||||
delete(f.shards[req.VolumeId], sid)
|
||||
}
|
||||
return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil
|
||||
}
|
||||
|
||||
// nodeId returns the "host:httpPort.grpcPort" node id form the cluster uses, so
|
||||
// ToGrpcAddress resolves to this server's listener.
|
||||
func (f *ecBalanceFakeServer) nodeId() string {
|
||||
host, port, err := net.SplitHostPort(f.grpcAddr)
|
||||
if err != nil {
|
||||
return f.grpcAddr
|
||||
}
|
||||
return fmt.Sprintf("%s:1.%s", host, port)
|
||||
}
|
||||
|
||||
func (f *ecBalanceFakeServer) shardCount(vid uint32) int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.shards[vid])
|
||||
}
|
||||
|
||||
func startEcBalanceFakeServer(t *testing.T, vid uint32, present ...uint32) *ecBalanceFakeServer {
|
||||
t.Helper()
|
||||
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
|
||||
f := &ecBalanceFakeServer{
|
||||
shards: map[uint32]map[uint32]bool{vid: {}},
|
||||
grpcAddr: listener.Addr().String(),
|
||||
}
|
||||
for _, sid := range present {
|
||||
f.shards[vid][sid] = true
|
||||
}
|
||||
|
||||
server := grpc.NewServer()
|
||||
volume_server_pb.RegisterVolumeServerServer(server, f)
|
||||
go func() { _ = server.Serve(listener) }()
|
||||
f.stop = server.Stop
|
||||
t.Cleanup(server.Stop)
|
||||
return f
|
||||
}
|
||||
|
||||
// fetchEcBalanceTopology builds a FetchTopology that reports the fake servers'
|
||||
// live shard inventory.
|
||||
func fetchEcBalanceTopology(servers []*ecBalanceFakeServer) func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
|
||||
return func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
|
||||
var nodes []*master_pb.DataNodeInfo
|
||||
for _, f := range servers {
|
||||
di := &master_pb.DiskInfo{Type: string(types.HardDriveType), MaxVolumeCount: 100}
|
||||
f.mu.Lock()
|
||||
for v, set := range f.shards {
|
||||
if len(set) == 0 {
|
||||
continue
|
||||
}
|
||||
si := erasure_coding.NewShardsInfo()
|
||||
for sid := range set {
|
||||
si.Set(erasure_coding.NewShardInfo(erasure_coding.ShardId(sid), 1))
|
||||
}
|
||||
di.EcShardInfos = append(di.EcShardInfos, &master_pb.VolumeEcShardInformationMessage{
|
||||
Id: v,
|
||||
Collection: "c1",
|
||||
EcIndexBits: si.Bitmap(),
|
||||
ShardSizes: si.SizesInt64(),
|
||||
DiskType: string(types.HardDriveType),
|
||||
})
|
||||
}
|
||||
f.mu.Unlock()
|
||||
nodes = append(nodes, &master_pb.DataNodeInfo{
|
||||
Id: f.nodeId(),
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{string(types.HardDriveType): di},
|
||||
})
|
||||
}
|
||||
return &master_pb.TopologyInfo{
|
||||
DataCenterInfos: []*master_pb.DataCenterInfo{{
|
||||
Id: "dc1",
|
||||
RackInfos: []*master_pb.RackInfo{{
|
||||
Id: "r1",
|
||||
DataNodeInfos: nodes,
|
||||
}},
|
||||
}},
|
||||
}, 0, nil
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcBalanceReplanAfterNoSpaceLeft reproduces seaweedfs/seaweedfs#11753:
|
||||
// the planner works from heartbeat free-slot counts that can disagree with the
|
||||
// volume server's own check, so a planned target can reject the copy with
|
||||
// "no space left" mid-run. The balance must re-plan without that node as a
|
||||
// target rather than fail the whole run.
|
||||
func TestEcBalanceReplanAfterNoSpaceLeft(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
var allShards []uint32
|
||||
for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
|
||||
allShards = append(allShards, uint32(sid))
|
||||
}
|
||||
|
||||
full := startEcBalanceFakeServer(t, vid)
|
||||
accept := startEcBalanceFakeServer(t, vid)
|
||||
source := startEcBalanceFakeServer(t, vid, allShards...)
|
||||
full.full = true
|
||||
|
||||
env := &Env{
|
||||
GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
FetchTopology: fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept}),
|
||||
}
|
||||
|
||||
err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("balance should re-plan around the full target, got: %v", err)
|
||||
}
|
||||
if full.copyCalls == 0 {
|
||||
t.Fatalf("expected the full server to have rejected at least one copy")
|
||||
}
|
||||
if got := full.shardCount(vid); got != 0 {
|
||||
t.Fatalf("full server should receive no shards, holds %d", got)
|
||||
}
|
||||
total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid)
|
||||
if total != len(allShards) {
|
||||
t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards))
|
||||
}
|
||||
if source.shardCount(vid) == len(allShards) {
|
||||
t.Fatalf("no shards were moved off the source")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot serves the pre-move
|
||||
// topology on every collection after the first, simulating a master heartbeat
|
||||
// that lags the landed moves. The re-plan must replay its completed moves onto
|
||||
// the stale snapshot instead of re-moving shards off the deleted source.
|
||||
func TestEcBalanceReplanAfterNoSpaceLeftStaleSnapshot(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
var allShards []uint32
|
||||
for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
|
||||
allShards = append(allShards, uint32(sid))
|
||||
}
|
||||
|
||||
full := startEcBalanceFakeServer(t, vid)
|
||||
accept := startEcBalanceFakeServer(t, vid)
|
||||
source := startEcBalanceFakeServer(t, vid, allShards...)
|
||||
full.full = true
|
||||
|
||||
live := fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full, accept})
|
||||
var stale *master_pb.TopologyInfo
|
||||
calls := 0
|
||||
fetchTopology := func(delay time.Duration) (*master_pb.TopologyInfo, uint64, error) {
|
||||
topo, size, err := live(delay)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
calls++
|
||||
if stale == nil {
|
||||
stale = topo
|
||||
}
|
||||
if calls > 1 {
|
||||
return stale, size, nil
|
||||
}
|
||||
return topo, size, nil
|
||||
}
|
||||
|
||||
env := &Env{
|
||||
GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
FetchTopology: fetchTopology,
|
||||
}
|
||||
|
||||
err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("balance should re-plan around the full target, got: %v", err)
|
||||
}
|
||||
if calls < 2 {
|
||||
t.Fatalf("expected a re-plan, topology collected %d time(s)", calls)
|
||||
}
|
||||
// Each shard lands on the accepting server exactly once; a stale re-plan
|
||||
// would copy already-moved shards again.
|
||||
if got, want := accept.copyCalls, accept.shardCount(vid); got != want {
|
||||
t.Fatalf("accepting server saw %d copies for %d held shards: completed moves were re-planned", got, want)
|
||||
}
|
||||
total := source.shardCount(vid) + full.shardCount(vid) + accept.shardCount(vid)
|
||||
if total != len(allShards) {
|
||||
t.Fatalf("shards lost or duplicated: total %d, want %d", total, len(allShards))
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcBalanceNoReceivingTargetFails surfaces the refusal when every
|
||||
// destination that could take a shard has rejected a copy, rather than
|
||||
// reporting success for a plan that moved nothing.
|
||||
func TestEcBalanceNoReceivingTargetFails(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
var allShards []uint32
|
||||
for sid := 0; sid < erasure_coding.DataShardsCount+erasure_coding.ParityShardsCount; sid++ {
|
||||
allShards = append(allShards, uint32(sid))
|
||||
}
|
||||
|
||||
full := startEcBalanceFakeServer(t, vid)
|
||||
source := startEcBalanceFakeServer(t, vid, allShards...)
|
||||
full.full = true
|
||||
|
||||
env := &Env{
|
||||
GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||||
FetchTopology: fetchEcBalanceTopology([]*ecBalanceFakeServer{source, full}),
|
||||
}
|
||||
|
||||
err := EcBalance(env, []string{"c1"}, "", nil, types.HardDriveType, 3, 0, true, nil, nil, nil)
|
||||
if err == nil {
|
||||
t.Fatalf("balance with no receiving target should fail")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "no space left") {
|
||||
t.Fatalf("expected the refusal to surface, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestApplyCompletedMovesReplaysDiskState checks that replaying a landed move
|
||||
// onto a stale snapshot carries the shard's disk attribution and the per-disk
|
||||
// slot accounting, so a re-plan does not treat the just-filled disk as empty.
|
||||
func TestApplyCompletedMovesReplaysDiskState(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
|
||||
source := newEcNode("dc1", "r1", "dn1", 99)
|
||||
source.Info.DiskInfos["hdd"] = &master_pb.DiskInfo{
|
||||
Type: "hdd",
|
||||
EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{{
|
||||
Id: vid,
|
||||
Collection: "c1",
|
||||
DiskId: 1,
|
||||
DiskType: "hdd",
|
||||
EcIndexBits: uint32(erasure_coding.ShardBits(0).Set(5)),
|
||||
}},
|
||||
}
|
||||
si := erasure_coding.NewShardsInfo()
|
||||
si.Set(erasure_coding.NewShardInfo(5, 1))
|
||||
source.Disks = map[uint32]*EcDisk{
|
||||
1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 99, EcShardCount: 1, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{1: si}},
|
||||
}
|
||||
|
||||
target := newEcNode("dc1", "r1", "dn2", 100)
|
||||
target.Disks = map[uint32]*EcDisk{
|
||||
1: {DiskId: 1, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}},
|
||||
2: {DiskId: 2, DiskType: "hdd", FreeEcSlots: 10, EcShardCount: 0, EcShards: map[needle.VolumeId]*erasure_coding.ShardsInfo{}},
|
||||
}
|
||||
|
||||
nodes := []*EcNode{source, target}
|
||||
move := ecbalancer.Move{
|
||||
VolumeID: vid,
|
||||
ShardID: 5,
|
||||
Collection: "c1",
|
||||
SourceNode: "dn1",
|
||||
TargetNode: "dn2",
|
||||
TargetDisk: 2,
|
||||
Phase: "within_rack",
|
||||
}
|
||||
applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType)
|
||||
|
||||
if ecNodeHasShard(source, needle.VolumeId(vid), 5) {
|
||||
t.Fatalf("source still holds shard 5")
|
||||
}
|
||||
if d := source.Disks[1]; d.EcShardCount != 0 || d.FreeEcSlots != 100 {
|
||||
t.Fatalf("source disk counts not restored: %+v", d)
|
||||
}
|
||||
if !ecNodeHasShard(target, needle.VolumeId(vid), 5) {
|
||||
t.Fatalf("target missing shard 5")
|
||||
}
|
||||
if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 {
|
||||
t.Fatalf("target disk 2 not charged: %+v", d)
|
||||
}
|
||||
if d := target.Disks[1]; d.EcShardCount != 0 {
|
||||
t.Fatalf("shard attributed to wrong disk: %+v", d)
|
||||
}
|
||||
if target.FreeEcSlot != 99 {
|
||||
t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot)
|
||||
}
|
||||
|
||||
// Replaying again (or onto a fresher snapshot) must not double-count.
|
||||
applyCompletedMoves(nodes, []ecbalancer.Move{move}, types.HardDriveType)
|
||||
if d := target.Disks[2]; d.EcShardCount != 1 || d.FreeEcSlots != 9 {
|
||||
t.Fatalf("double-applied move: %+v", d)
|
||||
}
|
||||
if target.FreeEcSlot != 99 {
|
||||
t.Fatalf("target free slots = %d, want 99", target.FreeEcSlot)
|
||||
}
|
||||
}
|
||||
+310
-49
@@ -8,6 +8,7 @@ import (
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
@@ -762,6 +763,26 @@ type ecBalancer struct {
|
||||
// migratingVolumeIds are mid-encode volumes whose shards are ingested from
|
||||
// every disk-type bucket; see EcBalance.
|
||||
migratingVolumeIds map[uint32]bool
|
||||
// noReceiveTargets are node ids the planner may not pick as destinations;
|
||||
// noReceiveDisks narrows that to individual disks on a node: a server-side
|
||||
// auto-pick refusal means no disk qualifies, but an explicit-disk refusal
|
||||
// only convicts that disk. See EcBalance.
|
||||
noReceiveTargets map[string]bool
|
||||
noReceiveDisks map[string]map[uint32]bool
|
||||
// mu guards the execution journal written by concurrent move tasks.
|
||||
mu sync.Mutex
|
||||
// rejectedTargets and rejectedDisks collect the destinations that refused a
|
||||
// copy this attempt; completedMoves records applied moves so a retry can
|
||||
// replay them onto a stale-heartbeat snapshot; sawOtherErr marks any
|
||||
// failure that is not a "no space left" refusal and makes the run
|
||||
// non-retryable.
|
||||
rejectedTargets map[string]bool
|
||||
rejectedDisks map[string]map[uint32]bool
|
||||
completedMoves []ecbalancer.Move
|
||||
sawOtherErr bool
|
||||
// plannedMoves is the move count of the last plan, for the empty-replan
|
||||
// check in EcBalance.
|
||||
plannedMoves int
|
||||
}
|
||||
|
||||
// excludeNodes is a set of server addresses kept out of the balance as copy/move
|
||||
@@ -782,32 +803,6 @@ type ecBalancer struct {
|
||||
// so a plain ec.balance -diskType=X never drags deliberately tiered shards of
|
||||
// other types onto X disks.
|
||||
func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, ioBytePerSecond int64, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId, migratingVolumeIds []needle.VolumeId) (err error) {
|
||||
// collect all ec nodes
|
||||
allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Drop excluded nodes (and the slots they contribute) before planning so they
|
||||
// can be neither a target nor a source for any move this balance plans.
|
||||
if len(excludeNodes) > 0 {
|
||||
kept := allEcNodes[:0]
|
||||
var excludedFreeSlots int
|
||||
for _, en := range allEcNodes {
|
||||
if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip {
|
||||
excludedFreeSlots += en.FreeEcSlot
|
||||
glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id)
|
||||
continue
|
||||
}
|
||||
kept = append(kept, en)
|
||||
}
|
||||
allEcNodes = kept
|
||||
totalFreeEcSlots -= excludedFreeSlots
|
||||
}
|
||||
|
||||
if totalFreeEcSlots < 1 {
|
||||
return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
|
||||
}
|
||||
|
||||
var volumeIdFilter map[uint32]bool
|
||||
if len(volumeIds) > 0 {
|
||||
@@ -824,22 +819,277 @@ func EcBalance(env *Env, collections []string, dc string, ecReplicaPlacement *su
|
||||
}
|
||||
}
|
||||
|
||||
ecb := &ecBalancer{
|
||||
env: env,
|
||||
ecNodes: allEcNodes,
|
||||
replicaPlacement: ecReplicaPlacement,
|
||||
applyBalancing: applyBalancing,
|
||||
maxParallelization: maxParallelization,
|
||||
ioBytePerSecond: ioBytePerSecond,
|
||||
diskType: diskType,
|
||||
volumeIds: volumeIdFilter,
|
||||
migratingVolumeIds: migrating,
|
||||
}
|
||||
// The planner's free-slot numbers come from heartbeat snapshots that can lag
|
||||
// the volume server's own accounting, so a planned target can refuse a copy
|
||||
// with "no space left" while shards move concurrently (issue #11753). The
|
||||
// server's refusal is authoritative: re-collect the topology, mark the
|
||||
// refusing targets no-receive, and plan again. They stay usable as sources.
|
||||
noReceiveTargets := make(map[string]bool)
|
||||
noReceiveDisks := make(map[string]map[uint32]bool)
|
||||
var completed []ecbalancer.Move
|
||||
var lastErr error
|
||||
for {
|
||||
// collect all ec nodes
|
||||
allEcNodes, totalFreeEcSlots, err := CollectEcNodesForDC(env, dc, diskType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(collections) == 0 {
|
||||
glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n")
|
||||
// Drop excluded nodes (and the slots they contribute) before planning so they
|
||||
// can be neither a target nor a source for any move this balance plans.
|
||||
if len(excludeNodes) > 0 {
|
||||
kept := allEcNodes[:0]
|
||||
var excludedFreeSlots int
|
||||
for _, en := range allEcNodes {
|
||||
if _, skip := excludeNodes[pb.NewServerAddressFromDataNode(en.Info)]; skip {
|
||||
excludedFreeSlots += en.FreeEcSlot
|
||||
glog.V(0).Infof("EC balance excluding node %s: skipped as unreachable by the encode orphan sweep", en.Info.Id)
|
||||
continue
|
||||
}
|
||||
kept = append(kept, en)
|
||||
}
|
||||
allEcNodes = kept
|
||||
totalFreeEcSlots -= excludedFreeSlots
|
||||
}
|
||||
// A refused destination's slots cannot take a copy, so they do not
|
||||
// count toward the free-space precheck.
|
||||
for _, en := range allEcNodes {
|
||||
if noReceiveTargets[en.Info.Id] {
|
||||
totalFreeEcSlots -= en.FreeEcSlot
|
||||
continue
|
||||
}
|
||||
for diskId := range noReceiveDisks[en.Info.Id] {
|
||||
if d := en.Disks[diskId]; d != nil {
|
||||
totalFreeEcSlots -= d.FreeEcSlots
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if totalFreeEcSlots < 1 {
|
||||
if lastErr != nil {
|
||||
return fmt.Errorf("no receiving targets remain: %w", lastErr)
|
||||
}
|
||||
return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots)
|
||||
}
|
||||
|
||||
// A fresh collection can still describe an applied move through a
|
||||
// heartbeat that predates its source delete; replay the journal.
|
||||
applyCompletedMoves(allEcNodes, completed, diskType)
|
||||
|
||||
ecb := &ecBalancer{
|
||||
env: env,
|
||||
ecNodes: allEcNodes,
|
||||
replicaPlacement: ecReplicaPlacement,
|
||||
applyBalancing: applyBalancing,
|
||||
maxParallelization: maxParallelization,
|
||||
ioBytePerSecond: ioBytePerSecond,
|
||||
diskType: diskType,
|
||||
volumeIds: volumeIdFilter,
|
||||
migratingVolumeIds: migrating,
|
||||
noReceiveTargets: noReceiveTargets,
|
||||
noReceiveDisks: noReceiveDisks,
|
||||
}
|
||||
|
||||
if len(collections) == 0 {
|
||||
glog.V(1).Infof("WARNING: No collections to balance EC volumes across.\n")
|
||||
}
|
||||
err = ecb.balance(collections)
|
||||
if !applyBalancing {
|
||||
return err
|
||||
}
|
||||
if err == nil {
|
||||
if lastErr != nil && ecb.plannedMoves == 0 {
|
||||
return fmt.Errorf("no receiving targets remain: %w", lastErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Replan while the failed attempt surfaced newly-refusing targets; a node
|
||||
// once marked can no longer be picked, so the loop terminates. Any other
|
||||
// failure aborts instead of being hidden by the retry.
|
||||
newRejects := ecb.mergeRejectedTargets(noReceiveTargets, noReceiveDisks)
|
||||
if len(newRejects) == 0 || ecb.hasOtherError() {
|
||||
return err
|
||||
}
|
||||
lastErr = err
|
||||
completed = append(completed, ecb.completedMoves...)
|
||||
fmt.Printf("ec.balance: %v reported no space left; re-collecting topology and re-planning without them as targets\n", newRejects)
|
||||
}
|
||||
}
|
||||
|
||||
// applyCompletedMoves replays already-applied moves onto a freshly collected
|
||||
// node list: heartbeats can lag, so the snapshot may still show a moved shard
|
||||
// on its deleted source or lack it on its mounted target.
|
||||
func applyCompletedMoves(ecNodes []*EcNode, moves []ecbalancer.Move, diskType types.DiskType) {
|
||||
if len(moves) == 0 {
|
||||
return
|
||||
}
|
||||
byID := make(map[string]*EcNode, len(ecNodes))
|
||||
for _, en := range ecNodes {
|
||||
byID[en.Info.Id] = en
|
||||
}
|
||||
for _, m := range moves {
|
||||
vid := needle.VolumeId(m.VolumeID)
|
||||
shardId := erasure_coding.ShardId(m.ShardID)
|
||||
if src := byID[m.SourceNode]; src != nil {
|
||||
removeEcShard(src, vid, shardId)
|
||||
}
|
||||
if m.Phase == "dedup" {
|
||||
continue
|
||||
}
|
||||
if dst := byID[m.TargetNode]; dst != nil && !ecNodeHasShard(dst, vid, shardId) {
|
||||
addEcShard(dst, m.Collection, vid, shardId, m.TargetDisk, diskType)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func ecNodeHasShard(ecNode *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) bool {
|
||||
for _, diskInfo := range ecNode.Info.DiskInfos {
|
||||
for _, eci := range diskInfo.EcShardInfos {
|
||||
if needle.VolumeId(eci.Id) == vid && erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// removeEcShard drops the shard from the node model, including the per-disk
|
||||
// bookkeeping of the disk that reported it.
|
||||
func removeEcShard(en *EcNode, vid needle.VolumeId, shardId erasure_coding.ShardId) {
|
||||
for _, diskInfo := range en.Info.DiskInfos {
|
||||
for _, eci := range diskInfo.EcShardInfos {
|
||||
if needle.VolumeId(eci.Id) != vid || !erasure_coding.ShardBits(eci.EcIndexBits).Has(shardId) {
|
||||
continue
|
||||
}
|
||||
if d := en.Disks[eci.DiskId]; d != nil {
|
||||
if si := d.EcShards[vid]; si != nil {
|
||||
si.Delete(shardId)
|
||||
}
|
||||
d.EcShardCount--
|
||||
d.FreeEcSlots++
|
||||
}
|
||||
}
|
||||
}
|
||||
en.DeleteEcVolumeShards(vid, []erasure_coding.ShardId{shardId})
|
||||
}
|
||||
|
||||
// addEcShard places the shard on the disk the move used when that disk is
|
||||
// known; otherwise (TargetDisk 0, server-picked) it falls back to the node's
|
||||
// disk-type bucket.
|
||||
func addEcShard(en *EcNode, collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, diskId uint32, diskType types.DiskType) {
|
||||
d := en.Disks[diskId]
|
||||
if diskId == 0 || d == nil {
|
||||
en.AddEcVolumeShards(vid, collection, []erasure_coding.ShardId{shardId}, diskType)
|
||||
return
|
||||
}
|
||||
diskInfo := en.Info.DiskInfos[string(diskType)]
|
||||
if diskInfo == nil {
|
||||
diskInfo = &master_pb.DiskInfo{Type: string(diskType)}
|
||||
en.Info.DiskInfos[string(diskType)] = diskInfo
|
||||
}
|
||||
var eci *master_pb.VolumeEcShardInformationMessage
|
||||
for _, e := range diskInfo.EcShardInfos {
|
||||
if needle.VolumeId(e.Id) == vid && e.DiskId == diskId {
|
||||
eci = e
|
||||
break
|
||||
}
|
||||
}
|
||||
if eci == nil {
|
||||
eci = &master_pb.VolumeEcShardInformationMessage{
|
||||
Id: uint32(vid),
|
||||
Collection: collection,
|
||||
DiskId: diskId,
|
||||
DiskType: string(diskType),
|
||||
}
|
||||
diskInfo.EcShardInfos = append(diskInfo.EcShardInfos, eci)
|
||||
}
|
||||
si := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(eci)
|
||||
before := si.Count()
|
||||
si.Set(erasure_coding.NewShardInfo(shardId, 0))
|
||||
eci.EcIndexBits = si.Bitmap()
|
||||
eci.ShardSizes = si.SizesInt64()
|
||||
added := si.Count() - before
|
||||
|
||||
en.FreeEcSlot -= added
|
||||
d.EcShardCount += added
|
||||
d.FreeEcSlots -= added
|
||||
if d.EcShards == nil {
|
||||
d.EcShards = make(map[needle.VolumeId]*erasure_coding.ShardsInfo)
|
||||
}
|
||||
if d.EcShards[vid] == nil {
|
||||
d.EcShards[vid] = erasure_coding.NewShardsInfo()
|
||||
}
|
||||
d.EcShards[vid].Set(erasure_coding.NewShardInfo(shardId, 0))
|
||||
}
|
||||
|
||||
// mergeRejectedTargets folds the destinations this run saw refuse into the
|
||||
// no-receive sets and returns the newly added ids; an empty result means no
|
||||
// progress is possible.
|
||||
func (ecb *ecBalancer) mergeRejectedTargets(noReceiveNodes map[string]bool, noReceiveDisks map[string]map[uint32]bool) (added []string) {
|
||||
ecb.mu.Lock()
|
||||
defer ecb.mu.Unlock()
|
||||
for id := range ecb.rejectedTargets {
|
||||
if !noReceiveNodes[id] {
|
||||
noReceiveNodes[id] = true
|
||||
added = append(added, id)
|
||||
}
|
||||
}
|
||||
for id, disks := range ecb.rejectedDisks {
|
||||
if noReceiveNodes[id] {
|
||||
continue
|
||||
}
|
||||
if noReceiveDisks[id] == nil {
|
||||
noReceiveDisks[id] = make(map[uint32]bool)
|
||||
}
|
||||
for diskId := range disks {
|
||||
if !noReceiveDisks[id][diskId] {
|
||||
noReceiveDisks[id][diskId] = true
|
||||
added = append(added, fmt.Sprintf("%s:disk%d", id, diskId))
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Strings(added)
|
||||
return
|
||||
}
|
||||
|
||||
func (ecb *ecBalancer) hasOtherError() bool {
|
||||
ecb.mu.Lock()
|
||||
defer ecb.mu.Unlock()
|
||||
return ecb.sawOtherErr
|
||||
}
|
||||
|
||||
// recordMoveResult journals an apply-mode move outcome: a success is replayed
|
||||
// onto the next attempt's snapshot, a "no space left" marks the target
|
||||
// no-receive, and any other failure makes the run non-retryable.
|
||||
func (ecb *ecBalancer) recordMoveResult(m ecbalancer.Move, err error) {
|
||||
ecb.mu.Lock()
|
||||
defer ecb.mu.Unlock()
|
||||
switch {
|
||||
case err == nil:
|
||||
ecb.completedMoves = append(ecb.completedMoves, m)
|
||||
case m.TargetNode != "" && volume_move.IsNoSpaceLeftError(err):
|
||||
if m.TargetDisk > 0 {
|
||||
// The refusal convicts only the named disk; the node's other
|
||||
// disks may still have room.
|
||||
if ecb.rejectedDisks == nil {
|
||||
ecb.rejectedDisks = make(map[string]map[uint32]bool)
|
||||
}
|
||||
if ecb.rejectedDisks[m.TargetNode] == nil {
|
||||
ecb.rejectedDisks[m.TargetNode] = make(map[uint32]bool)
|
||||
}
|
||||
ecb.rejectedDisks[m.TargetNode][m.TargetDisk] = true
|
||||
} else {
|
||||
// The server auto-picked and found no usable disk: the whole
|
||||
// node is out.
|
||||
if ecb.rejectedTargets == nil {
|
||||
ecb.rejectedTargets = make(map[string]bool)
|
||||
}
|
||||
ecb.rejectedTargets[m.TargetNode] = true
|
||||
}
|
||||
default:
|
||||
ecb.sawOtherErr = true
|
||||
}
|
||||
return ecb.balance(collections)
|
||||
}
|
||||
|
||||
// defaultECRatio resolves a collection's EC data/parity counts, defaulting to
|
||||
@@ -852,7 +1102,7 @@ func defaultECRatio(_ string) (int, int) {
|
||||
// balance plans EC shard moves with the shared planner and executes them. When
|
||||
// collections is empty all collections present are balanced.
|
||||
func (ecb *ecBalancer) balance(collections []string) error {
|
||||
topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds)
|
||||
topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds, ecb.migratingVolumeIds, ecb.noReceiveTargets, ecb.noReceiveDisks)
|
||||
if len(ecb.volumeIds) > 0 {
|
||||
requested := make([]uint32, 0, len(ecb.volumeIds))
|
||||
for vid := range ecb.volumeIds {
|
||||
@@ -896,6 +1146,7 @@ func (ecb *ecBalancer) balance(collections []string) error {
|
||||
}
|
||||
fmt.Printf("planned %d ec shard move(s) and %d ec shard deletion(s)\n", len(moves)-deletions, deletions)
|
||||
}
|
||||
ecb.plannedMoves = len(moves)
|
||||
return ecb.executeMoves(moves)
|
||||
}
|
||||
|
||||
@@ -907,7 +1158,7 @@ func (ecb *ecBalancer) balance(collections []string) error {
|
||||
// (0,0 when unreported, e.g. always in OSS), which Plan prefers over the
|
||||
// collection ratio for mixed-ratio clusters, and the set of volume ids that made
|
||||
// it into the topology.
|
||||
func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) {
|
||||
func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool, migratingVolumeIds map[uint32]bool, noReceiveTargets map[string]bool, noReceiveDisks map[string]map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) {
|
||||
allowed := make(map[string]bool, len(collections))
|
||||
for _, c := range collections {
|
||||
allowed[c] = true
|
||||
@@ -924,11 +1175,17 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.
|
||||
for _, en := range ecNodes {
|
||||
rackKey := string(en.DC) + ":" + string(en.Rack)
|
||||
node := topo.AddNode(en.Info.Id, string(en.DC), rackKey, en.FreeEcSlot)
|
||||
if noReceiveTargets[en.Info.Id] {
|
||||
node.SetNoReceive()
|
||||
}
|
||||
// Group by physical machine (host) so shards spread across machines, not just
|
||||
// nodes; the id stays the node identity used for moves.
|
||||
node.SetHost(pb.NewServerAddressFromDataNode(en.Info).ToHost())
|
||||
for diskId, d := range en.Disks {
|
||||
node.AddDisk(diskId, d.DiskType, d.FreeEcSlots, d.EcShardCount)
|
||||
if noReceiveDisks[en.Info.Id][diskId] {
|
||||
node.SetDiskNoReceive(diskId)
|
||||
}
|
||||
}
|
||||
for diskTypeKey, diskInfo := range en.Info.DiskInfos {
|
||||
if diskInfo == nil {
|
||||
@@ -1069,14 +1326,16 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e
|
||||
// thing making this safe -- and the plan saying so is not evidence. A
|
||||
// topology entry can name a location holding nothing, and deleting on
|
||||
// that basis removes the last copy. Confirm the keep node has it.
|
||||
if err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId); err != nil {
|
||||
return err
|
||||
}
|
||||
err := verifyEcShardOnKeepNode(grpcDialOption, m.Collection, vid, m.KeepNode, shardId)
|
||||
addr := pb.NewServerAddressFromDataNode(src.Info)
|
||||
if err := UnmountEcShards(grpcDialOption, vid, addr, shardIds); err != nil {
|
||||
return err
|
||||
if err == nil {
|
||||
err = UnmountEcShards(grpcDialOption, vid, addr, shardIds)
|
||||
}
|
||||
return SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds)
|
||||
if err == nil {
|
||||
err = SourceServerDeleteEcShards(grpcDialOption, m.Collection, vid, addr, shardIds)
|
||||
}
|
||||
ecb.recordMoveResult(m, err)
|
||||
return err
|
||||
}
|
||||
|
||||
dst := byID[m.TargetNode]
|
||||
@@ -1092,7 +1351,9 @@ func (ecb *ecBalancer) executeMove(byID map[string]*EcNode, m ecbalancer.Move) e
|
||||
// Dry-run: update the in-memory model only.
|
||||
return MoveMountedShardToEcNode(ecb.env, src, m.Collection, vid, shardId, dst, m.TargetDisk, false, ecb.diskType)
|
||||
}
|
||||
return ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk)
|
||||
err := ecb.applyShardMoveRPC(src, dst, m.Collection, vid, shardId, m.TargetDisk)
|
||||
ecb.recordMoveResult(m, err)
|
||||
return err
|
||||
}
|
||||
|
||||
// applyShardMoveRPC copies a shard to the destination disk, verifies the
|
||||
|
||||
@@ -390,6 +390,14 @@ func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpat
|
||||
return doDeleteFolderChildren()
|
||||
}
|
||||
|
||||
var likeEscaper = strings.NewReplacer("!", "!!", "%", "!%", "_", "!_")
|
||||
|
||||
// LikePattern escapes LIKE wildcards so the prefix matches literally;
|
||||
// '!' pairs with the ESCAPE '!' clause emitted by the SqlGenerators.
|
||||
func LikePattern(prefix string) string {
|
||||
return likeEscaper.Replace(prefix) + "%"
|
||||
}
|
||||
|
||||
func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
|
||||
|
||||
db, bucket, shortPath, err := store.getTxOrDB(ctx, dirPath, true)
|
||||
@@ -402,7 +410,7 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
|
||||
sqlText = store.GetSqlListInclusive(bucket)
|
||||
}
|
||||
|
||||
rows, err := db.QueryContext(ctx, sqlText, util.HashStringToLong(string(shortPath)), startFileName, string(shortPath), prefix+"%", limit+1)
|
||||
rows, err := db.QueryContext(ctx, sqlText, util.HashStringToLong(string(shortPath)), startFileName, string(shortPath), LikePattern(prefix), limit+1)
|
||||
if err != nil {
|
||||
return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
|
||||
}
|
||||
|
||||
@@ -52,6 +52,24 @@ func TestGetTxOrDBRealBucket(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLikePattern(t *testing.T) {
|
||||
cases := []struct {
|
||||
prefix, want string
|
||||
}{
|
||||
{"", "%"},
|
||||
{"plain", "plain%"},
|
||||
{"call_03_KNl", "call!_03!_KNl%"},
|
||||
{"100%", "100!%%"},
|
||||
{"a!b", "a!!b%"},
|
||||
{`a\b`, `a\b%`},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := LikePattern(c.prefix); got != c.want {
|
||||
t.Errorf("LikePattern(%q) = %q, want %q", c.prefix, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSplitPoolForKv(t *testing.T) {
|
||||
cases := []struct {
|
||||
maxOpen, mainOpen, kvOpen int
|
||||
|
||||
+1
-1
@@ -890,7 +890,7 @@ func (f *Filer) DeleteEmptyParentDirectories(ctx context.Context, dirPath util.F
|
||||
glog.V(2).InfofCtx(ctx, "DeleteEmptyParentDirectories: deleting empty directory %s", dirPath)
|
||||
parentDir, _ := dirPath.DirAndName()
|
||||
if dirEntry, findErr := f.FindEntry(ctx, dirPath); findErr == nil {
|
||||
if delErr := f.doDeleteEntryMetaAndData(ctx, dirEntry, false, false, nil); delErr == nil {
|
||||
if delErr := f.doDeleteEntryMetaAndData(f.emptyFolderDeleteContext(ctx, dirPath), dirEntry, false, false, nil); delErr == nil {
|
||||
// Successfully deleted, continue checking upwards
|
||||
f.DeleteEmptyParentDirectories(ctx, util.FullPath(parentDir), stopAtPath)
|
||||
} else {
|
||||
|
||||
@@ -80,6 +80,27 @@ func keepsRemoteObject(ctx context.Context) bool {
|
||||
return ctx.Value(keepRemoteObjectKey{}) != nil
|
||||
}
|
||||
|
||||
// emptyFolderDeleteContext makes deleting a folder that is empty in the filer
|
||||
// metadata-only when it is under a remote mount. A mount is populated lazily,
|
||||
// so the folder can still hold remote objects never read, and removing its
|
||||
// remote directory deletes every object under that prefix.
|
||||
func (f *Filer) emptyFolderDeleteContext(ctx context.Context, dirPath util.FullPath) context.Context {
|
||||
if f.IsUnderRemoteMount(dirPath) {
|
||||
return WithKeepRemoteObject(ctx)
|
||||
}
|
||||
return ctx
|
||||
}
|
||||
|
||||
// IsUnderRemoteMount reports whether p is a remote mount directory or lies under one.
|
||||
func (f *Filer) IsUnderRemoteMount(p util.FullPath) bool {
|
||||
if f.RemoteStorage == nil {
|
||||
return false
|
||||
}
|
||||
// Mount rules are keyed with a trailing slash, so this also matches the mount directory itself.
|
||||
_, remoteLoc := f.RemoteStorage.FindMountDirectory(util.FullPath(string(p) + "/"))
|
||||
return remoteLoc != nil
|
||||
}
|
||||
|
||||
func deleteEventEntry(ctx context.Context, entry *Entry) *Entry {
|
||||
if !keepsRemoteObject(ctx) {
|
||||
return entry
|
||||
@@ -108,6 +129,9 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
|
||||
if ifNotModifiedAfter > 0 && entry.Attr.Mtime.Unix() > ifNotModifiedAfter {
|
||||
return nil
|
||||
}
|
||||
if !isRecursive && entry.IsDirectory() {
|
||||
ctx = f.emptyFolderDeleteContext(ctx, p)
|
||||
}
|
||||
isDeleteCollection := f.IsBucket(entry)
|
||||
collectionName := ""
|
||||
if isDeleteCollection {
|
||||
@@ -170,7 +194,7 @@ func (f *Filer) DeleteEntryMetaAndData(ctx context.Context, p util.FullPath, isR
|
||||
func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry, isRecursive, ignoreRecursiveError, shouldDeleteChunks, isDeletingBucket, isFromOtherCluster bool, signatures []int32, onHardLinkIdsFn OnHardLinkIdsFunc) (err error) {
|
||||
|
||||
var dirTombstoneTs int64
|
||||
if isRecursive {
|
||||
if isRecursive && !keepsRemoteObject(ctx) {
|
||||
// Tombstone the directory before its children: when the store drops
|
||||
// the subtree without listing it, or a child error aborts the walk,
|
||||
// the ancestor tombstone still covers every descendant.
|
||||
@@ -201,7 +225,9 @@ func (f *Filer) doBatchDeleteFolderMetaAndData(ctx context.Context, entry *Entry
|
||||
|
||||
for _, sub := range entries {
|
||||
lastFileName = sub.Name()
|
||||
f.noteRemoteDeletion(sub.FullPath, sub.IsDirectory(), time.Now().UnixNano())
|
||||
if !keepsRemoteObject(ctx) {
|
||||
f.noteRemoteDeletion(sub.FullPath, sub.IsDirectory(), time.Now().UnixNano())
|
||||
}
|
||||
if sub.IsDirectory() {
|
||||
subIsDeletingBucket := f.IsBucket(sub)
|
||||
err = f.doBatchDeleteFolderMetaAndData(ctx, sub, isRecursive, ignoreRecursiveError, shouldDeleteChunks, subIsDeletingBucket, isFromOtherCluster, nil, onHardLinkIdsFn)
|
||||
@@ -264,7 +290,10 @@ func (f *Filer) doDeleteEntryMetaAndData(ctx context.Context, entry *Entry, shou
|
||||
}
|
||||
}
|
||||
|
||||
f.noteRemoteDeletion(entry.FullPath, entry.IsDirectory(), time.Now().UnixNano())
|
||||
// A metadata-only delete owes the remote nothing, so lazy reads stay open.
|
||||
if !keepsRemoteObject(ctx) {
|
||||
f.noteRemoteDeletion(entry.FullPath, entry.IsDirectory(), time.Now().UnixNano())
|
||||
}
|
||||
|
||||
if storeDeletionErr := f.Store.DeleteOneEntry(ctx, entry); storeDeletionErr != nil {
|
||||
return fmt.Errorf("filer store delete: %w", storeDeletionErr)
|
||||
|
||||
@@ -18,11 +18,15 @@ import (
|
||||
|
||||
type lazyFetchContextKey struct{}
|
||||
|
||||
// lazyParentsContextKey marks a FindEntry issued while materializing missing
|
||||
// parents: that fetch's own parent walk would only re-stat the same gaps.
|
||||
type lazyParentsContextKey struct{}
|
||||
|
||||
// maybeLazyFetchFromRemote is called by FindEntry when the store returns no
|
||||
// entry for p. If p is under a remote-storage mount, it stats the remote
|
||||
// object, builds a filer Entry from the result, and persists it via
|
||||
// CreateEntry with SkipCheckParentDirectory so phantom parent directories
|
||||
// under the mount are not required.
|
||||
// CreateEntry together with its missing parent directories: remote.uncache
|
||||
// and cache eviction find entries by walking down from the mount directory.
|
||||
//
|
||||
// On a CreateEntry failure after a successful StatFile the in-memory entry is
|
||||
// still returned (availability over consistency); the singleflight key is
|
||||
@@ -127,7 +131,10 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
|
||||
glog.V(2).InfofCtx(ctx, "maybeLazyFetchFromRemote: %s deleted during remote stat", p)
|
||||
return lazyFetchResult{nil}, nil
|
||||
}
|
||||
saveErr := f.CreateEntry(persistCtx, entry, nil, false, false, nil, true, f.MaxFilenameLength)
|
||||
if ctx.Value(lazyParentsContextKey{}) == nil {
|
||||
f.fetchMissingRemoteParents(persistBaseCtx, mountDir, p)
|
||||
}
|
||||
saveErr := f.CreateEntry(persistCtx, entry, nil, false, false, nil, false, f.MaxFilenameLength)
|
||||
if saveErr != nil {
|
||||
glog.Warningf("maybeLazyFetchFromRemote: failed to persist filer entry for %s: %v", p, saveErr)
|
||||
f.lazyFetchGroup.Forget(key)
|
||||
@@ -157,6 +164,25 @@ func (f *Filer) maybeLazyFetchFromRemote(ctx context.Context, p util.FullPath) (
|
||||
return result.entry, nil
|
||||
}
|
||||
|
||||
// fetchMissingRemoteParents looks up p's missing parents under mountDir, top
|
||||
// down, through lazy fetch. A remote object stored at a parent's key is then
|
||||
// materialized first, so creating p promotes it to a prefix object instead of
|
||||
// hiding it behind a plain directory. Each missing parent costs one remote
|
||||
// stat, once per directory.
|
||||
func (f *Filer) fetchMissingRemoteParents(ctx context.Context, mountDir, p util.FullPath) {
|
||||
var missing []util.FullPath
|
||||
for dir, _ := p.DirAndName(); len(dir) > len(mountDir); dir, _ = util.FullPath(dir).DirAndName() {
|
||||
if _, err := f.Store.FindEntry(ctx, util.FullPath(dir)); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
break
|
||||
}
|
||||
missing = append(missing, util.FullPath(dir))
|
||||
}
|
||||
parentCtx := context.WithValue(ctx, lazyParentsContextKey{}, true)
|
||||
for i := len(missing) - 1; i >= 0; i-- {
|
||||
f.FindEntry(parentCtx, missing[i])
|
||||
}
|
||||
}
|
||||
|
||||
// retractLazyRemoteEntry deletes the entry at entry.FullPath only when it is
|
||||
// still the entry a lazy remote read just materialized — a concurrent write
|
||||
// may have replaced it, and deleting by path alone would take that write down.
|
||||
|
||||
@@ -103,6 +103,10 @@ func (f *Filer) maybeLazyListFromRemote(ctx context.Context, p util.FullPath) {
|
||||
persistCtx := context.WithValue(context.WithoutCancel(ctx), lazyListContextKey{}, true)
|
||||
persistCtx = context.WithValue(persistCtx, lazyFetchContextKey{}, true)
|
||||
|
||||
// Entries listed under p must stay reachable by a walk from the mount
|
||||
// directory, the same reason the lazy fetch fills in missing parents.
|
||||
f.fetchMissingRemoteParents(context.WithoutCancel(ctx), mountDir, p)
|
||||
|
||||
listErr := client.ListDirectory(persistCtx, objectLoc, func(dir string, name string, isDirectory bool, remoteEntry *filer_pb.RemoteEntry) error {
|
||||
childPath := p.Child(name)
|
||||
|
||||
@@ -215,7 +219,7 @@ func (f *Filer) updateDirectoryListingSyncedAt(ctx context.Context, p util.FullP
|
||||
dirEntry.Extended = make(map[string][]byte)
|
||||
}
|
||||
dirEntry.Extended[xattrRemoteListingSyncedAt] = []byte(fmt.Sprintf("%d", syncTime.Unix()))
|
||||
if saveErr := f.CreateEntry(ctx, dirEntry, nil, false, false, nil, true, f.MaxFilenameLength); saveErr != nil {
|
||||
if saveErr := f.CreateEntry(ctx, dirEntry, nil, false, false, nil, false, f.MaxFilenameLength); saveErr != nil {
|
||||
glog.Warningf("maybeLazyListFromRemote: create dir synced_at for %s: %v", p, saveErr)
|
||||
}
|
||||
return
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/remote_storage"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
@@ -35,6 +36,7 @@ type stubFilerStore struct {
|
||||
kv map[string][]byte
|
||||
insertErr error
|
||||
findErr error
|
||||
findErrByPath map[string]error
|
||||
kvPutErr error
|
||||
kvGetErr error
|
||||
kvDeleteErr error
|
||||
@@ -200,6 +202,9 @@ func (s *stubFilerStore) FindEntry(_ context.Context, p util.FullPath) (*Entry,
|
||||
if s.findErr != nil {
|
||||
return nil, s.findErr
|
||||
}
|
||||
if err := s.findErrByPath[string(p)]; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e, ok := s.entries[string(p)]; ok {
|
||||
return e, nil
|
||||
}
|
||||
@@ -340,6 +345,11 @@ func putConfigEntry(store *stubFilerStore, path string, content []byte) {
|
||||
}
|
||||
}
|
||||
|
||||
// putDirEntry stores a plain directory entry, so a lazy fetch below it finds its parent in place.
|
||||
func putDirEntry(store *stubFilerStore, path string) {
|
||||
store.entries[path] = &Entry{FullPath: util.FullPath(path), Attr: Attr{Mode: os.ModeDir | 0755}}
|
||||
}
|
||||
|
||||
// --- tests ---
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
|
||||
@@ -375,6 +385,199 @@ func TestMaybeLazyFetchFromRemote_HitsRemoteAndPersists(t *testing.T) {
|
||||
assert.Equal(t, []byte("zstd"), stored.Extended["Content-Encoding"])
|
||||
}
|
||||
|
||||
// pathStatRemoteClient stats only the objects it holds, keyed by remote path.
|
||||
type pathStatRemoteClient struct {
|
||||
stubRemoteClient
|
||||
objects map[string]*filer_pb.RemoteEntry
|
||||
statCalls []string
|
||||
}
|
||||
|
||||
func (c *pathStatRemoteClient) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
|
||||
c.statCalls = append(c.statCalls, loc.Path)
|
||||
if remoteEntry, found := c.objects[loc.Path]; found {
|
||||
return remoteEntry, nil
|
||||
}
|
||||
return nil, remote_storage.ErrRemoteObjectNotFound
|
||||
}
|
||||
|
||||
func newPathStatFiler(t *testing.T, storageType string, client remote_storage.RemoteStorageClient) (*Filer, *stubFilerStore) {
|
||||
t.Helper()
|
||||
t.Cleanup(registerStubMaker(t, storageType, client))
|
||||
conf := &remote_pb.RemoteConf{Name: "mystore", Type: storageType}
|
||||
rs := NewFilerRemoteStorage()
|
||||
rs.storageNameToConf[conf.Name] = conf
|
||||
rs.mapDirectoryToRemoteStorage("/buckets/mybucket", &remote_pb.RemoteStorageLocation{
|
||||
Name: "mystore",
|
||||
Bucket: "mybucket",
|
||||
Path: "/",
|
||||
})
|
||||
store := newStubFilerStore()
|
||||
return newTestFiler(t, store, rs), store
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_PersistsParentDirectories(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
|
||||
"/a/b/file.bin": {RemoteMtime: 1700000000, RemoteSize: 1234},
|
||||
"/a/b/other.bin": {RemoteMtime: 1700000000, RemoteSize: 5678},
|
||||
}}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_parents", stub)
|
||||
|
||||
const filePath = "/buckets/mybucket/a/b/file.bin"
|
||||
entry, err := f.maybeLazyFetchFromRemote(context.Background(), filePath)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, entry)
|
||||
assert.Equal(t, []string{"/a/b/file.bin", "/a", "/a/b"}, stub.statCalls, "each missing parent is stat'ed once")
|
||||
|
||||
for _, dir := range []string{"/buckets/mybucket/a", "/buckets/mybucket/a/b"} {
|
||||
dirEntry := store.getEntry(dir)
|
||||
require.NotNil(t, dirEntry, "parent %s not persisted", dir)
|
||||
assert.True(t, dirEntry.IsDirectory(), "%s should be a directory", dir)
|
||||
assert.Nil(t, dirEntry.Remote, "%s holds no remote object", dir)
|
||||
}
|
||||
|
||||
// A sibling finds its parents in place and stats only itself.
|
||||
stub.statCalls = nil
|
||||
_, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/other.bin")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"/a/b/other.bin"}, stub.statCalls)
|
||||
|
||||
// Once the content is cached, eviction walks down from the mount and must reach it.
|
||||
stored := store.getEntry(filePath)
|
||||
require.NotNil(t, stored)
|
||||
stored.Chunks = []*filer_pb.FileChunk{{FileId: "1,01637037d6", Size: 1234}}
|
||||
stored.Remote.LastLocalSyncTsNs = time.Now().UnixNano()
|
||||
require.NoError(t, store.UpdateEntry(context.Background(), stored))
|
||||
|
||||
evictable := f.ListEvictableRemoteEntries(context.Background(), []util.FullPath{"/buckets/mybucket"}, 0)
|
||||
require.Len(t, evictable, 1)
|
||||
assert.Equal(t, util.FullPath(filePath), evictable[0].FullPath)
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_ParentLookupStopsAtStoreError(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
|
||||
"/a/b/file.bin": {RemoteMtime: 1700000000, RemoteSize: 1234},
|
||||
}}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_parent_store_error", stub)
|
||||
store.findErrByPath = map[string]error{"/buckets/mybucket/a/b": errors.New("store unavailable")}
|
||||
|
||||
_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/file.bin")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"/a/b/file.bin"}, stub.statCalls, "a parent the store cannot read is not treated as missing")
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_PromotesRemoteObjectAtParentKey(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
|
||||
"/reports": {RemoteMtime: 1700000000, RemoteSize: 10},
|
||||
"/reports/2026.csv": {RemoteMtime: 1700000000, RemoteSize: 20},
|
||||
}}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_prefix_object", stub)
|
||||
|
||||
_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/reports/2026.csv")
|
||||
require.NoError(t, err)
|
||||
|
||||
parent := store.getEntry("/buckets/mybucket/reports")
|
||||
require.NotNil(t, parent)
|
||||
assert.True(t, parent.IsDirectory())
|
||||
assert.Equal(t, []byte("true"), parent.Extended[s3_constants.SeaweedFSPrefixObject], "the remote object at the parent key must stay readable")
|
||||
require.NotNil(t, parent.Remote)
|
||||
assert.Equal(t, int64(10), parent.Remote.RemoteSize)
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_StatsEachParentOnce(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
|
||||
"/a/b/c/file.bin": {RemoteMtime: 1700000000, RemoteSize: 1234},
|
||||
}}
|
||||
f, _ := newPathStatFiler(t, "stub_lazy_parent_stats", stub)
|
||||
|
||||
_, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/a/b/c/file.bin")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, []string{"/a/b/c/file.bin", "/a", "/a/b", "/a/b/c"}, stub.statCalls,
|
||||
"a parent still missing after its own stat must not be re-stat'ed by the next lookup")
|
||||
}
|
||||
|
||||
func TestMaybeLazyListFromRemote_PersistsParentDirectories(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{}}
|
||||
stub.listDirFn = func(loc *remote_pb.RemoteStorageLocation, visitFn remote_storage.VisitFunc) error {
|
||||
return visitFn("/", "file.bin", false, &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 1234})
|
||||
}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_list_parents", stub)
|
||||
f.RemoteStorage.mapDirectoryToRemoteStorage("/buckets/listmount", &remote_pb.RemoteStorageLocation{
|
||||
Name: "mystore",
|
||||
Bucket: "listmount",
|
||||
Path: "/",
|
||||
ListingCacheTtlSeconds: 300,
|
||||
})
|
||||
|
||||
f.maybeLazyListFromRemote(context.Background(), util.FullPath("/buckets/listmount/a/b"))
|
||||
|
||||
assert.Equal(t, []string{"/a"}, stub.statCalls, "each missing parent is stat'ed once")
|
||||
for _, dir := range []string{"/buckets/listmount", "/buckets/listmount/a", "/buckets/listmount/a/b"} {
|
||||
dirEntry := store.getEntry(dir)
|
||||
require.NotNil(t, dirEntry, "%s not persisted", dir)
|
||||
assert.True(t, dirEntry.IsDirectory(), "%s should be a directory", dir)
|
||||
}
|
||||
|
||||
child := store.getEntry("/buckets/listmount/a/b/file.bin")
|
||||
require.NotNil(t, child)
|
||||
child.Chunks = []*filer_pb.FileChunk{{FileId: "1,01637037d6", Size: 1234}}
|
||||
child.Remote.LastLocalSyncTsNs = time.Now().UnixNano()
|
||||
require.NoError(t, store.UpdateEntry(context.Background(), child))
|
||||
|
||||
evictable := f.ListEvictableRemoteEntries(context.Background(), []util.FullPath{"/buckets/listmount"}, 0)
|
||||
require.Len(t, evictable, 1)
|
||||
assert.Equal(t, util.FullPath("/buckets/listmount/a/b/file.bin"), evictable[0].FullPath)
|
||||
}
|
||||
|
||||
func TestDeleteEntryMetaAndData_EmptyFolderUnderMountKeepsRemotePrefix(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{objects: map[string]*filer_pb.RemoteEntry{
|
||||
"/dir/unread.bin": {RemoteMtime: 1700000000, RemoteSize: 1},
|
||||
}}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_empty_folder", stub)
|
||||
dirPath := util.FullPath("/buckets/mybucket/dir")
|
||||
putDirEntry(store, string(dirPath))
|
||||
|
||||
require.NoError(t, f.DeleteEntryMetaAndData(context.Background(), dirPath, false, false, false, false, nil, 0))
|
||||
|
||||
assert.Empty(t, stub.removeCalls, "an empty local folder must not remove its remote prefix")
|
||||
assert.Nil(t, store.getEntry(string(dirPath)))
|
||||
entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/unread.bin")
|
||||
require.NoError(t, err)
|
||||
assert.NotNil(t, entry, "a metadata-only delete must not block lazy reads under the folder")
|
||||
}
|
||||
|
||||
func TestDeleteEmptyParentDirectories_UnderMountKeepsRemotePrefix(t *testing.T) {
|
||||
stub := &pathStatRemoteClient{}
|
||||
f, store := newPathStatFiler(t, "stub_lazy_expired_parents", stub)
|
||||
f.DirBucketsPath = "/buckets"
|
||||
putDirEntry(store, "/buckets/mybucket/a")
|
||||
putDirEntry(store, "/buckets/mybucket/a/b")
|
||||
|
||||
f.DeleteEmptyParentDirectories(context.Background(), "/buckets/mybucket/a/b", "/buckets")
|
||||
|
||||
assert.Empty(t, stub.removeCalls, "an empty local folder must not remove its remote prefix")
|
||||
assert.Nil(t, store.getEntry("/buckets/mybucket/a/b"))
|
||||
assert.Nil(t, store.getEntry("/buckets/mybucket/a"))
|
||||
}
|
||||
|
||||
func TestIsUnderRemoteMount(t *testing.T) {
|
||||
rs := NewFilerRemoteStorage()
|
||||
rs.mapDirectoryToRemoteStorage("/buckets/mirror", &remote_pb.RemoteStorageLocation{
|
||||
Name: "mystore",
|
||||
Bucket: "mirror",
|
||||
Path: "/",
|
||||
})
|
||||
f := newTestFiler(t, newStubFilerStore(), rs)
|
||||
|
||||
for path, want := range map[util.FullPath]bool{
|
||||
"/buckets/mirror": true,
|
||||
"/buckets/mirror/a/b": true,
|
||||
"/buckets/mirror2/a": false,
|
||||
"/buckets/local/a": false,
|
||||
} {
|
||||
assert.Equal(t, want, f.IsUnderRemoteMount(path), path)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaybeLazyFetchFromRemote_NotUnderMount(t *testing.T) {
|
||||
rs := NewFilerRemoteStorage()
|
||||
store := newStubFilerStore()
|
||||
|
||||
@@ -407,7 +407,7 @@ func (f *Filer) onRemoteDeletionEvents(event *filer_pb.SubscribeMetadataResponse
|
||||
if message == nil {
|
||||
return
|
||||
}
|
||||
if message.OldEntry != nil {
|
||||
if message.OldEntry != nil && !IsMetadataOnlyDelete(message.OldEntry) {
|
||||
sourcePath := filer_pb.MetadataEventSourceFullPath(event)
|
||||
if message.NewEntry == nil || sourcePath != filer_pb.MetadataEventTargetFullPath(event) {
|
||||
if f.RemoteStorage != nil && f.remoteTombstones != nil {
|
||||
|
||||
@@ -187,6 +187,7 @@ func TestMaybeLazyFetchFromRemote_SyncOffsetReleasesTombstone(t *testing.T) {
|
||||
},
|
||||
}
|
||||
f, store := newMountedTestFiler(t, storageType, stub, 0)
|
||||
putDirEntry(store, "/buckets/mybucket/dir")
|
||||
|
||||
filePath := util.FullPath("/buckets/mybucket/dir/a.txt")
|
||||
deleteTsNs := time.Now().UnixNano()
|
||||
@@ -222,6 +223,7 @@ func TestDeleteEntryMetaAndData_TombstonesPath(t *testing.T) {
|
||||
},
|
||||
}
|
||||
f, store := newMountedTestFiler(t, storageType, stub, 0)
|
||||
putDirEntry(store, "/buckets/mybucket/dir")
|
||||
|
||||
filePath := util.FullPath("/buckets/mybucket/dir/a.txt")
|
||||
store.entries[string(filePath)] = &Entry{
|
||||
@@ -257,6 +259,35 @@ func TestDeleteEntryMetaAndData_TombstonesPath(t *testing.T) {
|
||||
assert.Equal(t, 1, stub.statCalls)
|
||||
}
|
||||
|
||||
func TestOnMetadataChangeEvent_MetadataOnlyDeleteSkipsTombstone(t *testing.T) {
|
||||
const storageType = "stub_tomb_metadata_only"
|
||||
stub := &countingRemoteClient{
|
||||
stubRemoteClient: stubRemoteClient{
|
||||
statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11},
|
||||
},
|
||||
}
|
||||
f, store := newMountedTestFiler(t, storageType, stub, 0)
|
||||
putDirEntry(store, "/buckets/mybucket/dir")
|
||||
|
||||
f.onMetadataChangeEvent(&filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/buckets/mybucket",
|
||||
TsNs: time.Now().UnixNano(),
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
OldEntry: &filer_pb.Entry{
|
||||
Name: "dir",
|
||||
IsDirectory: true,
|
||||
Extended: map[string][]byte{ExtKeepRemoteObjectKey: []byte("true")},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
// nothing is owed to the remote, so objects under the folder stay readable
|
||||
entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/a.txt")
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, entry)
|
||||
assert.Equal(t, 1, stub.statCalls)
|
||||
}
|
||||
|
||||
func TestOnMetadataChangeEvent_PeerDeleteTombstones(t *testing.T) {
|
||||
const storageType = "stub_tomb_peer"
|
||||
stub := &countingRemoteClient{
|
||||
|
||||
@@ -64,12 +64,12 @@ func (gen *SqlGenMysql) nameExpr() string {
|
||||
|
||||
func (gen *SqlGenMysql) GetSqlListExclusive(tableName string) string {
|
||||
name := gen.nameExpr()
|
||||
return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s > ? AND `directory` = ? AND %s LIKE ? ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
|
||||
return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s > ? AND `directory` = ? AND %s LIKE ? ESCAPE '!' ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
|
||||
}
|
||||
|
||||
func (gen *SqlGenMysql) GetSqlListInclusive(tableName string) string {
|
||||
name := gen.nameExpr()
|
||||
return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s >= ? AND `directory` = ? AND %s LIKE ? ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
|
||||
return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s >= ? AND `directory` = ? AND %s LIKE ? ESCAPE '!' ORDER BY %s ASC LIMIT ?", tableName, name, name, name)
|
||||
}
|
||||
|
||||
func (gen *SqlGenMysql) GetSqlCreateTable(tableName string) string {
|
||||
|
||||
@@ -65,6 +65,15 @@ func TestListSqlBinaryOrderingOnNonBinaryColumn(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestListSqlEscapesLikeWildcards(t *testing.T) {
|
||||
gen := &SqlGenMysql{}
|
||||
for _, got := range []string{gen.GetSqlListExclusive("filemeta"), gen.GetSqlListInclusive("filemeta")} {
|
||||
if !strings.Contains(got, "LIKE ? ESCAPE '!'") {
|
||||
t.Fatalf("list query must declare the escape char matching LikePattern, got: %s", got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsBinaryCollation(t *testing.T) {
|
||||
binary := []string{"", "binary", "utf8mb4_bin", "utf8mb3_bin", "latin1_bin", "UTF8MB4_BIN"}
|
||||
for _, c := range binary {
|
||||
|
||||
@@ -77,12 +77,12 @@ func (gen *SqlGenPostgres) nameExpr() string {
|
||||
|
||||
func (gen *SqlGenPostgres) GetSqlListExclusive(tableName string) string {
|
||||
name := gen.nameExpr()
|
||||
return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>$2 AND directory=$3 AND %s like $4 ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
|
||||
return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>$2 AND directory=$3 AND %s like $4 ESCAPE '!' ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
|
||||
}
|
||||
|
||||
func (gen *SqlGenPostgres) GetSqlListInclusive(tableName string) string {
|
||||
name := gen.nameExpr()
|
||||
return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>=$2 AND directory=$3 AND %s like $4 ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
|
||||
return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>=$2 AND directory=$3 AND %s like $4 ESCAPE '!' ORDER BY %s ASC LIMIT $5`, tableName, name, name, name)
|
||||
}
|
||||
|
||||
func (gen *SqlGenPostgres) GetSqlCreateTable(tableName string) string {
|
||||
|
||||
@@ -62,6 +62,15 @@ func TestListSqlBinaryOrderingOnNonBinaryColumn(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestListSqlEscapesLikeWildcards(t *testing.T) {
|
||||
gen := &SqlGenPostgres{}
|
||||
for _, got := range []string{gen.GetSqlListExclusive("filemeta"), gen.GetSqlListInclusive("filemeta")} {
|
||||
if !strings.Contains(got, "like $4 ESCAPE '!'") {
|
||||
t.Fatalf("list query must declare the escape char matching LikePattern, got: %s", got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsByteOrderedCollation(t *testing.T) {
|
||||
byteOrdered := []string{"C", "POSIX", "c", "C.UTF-8", "C.utf8"}
|
||||
for _, c := range byteOrdered {
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
//go:build sqlite
|
||||
|
||||
package sqlite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer/mysql"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
func TestListDirectoryPrefixedEntriesEscapesLikeWildcards(t *testing.T) {
|
||||
db, err := sql.Open("sqlite", "file::memory:?cache=shared")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
_, err = db.Exec(`CREATE TABLE filemeta (dirhash BIGINT, name VARCHAR(1000), directory TEXT, meta BLOB, PRIMARY KEY (dirhash, name))`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
store := &SqliteStore{}
|
||||
store.SqlGenerator = &mysql.SqlGenMysql{}
|
||||
store.DB = db
|
||||
|
||||
ctx := context.Background()
|
||||
names := []string{
|
||||
"call_03_KNlHPt6grkZMe4WZVwQi9710", // target: literal _ prefix
|
||||
"callX03XKNlZZZ", // would match if _ were a wildcard
|
||||
"call_99_other",
|
||||
"100%_done", // % and _ in the name
|
||||
"100aXdone", // would match prefix "100%_" if unescaped
|
||||
"bang!file",
|
||||
}
|
||||
for _, n := range names {
|
||||
entry := &filer.Entry{FullPath: util.FullPath("/dir/" + n)}
|
||||
if err := store.InsertEntry(ctx, entry); err != nil {
|
||||
t.Fatalf("insert %s: %v", n, err)
|
||||
}
|
||||
}
|
||||
|
||||
list := func(prefix string) []string {
|
||||
var got []string
|
||||
_, err := store.ListDirectoryPrefixedEntries(ctx, "/dir", "", false, 100, prefix, func(e *filer.Entry) (bool, error) {
|
||||
got = append(got, e.Name())
|
||||
return true, nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("list %q: %v", prefix, err)
|
||||
}
|
||||
return got
|
||||
}
|
||||
|
||||
if got := list("call_03_"); len(got) != 1 || got[0] != "call_03_KNlHPt6grkZMe4WZVwQi9710" {
|
||||
t.Fatalf("prefix with _: got %v", got)
|
||||
}
|
||||
if got := list("call_"); len(got) != 2 {
|
||||
t.Fatalf("prefix call_: got %v", got)
|
||||
}
|
||||
if got := list("100%_"); len(got) != 1 || got[0] != "100%_done" {
|
||||
t.Fatalf("prefix with %%: got %v", got)
|
||||
}
|
||||
if got := list("bang!"); len(got) != 1 || got[0] != "bang!file" {
|
||||
t.Fatalf("prefix with !: got %v", got)
|
||||
}
|
||||
if got := list(""); len(got) != len(names) {
|
||||
t.Fatalf("empty prefix: got %v", got)
|
||||
}
|
||||
}
|
||||
@@ -55,7 +55,7 @@ const (
|
||||
|
||||
SELECT name, meta
|
||||
FROM ` + asql.DEFAULT_TABLE + `
|
||||
WHERE dir_hash = $dir_hash AND directory = $directory and name > $start_name and name LIKE $prefix
|
||||
WHERE dir_hash = $dir_hash AND directory = $directory and name > $start_name and name LIKE $prefix ESCAPE '!'
|
||||
ORDER BY name ASC LIMIT $limit;`
|
||||
|
||||
listInclusiveDirectoryQuery = `
|
||||
@@ -68,6 +68,6 @@ const (
|
||||
|
||||
SELECT name, meta
|
||||
FROM ` + asql.DEFAULT_TABLE + `
|
||||
WHERE dir_hash = $dir_hash AND directory = $directory and name >= $start_name and name LIKE $prefix
|
||||
WHERE dir_hash = $dir_hash AND directory = $directory and name >= $start_name and name LIKE $prefix ESCAPE '!'
|
||||
ORDER BY name ASC LIMIT $limit;`
|
||||
)
|
||||
|
||||
@@ -288,7 +288,7 @@ func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath
|
||||
table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(*shortDir))),
|
||||
table.ValueParam("$directory", types.UTF8Value(*shortDir)),
|
||||
table.ValueParam("$start_name", types.UTF8Value(startFileName)),
|
||||
table.ValueParam("$prefix", types.UTF8Value(prefix+"%")),
|
||||
table.ValueParam("$prefix", types.UTF8Value(abstract_sql.LikePattern(prefix))),
|
||||
table.ValueParam("$limit", types.Uint64Value(uint64(chunkLimit))),
|
||||
)
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
@@ -202,6 +203,14 @@ func (m *Mover) DeleteEcShards(ctx context.Context, volumeId needle.VolumeId, co
|
||||
})
|
||||
}
|
||||
|
||||
// IsNoSpaceLeftError matches the volume server's refusal when no local disk
|
||||
// can take the copied shard ("no space left" from FindEcShardTargetLocation,
|
||||
// an ENOSPC writing to a caller-pinned disk, and the Rust server's same
|
||||
// reply), however the move chain wraps it.
|
||||
func IsNoSpaceLeftError(err error) bool {
|
||||
return err != nil && strings.Contains(err.Error(), "no space left")
|
||||
}
|
||||
|
||||
// RemoveEcShards unmounts then deletes shards in place — the dedup path for a
|
||||
// shard that already has a copy elsewhere.
|
||||
func (m *Mover) RemoveEcShards(ctx context.Context, volumeId needle.VolumeId, collection string, server pb.ServerAddress, shardIds []erasure_coding.ShardId) error {
|
||||
|
||||
@@ -58,7 +58,10 @@ func (fs *FilerServer) autoChunk(ctx context.Context, w http.ResponseWriter, r *
|
||||
writeJsonError(w, r, util.HttpStatusCancelled, err)
|
||||
case strings.HasSuffix(errStr, "is a file") || strings.HasSuffix(errStr, "already exists"):
|
||||
writeJsonError(w, r, http.StatusConflict, err)
|
||||
case errStr == constants.ErrMsgBadDigest:
|
||||
case errStr == constants.ErrMsgBadDigest,
|
||||
errStr == constants.ErrMsgAppendSaveInsideChunked,
|
||||
errStr == constants.ErrMsgAppendInlineMode,
|
||||
errStr == constants.ErrMsgOffsetWriteInline:
|
||||
writeJsonError(w, r, http.StatusBadRequest, err)
|
||||
default:
|
||||
writeJsonError(w, r, http.StatusInternalServerError, err)
|
||||
@@ -99,8 +102,17 @@ func (fs *FilerServer) doPostAutoChunk(ctx context.Context, w http.ResponseWrite
|
||||
if so.SaveInside {
|
||||
buf := bufPool.Get().(*bytes.Buffer)
|
||||
buf.Reset()
|
||||
buf.ReadFrom(part1)
|
||||
filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, nil, nil, 0, buf.Bytes())
|
||||
if _, err := buf.ReadFrom(part1); err != nil {
|
||||
bufPool.Put(buf)
|
||||
return nil, nil, fmt.Errorf("read input: %w", err)
|
||||
}
|
||||
md5bytes = util.Md5(buf.Bytes())
|
||||
headerMd5 := r.Header.Get("Content-Md5")
|
||||
if headerMd5 != "" && !(util.Base64Encode(md5bytes) == headerMd5 || fmt.Sprintf("%x", md5bytes) == headerMd5) {
|
||||
bufPool.Put(buf)
|
||||
return nil, nil, errors.New(constants.ErrMsgBadDigest)
|
||||
}
|
||||
filerResult, replyerr, _ = fs.saveMetaData(ctx, r, fileName, contentType, so, chunkSize, md5bytes, nil, int64(buf.Len()), buf.Bytes())
|
||||
bufPool.Put(buf)
|
||||
return
|
||||
}
|
||||
@@ -293,7 +305,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
|
||||
if entry != nil {
|
||||
inlineCandidate := len(entry.Content) > 0 ||
|
||||
(isAppend && entry.FileSize == 0 && entry.Remote == nil && len(entry.HardLinkId) == 0 && len(entry.GetChunks()) == 0)
|
||||
if isAppend && !so.SaveInside && content != nil && !inlineCandidate {
|
||||
if isAppend && content != nil && !inlineCandidate {
|
||||
if so.SaveInside {
|
||||
replyerr = errors.New(constants.ErrMsgAppendSaveInsideChunked)
|
||||
return
|
||||
}
|
||||
replyerr = fmt.Errorf("inline file changed while preparing append; retry the request")
|
||||
return
|
||||
}
|
||||
@@ -301,11 +317,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
|
||||
if inlineCandidate {
|
||||
if isOffsetWrite {
|
||||
// TODO: support inline offset writes separately from append semantics.
|
||||
replyerr = fmt.Errorf("offset write to inline small file is not supported yet")
|
||||
replyerr = errors.New(constants.ErrMsgOffsetWriteInline)
|
||||
return
|
||||
}
|
||||
if !isAppend || so.SaveInside || entry.Remote != nil || len(entry.HardLinkId) != 0 || len(entry.GetChunks()) != 0 {
|
||||
replyerr = fmt.Errorf("append to inline content with this storage mode is not supported")
|
||||
replyerr = errors.New(constants.ErrMsgAppendInlineMode)
|
||||
return
|
||||
}
|
||||
if entry.FileSize > uint64(len(entry.Content)) {
|
||||
|
||||
@@ -270,6 +270,164 @@ func TestFilerInlineAppendMultipartPostStaysInline(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilerSaveInsideMultipartPostRecordsSize(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
content := []byte(`{"identities":[]}`)
|
||||
store := newRenameTestStore()
|
||||
f := newRenameTestFiler(t, store)
|
||||
fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
|
||||
|
||||
var body bytes.Buffer
|
||||
multipartWriter := multipart.NewWriter(&body)
|
||||
part, err := multipartWriter.CreateFormFile("file", "identity.json")
|
||||
if err != nil {
|
||||
t.Fatalf("create multipart file part: %v", err)
|
||||
}
|
||||
if _, err = part.Write(content); err != nil {
|
||||
t.Fatalf("write multipart file part: %v", err)
|
||||
}
|
||||
if err = multipartWriter.Close(); err != nil {
|
||||
t.Fatalf("close multipart body: %v", err)
|
||||
}
|
||||
r := httptest.NewRequest(http.MethodPost, "/", &body)
|
||||
r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
|
||||
result, requestMD5, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, int64(body.Len()), &operation.StorageOption{SaveInside: true})
|
||||
if err != nil {
|
||||
t.Fatalf("POST saveInside: %v", err)
|
||||
}
|
||||
if result == nil || result.Size != int64(len(content)) {
|
||||
t.Fatalf("result = %#v, want size %d", result, len(content))
|
||||
}
|
||||
wantMD5 := md5.Sum(content)
|
||||
if !bytes.Equal(requestMD5, wantMD5[:]) {
|
||||
t.Fatalf("request MD5 = %x, want %x", requestMD5, wantMD5)
|
||||
}
|
||||
saved, err := f.FindEntry(ctx, "/identity.json")
|
||||
if err != nil {
|
||||
t.Fatalf("find saved entry: %v", err)
|
||||
}
|
||||
if !bytes.Equal(saved.Content, content) || saved.FileSize != uint64(len(content)) {
|
||||
t.Fatalf("saved entry: content=%q size=%d, want %q size %d", saved.Content, saved.FileSize, content, len(content))
|
||||
}
|
||||
if !bytes.Equal(saved.Md5, wantMD5[:]) {
|
||||
t.Fatalf("saved MD5 = %x, want %x", saved.Md5, wantMD5)
|
||||
}
|
||||
}
|
||||
|
||||
func saveInsideMultipartRequest(t *testing.T, target, fileName string, content []byte) *http.Request {
|
||||
t.Helper()
|
||||
var body bytes.Buffer
|
||||
multipartWriter := multipart.NewWriter(&body)
|
||||
part, err := multipartWriter.CreateFormFile("file", fileName)
|
||||
if err != nil {
|
||||
t.Fatalf("create multipart file part: %v", err)
|
||||
}
|
||||
if _, err = part.Write(content); err != nil {
|
||||
t.Fatalf("write multipart file part: %v", err)
|
||||
}
|
||||
if err = multipartWriter.Close(); err != nil {
|
||||
t.Fatalf("close multipart body: %v", err)
|
||||
}
|
||||
r := httptest.NewRequest(http.MethodPost, target, &body)
|
||||
r.Header.Set("Content-Type", multipartWriter.FormDataContentType())
|
||||
return r
|
||||
}
|
||||
|
||||
func TestFilerSaveInsideMultipartPostRejectsBadDigest(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
content := []byte(`{"identities":[]}`)
|
||||
store := newRenameTestStore()
|
||||
f := newRenameTestFiler(t, store)
|
||||
fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
|
||||
|
||||
r := saveInsideMultipartRequest(t, "/", "identity.json", content)
|
||||
r.Header.Set("Content-Md5", util.Base64Encode([]byte("not the digest")))
|
||||
if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil || !strings.Contains(err.Error(), "Content-Md5") {
|
||||
t.Fatalf("POST saveInside error = %v, want bad Content-MD5 error", err)
|
||||
}
|
||||
if _, err := f.FindEntry(ctx, "/identity.json"); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("bad digest still saved the entry: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilerSaveInsideMultipartAppendRejected(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const filePath = "/append-target.txt"
|
||||
appendContent := []byte("new")
|
||||
|
||||
t.Run("chunk backed target", func(t *testing.T) {
|
||||
oldChunk := &filer_pb.FileChunk{FileId: "1,00000001", Offset: 0, Size: 4}
|
||||
store := newRenameTestStore()
|
||||
f := newRenameTestFiler(t, store)
|
||||
if err := store.InsertEntry(ctx, &filer.Entry{
|
||||
FullPath: util.FullPath(filePath),
|
||||
Attr: filer.Attr{
|
||||
Mtime: time.Unix(1, 0),
|
||||
Crtime: time.Unix(1, 0),
|
||||
FileSize: 4,
|
||||
},
|
||||
Chunks: []*filer_pb.FileChunk{oldChunk},
|
||||
}); err != nil {
|
||||
t.Fatalf("insert chunk-backed entry: %v", err)
|
||||
}
|
||||
|
||||
fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
|
||||
r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
|
||||
if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
|
||||
t.Fatal("saveInside append to chunk-backed entry succeeded")
|
||||
}
|
||||
recorder := httptest.NewRecorder()
|
||||
r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
|
||||
fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
|
||||
if recorder.Code != http.StatusBadRequest {
|
||||
t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
|
||||
}
|
||||
unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
|
||||
if err != nil {
|
||||
t.Fatalf("find entry after rejected append: %v", err)
|
||||
}
|
||||
if unchanged.FileSize != 4 || len(unchanged.Chunks) != 1 || len(unchanged.Content) != 0 {
|
||||
t.Fatalf("rejected append changed entry: size=%d chunks=%d content=%q", unchanged.FileSize, len(unchanged.Chunks), unchanged.Content)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("inline target", func(t *testing.T) {
|
||||
oldContent := []byte("old-")
|
||||
store := newRenameTestStore()
|
||||
f := newRenameTestFiler(t, store)
|
||||
if err := store.InsertEntry(ctx, &filer.Entry{
|
||||
FullPath: util.FullPath(filePath),
|
||||
Attr: filer.Attr{
|
||||
Mtime: time.Unix(1, 0),
|
||||
Crtime: time.Unix(1, 0),
|
||||
FileSize: uint64(len(oldContent)),
|
||||
},
|
||||
Content: append([]byte(nil), oldContent...),
|
||||
}); err != nil {
|
||||
t.Fatalf("insert inline entry: %v", err)
|
||||
}
|
||||
|
||||
fs := &FilerServer{filer: f, option: &FilerOption{MaxMB: 1}}
|
||||
r := saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
|
||||
if _, _, err := fs.doPostAutoChunk(ctx, httptest.NewRecorder(), r, 4, r.ContentLength, &operation.StorageOption{SaveInside: true}); err == nil {
|
||||
t.Fatal("saveInside append to inline entry succeeded")
|
||||
}
|
||||
recorder := httptest.NewRecorder()
|
||||
r = saveInsideMultipartRequest(t, filePath+"?op=append", "append-target.txt", appendContent)
|
||||
fs.autoChunk(ctx, recorder, r, r.ContentLength, &operation.StorageOption{SaveInside: true})
|
||||
if recorder.Code != http.StatusBadRequest {
|
||||
t.Fatalf("rejected saveInside append status = %d, want %d", recorder.Code, http.StatusBadRequest)
|
||||
}
|
||||
unchanged, err := f.FindEntry(ctx, util.FullPath(filePath))
|
||||
if err != nil {
|
||||
t.Fatalf("find entry after rejected append: %v", err)
|
||||
}
|
||||
if !bytes.Equal(unchanged.Content, oldContent) || unchanged.FileSize != uint64(len(oldContent)) {
|
||||
t.Fatalf("rejected append changed entry: content=%q size=%d", unchanged.Content, unchanged.FileSize)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestFilerInlineOffsetWriteRemainsUnsupported(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const filePath = "/offset.txt"
|
||||
|
||||
@@ -394,7 +394,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv
|
||||
}
|
||||
location = vs.store.FindEcShardTargetLocation(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.DataShardsCount, shardIds...)
|
||||
if location == nil {
|
||||
return nil, fmt.Errorf("no space left")
|
||||
return nil, status.Error(codes.ResourceExhausted, "no space left")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -570,6 +570,14 @@ func (f *WebDavFile) Read(p []byte) (readSize int, err error) {
|
||||
if fileSize == 0 {
|
||||
return 0, io.EOF
|
||||
}
|
||||
if len(f.entry.Content) > 0 {
|
||||
if f.off >= int64(len(f.entry.Content)) {
|
||||
return 0, io.EOF
|
||||
}
|
||||
readSize = copy(p, f.entry.Content[f.off:])
|
||||
f.off += int64(readSize)
|
||||
return readSize, nil
|
||||
}
|
||||
if f.visibleIntervals == nil {
|
||||
f.visibleIntervals, err = filer.NonOverlappingVisibleIntervals(f.ctx, f.fs.filerClient.GetLookupFileIdFunction(), f.entry.GetChunks(), 0, fileSize)
|
||||
if err != nil {
|
||||
|
||||
@@ -3,6 +3,7 @@ package weed_server
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"testing"
|
||||
@@ -104,3 +105,28 @@ func TestWebDavFile_Read_ManifestResolveFailure(t *testing.T) {
|
||||
assert.Error(t, err)
|
||||
assert.Equal(t, 0, n)
|
||||
}
|
||||
|
||||
// Small files stored inline in the entry have no chunks to read from.
|
||||
func TestWebDavFile_Read_InlineContent(t *testing.T) {
|
||||
content := []byte("hello inline")
|
||||
entry := &filer_pb.Entry{
|
||||
Name: "small.txt",
|
||||
Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(content))},
|
||||
Content: content,
|
||||
}
|
||||
filerAddr := startFakeWebDavFiler(t, &noVolumeFiler{})
|
||||
fc := wdclient.NewFilerClient([]pb.ServerAddress{filerAddr}, grpc.WithTransportCredentials(insecure.NewCredentials()), "")
|
||||
t.Cleanup(fc.Close)
|
||||
f := &WebDavFile{fs: &WebDavFileSystem{filerClient: fc}, name: "/small.txt", entry: entry, ctx: context.Background()}
|
||||
|
||||
got, err := io.ReadAll(f)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, content, got)
|
||||
|
||||
_, err = f.Seek(6, io.SeekStart)
|
||||
assert.NoError(t, err)
|
||||
buf := make([]byte, 3)
|
||||
n, err := f.Read(buf)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "inl", string(buf[:n]))
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package shell
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -273,7 +274,7 @@ func moveAwayOneNormalVolume(commandEnv *CommandEnv, volumeReplicas map[uint32][
|
||||
}
|
||||
// most empty one is in the front
|
||||
slices.SortFunc(otherNodes, func(a, b *Node) int {
|
||||
return int(a.localVolumeRatio(maxVolumeCountFn) - b.localVolumeRatio(maxVolumeCountFn))
|
||||
return cmp.Compare(a.localVolumeRatio(maxVolumeCountFn), b.localVolumeRatio(maxVolumeCountFn))
|
||||
})
|
||||
for i := 0; i < len(otherNodes); i++ {
|
||||
emptyNode := otherNodes[i]
|
||||
|
||||
@@ -5,6 +5,22 @@ import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestEvacuatePrefersLowerVolumeRatio(t *testing.T) {
|
||||
volumes := mkByteVolumes(1, 1, 1)
|
||||
source := makeByteNode("source", 10, 0, 0, volumes)
|
||||
fuller := makeByteNode("fuller", 10, 0, 0, mkByteVolumes(10, 4, 1))
|
||||
emptier := makeByteNode("emptier", 10, 0, 0, mkByteVolumes(20, 2, 1))
|
||||
|
||||
// Put the 40%-full server first: truncating 0.2 - 0.4 to int makes the ratios tie.
|
||||
moved, err := moveAwayOneNormalVolume(nil, nil, volumes[0], source, []*Node{fuller, emptier}, false)
|
||||
if err != nil || !moved {
|
||||
t.Fatalf("evacuate: moved=%v, err=%v", moved, err)
|
||||
}
|
||||
if _, found := emptier.selectedVolumes[volumes[0].Id]; !found {
|
||||
t.Fatal("expected evacuation to choose the server with lower occupancy")
|
||||
}
|
||||
}
|
||||
|
||||
func TestVolumeServerEvacuate(t *testing.T) {
|
||||
targetServer := ""
|
||||
volumeRack := "rack1"
|
||||
|
||||
@@ -38,6 +38,10 @@ type Node struct {
|
||||
rack string // composite rack key (e.g. "dc1:rack1")
|
||||
|
||||
freeSlots int
|
||||
// noReceive marks a node that refused shards (e.g. answered "no space
|
||||
// left"): it stays usable as a move source but is never picked as a target,
|
||||
// regardless of what its freeSlots say.
|
||||
noReceive bool
|
||||
disks map[uint32]*disk
|
||||
shards map[volKey]*volumeShards
|
||||
}
|
||||
@@ -47,7 +51,8 @@ type disk struct {
|
||||
diskType string
|
||||
tags []string // placement tags, for preferred-tag tiering
|
||||
freeSlots int
|
||||
shardCount int // total EC shards on this disk across all volumes
|
||||
shardCount int // total EC shards on this disk across all volumes
|
||||
noReceive bool // this disk refused a shard; siblings may still receive
|
||||
}
|
||||
|
||||
type volumeShards struct {
|
||||
@@ -156,6 +161,39 @@ func (n *Node) SetHost(host string) {
|
||||
}
|
||||
}
|
||||
|
||||
// SetNoReceive marks the node ineligible as a move destination. Use when the
|
||||
// node itself refused shards (e.g. answered "no space left") so the heartbeat's
|
||||
// free-slot count cannot be trusted; the node can still shed shards as a source.
|
||||
func (n *Node) SetNoReceive() {
|
||||
n.noReceive = true
|
||||
}
|
||||
|
||||
// SetDiskNoReceive marks one disk on the node ineligible as a move destination.
|
||||
// An explicit-disk copy can be refused while the node's other disks still have
|
||||
// room; a disk_id 0 refusal means the server searched every disk and is a
|
||||
// whole-node refusal (SetNoReceive) instead.
|
||||
func (n *Node) SetDiskNoReceive(diskID uint32) {
|
||||
if d, ok := n.disks[diskID]; ok {
|
||||
d.noReceive = true
|
||||
}
|
||||
}
|
||||
|
||||
// receiveSlots is the node's receiving capacity: its aggregate free slots with
|
||||
// refused disks subtracted, and 0 for a refused node. Slots a disk just
|
||||
// refused are phantom capacity whatever the snapshot reported.
|
||||
func (n *Node) receiveSlots() int {
|
||||
if n.noReceive {
|
||||
return 0
|
||||
}
|
||||
free := n.freeSlots
|
||||
for _, d := range n.disks {
|
||||
if d.noReceive {
|
||||
free -= d.freeSlots
|
||||
}
|
||||
}
|
||||
return max(free, 0)
|
||||
}
|
||||
|
||||
// AddDisk registers a physical disk. shardCount is the disk's total EC shard
|
||||
// count across all volumes (used for disk scoring); freeSlots is the per-disk
|
||||
// free EC shard slots.
|
||||
@@ -296,7 +334,7 @@ func buildRacks(nodes map[string]*Node) map[string]*rack {
|
||||
racks[n.rack] = r
|
||||
}
|
||||
r.nodes[n.id] = n
|
||||
r.freeSlots += n.freeSlots
|
||||
r.freeSlots += n.receiveSlots()
|
||||
}
|
||||
return racks
|
||||
}
|
||||
@@ -440,7 +478,7 @@ func planRackTotalCap(vk volKey, racks map[string]*rack, rackShardCount map[stri
|
||||
continue
|
||||
}
|
||||
for _, n := range r.nodes {
|
||||
room[rk] += min(max(n.freeSlots, 0), max(rp.SameRackCount-volumeShardCount(n, vk), 0))
|
||||
room[rk] += min(n.receiveSlots(), max(rp.SameRackCount-volumeShardCount(n, vk), 0))
|
||||
}
|
||||
}
|
||||
return rackTotalCap(sortedKeys(racks), nil, room, totalShards)
|
||||
@@ -575,12 +613,15 @@ func pickBestNodeForVolumeOnDisk(nodes []*Node, vk volKey, rp *super_block.Repli
|
||||
// metadata is authoritative when present; without it the node's aggregate
|
||||
// stands in so the volume server's own disk selection can place the shard.
|
||||
func freeSlotsForType(n *Node, diskType string) int {
|
||||
if n.noReceive {
|
||||
return 0
|
||||
}
|
||||
if len(n.disks) == 0 {
|
||||
return n.freeSlots
|
||||
}
|
||||
free := 0
|
||||
for _, d := range n.disks {
|
||||
if d.freeSlots > 0 && (diskType == "" || d.diskType == diskType) {
|
||||
if d.freeSlots > 0 && !d.noReceive && (diskType == "" || d.diskType == diskType) {
|
||||
free += d.freeSlots
|
||||
}
|
||||
}
|
||||
@@ -604,7 +645,7 @@ func pickBestNodeForVolume(nodes []*Node, vk volKey, rp *super_block.ReplicaPlac
|
||||
bestFree := -1
|
||||
bestScore := -1
|
||||
for _, node := range nodes {
|
||||
if node.freeSlots <= 0 {
|
||||
if node.receiveSlots() <= 0 {
|
||||
continue
|
||||
}
|
||||
count := volumeShardCount(node, vk)
|
||||
@@ -909,7 +950,7 @@ func detectGlobalImbalance(nodes map[string]*Node, racks map[string]*rack, diskT
|
||||
continue
|
||||
}
|
||||
util := float64(count) / float64(capacity)
|
||||
if util < minUtil && node.freeSlots > 0 {
|
||||
if util < minUtil && node.receiveSlots() > 0 {
|
||||
minUtil, minCount, minNode = util, count, node
|
||||
}
|
||||
if util > maxUtil {
|
||||
@@ -1091,7 +1132,7 @@ func pickBestDiskOnNode(node *Node, vk volKey, diskType string, shardID, dataSha
|
||||
if diskType != "" && d.diskType != diskType {
|
||||
continue
|
||||
}
|
||||
if d.freeSlots <= 0 {
|
||||
if d.freeSlots <= 0 || d.noReceive {
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
@@ -926,3 +926,67 @@ func TestGlobalPrefersVolumeAbsentFromDestinationMachine(t *testing.T) {
|
||||
t.Errorf("first global move is volume %d, want 200 (vol100 already on the destination machine)", moves[0].volumeID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoReceiveNodeNeverTargeted: a node that refused a copy ("no space left")
|
||||
// must not be picked as a destination no matter how many free slots it reports,
|
||||
// but it still sheds its own over-cap shards as a source.
|
||||
func TestNoReceiveNodeNeverTargeted(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
topo := NewTopology()
|
||||
src := topo.AddNode("n1", "dc1", "dc1:r1", 50)
|
||||
src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))
|
||||
full := topo.AddNode("n2", "dc1", "dc1:r1", 50)
|
||||
full.AddShards(vid, "c", 0, bits(10, 11, 12, 13)) // all parity, over the 2/node cap
|
||||
full.SetNoReceive()
|
||||
topo.AddNode("n3", "dc1", "dc1:r1", 50)
|
||||
|
||||
moves := Plan(topo, Options{Ratio: ratio(10, 4)})
|
||||
if len(moves) == 0 {
|
||||
t.Fatal("expected moves off the over-cap nodes")
|
||||
}
|
||||
shed := false
|
||||
for _, m := range moves {
|
||||
if m.TargetNode == "n2" {
|
||||
t.Fatalf("move targets the refusing node: %+v", m)
|
||||
}
|
||||
if m.SourceNode == "n2" {
|
||||
shed = true
|
||||
}
|
||||
}
|
||||
if !shed {
|
||||
t.Error("the refusing node should still shed its over-cap shards as a source")
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoReceiveDiskLeavesSiblingsEligible: an explicit-disk refusal marks only
|
||||
// that disk, not the node — the planner may still target the node's other
|
||||
// disks, but never the refused one.
|
||||
func TestNoReceiveDiskLeavesSiblingsEligible(t *testing.T) {
|
||||
const vid = uint32(1)
|
||||
topo := NewTopology()
|
||||
src := topo.AddNode("n1", "dc1", "dc1:r1", 50)
|
||||
src.AddShards(vid, "c", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))
|
||||
n2 := topo.AddNode("n2", "dc1", "dc1:r1", 50)
|
||||
// The refused disk scores better (empty); without the noReceive check the
|
||||
// picker would choose it again.
|
||||
n2.AddDisk(1, "hdd", 50, 0)
|
||||
n2.AddDisk(2, "hdd", 50, 50)
|
||||
n2.AddShards(vid, "c", 2, bits(10, 11, 12, 13))
|
||||
n2.SetDiskNoReceive(1)
|
||||
topo.AddNode("n3", "dc1", "dc1:r1", 50)
|
||||
|
||||
moves := Plan(topo, Options{Ratio: ratio(10, 4)})
|
||||
targeted := false
|
||||
for _, m := range moves {
|
||||
if m.TargetNode != "n2" {
|
||||
continue
|
||||
}
|
||||
targeted = true
|
||||
if m.TargetDisk == 1 {
|
||||
t.Fatalf("move targets the refused disk: %+v", m)
|
||||
}
|
||||
}
|
||||
if !targeted {
|
||||
t.Fatal("a node with one refused disk should still receive on its other disks")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -430,6 +430,9 @@ func rackTotalCap(rackKeys []string, held, room map[string]int, totalShards int)
|
||||
// its free slots, bounded by the per-disk durability cap (maxPerDisk shards of
|
||||
// one volume per disk; <= 0 disables it).
|
||||
func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int {
|
||||
if d.noReceive {
|
||||
return 0
|
||||
}
|
||||
room := d.freeSlots
|
||||
if maxPerDisk > 0 {
|
||||
held := 0
|
||||
@@ -444,13 +447,16 @@ func diskShardRoom(n *Node, d *disk, vk volKey, maxPerDisk int) int {
|
||||
// nodeShardRoom returns how many more shards of the volume the node's eligible
|
||||
// disks can take, bounded by the node's free slots.
|
||||
func nodeShardRoom(n *Node, vk volKey, eligible func(*disk) bool, maxPerDisk int) int {
|
||||
if n.noReceive {
|
||||
return 0
|
||||
}
|
||||
room := 0
|
||||
for _, d := range n.disks {
|
||||
if eligible(d) {
|
||||
room += diskShardRoom(n, d, vk, maxPerDisk)
|
||||
}
|
||||
}
|
||||
return max(min(room, n.freeSlots), 0)
|
||||
return max(min(room, n.receiveSlots()), 0)
|
||||
}
|
||||
|
||||
// rackShardRoom returns how many more shards of the volume the rack can take.
|
||||
@@ -532,7 +538,7 @@ func pickBestDiskEligible(node *Node, vk volKey, eligible func(*disk) bool, pref
|
||||
bestPreferred := false
|
||||
for _, diskID := range sortedDiskKeys(node.disks) {
|
||||
d := node.disks[diskID]
|
||||
if !eligible(d) || d.freeSlots <= 0 {
|
||||
if !eligible(d) || d.freeSlots <= 0 || d.noReceive {
|
||||
continue
|
||||
}
|
||||
existingShards := 0
|
||||
|
||||
@@ -7,6 +7,10 @@ const (
|
||||
ErrMsgOperationNotPermitted = "operation not permitted"
|
||||
ErrMsgBadDigest = "The Content-Md5 you specified did not match what we received."
|
||||
ErrMsgEntryNameTooLong = "entry name too long"
|
||||
// Unsupported write-mode combinations; deterministic client errors.
|
||||
ErrMsgAppendSaveInsideChunked = "append saved inside the entry to a chunked file is not supported"
|
||||
ErrMsgAppendInlineMode = "append to inline content with this storage mode is not supported"
|
||||
ErrMsgOffsetWriteInline = "offset write to inline small file is not supported yet"
|
||||
// Suffix appended to a path, e.g. "/buckets/b/foo/bar is a file"
|
||||
ErrMsgIsAFile = " is a file"
|
||||
// Prefix + suffix, e.g. "existing /buckets/b/foo is a directory"
|
||||
|
||||
@@ -116,17 +116,29 @@ func (t *ECBalanceTask) Execute(ctx context.Context, params *worker_pb.TaskParam
|
||||
return nil
|
||||
}
|
||||
|
||||
err := mover.MoveEcShards(ctx, volume_move.EcShardMove{
|
||||
move := volume_move.EcShardMove{
|
||||
VolumeId: needle.VolumeId(params.VolumeId),
|
||||
Collection: params.Collection,
|
||||
ShardIds: shardIds,
|
||||
Source: sourceAddr,
|
||||
Target: targetAddr,
|
||||
TargetDisk: target.DiskId,
|
||||
}, volume_move.EcMoveOptions{
|
||||
}
|
||||
opts := volume_move.EcMoveOptions{
|
||||
IoBytePerSecond: ecParams.GetIoBytePerSecond(),
|
||||
Progress: t.reportProgress,
|
||||
})
|
||||
}
|
||||
err := mover.MoveEcShards(ctx, move, opts)
|
||||
// A "no space left" refusal to a pinned disk convicts only that disk;
|
||||
// retry once letting the server pick any disk on the same node. The
|
||||
// planner's node choice stands; only the intra-node disk defers to the
|
||||
// server's live view. When the node is truly full, the retry fails with
|
||||
// the same error.
|
||||
if err != nil && target.DiskId > 0 && volume_move.IsNoSpaceLeftError(err) {
|
||||
glog.Infof("EC balance: %s disk %d has no space left; retrying with server-picked disk", target.Node, target.DiskId)
|
||||
move.TargetDisk = 0
|
||||
err = mover.MoveEcShards(ctx, move, opts)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials/insecure"
|
||||
)
|
||||
|
||||
func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) {
|
||||
@@ -25,3 +27,32 @@ func TestECBalanceTaskRejectsOutOfRangeShardIds(t *testing.T) {
|
||||
t.Fatalf("Execute: expected out-of-range rejection, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A refusal against the pinned disk convicts only that disk, not the node:
|
||||
// the task retries once letting the server pick a disk, and the move completes.
|
||||
func TestECBalanceTaskRetriesServerPickOnPinnedDiskRefusal(t *testing.T) {
|
||||
source := startFakeEcVolumeServer(t, 7, 5)
|
||||
target := startFakeEcVolumeServer(t, 7)
|
||||
target.refuseCopyOnDisk = 2
|
||||
|
||||
task := NewECBalanceTask("t2", 7, "c1",
|
||||
grpc.WithTransportCredentials(insecure.NewCredentials()))
|
||||
params := &worker_pb.TaskParams{
|
||||
VolumeId: 7,
|
||||
Sources: []*worker_pb.TaskSource{{Node: source.address(), ShardIds: []uint32{5}}},
|
||||
Targets: []*worker_pb.TaskTarget{{Node: target.address(), DiskId: 2, ShardIds: []uint32{5}}},
|
||||
}
|
||||
|
||||
if err := task.Execute(context.Background(), params); err != nil {
|
||||
t.Fatalf("Execute: %v", err)
|
||||
}
|
||||
if got := target.copyDiskIds; len(got) != 2 || got[0] != 2 || got[1] != 0 {
|
||||
t.Fatalf("expected copies to disk 2 then server-picked 0, got %v", got)
|
||||
}
|
||||
if source.has(7, 5) {
|
||||
t.Fatalf("source still holds shard 5")
|
||||
}
|
||||
if !target.has(7, 5) {
|
||||
t.Fatalf("target never received shard 5")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,6 +32,12 @@ type fakeEcVolumeServer struct {
|
||||
// the request even when it was a no-op against the inventory.
|
||||
deleted []string
|
||||
unmounted []string
|
||||
// refuseCopyOnDisk, when nonzero, makes VolumeEcShardsCopy fail requests
|
||||
// pinning that disk id with "no space left" — a full disk; a request with
|
||||
// DiskId 0 (server picks) still succeeds.
|
||||
refuseCopyOnDisk uint32
|
||||
// copyDiskIds records the DiskId each copy request carried.
|
||||
copyDiskIds []uint32
|
||||
// grpcAddr is host:port; the code under test dials by address string.
|
||||
grpcAddr string
|
||||
stop func()
|
||||
@@ -46,11 +52,31 @@ func (f *fakeEcVolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume
|
||||
VolumeId: req.VolumeId,
|
||||
ShardId: sid,
|
||||
Collection: f.collection,
|
||||
Size: 1,
|
||||
})
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// VolumeEcShardsCopy records the request's DiskId and adds the shards to this
|
||||
// server's inventory, unless the pinned disk is the configured refusing disk.
|
||||
func (f *fakeEcVolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.copyDiskIds = append(f.copyDiskIds, req.DiskId)
|
||||
if f.refuseCopyOnDisk != 0 && req.DiskId == f.refuseCopyOnDisk {
|
||||
return nil, fmt.Errorf("disk %d: no space left", req.DiskId)
|
||||
}
|
||||
for _, sid := range req.ShardIds {
|
||||
f.shards[req.VolumeId][sid] = true
|
||||
}
|
||||
return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeEcVolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_server_pb.VolumeEcShardsMountRequest) (*volume_server_pb.VolumeEcShardsMountResponse, error) {
|
||||
return &volume_server_pb.VolumeEcShardsMountResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeEcVolumeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
|
||||
Reference in New Issue
Block a user