mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-17 03:55:50 +00:00
Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  
|
6da432b8bf | ||
|
|
da3b7feeae | ||
|
|
00804b32fa | ||
|
|
44657988ea | ||
|
|
2d06e2d6c1 | ||
|
|
446e6f3444 | ||
|
|
348116b2c4 | ||
|
|
5b4a707c43 | ||
|
|
d3a97f76bd | ||
 github-actions[bot]
|
cdc6976e3a | ||
|
Chris LuandGitHub
|
921535411b |
+1071
-1080
File diff suppressed because it is too large
Load Diff
|
Before Width: | Height: | Size: 54 KiB After Width: | Height: | Size: 54 KiB |
@@ -620,11 +620,20 @@ impl DiskLocation {
|
||||
return Err(v.compacting_error());
|
||||
}
|
||||
if let Some(mut v) = self.volumes.remove(&vid) {
|
||||
crate::metrics::VOLUME_GAUGE
|
||||
.with_label_values(&[&v.collection, "volume"])
|
||||
.dec();
|
||||
v.destroy(only_empty, only_garbage, keep_remote_data)?;
|
||||
Ok(())
|
||||
match v.destroy(only_empty, only_garbage, keep_remote_data) {
|
||||
Ok(()) => {
|
||||
crate::metrics::VOLUME_GAUGE
|
||||
.with_label_values(&[&v.collection, "volume"])
|
||||
.dec();
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
// A failed destroy left the local files in place, so the
|
||||
// replica stays registered rather than becoming a phantom.
|
||||
self.volumes.insert(vid, v);
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
Err(VolumeError::NotFound)
|
||||
}
|
||||
@@ -641,11 +650,13 @@ impl DiskLocation {
|
||||
|
||||
for vid in vids {
|
||||
if let Some(mut v) = self.volumes.remove(&vid) {
|
||||
crate::metrics::VOLUME_GAUGE
|
||||
.with_label_values(&[&v.collection, "volume"])
|
||||
.dec();
|
||||
if let Err(e) = v.destroy(false, false, false) {
|
||||
warn!(volume_id = vid.0, error = %e, "delete collection: failed to destroy volume");
|
||||
self.volumes.insert(vid, v);
|
||||
} else {
|
||||
crate::metrics::VOLUME_GAUGE
|
||||
.with_label_values(&[&v.collection, "volume"])
|
||||
.dec();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5006,13 +5006,12 @@ impl Volume {
|
||||
.get(&storage_name);
|
||||
if let Some(backend) = backend {
|
||||
if let Err(e) = backend.delete_file_blocking(&storage_key) {
|
||||
warn!(
|
||||
volume_id = self.id.0,
|
||||
storage_name,
|
||||
storage_key,
|
||||
error = %e,
|
||||
"failed to delete remote tier file during destroy"
|
||||
);
|
||||
// Stop short of local cleanup: a failed remote delete must
|
||||
// not strand a registered replica with no local files.
|
||||
return Err(VolumeError::Io(io::Error::other(format!(
|
||||
"volume {} remote object {}/{} delete failed: {}",
|
||||
self.id, storage_name, storage_key, e
|
||||
))));
|
||||
}
|
||||
} else {
|
||||
warn!(
|
||||
|
||||
@@ -114,7 +114,7 @@ func TestS3Integration(t *testing.T) {
|
||||
// findAvailablePort finds an available port by binding to port 0
|
||||
|
||||
// startMiniCluster starts a weed mini instance directly without exec.
|
||||
// Extra flags (e.g. "-s3.allowDeleteBucketNotEmpty=false") can be appended via extraArgs.
|
||||
// Extra flags can be appended via extraArgs.
|
||||
func startMiniCluster(t *testing.T, extraArgs ...string) (*TestCluster, error) {
|
||||
// Allocate non-colliding ports (including gRPC offsets) for weed mini.
|
||||
ports := testutil.MustFreeMiniPorts(t, []string{"Master", "Volume", "Filer", "S3"})
|
||||
@@ -872,7 +872,7 @@ func TestS3DeleteBucketNotEmpty(t *testing.T) {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
cluster, err := startMiniCluster(t, "-s3.allowDeleteBucketNotEmpty=false")
|
||||
cluster, err := startMiniCluster(t)
|
||||
require.NoError(t, err)
|
||||
defer cluster.Stop()
|
||||
|
||||
|
||||
@@ -228,8 +228,8 @@ test-with-server: start-server
|
||||
# Test with different configurations
|
||||
test-versioning-with-configs: check-deps
|
||||
@echo "Testing with different S3 configurations..."
|
||||
@echo "Testing with empty folder allowed..."
|
||||
@$(WEED_BINARY) mini -s3.port=$(S3_PORT) > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid
|
||||
@echo "Testing with delete bucket not empty allowed..."
|
||||
@$(WEED_BINARY) mini -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=true > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid
|
||||
@sleep 5
|
||||
@go test -v -timeout=5m -run "TestVersioningBasicWorkflow" . || true
|
||||
@if [ -f weed-config1.pid ]; then kill -TERM $$(cat weed-config1.pid) 2>/dev/null || true; rm -f weed-config1.pid; fi
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/seaweedfs/seaweedfs/weed/admin/maintenance"
|
||||
adminplugin "github.com/seaweedfs/seaweedfs/weed/admin/plugin"
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
@@ -1155,6 +1156,7 @@ func (s *AdminServer) SetBucketLifecycle(bucketName string, rules []BucketLifecy
|
||||
bucketPath := filerConfig.BucketsPath + "/" + bucketName
|
||||
resp, err := client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: bucketPath,
|
||||
RequestId: uuid.NewString(),
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + bucketPath,
|
||||
Mutations: []*filer_pb.ObjectMutation{bucketLifecycleMutation(filerConfig.BucketsPath, bucketName, lifecycleXML)},
|
||||
})
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
|
||||
@@ -126,6 +127,7 @@ func (s *AdminServer) writeBucketPolicy(bucketName string, policyJSON []byte) er
|
||||
bucketPath := filerConfig.BucketsPath + "/" + bucketName
|
||||
resp, err := client.ObjectTransaction(context.Background(), &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: bucketPath,
|
||||
RequestId: uuid.NewString(),
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + bucketPath,
|
||||
Mutations: []*filer_pb.ObjectMutation{bucketPolicyMutation(filerConfig.BucketsPath, bucketName, policyJSON)},
|
||||
})
|
||||
|
||||
@@ -256,11 +256,13 @@ templ Admin(data dash.AdminData) {
|
||||
if tier.IsRemote {
|
||||
<span class="text-muted">-</span>
|
||||
} else {
|
||||
<div class="progress" style="height: 20px;">
|
||||
<div class="progress-bar" role="progressbar"
|
||||
style={fmt.Sprintf("width: %d%%", tier.UsagePercent())}>
|
||||
{fmt.Sprintf("%d%%", tier.UsagePercent())}
|
||||
<div class="d-flex align-items-center">
|
||||
<div class="progress flex-grow-1" style="height: 20px; min-width: 48px;">
|
||||
<div class="progress-bar" role="progressbar"
|
||||
style={fmt.Sprintf("width: %d%%", tier.UsagePercent())}>
|
||||
</div>
|
||||
</div>
|
||||
<span class="ms-2 text-nowrap">{fmt.Sprintf("%d%%", tier.UsagePercent())}</span>
|
||||
</div>
|
||||
}
|
||||
</td>
|
||||
@@ -445,11 +447,13 @@ templ Admin(data dash.AdminData) {
|
||||
<td>{vs.DataCenter}</td>
|
||||
<td>{vs.Rack}</td>
|
||||
<td>
|
||||
<div class="progress" style="height: 20px;">
|
||||
<div class="progress-bar" role="progressbar"
|
||||
style={fmt.Sprintf("width: %d%%", calculatePercent(vs.Volumes, vs.MaxVolumes))}>
|
||||
{fmt.Sprintf("%d/%d", vs.Volumes, vs.MaxVolumes)}
|
||||
<div class="d-flex align-items-center">
|
||||
<div class="progress flex-grow-1" style="height: 20px; min-width: 48px;">
|
||||
<div class="progress-bar" role="progressbar"
|
||||
style={fmt.Sprintf("width: %d%%", calculatePercent(vs.Volumes, vs.MaxVolumes))}>
|
||||
</div>
|
||||
</div>
|
||||
<span class="ms-2 text-nowrap">{fmt.Sprintf("%d/%d", vs.Volumes, vs.MaxVolumes)}</span>
|
||||
</div>
|
||||
</td>
|
||||
<td>
|
||||
|
||||
@@ -342,33 +342,33 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
} else {
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 32, "<div class=\"progress\" style=\"height: 20px;\"><div class=\"progress-bar\" role=\"progressbar\" style=\"")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 32, "<div class=\"d-flex align-items-center\"><div class=\"progress flex-grow-1\" style=\"height: 20px; min-width: 48px;\"><div class=\"progress-bar\" role=\"progressbar\" style=\"")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
var templ_7745c5c3_Var18 string
|
||||
templ_7745c5c3_Var18, templ_7745c5c3_Err = templruntime.SanitizeStyleAttributeValues(fmt.Sprintf("width: %d%%", tier.UsagePercent()))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 261, Col: 119}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 262, Col: 123}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var18))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 33, "\">")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 33, "\"></div></div><span class=\"ms-2 text-nowrap\">")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
var templ_7745c5c3_Var19 string
|
||||
templ_7745c5c3_Var19, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d%%", tier.UsagePercent()))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 262, Col: 105}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 265, Col: 132}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var19))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 34, "</div></div>")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 34, "</span></div>")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
@@ -390,7 +390,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var20 templ.SafeURL
|
||||
templ_7745c5c3_Var20, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/cluster/mount-clients"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 286, Col: 74}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 288, Col: 74}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var20))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -403,7 +403,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var21 string
|
||||
templ_7745c5c3_Var21, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", data.TotalMountClients))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 294, Col: 126}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 296, Col: 126}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var21))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -421,7 +421,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var22 templ.SafeURL
|
||||
templ_7745c5c3_Var22, templ_7745c5c3_Err = templ.JoinURLErrs(templ.SafeURL(fmt.Sprintf("//%s/ui/index.html", master.Address)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 325, Col: 121}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 327, Col: 121}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var22))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -434,7 +434,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var23 string
|
||||
templ_7745c5c3_Var23, templ_7745c5c3_Err = templ.JoinStringErrs(master.Address)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 326, Col: 67}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 328, Col: 67}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var23))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -467,7 +467,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var24 string
|
||||
templ_7745c5c3_Var24, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.MasterNodes)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 359, Col: 85}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 361, Col: 85}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var24))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -480,7 +480,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var25 string
|
||||
templ_7745c5c3_Var25, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.VolumeServers)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 367, Col: 87}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 369, Col: 87}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var25))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -493,7 +493,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var26 string
|
||||
templ_7745c5c3_Var26, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.FilerNodes)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 375, Col: 84}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 377, Col: 84}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var26))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -506,7 +506,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var27 string
|
||||
templ_7745c5c3_Var27, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.S3Nodes)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 383, Col: 81}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 385, Col: 81}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var27))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -519,7 +519,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var28 string
|
||||
templ_7745c5c3_Var28, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", len(data.MessageBrokers)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 391, Col: 88}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 393, Col: 88}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var28))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -532,7 +532,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var29 templ.SafeURL
|
||||
templ_7745c5c3_Var29, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/volumes"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 416, Col: 90}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 418, Col: 90}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var29))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -545,7 +545,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var30 templ.SafeURL
|
||||
templ_7745c5c3_Var30, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/cluster"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 417, Col: 90}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 419, Col: 90}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var30))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -563,7 +563,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var31 string
|
||||
templ_7745c5c3_Var31, templ_7745c5c3_Err = templ.JoinStringErrs(vs.ID)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 438, Col: 54}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 440, Col: 54}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var31))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -576,7 +576,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var32 templ.SafeURL
|
||||
templ_7745c5c3_Var32, templ_7745c5c3_Err = templ.JoinURLErrs(templ.SafeURL(fmt.Sprintf("//%s/ui/index.html", vs.GetDisplayAddress())))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 440, Col: 129}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 442, Col: 129}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var32))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -589,7 +589,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var33 string
|
||||
templ_7745c5c3_Var33, templ_7745c5c3_Err = templ.JoinStringErrs(vs.GetDisplayAddress())
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 441, Col: 75}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 443, Col: 75}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var33))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -602,7 +602,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var34 string
|
||||
templ_7745c5c3_Var34, templ_7745c5c3_Err = templ.JoinStringErrs(vs.DataCenter)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 445, Col: 62}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 447, Col: 62}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var34))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -615,39 +615,39 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var35 string
|
||||
templ_7745c5c3_Var35, templ_7745c5c3_Err = templ.JoinStringErrs(vs.Rack)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 446, Col: 56}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 448, Col: 56}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var35))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 59, "</td><td><div class=\"progress\" style=\"height: 20px;\"><div class=\"progress-bar\" role=\"progressbar\" style=\"")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 59, "</td><td><div class=\"d-flex align-items-center\"><div class=\"progress flex-grow-1\" style=\"height: 20px; min-width: 48px;\"><div class=\"progress-bar\" role=\"progressbar\" style=\"")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
var templ_7745c5c3_Var36 string
|
||||
templ_7745c5c3_Var36, templ_7745c5c3_Err = templruntime.SanitizeStyleAttributeValues(fmt.Sprintf("width: %d%%", calculatePercent(vs.Volumes, vs.MaxVolumes)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 450, Col: 135}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 453, Col: 139}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var36))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 60, "\">")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 60, "\"></div></div><span class=\"ms-2 text-nowrap\">")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
var templ_7745c5c3_Var37 string
|
||||
templ_7745c5c3_Var37, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d/%d", vs.Volumes, vs.MaxVolumes))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 451, Col: 104}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 456, Col: 131}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var37))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 61, "</div></div></td><td>")
|
||||
templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 61, "</span></div></td><td>")
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ_7745c5c3_Err
|
||||
}
|
||||
@@ -659,7 +659,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var38 string
|
||||
templ_7745c5c3_Var38, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d", vs.EcShards))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 457, Col: 127}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 461, Col: 127}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var38))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -677,7 +677,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var39 string
|
||||
templ_7745c5c3_Var39, templ_7745c5c3_Err = templ.JoinStringErrs(fmt.Sprintf("%d vol", vs.EcVolumes))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 459, Col: 119}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 463, Col: 119}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var39))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -701,7 +701,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var40 string
|
||||
templ_7745c5c3_Var40, templ_7745c5c3_Err = templ.JoinStringErrs(formatBytes(vs.DiskUsage))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 466, Col: 74}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 470, Col: 74}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var40))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -714,7 +714,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var41 string
|
||||
templ_7745c5c3_Var41, templ_7745c5c3_Err = templ.JoinStringErrs(formatBytes(vs.DiskCapacity))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 466, Col: 107}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 470, Col: 107}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var41))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -732,7 +732,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var42 string
|
||||
templ_7745c5c3_Var42, templ_7745c5c3_Err = templ.JoinStringErrs(formatBytes(vs.RemoteSize))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 469, Col: 137}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 473, Col: 137}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var42))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -761,7 +761,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var43 templ.SafeURL
|
||||
templ_7745c5c3_Var43, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/filer"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 504, Col: 88}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 508, Col: 88}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var43))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -774,7 +774,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var44 templ.SafeURL
|
||||
templ_7745c5c3_Var44, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/cluster"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 505, Col: 90}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 509, Col: 90}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var44))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -792,7 +792,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var45 templ.SafeURL
|
||||
templ_7745c5c3_Var45, templ_7745c5c3_Err = templ.JoinURLErrs(templ.SafeURL(fmt.Sprintf("//%s", filer.Address)))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 524, Col: 106}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 528, Col: 106}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var45))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -805,7 +805,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var46 string
|
||||
templ_7745c5c3_Var46, templ_7745c5c3_Err = templ.JoinStringErrs(filer.Address)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 525, Col: 66}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 529, Col: 66}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var46))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -818,7 +818,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var47 string
|
||||
templ_7745c5c3_Var47, templ_7745c5c3_Err = templ.JoinStringErrs(filer.DataCenter)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 529, Col: 65}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 533, Col: 65}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var47))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -831,7 +831,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var48 string
|
||||
templ_7745c5c3_Var48, templ_7745c5c3_Err = templ.JoinStringErrs(filer.Rack)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 530, Col: 59}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 534, Col: 59}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var48))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -844,7 +844,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var49 string
|
||||
templ_7745c5c3_Var49, templ_7745c5c3_Err = templ.JoinStringErrs(filer.LastUpdated.Format("2006-01-02 15:04:05"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 531, Col: 96}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 535, Col: 96}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var49))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -868,7 +868,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var50 templ.SafeURL
|
||||
templ_7745c5c3_Var50, templ_7745c5c3_Err = templ.JoinURLErrs(dash.PUrl(ctx, "/cluster/s3"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 564, Col: 93}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 568, Col: 93}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var50))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -886,7 +886,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var51 string
|
||||
templ_7745c5c3_Var51, templ_7745c5c3_Err = templ.JoinStringErrs(s3Node.Address)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 581, Col: 63}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 585, Col: 63}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var51))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -899,7 +899,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var52 string
|
||||
templ_7745c5c3_Var52, templ_7745c5c3_Err = templ.JoinStringErrs(s3Node.DataCenter)
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 582, Col: 66}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 586, Col: 66}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var52))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -912,7 +912,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var53 string
|
||||
templ_7745c5c3_Var53, templ_7745c5c3_Err = templ.JoinStringErrs(s3Node.LastUpdated.Format("2006-01-02 15:04:05"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 583, Col: 97}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 587, Col: 97}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var53))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
@@ -936,7 +936,7 @@ func Admin(data dash.AdminData) templ.Component {
|
||||
var templ_7745c5c3_Var54 string
|
||||
templ_7745c5c3_Var54, templ_7745c5c3_Err = templ.JoinStringErrs(data.LastUpdated.Format("2006-01-02 15:04:05"))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 607, Col: 81}
|
||||
return templ.Error{Err: templ_7745c5c3_Err, FileName: `view/app/admin.templ`, Line: 611, Col: 81}
|
||||
}
|
||||
_, templ_7745c5c3_Err = templ_7745c5c3_Buffer.WriteString(templ.EscapeString(templ_7745c5c3_Var54))
|
||||
if templ_7745c5c3_Err != nil {
|
||||
|
||||
@@ -154,7 +154,7 @@ func init() {
|
||||
filerS3Options.metricsHttpPort = cmdFiler.Flag.Int("s3.metricsPort", 0, "Prometheus metrics listen port")
|
||||
filerS3Options.metricsHttpIp = cmdFiler.Flag.String("s3.metricsIp", "", "metrics listen ip. If empty, default to same as -s3.ip.bind option.")
|
||||
cmdFiler.Flag.Bool("s3.allowEmptyFolder", true, "deprecated, ignored. Empty folder cleanup is now automatic.")
|
||||
filerS3Options.allowDeleteBucketNotEmpty = cmdFiler.Flag.Bool("s3.allowDeleteBucketNotEmpty", true, "allow recursive deleting all entries along with bucket")
|
||||
filerS3Options.allowDeleteBucketNotEmpty = cmdFiler.Flag.Bool("s3.allowDeleteBucketNotEmpty", false, "allow recursive deleting all entries along with bucket")
|
||||
filerS3Options.autoCreateBucket = cmdFiler.Flag.Bool("s3.autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
|
||||
filerS3Options.localSocket = cmdFiler.Flag.String("s3.localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
|
||||
filerS3Options.tlsCACertificate = cmdFiler.Flag.String("s3.cacert.file", "", "path to the TLS CA certificate file")
|
||||
|
||||
@@ -407,7 +407,7 @@ var (
|
||||
miniMetricsHttpIp = cmdMini.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
|
||||
miniS3Config = cmdMini.Flag.String("s3.config", "", "path to the S3 config file")
|
||||
miniIamConfig = cmdMini.Flag.String("s3.iam.config", "", "path to the advanced IAM config file for S3")
|
||||
miniS3AllowDeleteBucketNotEmpty = cmdMini.Flag.Bool("s3.allowDeleteBucketNotEmpty", true, "allow recursive deleting all entries along with bucket")
|
||||
miniS3AllowDeleteBucketNotEmpty = cmdMini.Flag.Bool("s3.allowDeleteBucketNotEmpty", false, "allow recursive deleting all entries along with bucket")
|
||||
miniS3AutoCreateBucket = cmdMini.Flag.Bool("s3.autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
|
||||
miniBucket = cmdMini.Flag.String("bucket", "", "comma-separated S3 bucket names to create on startup if they do not already exist; leave empty to skip. Falls back to S3_BUCKET env var.")
|
||||
miniTableBucket = cmdMini.Flag.String("tableBucket", "", "comma-separated S3 Tables buckets to create on startup if they do not already exist, each name[:FORMAT] with FORMAT one of ICEBERG (default) or LANCE, e.g. warehouse,vectors:LANCE; leave empty to skip. Falls back to S3_TABLE_BUCKET env var.")
|
||||
|
||||
+1
-1
@@ -114,7 +114,7 @@ func init() {
|
||||
s3StandaloneOptions.metricsHttpPort = cmdS3.Flag.Int("metricsPort", 0, "Prometheus metrics listen port")
|
||||
s3StandaloneOptions.metricsHttpIp = cmdS3.Flag.String("metricsIp", "", "metrics listen ip. If empty, default to same as -ip.bind option.")
|
||||
cmdS3.Flag.Bool("allowEmptyFolder", true, "deprecated, ignored. Empty folder cleanup is now automatic.")
|
||||
s3StandaloneOptions.allowDeleteBucketNotEmpty = cmdS3.Flag.Bool("allowDeleteBucketNotEmpty", true, "allow recursive deleting all entries along with bucket")
|
||||
s3StandaloneOptions.allowDeleteBucketNotEmpty = cmdS3.Flag.Bool("allowDeleteBucketNotEmpty", false, "allow recursive deleting all entries along with bucket")
|
||||
s3StandaloneOptions.autoCreateBucket = cmdS3.Flag.Bool("autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
|
||||
s3StandaloneOptions.localFilerSocket = cmdS3.Flag.String("localFilerSocket", "", "local filer socket path")
|
||||
s3StandaloneOptions.localSocket = cmdS3.Flag.String("localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
|
||||
|
||||
@@ -182,7 +182,7 @@ func init() {
|
||||
s3Options.iamConfig = cmdServer.Flag.String("s3.iam.config", "", "path to the advanced IAM config file for S3. Overrides -iam.config if both are provided.")
|
||||
s3Options.auditLogConfig = cmdServer.Flag.String("s3.auditLogConfig", "", "path to the audit log config file")
|
||||
cmdServer.Flag.Bool("s3.allowEmptyFolder", true, "deprecated, ignored. Empty folder cleanup is now automatic.")
|
||||
s3Options.allowDeleteBucketNotEmpty = cmdServer.Flag.Bool("s3.allowDeleteBucketNotEmpty", true, "allow recursive deleting all entries along with bucket")
|
||||
s3Options.allowDeleteBucketNotEmpty = cmdServer.Flag.Bool("s3.allowDeleteBucketNotEmpty", false, "allow recursive deleting all entries along with bucket")
|
||||
s3Options.autoCreateBucket = cmdServer.Flag.Bool("s3.autoCreateBucket", true, "create the bucket on upload if it does not exist, for admin identities only")
|
||||
s3Options.localSocket = cmdServer.Flag.String("s3.localSocket", "", "default to /tmp/seaweedfs-s3-<port>.sock")
|
||||
s3Options.bindIp = cmdServer.Flag.String("s3.ip.bind", "", "ip address to bind to. If empty, default to same as -ip.bind option.")
|
||||
|
||||
@@ -3,6 +3,7 @@ package filer
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -708,3 +709,172 @@ func TestPendingDeleteRecordKeepsItsTime(t *testing.T) {
|
||||
t.Fatalf("MpDl = %v, %v, %v; want 100", ts, found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A replayed directory delete must remove the local subtree too: the origin
|
||||
// only emits the directory's own key, and children the peer holds that the
|
||||
// origin never listed would orphan under the missing parent.
|
||||
func TestReplayDirectoryDeleteSweepsLocalChildren(t *testing.T) {
|
||||
store := newStubFilerStore()
|
||||
f := newReplayTestFiler(store)
|
||||
ctx := context.Background()
|
||||
|
||||
put := func(path string, isDir bool) {
|
||||
e := &Entry{FullPath: util.FullPath(path)}
|
||||
if isDir {
|
||||
e.Attr.Mode = os.ModeDir
|
||||
}
|
||||
if err := store.InsertEntry(ctx, e); err != nil {
|
||||
t.Fatalf("insert %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
put("/d", true)
|
||||
put("/d/a.txt", false)
|
||||
put("/d/sub", true)
|
||||
put("/d/sub/nested.txt", false)
|
||||
|
||||
event := &filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/",
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true},
|
||||
DeleteChunks: false,
|
||||
},
|
||||
TsNs: time.Now().UnixNano(),
|
||||
}
|
||||
if err := f.Replay(event); err != nil {
|
||||
t.Fatalf("replay: %v", err)
|
||||
}
|
||||
|
||||
for _, path := range []string{"/d", "/d/a.txt", "/d/sub", "/d/sub/nested.txt"} {
|
||||
if _, err := store.FindEntry(ctx, util.FullPath(path)); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("expected %s gone after replayed directory delete, got %v", path, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayFileDeleteKeepsSiblingEntries(t *testing.T) {
|
||||
store := newStubFilerStore()
|
||||
f := newReplayTestFiler(store)
|
||||
ctx := context.Background()
|
||||
|
||||
dir := &Entry{FullPath: "/d", Attr: Attr{Mode: os.ModeDir}}
|
||||
if err := store.InsertEntry(ctx, dir); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, name := range []string{"gone.txt", "kept.txt"} {
|
||||
if err := store.InsertEntry(ctx, &Entry{FullPath: util.NewFullPath("/d", name)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
event := &filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/d",
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
OldEntry: &filer_pb.Entry{Name: "gone.txt"},
|
||||
},
|
||||
TsNs: time.Now().UnixNano(),
|
||||
}
|
||||
if err := f.Replay(event); err != nil {
|
||||
t.Fatalf("replay: %v", err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, "/d/gone.txt"); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("expected deleted file gone, got %v", err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, "/d/kept.txt"); err != nil {
|
||||
t.Fatalf("expected sibling kept, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReplayDirectoryDeleteKeepsNewerChildren(t *testing.T) {
|
||||
store := newStubFilerStore()
|
||||
f := newReplayTestFiler(store)
|
||||
ctx := context.Background()
|
||||
|
||||
eventTs := time.Now().UnixNano()
|
||||
put := func(path string, isDir bool, mtime time.Time) {
|
||||
e := &Entry{FullPath: util.FullPath(path), Attr: Attr{Mtime: mtime}}
|
||||
if isDir {
|
||||
e.Attr.Mode = os.ModeDir
|
||||
}
|
||||
if err := store.InsertEntry(ctx, e); err != nil {
|
||||
t.Fatalf("insert %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
old := time.Unix(1600000000, 0)
|
||||
newer := time.Unix(0, eventTs).Add(time.Second)
|
||||
put("/d", true, old)
|
||||
put("/d/old.txt", false, old)
|
||||
put("/d/sub", true, old)
|
||||
put("/d/sub/newer.txt", false, newer)
|
||||
|
||||
event := &filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/",
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true},
|
||||
DeleteChunks: false,
|
||||
},
|
||||
TsNs: eventTs,
|
||||
}
|
||||
if err := f.Replay(event); err != nil {
|
||||
t.Fatalf("replay: %v", err)
|
||||
}
|
||||
|
||||
if _, err := store.FindEntry(ctx, "/d/old.txt"); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatalf("older child should be swept, got %v", err)
|
||||
}
|
||||
for _, path := range []string{"/d", "/d/sub", "/d/sub/newer.txt"} {
|
||||
if _, err := store.FindEntry(ctx, util.FullPath(path)); err != nil {
|
||||
t.Fatalf("newer child and its ancestors should survive: %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A directory kept for newer descendants still records the delete: a stale
|
||||
// create replayed later for another child must be fenced by it.
|
||||
func TestReplayKeptDirectoryDeleteFencesStaleCreates(t *testing.T) {
|
||||
store := newStubFilerStore()
|
||||
f := newReplayTestFiler(store)
|
||||
ctx := context.Background()
|
||||
|
||||
eventTs := time.Now().UnixNano()
|
||||
old := time.Unix(1600000000, 0)
|
||||
newer := time.Unix(0, eventTs).Add(time.Second)
|
||||
put := func(path string, isDir bool, mtime time.Time) {
|
||||
e := &Entry{FullPath: util.FullPath(path), Attr: Attr{Mtime: mtime}}
|
||||
if isDir {
|
||||
e.Attr.Mode = os.ModeDir
|
||||
}
|
||||
if err := store.InsertEntry(ctx, e); err != nil {
|
||||
t.Fatalf("insert %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
put("/d", true, old)
|
||||
put("/d/newer.txt", false, newer)
|
||||
|
||||
event := &filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/",
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
OldEntry: &filer_pb.Entry{Name: "d", IsDirectory: true},
|
||||
},
|
||||
TsNs: eventTs,
|
||||
}
|
||||
if err := f.Replay(event); err != nil {
|
||||
t.Fatalf("replay: %v", err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, "/d"); err != nil {
|
||||
t.Fatalf("newer child should keep the directory: %v", err)
|
||||
}
|
||||
|
||||
stale := &filer_pb.SubscribeMetadataResponse{
|
||||
Directory: "/d",
|
||||
EventNotification: &filer_pb.EventNotification{
|
||||
NewEntry: &filer_pb.Entry{Name: "late.txt"},
|
||||
},
|
||||
TsNs: eventTs - int64(time.Minute),
|
||||
}
|
||||
if err := f.Replay(stale); err != nil {
|
||||
t.Fatalf("replay stale create: %v", err)
|
||||
}
|
||||
if _, err := store.FindEntry(ctx, "/d/late.txt"); !errors.Is(err, filer_pb.ErrNotFound) {
|
||||
t.Fatal("stale create resurrected a child under a directory the origin deleted")
|
||||
}
|
||||
}
|
||||
|
||||
+113
-3
@@ -2,6 +2,7 @@ package filer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
@@ -78,10 +79,26 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
|
||||
return err
|
||||
}
|
||||
if !newer {
|
||||
glog.V(4).Infof("deleting %v", oldPath)
|
||||
if err := f.Store.DeleteEntry(ctx, oldPath); err != nil {
|
||||
return err
|
||||
// A directory delete on the origin removed the whole subtree, but
|
||||
// only the directory's own key replicates. Sweep local children the
|
||||
// origin never listed, or they orphan under a missing parent.
|
||||
keepDir := false
|
||||
if message.OldEntry.IsDirectory {
|
||||
var err error
|
||||
keepDir, err = f.deleteReplayedFolderChildren(ctx, oldPath, message.DeleteChunks, resp.TsNs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if !keepDir {
|
||||
glog.V(4).Infof("deleting %v", oldPath)
|
||||
if err := f.Store.DeleteEntry(ctx, oldPath); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// The origin deleted the directory even when a newer descendant
|
||||
// keeps its entry here — recording the delete fences stale child
|
||||
// creates that would otherwise resurrect under it.
|
||||
f.notePathChange(oldPath, resp.TsNs, true)
|
||||
}
|
||||
}
|
||||
@@ -104,6 +121,99 @@ func (f *Filer) Replay(resp *filer_pb.SubscribeMetadataResponse) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// deleteReplayedFolderChildren removes local entries under dir, and queues
|
||||
// the chunks of removed files for deletion. Children this filer committed
|
||||
// after the event stay — a newer local write wins over a delayed delete —
|
||||
// along with the directories that hold them.
|
||||
func (f *Filer) deleteReplayedFolderChildren(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64) (bool, error) {
|
||||
sweep := &folderSweep{chunksByPath: make(map[util.FullPath][]*filer_pb.FileChunk)}
|
||||
kept, err := f.sweepReplayedFolder(ctx, dir, deleteChunks, tsNs, sweep)
|
||||
// Flush what was collected even on error: the entries already deleted
|
||||
// cannot be listed again on retry, so their chunks must be queued now.
|
||||
sweep.flush(ctx, f)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
// folderSweep accumulates the chunks of removed children for deletion. It
|
||||
// flushes in batches so a large subtree cannot hold every chunk reference in
|
||||
// memory at once, and so chunk queueing overlaps the listing work.
|
||||
type folderSweep struct {
|
||||
chunksByPath map[util.FullPath][]*filer_pb.FileChunk
|
||||
count int
|
||||
}
|
||||
|
||||
const folderSweepChunkBatch = 4096
|
||||
|
||||
func (s *folderSweep) add(path util.FullPath, chunks []*filer_pb.FileChunk) {
|
||||
s.chunksByPath[path] = append(s.chunksByPath[path], chunks...)
|
||||
s.count += len(chunks)
|
||||
}
|
||||
|
||||
func (s *folderSweep) flush(ctx context.Context, f *Filer) {
|
||||
for path, chunks := range s.chunksByPath {
|
||||
f.DeleteChunks(ctx, path, chunks)
|
||||
}
|
||||
s.chunksByPath = make(map[util.FullPath][]*filer_pb.FileChunk)
|
||||
s.count = 0
|
||||
}
|
||||
|
||||
// sweepReplayedFolder deletes every child of dir whose local commit is not
|
||||
// newer than the event, working subtree-first so nested entries go too.
|
||||
// It reports whether a newer descendant keeps dir alive — the caller then
|
||||
// keeps the dir entry itself.
|
||||
func (f *Filer) sweepReplayedFolder(ctx context.Context, dir util.FullPath, deleteChunks bool, tsNs int64, sweep *folderSweep) (kept bool, err error) {
|
||||
lastFileName := ""
|
||||
for {
|
||||
entries, _, lerr := f.ListDirectoryEntries(ctx, dir, lastFileName, false, PaginationSize, "", "", "")
|
||||
if lerr != nil {
|
||||
return false, fmt.Errorf("list folder %s: %w", dir, lerr)
|
||||
}
|
||||
for _, sub := range entries {
|
||||
lastFileName = sub.Name()
|
||||
childKept := false
|
||||
if tsNs > 0 {
|
||||
localTsNs, tsErr := f.localCommitTsNs(ctx, sub.FullPath)
|
||||
if tsErr != nil {
|
||||
return false, tsErr
|
||||
}
|
||||
childKept = localTsNs > tsNs
|
||||
}
|
||||
if !childKept && sub.IsDirectory() {
|
||||
childKept, err = f.sweepReplayedFolder(ctx, sub.FullPath, deleteChunks, tsNs, sweep)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
if childKept {
|
||||
kept = true
|
||||
continue
|
||||
}
|
||||
if derr := f.Store.DeleteEntry(ctx, sub.FullPath); derr != nil {
|
||||
return false, fmt.Errorf("delete %s: %w", sub.FullPath, derr)
|
||||
}
|
||||
f.notePathChange(sub.FullPath, tsNs, true)
|
||||
// hard-linked chunks are owned by the shared inode record; the
|
||||
// origin's own events decide their deletion. A directory entry
|
||||
// can still carry chunks of its own — an S3 child promotes an
|
||||
// existing file to a directory without dropping them — so the
|
||||
// entry's chunks are keyed to its own path whatever it is.
|
||||
if deleteChunks && len(sub.HardLinkId) == 0 && len(sub.GetChunks()) > 0 {
|
||||
sweep.add(sub.FullPath, sub.GetChunks())
|
||||
if sweep.count >= folderSweepChunkBatch {
|
||||
sweep.flush(ctx, f)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(entries) < PaginationSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
// replayBlockedByLocal reports whether the local store already holds a
|
||||
// version of path committed after the event's timestamp. Lookup failures are
|
||||
// propagated: a replayed change must not be applied on the guess that the
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
@@ -563,8 +564,9 @@ func (f *FilerOIDCProviderStore) UpdateProvider(ctx context.Context, filerAddres
|
||||
func (f *FilerOIDCProviderStore) deleteProviderEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry) (bool, error) {
|
||||
fullPath := f.basePath + "/" + entry.Name
|
||||
resp, err := client.ObjectTransaction(ctx, &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: fullPath,
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath,
|
||||
LockKey: fullPath,
|
||||
RequestId: uuid.NewString(),
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath,
|
||||
Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{
|
||||
Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: entry,
|
||||
}}},
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/karlseguin/ccache/v2"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/iam/policy"
|
||||
@@ -380,8 +381,9 @@ func (f *FilerRoleStore) UpdateRole(ctx context.Context, filerAddress string, ro
|
||||
func (f *FilerRoleStore) deleteRoleEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, entry *filer_pb.Entry) (bool, error) {
|
||||
fullPath := f.basePath + "/" + entry.Name
|
||||
resp, err := client.ObjectTransaction(ctx, &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: fullPath,
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath,
|
||||
LockKey: fullPath,
|
||||
RequestId: uuid.NewString(),
|
||||
RouteKey: s3_constants.ObjectWriteRouteKeyPrefix + fullPath,
|
||||
Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{
|
||||
Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: entry,
|
||||
}}},
|
||||
|
||||
@@ -392,6 +392,8 @@ message ObjectTransactionRequest {
|
||||
string condition_key = 6; // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key)
|
||||
string route_key = 7; // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
|
||||
bool is_moved = 8; // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
|
||||
string request_id = 9; // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying
|
||||
string condition_alt_key = 10; // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version)
|
||||
}
|
||||
|
||||
message ObjectTransactionResponse {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
// Code generated by protoc-gen-go. DO NOT EDIT.
|
||||
// versions:
|
||||
// protoc-gen-go v1.36.6
|
||||
// protoc v3.21.12
|
||||
// protoc v7.35.0
|
||||
// source: filer.proto
|
||||
|
||||
package filer_pb
|
||||
@@ -1811,9 +1811,11 @@ type ObjectTransactionRequest struct {
|
||||
Mutations []*ObjectMutation `protobuf:"bytes,3,rep,name=mutations,proto3" json:"mutations,omitempty"`
|
||||
IsFromOtherCluster bool `protobuf:"varint,4,opt,name=is_from_other_cluster,json=isFromOtherCluster,proto3" json:"is_from_other_cluster,omitempty"`
|
||||
Signatures []int32 `protobuf:"varint,5,rep,packed,name=signatures,proto3" json:"signatures,omitempty"`
|
||||
ConditionKey string `protobuf:"bytes,6,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key)
|
||||
RouteKey string `protobuf:"bytes,7,opt,name=route_key,json=routeKey,proto3" json:"route_key,omitempty"` // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
|
||||
IsMoved bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
|
||||
ConditionKey string `protobuf:"bytes,6,opt,name=condition_key,json=conditionKey,proto3" json:"condition_key,omitempty"` // if set, evaluate the condition against this entry instead of lock_key (still locking lock_key)
|
||||
RouteKey string `protobuf:"bytes,7,opt,name=route_key,json=routeKey,proto3" json:"route_key,omitempty"` // ring key identifying the owner filer; a non-owner forwards the whole transaction to it
|
||||
IsMoved bool `protobuf:"varint,8,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded transaction so the receiver applies it locally instead of forwarding again
|
||||
RequestId string `protobuf:"bytes,9,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` // idempotency token: the owner records the outcome and replays it on a retried request instead of re-applying
|
||||
ConditionAltKey string `protobuf:"bytes,10,opt,name=condition_alt_key,json=conditionAltKey,proto3" json:"condition_alt_key,omitempty"` // if the condition's entry is absent, evaluate it against this entry instead (e.g. null slot, else latest version)
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -1904,6 +1906,20 @@ func (x *ObjectTransactionRequest) GetIsMoved() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (x *ObjectTransactionRequest) GetRequestId() string {
|
||||
if x != nil {
|
||||
return x.RequestId
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *ObjectTransactionRequest) GetConditionAltKey() string {
|
||||
if x != nil {
|
||||
return x.ConditionAltKey
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type ObjectTransactionResponse struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
|
||||
@@ -7433,7 +7449,7 @@ const file_filer_proto_rawDesc = "" +
|
||||
"\fexclude_name\x18\t \x01(\tR\vexcludeName\x1a?\n" +
|
||||
"\x11CopyExtendedEntry\x12\x10\n" +
|
||||
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xd5\x02\n" +
|
||||
"\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xa0\x03\n" +
|
||||
"\x18ObjectTransactionRequest\x12\x19\n" +
|
||||
"\block_key\x18\x01 \x01(\tR\alockKey\x126\n" +
|
||||
"\tcondition\x18\x02 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x126\n" +
|
||||
@@ -7444,7 +7460,11 @@ const file_filer_proto_rawDesc = "" +
|
||||
"signatures\x12#\n" +
|
||||
"\rcondition_key\x18\x06 \x01(\tR\fconditionKey\x12\x1b\n" +
|
||||
"\troute_key\x18\a \x01(\tR\brouteKey\x12\x19\n" +
|
||||
"\bis_moved\x18\b \x01(\bR\aisMoved\"f\n" +
|
||||
"\bis_moved\x18\b \x01(\bR\aisMoved\x12\x1d\n" +
|
||||
"\n" +
|
||||
"request_id\x18\t \x01(\tR\trequestId\x12*\n" +
|
||||
"\x11condition_alt_key\x18\n" +
|
||||
" \x01(\tR\x0fconditionAltKey\"f\n" +
|
||||
"\x19ObjectTransactionResponse\x12\x14\n" +
|
||||
"\x05error\x18\x01 \x01(\tR\x05error\x123\n" +
|
||||
"\n" +
|
||||
|
||||
@@ -1567,6 +1567,20 @@ func (m *ObjectTransactionRequest) MarshalToSizedBufferVT(dAtA []byte) (int, err
|
||||
i -= len(m.unknownFields)
|
||||
copy(dAtA[i:], m.unknownFields)
|
||||
}
|
||||
if len(m.ConditionAltKey) > 0 {
|
||||
i -= len(m.ConditionAltKey)
|
||||
copy(dAtA[i:], m.ConditionAltKey)
|
||||
i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.ConditionAltKey)))
|
||||
i--
|
||||
dAtA[i] = 0x52
|
||||
}
|
||||
if len(m.RequestId) > 0 {
|
||||
i -= len(m.RequestId)
|
||||
copy(dAtA[i:], m.RequestId)
|
||||
i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.RequestId)))
|
||||
i--
|
||||
dAtA[i] = 0x4a
|
||||
}
|
||||
if m.IsMoved {
|
||||
i--
|
||||
if m.IsMoved {
|
||||
@@ -7188,6 +7202,14 @@ func (m *ObjectTransactionRequest) SizeVT() (n int) {
|
||||
if m.IsMoved {
|
||||
n += 2
|
||||
}
|
||||
l = len(m.RequestId)
|
||||
if l > 0 {
|
||||
n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
|
||||
}
|
||||
l = len(m.ConditionAltKey)
|
||||
if l > 0 {
|
||||
n += 1 + l + protohelpers.SizeOfVarint(uint64(l))
|
||||
}
|
||||
n += len(m.unknownFields)
|
||||
return n
|
||||
}
|
||||
@@ -13841,6 +13863,70 @@ func (m *ObjectTransactionRequest) UnmarshalVT(dAtA []byte) error {
|
||||
}
|
||||
}
|
||||
m.IsMoved = bool(v != 0)
|
||||
case 9:
|
||||
if wireType != 2 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field RequestId", wireType)
|
||||
}
|
||||
var stringLen uint64
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return protohelpers.ErrIntOverflow
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
stringLen |= uint64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
intStringLen := int(stringLen)
|
||||
if intStringLen < 0 {
|
||||
return protohelpers.ErrInvalidLength
|
||||
}
|
||||
postIndex := iNdEx + intStringLen
|
||||
if postIndex < 0 {
|
||||
return protohelpers.ErrInvalidLength
|
||||
}
|
||||
if postIndex > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
m.RequestId = string(dAtA[iNdEx:postIndex])
|
||||
iNdEx = postIndex
|
||||
case 10:
|
||||
if wireType != 2 {
|
||||
return fmt.Errorf("proto: wrong wireType = %d for field ConditionAltKey", wireType)
|
||||
}
|
||||
var stringLen uint64
|
||||
for shift := uint(0); ; shift += 7 {
|
||||
if shift >= 64 {
|
||||
return protohelpers.ErrIntOverflow
|
||||
}
|
||||
if iNdEx >= l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
b := dAtA[iNdEx]
|
||||
iNdEx++
|
||||
stringLen |= uint64(b&0x7F) << shift
|
||||
if b < 0x80 {
|
||||
break
|
||||
}
|
||||
}
|
||||
intStringLen := int(stringLen)
|
||||
if intStringLen < 0 {
|
||||
return protohelpers.ErrInvalidLength
|
||||
}
|
||||
postIndex := iNdEx + intStringLen
|
||||
if postIndex < 0 {
|
||||
return protohelpers.ErrInvalidLength
|
||||
}
|
||||
if postIndex > l {
|
||||
return io.ErrUnexpectedEOF
|
||||
}
|
||||
m.ConditionAltKey = string(dAtA[iNdEx:postIndex])
|
||||
iNdEx = postIndex
|
||||
default:
|
||||
iNdEx = preIndex
|
||||
skippy, err := protohelpers.Skip(dAtA[iNdEx:])
|
||||
|
||||
@@ -2916,16 +2916,12 @@ func (iam *IdentityAccessManagement) bucketsNamedByAttachedPolicies(r *http.Requ
|
||||
// its query describe. versionId, when set, and the STS session token ride along
|
||||
// for policy conditions. Returns ErrNone when auth is disabled (checked by the
|
||||
// callers) or the identity is an admin.
|
||||
func (iam *IdentityAccessManagement) authorizeObjectKeyAction(r *http.Request, identity *Identity, method string, action Action, bucket, objectKey, versionId string) s3err.ErrorCode {
|
||||
if identity == nil {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
if identity.isAdmin() {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
|
||||
// Shallow copy: authorization only reads headers, so sharing the header map
|
||||
// with the original request is safe, and this can run once per key.
|
||||
// objectKeyPolicyRequest synthesizes a <method> /<bucket>/<objectKey> request
|
||||
// for policy evaluation against a key the real URL does not name. The query is
|
||||
// built from scratch so a param on the real request (delete, uploadId,
|
||||
// partNumber, ...) cannot steer ResolveS3Action off the intended action; the
|
||||
// STS token and versionId ride along for policy conditions.
|
||||
func objectKeyPolicyRequest(r *http.Request, method, bucket, objectKey, versionId string) *http.Request {
|
||||
keyReq := new(http.Request)
|
||||
*keyReq = *r
|
||||
keyURL := &url.URL{
|
||||
@@ -2933,9 +2929,6 @@ func (iam *IdentityAccessManagement) authorizeObjectKeyAction(r *http.Request, i
|
||||
Host: r.URL.Host,
|
||||
Path: "/" + bucket + "/" + objectKey,
|
||||
}
|
||||
// Build the query from scratch so a param on the real request (delete,
|
||||
// uploadId, partNumber, ...) cannot steer ResolveS3Action off the intended
|
||||
// action; keep the STS token and per-key versionId for policy conditions.
|
||||
keyQuery := make(url.Values)
|
||||
if versionId != "" {
|
||||
keyQuery.Set("versionId", versionId)
|
||||
@@ -2954,6 +2947,18 @@ func (iam *IdentityAccessManagement) authorizeObjectKeyAction(r *http.Request, i
|
||||
keyReq.Body = nil
|
||||
keyReq.GetBody = nil
|
||||
keyReq.ContentLength = 0
|
||||
return keyReq
|
||||
}
|
||||
|
||||
func (iam *IdentityAccessManagement) authorizeObjectKeyAction(r *http.Request, identity *Identity, method string, action Action, bucket, objectKey, versionId string) s3err.ErrorCode {
|
||||
if identity == nil {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
if identity.isAdmin() {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
|
||||
keyReq := objectKeyPolicyRequest(r, method, bucket, objectKey, versionId)
|
||||
|
||||
if iam.policyEngine != nil {
|
||||
principal := buildPrincipalARN(identity, keyReq)
|
||||
|
||||
@@ -4,6 +4,9 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
@@ -257,3 +260,112 @@ func TestHasPolicyForBucket(t *testing.T) {
|
||||
// No longer has policy
|
||||
assert.False(t, engine.HasPolicyForBucket("test-bucket"))
|
||||
}
|
||||
|
||||
// TestRecheckCopySourcePolicy verifies that a tag-conditioned Deny on
|
||||
// s3:GetObject also blocks CopyObject-style reads of the source object. The
|
||||
// destination request keeps its PUT method while the policy is evaluated
|
||||
// against a synthesized GET on the source key.
|
||||
func TestRecheckCopySourcePolicy(t *testing.T) {
|
||||
s3a := &S3ApiServer{
|
||||
policyEngine: NewBucketPolicyEngine(),
|
||||
}
|
||||
|
||||
policyJSON := `{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [
|
||||
{
|
||||
"Effect": "Allow",
|
||||
"Principal": "*",
|
||||
"Action": "s3:*",
|
||||
"Resource": ["arn:aws:s3:::src-bucket", "arn:aws:s3:::src-bucket/*", "arn:aws:s3:::dst-bucket", "arn:aws:s3:::dst-bucket/*"]
|
||||
},
|
||||
{
|
||||
"Effect": "Deny",
|
||||
"Principal": "*",
|
||||
"Action": "s3:GetObject",
|
||||
"Resource": "arn:aws:s3:::src-bucket/*",
|
||||
"Condition": {
|
||||
"StringEquals": {
|
||||
"s3:ExistingObjectTag/status": ["private"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}`
|
||||
require.NoError(t, s3a.policyEngine.engine.SetBucketPolicy("src-bucket", policyJSON))
|
||||
|
||||
identity := &Identity{Name: "denied-user", PrincipalArn: "arn:aws:iam::123456789012:user/denied"}
|
||||
|
||||
req := httptest.NewRequest(http.MethodPut, "/dst-bucket/dst-key", nil)
|
||||
req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), identity))
|
||||
|
||||
privateEntry := map[string][]byte{
|
||||
s3_constants.AmzObjectTaggingPrefix + "status": []byte("private"),
|
||||
}
|
||||
assert.Equal(t, s3err.ErrAccessDenied,
|
||||
s3a.recheckCopySourcePolicy(req, "src-bucket", "src-key", "", privateEntry, "test"))
|
||||
|
||||
publicEntry := map[string][]byte{
|
||||
s3_constants.AmzObjectTaggingPrefix + "status": []byte("public"),
|
||||
}
|
||||
assert.Equal(t, s3err.ErrNone,
|
||||
s3a.recheckCopySourcePolicy(req, "src-bucket", "src-key", "", publicEntry, "test"))
|
||||
}
|
||||
|
||||
// A source read whose only grant is conditioned on s3:ExistingObjectTag denies
|
||||
// in the tagless first phase; the deferred lookup recheck must still grant it
|
||||
// when the resolved entry's tags satisfy the Allow, and keep denying when they
|
||||
// do not. The same recheck with a synthesized DELETE is what a rename applies
|
||||
// to the source removal.
|
||||
func TestRecheckPolicyForKeyWithLookup(t *testing.T) {
|
||||
filerImpl := &ambiguousPutFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/src-bucket/src-key": {
|
||||
Name: "src-key",
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.AmzObjectTaggingPrefix + "status": []byte("public"),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
|
||||
s3a.policyEngine = NewBucketPolicyEngine()
|
||||
s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
|
||||
s3a.bucketConfigCache.Set("src-bucket", &BucketConfig{Name: "src-bucket"})
|
||||
|
||||
policyJSON := `{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [
|
||||
{
|
||||
"Effect": "Allow",
|
||||
"Principal": "*",
|
||||
"Action": "s3:GetObject",
|
||||
"Resource": "arn:aws:s3:::src-bucket/*",
|
||||
"Condition": {
|
||||
"StringEquals": {
|
||||
"s3:ExistingObjectTag/status": ["public"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}`
|
||||
require.NoError(t, s3a.policyEngine.engine.SetBucketPolicy("src-bucket", policyJSON))
|
||||
|
||||
identity := &Identity{Name: "tagged-user", PrincipalArn: "arn:aws:iam::123456789012:user/tagged"}
|
||||
req := httptest.NewRequest(http.MethodPut, "/dst-bucket/dst-key", nil)
|
||||
req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), identity))
|
||||
|
||||
assert.Equal(t, s3err.ErrNone,
|
||||
s3a.recheckPolicyForKeyWithLookup(req, http.MethodGet, "src-bucket", "src-key", "", string(s3_constants.ACTION_READ), "test"),
|
||||
"a tag-conditioned Allow on the resolved entry must still grant the read")
|
||||
|
||||
filerImpl.entries["/buckets/src-bucket/src-key"].Extended[s3_constants.AmzObjectTaggingPrefix+"status"] = []byte("private")
|
||||
assert.Equal(t, s3err.ErrAccessDenied,
|
||||
s3a.recheckPolicyForKeyWithLookup(req, http.MethodGet, "src-bucket", "src-key", "", string(s3_constants.ACTION_READ), "test"),
|
||||
"non-matching tags must keep the first-phase deny")
|
||||
|
||||
// Without a policy on the bucket the recheck must not widen anything.
|
||||
require.NoError(t, s3a.policyEngine.engine.DeleteBucketPolicy("src-bucket"))
|
||||
assert.Equal(t, s3err.ErrAccessDenied,
|
||||
s3a.recheckPolicyForKeyWithLookup(req, http.MethodGet, "src-bucket", "src-key", "", string(s3_constants.ACTION_READ), "test"))
|
||||
}
|
||||
|
||||
@@ -6,8 +6,10 @@ import (
|
||||
"crypto/rsa"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -148,3 +150,159 @@ func TestOIDCBearerEnforcesRoleTrustPolicy(t *testing.T) {
|
||||
require.Equal(t, s3err.ErrNone, errCode)
|
||||
assert.Equal(t, "arn:aws:iam::role/BearerTrustedRole", identity.Principal)
|
||||
}
|
||||
|
||||
// An OIDC bearer token's aws:userid claim is server-asserted from the
|
||||
// verified (sub, iss) pair: a token-supplied value is overwritten, and a
|
||||
// token without a subject is rejected outright.
|
||||
func TestOIDCBearerUserIDIsIssuerScoped(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
key, err := rsa.GenerateKey(rand.Reader, 2048)
|
||||
require.NoError(t, err)
|
||||
|
||||
jwks := map[string]interface{}{
|
||||
"keys": []map[string]interface{}{{
|
||||
"kty": "RSA",
|
||||
"kid": "bearer-test-key",
|
||||
"use": "sig",
|
||||
"alg": "RS256",
|
||||
"n": base64.RawURLEncoding.EncodeToString(key.PublicKey.N.Bytes()),
|
||||
"e": "AQAB",
|
||||
}},
|
||||
}
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/jwks" {
|
||||
json.NewEncoder(w).Encode(jwks)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
provider := oidc.NewOIDCProvider("test-oidc")
|
||||
require.NoError(t, provider.Initialize(&oidc.OIDCConfig{
|
||||
Issuer: server.URL,
|
||||
ClientID: "test-client",
|
||||
JWKSUri: server.URL + "/jwks",
|
||||
RoleMapping: &providers.RoleMapping{
|
||||
Rules: []providers.MappingRule{
|
||||
{Claim: "groups", Value: "trusted-group", Role: "arn:aws:iam::role/BearerTrustedRole"},
|
||||
},
|
||||
},
|
||||
}))
|
||||
|
||||
iamManager := integration.NewIAMManager()
|
||||
require.NoError(t, iamManager.Initialize(&integration.IAMConfig{
|
||||
STS: &sts.STSConfig{
|
||||
TokenDuration: sts.FlexibleDuration{Duration: time.Hour},
|
||||
MaxSessionLength: sts.FlexibleDuration{Duration: 12 * time.Hour},
|
||||
Issuer: "test-sts",
|
||||
SigningKey: []byte("test-signing-key-32-characters-long"),
|
||||
},
|
||||
Policy: &policy.PolicyEngineConfig{DefaultEffect: "Deny", StoreType: "memory"},
|
||||
Roles: &integration.RoleStoreConfig{StoreType: "memory"},
|
||||
}, func() string { return "localhost:8888" }))
|
||||
require.NoError(t, iamManager.RegisterIdentityProvider(provider))
|
||||
require.NoError(t, iamManager.CreatePolicy(ctx, "", "BearerTestPolicy", &policy.PolicyDocument{
|
||||
Version: "2012-10-17",
|
||||
Statement: []policy.Statement{{
|
||||
Effect: "Allow",
|
||||
Action: []string{"s3:*"},
|
||||
Resource: []string{"arn:aws:s3:::*", "arn:aws:s3:::*/*"},
|
||||
}},
|
||||
}))
|
||||
require.NoError(t, iamManager.CreateRole(ctx, "", "BearerTrustedRole", &integration.RoleDefinition{
|
||||
RoleName: "BearerTrustedRole",
|
||||
TrustPolicy: &policy.PolicyDocument{
|
||||
Version: "2012-10-17",
|
||||
Statement: []policy.Statement{{
|
||||
Effect: "Allow",
|
||||
Principal: map[string]interface{}{"Federated": "test-oidc"},
|
||||
Action: []string{"sts:AssumeRoleWithWebIdentity"},
|
||||
}},
|
||||
},
|
||||
AttachedPolicies: []string{"BearerTestPolicy"},
|
||||
}))
|
||||
|
||||
sign := func(extra map[string]interface{}) string {
|
||||
claims := jwt.MapClaims{
|
||||
"iss": server.URL,
|
||||
"aud": "test-client",
|
||||
"exp": time.Now().Add(time.Hour).Unix(),
|
||||
"iat": time.Now().Unix(),
|
||||
"groups": []string{"trusted-group"},
|
||||
}
|
||||
for k, v := range extra {
|
||||
claims[k] = v
|
||||
}
|
||||
token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims)
|
||||
token.Header["kid"] = "bearer-test-key"
|
||||
signed, err := token.SignedString(key)
|
||||
require.NoError(t, err)
|
||||
return signed
|
||||
}
|
||||
|
||||
s3iam := NewS3IAMIntegration(iamManager, "localhost:8888")
|
||||
authenticate := func(token string) (*IAMIdentity, s3err.ErrorCode) {
|
||||
req := httptest.NewRequest("GET", "/test-bucket/test-object", http.NoBody)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
return s3iam.AuthenticateJWT(ctx, req)
|
||||
}
|
||||
|
||||
identity, errCode := authenticate(sign(map[string]interface{}{
|
||||
"sub": "bearer-user",
|
||||
"aws:userid": "admin",
|
||||
}))
|
||||
require.Equal(t, s3err.ErrNone, errCode)
|
||||
assert.Equal(t,
|
||||
sts.ComputeParentUser("bearer-user", server.URL),
|
||||
identity.Claims["aws:userid"],
|
||||
"the token-supplied aws:userid must be overwritten with the verified (sub, iss) hash")
|
||||
assert.Equal(t, "bearer-user", identity.Claims["sub"])
|
||||
assert.Equal(t, server.URL, identity.Claims["iss"])
|
||||
assert.Equal(t, server.URL, identity.Claims["seaweed:oidc-issuers"],
|
||||
"the registered provider's issuer must be asserted as trusted for legacy-owner anchoring")
|
||||
|
||||
_, errCode = authenticate(sign(map[string]interface{}{"sub": ""}))
|
||||
assert.Equal(t, s3err.ErrAccessDenied, errCode, "a token with no subject must not authenticate")
|
||||
}
|
||||
|
||||
// flakyListOIDCStore fails ListProviders on demand, standing in for a
|
||||
// filer-backed provider store that cannot be read.
|
||||
type flakyListOIDCStore struct {
|
||||
integration.OIDCProviderStore
|
||||
fail atomic.Bool
|
||||
}
|
||||
|
||||
func (s *flakyListOIDCStore) ListProviders(ctx context.Context, filerAddress string) ([]*integration.OIDCProviderRecord, error) {
|
||||
if s.fail.Load() {
|
||||
return nil, fmt.Errorf("list OIDC providers: simulated outage")
|
||||
}
|
||||
return s.OIDCProviderStore.ListProviders(ctx, filerAddress)
|
||||
}
|
||||
|
||||
// A failed store listing must not pass the registered providers off as the
|
||||
// complete trusted set: hidden issuers could shrink the reported set to one
|
||||
// entry, the only shape under which a legacy bare subject re-anchors.
|
||||
func TestTrustedOIDCIssuersFailsClosedOnStoreError(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
iamManager := integration.NewIAMManager()
|
||||
s3iam := NewS3IAMIntegration(iamManager, "")
|
||||
|
||||
// No provider store at all: registered providers are the complete set —
|
||||
// there is nowhere an issuer could hide behind a listing failure.
|
||||
assert.Empty(t, s3iam.trustedOIDCIssuers(ctx))
|
||||
|
||||
mem := integration.NewMemoryOIDCProviderStore()
|
||||
require.NoError(t, mem.StoreProvider(ctx, "", &integration.OIDCProviderRecord{
|
||||
ARN: "arn:aws:iam::111122223333:oidc-provider/idp.example.com",
|
||||
URL: "https://idp.example.com",
|
||||
}))
|
||||
store := &flakyListOIDCStore{OIDCProviderStore: mem}
|
||||
iamManager.SetOIDCProviderStore(store)
|
||||
|
||||
assert.Equal(t, []string{"https://idp.example.com"}, s3iam.trustedOIDCIssuers(ctx))
|
||||
|
||||
store.fail.Store(true)
|
||||
assert.Empty(t, s3iam.trustedOIDCIssuers(ctx),
|
||||
"a failed listing must report no trusted issuers, not the partial registered set")
|
||||
}
|
||||
|
||||
@@ -126,6 +126,12 @@ func (s3iam *S3IAMIntegration) AuthenticateJWT(ctx context.Context, r *http.Requ
|
||||
return nil, s3err.ErrAccessDenied
|
||||
}
|
||||
|
||||
// A token without a subject has no issuer-scoped principal; letting it
|
||||
// through would make the client-supplied claims below stand in for one.
|
||||
if identity.UserID == "" {
|
||||
return nil, s3err.ErrAccessDenied
|
||||
}
|
||||
|
||||
// Extract role from OIDC identity
|
||||
if identity.RoleArn == "" {
|
||||
return nil, s3err.ErrAccessDenied
|
||||
@@ -162,6 +168,22 @@ func (s3iam *S3IAMIntegration) AuthenticateJWT(ctx context.Context, r *http.Requ
|
||||
// Set critical claims explicitly, overwriting any from attributes to ensure correctness.
|
||||
claims["sub"] = identity.UserID
|
||||
claims["role"] = identity.RoleArn
|
||||
claims["iss"] = identity.Issuer
|
||||
|
||||
// The bare sub collides across issuers; aws:userid carries the same
|
||||
// (sub, iss) hash STS sessions already expose, so downstream ownership
|
||||
// keys stay namespaced per identity provider. A claim the token itself
|
||||
// supplies is always scrubbed, never adopted.
|
||||
if parentUser := sts.ComputeParentUser(identity.UserID, identity.Issuer); parentUser != "" {
|
||||
claims["aws:userid"] = parentUser
|
||||
} else {
|
||||
delete(claims, "aws:userid")
|
||||
}
|
||||
if issuers := s3iam.trustedOIDCIssuers(ctx); len(issuers) > 0 {
|
||||
claims["seaweed:oidc-issuers"] = strings.Join(issuers, " ")
|
||||
} else {
|
||||
delete(claims, "seaweed:oidc-issuers")
|
||||
}
|
||||
|
||||
// Use real email address if available
|
||||
emailAddress := identity.UserID + "@oidc.local"
|
||||
@@ -425,6 +447,7 @@ type OIDCIdentity struct {
|
||||
UserID string
|
||||
RoleArn string
|
||||
Provider string
|
||||
Issuer string
|
||||
Email string
|
||||
DisplayName string
|
||||
Groups []string
|
||||
@@ -488,6 +511,7 @@ func (s3iam *S3IAMIntegration) validateExternalOIDCToken(ctx context.Context, to
|
||||
UserID: externalIdentity.UserID,
|
||||
RoleArn: roleArn,
|
||||
Provider: fmt.Sprintf("%T", provider), // Use provider type as identifier
|
||||
Issuer: externalIdentity.Issuer,
|
||||
Email: externalIdentity.Email,
|
||||
DisplayName: externalIdentity.DisplayName,
|
||||
Groups: externalIdentity.Groups,
|
||||
@@ -495,6 +519,54 @@ func (s3iam *S3IAMIntegration) validateExternalOIDCToken(ctx context.Context, to
|
||||
}, nil
|
||||
}
|
||||
|
||||
// trustedOIDCIssuers returns the issuer URLs of the OIDC providers the
|
||||
// deployment is configured to trust. It is what a legacy (pre-scoping) bare
|
||||
// subject can be re-anchored to: exactly one issuer makes the mapping
|
||||
// unambiguous; anything else leaves it unknown, which callers must treat as
|
||||
// fail-closed rather than guessing from the caller's issuer.
|
||||
//
|
||||
// The list is only safe while complete. Without a provider store the
|
||||
// registered providers are the whole set — there is nowhere an issuer could
|
||||
// hide. With a store, a failed listing leaves the real set unknown, and
|
||||
// passing the registered subset off as complete could collapse a
|
||||
// multi-issuer deployment to a single issuer: the one shape a bare subject
|
||||
// re-anchors under, so a caller from that issuer could claim a stored
|
||||
// subject that belongs to a provider hidden behind the error. Fail closed.
|
||||
func (s3iam *S3IAMIntegration) trustedOIDCIssuers(ctx context.Context) []string {
|
||||
if s3iam.iamManager == nil {
|
||||
return nil
|
||||
}
|
||||
seen := make(map[string]bool)
|
||||
var issuers []string
|
||||
// Configured identity providers registered with STS cover deployments
|
||||
// whose OIDC setup lives in config rather than the provider store.
|
||||
if stsService := s3iam.iamManager.GetSTSService(); stsService != nil {
|
||||
for _, p := range stsService.GetProviders() {
|
||||
if issuer, ok := p.(interface{ GetIssuer() string }); ok {
|
||||
if iss := issuer.GetIssuer(); iss != "" && !seen[iss] {
|
||||
seen[iss] = true
|
||||
issuers = append(issuers, iss)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if s3iam.iamManager.GetOIDCProviderStore() == nil {
|
||||
return issuers
|
||||
}
|
||||
records, err := s3iam.iamManager.ListOIDCProviders(ctx)
|
||||
if err != nil {
|
||||
glog.V(2).Infof("trustedOIDCIssuers: %v", err)
|
||||
return nil
|
||||
}
|
||||
for _, rec := range records {
|
||||
if rec.URL != "" && !seen[rec.URL] {
|
||||
seen[rec.URL] = true
|
||||
issuers = append(issuers, rec.URL)
|
||||
}
|
||||
}
|
||||
return issuers
|
||||
}
|
||||
|
||||
// selectPrimaryRole simply picks the first role from the list
|
||||
// The OIDC provider should return roles in priority order (most important first)
|
||||
func (s3iam *S3IAMIntegration) selectPrimaryRole(roles []string, externalIdentity *providers.ExternalIdentity) string {
|
||||
|
||||
@@ -3,10 +3,13 @@ package s3api
|
||||
import (
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
@@ -180,3 +183,31 @@ func TestAuthorizeObjectKeyActionBucketAllowDoesNotOverrideIdentityDeny(t *testi
|
||||
iam.AuthorizeCopySource(req, reader, bpdBucket, "other/data.txt", ""),
|
||||
"AuthorizeCopySource must preserve the bucket Allow over an implicit identity denial")
|
||||
}
|
||||
|
||||
// A deferred tag-aware recheck (copy/rename source) re-runs only the bucket
|
||||
// policy; it must still honor an explicit Deny in the caller's attached or
|
||||
// session policies, or the phase-1 veto on secret/* could be bypassed.
|
||||
func TestRecheckPolicyForKeyWithLookupHonorsIdentityDeny(t *testing.T) {
|
||||
filerImpl := &ambiguousPutFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/" + bpdBucket + "/secret/payroll.txt": {Name: "payroll.txt"},
|
||||
"/buckets/" + bpdBucket + "/other/data.txt": {Name: "data.txt"},
|
||||
},
|
||||
}
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
|
||||
iam := newBucketPolicyDenyIAM(t, bpdBucketPolicyDoc(t, "secret/*", "other/*"))
|
||||
s3a.iam = iam
|
||||
s3a.policyEngine = iam.policyEngine
|
||||
s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
|
||||
s3a.bucketConfigCache.Set(bpdBucket, &BucketConfig{Name: bpdBucket})
|
||||
|
||||
req := httptest.NewRequest(http.MethodPut, "/victim-bucket/dst-key", nil)
|
||||
req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), bpdReader(t, iam)))
|
||||
|
||||
assert.Equal(t, s3err.ErrAccessDenied,
|
||||
s3a.recheckPolicyForKeyWithLookup(req, http.MethodGet, bpdBucket, "secret/payroll.txt", "", s3_constants.ACTION_READ, "test"),
|
||||
"an explicit identity-policy Deny must survive the bucket-policy recheck")
|
||||
assert.Equal(t, s3err.ErrNone,
|
||||
s3a.recheckPolicyForKeyWithLookup(req, http.MethodGet, bpdBucket, "other/data.txt", "", s3_constants.ACTION_READ, "test"),
|
||||
"a bucket-policy Allow on a prefix the identity does not deny must still grant")
|
||||
}
|
||||
|
||||
@@ -219,6 +219,11 @@ func (s3a *S3ApiServer) GetObjectAttributesHandler(w http.ResponseWriter, r *htt
|
||||
}
|
||||
}
|
||||
|
||||
if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, s3_constants.S3_ACTION_GET_OBJECT_ATTRIBUTES, entry.Extended, "GetObjectAttributesHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// Evaluate conditional headers against the resolved entry (after version resolution)
|
||||
// This ensures conditions are checked against the correct version, not always the latest
|
||||
if s3a.hasConditionalHeaders(r) {
|
||||
|
||||
@@ -202,10 +202,14 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
|
||||
// CopyObject requires s3:GetObject on the source in addition to s3:PutObject
|
||||
// on the destination. The Auth middleware only checked the destination, so
|
||||
// verify the source explicitly here.
|
||||
// verify the source explicitly here. A first-phase deny decided without
|
||||
// object tags can still flip on a tag-conditioned Allow, so when a bucket
|
||||
// policy exists the tag-aware recheck gets the last word.
|
||||
if errCode := s3a.authorizeCopySource(r, srcBucket, srcObject, srcVersionId); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
if s3a.recheckPolicyForKeyWithLookup(r, http.MethodGet, srcBucket, srcObject, srcVersionId, string(s3_constants.ACTION_READ), "CopyObjectHandler") != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// A copy creates destination ownership and grants independently of the
|
||||
@@ -238,6 +242,11 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
return
|
||||
}
|
||||
|
||||
if errCode := s3a.recheckCopySourcePolicy(r, srcBucket, srcObject, srcVersionId, entry.Extended, "CopyObjectHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// Cache remote-only sources before copying; otherwise the copy path below
|
||||
// writes a destination with FileSize > 0 but no chunks/content.
|
||||
if entry.IsInRemoteOnly() {
|
||||
@@ -343,6 +352,12 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
updatedEntry = cloneProtoEntry(rawEntry)
|
||||
}
|
||||
updatedEntry.Extended = mergeCopyMetadata(updatedEntry.Extended, updatedMetadata)
|
||||
// The update keeps the object's existing lock state, but request
|
||||
// lock headers still need validating so a disallowed combination
|
||||
// is rejected rather than silently ignored.
|
||||
if errCode := s3a.applyCopyObjectLockHeaders(r, dstBucket, dstObject, updatedEntry); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
}
|
||||
if updatedEntry.Attributes == nil {
|
||||
updatedEntry.Attributes = &filer_pb.FuseAttributes{}
|
||||
}
|
||||
@@ -446,6 +461,12 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
if k == s3_constants.ExtAmzOwnerKey || k == s3_constants.ExtAmzAclKey {
|
||||
continue
|
||||
}
|
||||
// Retention and legal hold are destination settings, not object
|
||||
// metadata: the copy takes its own lock headers or the destination
|
||||
// bucket's default instead.
|
||||
if isCopyLockKey(k) {
|
||||
continue
|
||||
}
|
||||
// Skip encryption-specific headers that might conflict with destination encryption type
|
||||
skipHeader := false
|
||||
|
||||
@@ -482,6 +503,11 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request
|
||||
// so REPLACE doesn't leak source values through the merge, as on self-copies.
|
||||
dstEntry.Extended = mergeCopyMetadata(dstEntry.Extended, processedMetadata)
|
||||
|
||||
if errCode := s3a.applyCopyObjectLockHeaders(r, dstBucket, dstObject, dstEntry); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// For zero-size files or files without chunks, handle inline content
|
||||
// This includes encrypted inline files that need decryption/re-encryption
|
||||
if entry.Attributes.FileSize == 0 || len(entry.GetChunks()) == 0 {
|
||||
@@ -590,6 +616,39 @@ func copyReplacesSourceEntry(sameDestination bool, dstVersioningState, srcVersio
|
||||
return sameDestination && dstVersioningState == "" && srcVersionId == ""
|
||||
}
|
||||
|
||||
// isCopyLockKey marks the extended attributes that carry object lock state; a
|
||||
// copy never inherits them from the source.
|
||||
func isCopyLockKey(key string) bool {
|
||||
switch key {
|
||||
case s3_constants.ExtObjectLockModeKey, s3_constants.ExtRetentionUntilDateKey, s3_constants.ExtLegalHoldKey:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// applyCopyObjectLockHeaders resolves the copied object's retention: explicit
|
||||
// copy-request lock headers win, then the destination bucket's default, and
|
||||
// the source's own retention never carries over.
|
||||
func (s3a *S3ApiServer) applyCopyObjectLockHeaders(r *http.Request, dstBucket, dstObject string, dstEntry *filer_pb.Entry) s3err.ErrorCode {
|
||||
objectLockEnabled, lockErr := s3a.isObjectLockEnabled(dstBucket)
|
||||
if lockErr != nil && !errors.Is(lockErr, filer_pb.ErrNotFound) {
|
||||
glog.Errorf("CopyObjectHandler: failed to check object lock for bucket %s: %v", dstBucket, lockErr)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
if validationErr := s3a.validateObjectLockHeaders(r, dstBucket, dstObject, objectLockEnabled); validationErr != nil {
|
||||
glog.V(2).Infof("CopyObjectHandler: object lock header validation failed for %s/%s: %v", dstBucket, dstObject, validationErr)
|
||||
return mapValidationErrorToS3Error(validationErr)
|
||||
}
|
||||
if !objectLockEnabled {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
if err := s3a.extractObjectLockMetadataFromRequest(r, dstEntry); err != nil {
|
||||
glog.Errorf("CopyObjectHandler: failed to extract object lock metadata for %s/%s: %v", dstBucket, dstObject, err)
|
||||
return mapValidationErrorToS3Error(err)
|
||||
}
|
||||
return s3err.ErrNone
|
||||
}
|
||||
|
||||
func cloneProtoEntry(entry *filer_pb.Entry) *filer_pb.Entry {
|
||||
if entry == nil {
|
||||
return nil
|
||||
@@ -931,8 +990,10 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
||||
// UploadPartCopy requires s3:GetObject on the source. The Auth middleware
|
||||
// only verified s3:PutObject (s3:UploadPart) on the destination.
|
||||
if errCode := s3a.authorizeCopySource(r, srcBucket, srcObject, srcVersionId); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
if s3a.recheckPolicyForKeyWithLookup(r, http.MethodGet, srcBucket, srcObject, srcVersionId, string(s3_constants.ACTION_READ), "CopyObjectPartHandler") != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
partIDString := r.URL.Query().Get("partNumber")
|
||||
@@ -1001,6 +1062,11 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req
|
||||
return
|
||||
}
|
||||
|
||||
if errCode := s3a.recheckCopySourcePolicy(r, srcBucket, srcObject, srcVersionId, entry.Extended, "CopyObjectPartHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// Cache remote-only sources before copying; the part-copy paths below
|
||||
// iterate entry.GetChunks() and would otherwise produce an empty part.
|
||||
if entry.IsInRemoteOnly() {
|
||||
|
||||
@@ -478,6 +478,10 @@ type putFinalize struct {
|
||||
lockKey string
|
||||
mutations []*filer_pb.ObjectMutation
|
||||
afterCreate func(entry *filer_pb.Entry) s3err.ErrorCode
|
||||
// conditionAltKey names the entry a routed WriteCondition falls back to
|
||||
// when the write path is absent — a suspended-versioning write points it
|
||||
// at the latest version so check and write stay under the filer lock
|
||||
conditionAltKey string
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -1010,15 +1014,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader
|
||||
routed := false
|
||||
routedAmbiguous := false
|
||||
var owner pb.ServerAddress
|
||||
condAltKey := ""
|
||||
if finalize != nil {
|
||||
condAltKey = finalize.conditionAltKey
|
||||
}
|
||||
if owner = s3a.routableWriteOwner(bucket, object); owner != "" {
|
||||
if cond, ok := routeWriteCondition(r, uniqueWritePath); ok {
|
||||
if cond, ok := routeWriteCondition(r, uniqueWritePath && condAltKey == ""); ok {
|
||||
// Routed mutations ride in the PUT's transaction (committing atomically),
|
||||
// so lockKey is the object path they carry, not the version file path.
|
||||
lockKey, finalizeMutations := filePath, []*filer_pb.ObjectMutation(nil)
|
||||
if finalize != nil && len(finalize.mutations) > 0 {
|
||||
lockKey, finalizeMutations = finalize.lockKey, finalize.mutations
|
||||
}
|
||||
resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", finalizeMutations)
|
||||
resp, err := s3a.routedPut(owner, s3a.objectRouteKey(bucket, object), lockKey, filePath, entry, cond, "", condAltKey, finalizeMutations)
|
||||
switch {
|
||||
case err != nil:
|
||||
routedAmbiguous = true
|
||||
@@ -1824,7 +1832,19 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
|
||||
// naming a marker that is gone, and the read path heals that by promoting an older
|
||||
// version, republishing the deleted key. Best-effort, as a stale flag self-heals on
|
||||
// the next list.
|
||||
// A conditional suspended write checks "the null slot if present, else the
|
||||
// latest version". Naming the pointed-at version as the fallback lets the
|
||||
// routed transaction evaluate that rule under the filer lock instead of
|
||||
// splitting check and write across a gateway lock that routed writes skip.
|
||||
conditionAltKey := ""
|
||||
if headers, headersErr := parseConditionalHeaders(r); headersErr == s3err.ErrNone && headers.isSet {
|
||||
// The filer resolves the latest pointer under the same lock that
|
||||
// serializes the write, so the condition cannot read a version a
|
||||
// concurrent delete has already displaced.
|
||||
conditionAltKey = s3a.toFilerPath(bucket, normalizedObject+s3_constants.VersionsFolder)
|
||||
}
|
||||
etag, errCode, sseMetadata = s3a.putToFiler(r, filePath, body, bucket, normalizedObject, 1, 0, &putFinalize{
|
||||
conditionAltKey: conditionAltKey,
|
||||
afterCreate: func(entry *filer_pb.Entry) s3err.ErrorCode {
|
||||
writtenETag := string(entry.Extended[s3_constants.ExtETagKey])
|
||||
if err := s3a.finalizeSuspendedNullWrite("", bucket, normalizedObject, s3_constants.ExtETagKey, writtenETag); err != nil {
|
||||
@@ -1832,7 +1852,7 @@ func (s3a *S3ApiServer) putSuspendedVersioningObject(r *http.Request, bucket, ob
|
||||
}
|
||||
return s3err.ErrNone
|
||||
},
|
||||
}, false, "")
|
||||
}, true, "")
|
||||
if errCode != s3err.ErrNone {
|
||||
glog.Errorf("putSuspendedVersioningObject: failed to upload object: %v", errCode)
|
||||
return "", errCode, SSEResponseMetadata{}
|
||||
|
||||
@@ -8,9 +8,12 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
@@ -18,10 +21,12 @@ import (
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/proto"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
@@ -112,6 +117,8 @@ type ambiguousPutFiler struct {
|
||||
|
||||
mu sync.Mutex
|
||||
entries map[string]*filer_pb.Entry
|
||||
txnCount int
|
||||
txnCheck func(req *filer_pb.ObjectTransactionRequest)
|
||||
apply bool
|
||||
createErr error
|
||||
respError string
|
||||
@@ -163,6 +170,92 @@ func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_p
|
||||
return &filer_pb.LookupDirectoryEntryResponse{}, nil
|
||||
}
|
||||
|
||||
// ObjectTransaction applies the WriteCondition against the entry the condition
|
||||
// key names, the way a real filer evaluates it under the path lock, then
|
||||
// applies PUT mutations. Routed-write tests use it to catch a condition sent
|
||||
// for an entry the gateway's conditional check never looked at.
|
||||
func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.txnCount++
|
||||
if f.txnCheck != nil {
|
||||
f.txnCheck(req)
|
||||
}
|
||||
key := req.LockKey
|
||||
if req.ConditionKey != "" {
|
||||
key = req.ConditionKey
|
||||
}
|
||||
current := f.entries[key]
|
||||
if current == nil && req.ConditionAltKey != "" {
|
||||
// Mirrors the filer's fallback: the alt key is the object's .versions
|
||||
// directory and the entry under test is the one its latest pointer
|
||||
// names; a delete marker resolves to absent.
|
||||
if versionsEntry := f.entries[req.ConditionAltKey]; versionsEntry != nil {
|
||||
if latest := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey]); latest != "" {
|
||||
key = req.ConditionAltKey + "/" + latest
|
||||
current = f.entries[key]
|
||||
if current != nil && string(current.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" {
|
||||
current = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, clause := range req.GetCondition().GetClauses() {
|
||||
var md5Hex string
|
||||
if current != nil && current.Attributes != nil {
|
||||
md5Hex = fmt.Sprintf("%x", current.Attributes.Md5)
|
||||
}
|
||||
ok := true
|
||||
switch clause.Kind {
|
||||
case filer_pb.WriteCondition_IF_NOT_EXISTS:
|
||||
ok = current == nil
|
||||
case filer_pb.WriteCondition_IF_EXISTS:
|
||||
ok = current != nil
|
||||
case filer_pb.WriteCondition_IF_ETAG_MATCH:
|
||||
ok = current != nil && slices.Contains(clause.Etags, md5Hex)
|
||||
case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH:
|
||||
ok = current == nil || !slices.Contains(clause.Etags, md5Hex)
|
||||
}
|
||||
if !ok {
|
||||
return &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil
|
||||
}
|
||||
}
|
||||
for _, m := range req.Mutations {
|
||||
if m.Type == filer_pb.ObjectMutation_PUT {
|
||||
f.entries[m.Directory+"/"+m.Entry.Name] = proto.Clone(m.Entry).(*filer_pb.Entry)
|
||||
}
|
||||
}
|
||||
return &filer_pb.ObjectTransactionResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
for fullPath, entry := range f.entries {
|
||||
if path.Dir(fullPath) != req.Directory {
|
||||
continue
|
||||
}
|
||||
if req.Prefix != "" && !strings.HasPrefix(entry.Name, req.Prefix) {
|
||||
continue
|
||||
}
|
||||
if req.StartFromFileName != "" && (entry.Name < req.StartFromFileName ||
|
||||
(entry.Name == req.StartFromFileName && !req.InclusiveStartFrom)) {
|
||||
continue
|
||||
}
|
||||
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.entries[req.Directory+"/"+req.Entry.Name] = proto.Clone(req.Entry).(*filer_pb.Entry)
|
||||
return &filer_pb.UpdateEntryResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *ambiguousPutFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) {
|
||||
resp := &filer_pb.LookupVolumeResponse{LocationsMap: map[string]*filer_pb.Locations{}}
|
||||
for _, vid := range req.VolumeIds {
|
||||
@@ -407,3 +500,115 @@ func TestFetchChunkDataDecryptsVolumeCipher(t *testing.T) {
|
||||
t.Fatalf("ranged plain read = %q, want %q", got, plaintext[5:9])
|
||||
}
|
||||
}
|
||||
|
||||
// A suspended-versioning overwrite resolves the conditional check against the
|
||||
// entry the read path serves: the null slot when present, else the latest
|
||||
// version under .versions/. The routed transaction carries that version as the
|
||||
// condition's fallback key so check and write stay under the filer lock —
|
||||
// checking the empty null-version slot alone would reject a matching If-Match.
|
||||
func TestSuspendedPutIfMatchRoutesWithConditionAltKey(t *testing.T) {
|
||||
volume := startFakeVolumeServer(t)
|
||||
filerImpl := &ambiguousPutFiler{
|
||||
volume: volume,
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
Name: "b",
|
||||
IsDirectory: true,
|
||||
Attributes: &filer_pb.FuseAttributes{},
|
||||
Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningSuspended)},
|
||||
},
|
||||
"/buckets/b/k.versions": {
|
||||
Name: "k.versions",
|
||||
IsDirectory: true,
|
||||
Attributes: &filer_pb.FuseAttributes{},
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.ExtLatestVersionIdKey: []byte("v1"),
|
||||
s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"),
|
||||
},
|
||||
},
|
||||
"/buckets/b/k.versions/v_v1": {
|
||||
Name: "v_v1",
|
||||
Attributes: &filer_pb.FuseAttributes{Md5: []byte{0xab, 0xc1, 0x23}, FileSize: 1},
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.ExtETagKey: []byte("abc123"),
|
||||
s3_constants.ExtVersionIdKey: []byte("v1"),
|
||||
},
|
||||
},
|
||||
},
|
||||
apply: true,
|
||||
}
|
||||
filerAddr := startFakeFiler(t, filerImpl)
|
||||
s3a := newPutTestServer(t, filerAddr)
|
||||
s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
|
||||
s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr)
|
||||
s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1)
|
||||
|
||||
filerImpl.txnCheck = func(req *filer_pb.ObjectTransactionRequest) {
|
||||
if req.ConditionAltKey != "/buckets/b/k.versions" {
|
||||
t.Fatalf("condition alt key = %q, want /buckets/b/k.versions", req.ConditionAltKey)
|
||||
}
|
||||
}
|
||||
|
||||
r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new"))
|
||||
r.Header.Set(s3_constants.IfMatch, `"abc123"`)
|
||||
_, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "")
|
||||
if code != s3err.ErrNone {
|
||||
t.Fatalf("suspended If-Match PUT = %v", code)
|
||||
}
|
||||
if filerImpl.entries["/buckets/b/k"] == nil {
|
||||
t.Fatal("null-version entry was not created")
|
||||
}
|
||||
if filerImpl.txnCount != 1 {
|
||||
t.Fatalf("conditional suspended write should route one transaction, got %d", filerImpl.txnCount)
|
||||
}
|
||||
}
|
||||
|
||||
// When the latest version is a delete marker, a suspended If-None-Match: *
|
||||
// write must treat the object as absent — the fallback resolves a tombstone
|
||||
// to "no current version", not an existing one.
|
||||
func TestSuspendedPutIfNoneMatchWithDeleteMarkerLatest(t *testing.T) {
|
||||
volume := startFakeVolumeServer(t)
|
||||
filerImpl := &ambiguousPutFiler{
|
||||
volume: volume,
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/b": {
|
||||
Name: "b",
|
||||
IsDirectory: true,
|
||||
Attributes: &filer_pb.FuseAttributes{},
|
||||
Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningSuspended)},
|
||||
},
|
||||
"/buckets/b/k.versions": {
|
||||
Name: "k.versions",
|
||||
IsDirectory: true,
|
||||
Attributes: &filer_pb.FuseAttributes{},
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"),
|
||||
},
|
||||
},
|
||||
"/buckets/b/k.versions/v_v1": {
|
||||
Name: "v_v1",
|
||||
Attributes: &filer_pb.FuseAttributes{FileSize: 0},
|
||||
Extended: map[string][]byte{
|
||||
s3_constants.ExtVersionIdKey: []byte("v1"),
|
||||
s3_constants.ExtDeleteMarkerKey: []byte("true"),
|
||||
},
|
||||
},
|
||||
},
|
||||
apply: true,
|
||||
}
|
||||
filerAddr := startFakeFiler(t, filerImpl)
|
||||
s3a := newPutTestServer(t, filerAddr)
|
||||
s3a.bucketConfigCache = NewBucketConfigCache(time.Minute)
|
||||
s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr)
|
||||
s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1)
|
||||
|
||||
r := httptest.NewRequest(http.MethodPut, "/b/k", strings.NewReader("new"))
|
||||
r.Header.Set(s3_constants.IfNoneMatch, "*")
|
||||
_, code, _ := s3a.putSuspendedVersioningObject(r, "b", "k", r.Body, "")
|
||||
if code != s3err.ErrNone {
|
||||
t.Fatalf("suspended If-None-Match PUT over a delete-marker latest = %v, want success", code)
|
||||
}
|
||||
if filerImpl.entries["/buckets/b/k"] == nil {
|
||||
t.Fatal("null-version entry was not created")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -77,8 +77,14 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
|
||||
// is what the request URL names. The source arrives in a header and loses
|
||||
// its key, so it needs both read and delete permission checked here.
|
||||
if errCode := s3a.authorizeRenameSource(r, bucket, srcObject); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
// Same tag-conditioned-Allow deferral as the copy path: the first
|
||||
// phase ran without the source's tags, so a tag-gated Allow can still
|
||||
// grant both the read and the delete the rename performs.
|
||||
if s3a.recheckPolicyForKeyWithLookup(r, http.MethodGet, bucket, srcObject, "", string(s3_constants.ACTION_READ), "RenameObjectHandler") != s3err.ErrNone ||
|
||||
s3a.recheckPolicyForKeyWithLookup(r, http.MethodDelete, bucket, srcObject, "", string(s3_constants.ACTION_WRITE), "RenameObjectHandler") != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
versioningState, err := s3a.getVersioningState(bucket)
|
||||
@@ -115,6 +121,14 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque
|
||||
if srcErrCode != s3err.ErrNone {
|
||||
return srcErrCode
|
||||
}
|
||||
if errCode := s3a.recheckCopySourcePolicy(r, bucket, srcObject, "", entry.Extended, "RenameObjectHandler"); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
}
|
||||
// The rename also deletes the source; a tag-gated Deny on the delete
|
||||
// applies the same way it does to a direct DELETE.
|
||||
if errCode := s3a.recheckPolicyWithObjectEntry(objectKeyPolicyRequest(r, http.MethodDelete, bucket, srcObject, ""), bucket, srcObject, string(s3_constants.ACTION_WRITE), entry.Extended, "RenameObjectHandler"); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
}
|
||||
if errCode := validateSourceConditionalHeaders(r, entry, renameSourceConditionalHeaders); errCode != s3err.ErrNone {
|
||||
return errCode
|
||||
}
|
||||
|
||||
@@ -74,7 +74,7 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
tags, err := s3a.getTags(dir, name)
|
||||
entry, err = s3a.getEntry(dir, name)
|
||||
if err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
|
||||
@@ -85,8 +85,10 @@ func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.R
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
writeSuccessResponseXML(w, r, FromTags(tags))
|
||||
if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, s3_constants.S3_ACTION_GET_OBJECT_TAGGING, entry.Extended, "GetObjectTaggingHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -183,6 +185,11 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
if errCode := s3a.recheckTagConditionedPolicyWithLookup(r, bucket, object, s3_constants.S3_ACTION_PUT_OBJECT_TAGGING, "PutObjectTaggingHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
if err = s3a.setTags(dir, name, tags); err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
|
||||
@@ -199,6 +206,11 @@ func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.R
|
||||
return
|
||||
}
|
||||
|
||||
if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, s3_constants.S3_ACTION_PUT_OBJECT_TAGGING, entry.Extended, "PutObjectTaggingHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// For versioned objects, resolve the directory holding the entry to update. A null
|
||||
// version lives under its full key's parent directory, so a nested key must not fall
|
||||
// back to the bucket root and rewrite a different object sharing the same basename.
|
||||
@@ -301,6 +313,11 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
if errCode := s3a.recheckTagConditionedPolicyWithLookup(r, bucket, object, s3_constants.S3_ACTION_DELETE_OBJECT_TAGGING, "DeleteObjectTaggingHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
err := s3a.rmTags(dir, name)
|
||||
if err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
@@ -318,6 +335,11 @@ func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *htt
|
||||
return
|
||||
}
|
||||
|
||||
if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, s3_constants.S3_ACTION_DELETE_OBJECT_TAGGING, entry.Extended, "DeleteObjectTaggingHandler"); errCode != s3err.ErrNone {
|
||||
s3err.WriteErrorResponse(w, r, errCode)
|
||||
return
|
||||
}
|
||||
|
||||
// For versioned objects, resolve the directory holding the entry to update. A null
|
||||
// version lives under its full key's parent directory, so a nested key must not fall
|
||||
// back to the bucket root and rewrite a different object sharing the same basename.
|
||||
|
||||
@@ -79,3 +79,20 @@ func TestVeeamObjectLockBugFix(t *testing.T) {
|
||||
assert.True(t, enabled, "Object Lock should be enabled when constant is used")
|
||||
})
|
||||
}
|
||||
|
||||
// Copied objects get their own retention from the request or the destination
|
||||
// bucket's default; the source's lock keys never carry into the destination
|
||||
// entry.
|
||||
func TestCopyDoesNotInheritLockKeys(t *testing.T) {
|
||||
// The destination-Extended copy loop filters on isCopyLockKey, so
|
||||
// retention and legal hold never carry over from the source entry.
|
||||
for _, key := range []string{
|
||||
s3_constants.ExtObjectLockModeKey,
|
||||
s3_constants.ExtRetentionUntilDateKey,
|
||||
s3_constants.ExtLegalHoldKey,
|
||||
} {
|
||||
assert.True(t, isCopyLockKey(key), "%s must be filtered from a copy destination", key)
|
||||
}
|
||||
assert.False(t, isCopyLockKey(s3_constants.ExtAmzOwnerKey))
|
||||
assert.False(t, isCopyLockKey("custom-key"))
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
@@ -157,6 +158,11 @@ func singleStrongETag(v string) (string, bool) {
|
||||
// the write. The reached filer forwards to the real owner by route_key, keeping its
|
||||
// per-path lock authoritative. Mirrors getObjectEntryRoutedByKey on the read path.
|
||||
func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
// One id per logical transaction: failover retries reuse it so the owner
|
||||
// replays the recorded outcome instead of re-applying the mutations.
|
||||
if req.RequestId == "" {
|
||||
req.RequestId = uuid.NewString()
|
||||
}
|
||||
var resp *filer_pb.ObjectTransactionResponse
|
||||
txn := func(client filer_pb.SeaweedFilerClient) error {
|
||||
var e error
|
||||
@@ -183,7 +189,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O
|
||||
// object path plus a RECOMPUTE_LATEST finalize, so the version's PUT and its
|
||||
// .versions pointer flip commit atomically (the recompute scans .versions/ after
|
||||
// the PUT and sees the new version).
|
||||
func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, filePath string, entry *filer_pb.Entry, cond *filer_pb.WriteCondition, conditionKey, conditionAltKey string, finalize []*filer_pb.ObjectMutation) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
mutations := make([]*filer_pb.ObjectMutation, 0, 1+len(finalize))
|
||||
mutations = append(mutations, &filer_pb.ObjectMutation{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
@@ -192,11 +198,12 @@ func (s3a *S3ApiServer) routedPut(owner pb.ServerAddress, routeKey, lockKey, fil
|
||||
})
|
||||
mutations = append(mutations, finalize...)
|
||||
return s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
Mutations: mutations,
|
||||
LockKey: lockKey,
|
||||
RouteKey: routeKey,
|
||||
Condition: cond,
|
||||
ConditionKey: conditionKey,
|
||||
ConditionAltKey: conditionAltKey,
|
||||
Mutations: mutations,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -227,7 +234,7 @@ func (s3a *S3ApiServer) routedMkFile(owner pb.ServerAddress, routeKey, parentDir
|
||||
cond, conditionKey = removal.condition, removal.conditionKey
|
||||
finalize = []*filer_pb.ObjectMutation{removal.mutation}
|
||||
}
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, finalize)
|
||||
resp, err := s3a.routedPut(owner, routeKey, filePath, filePath, entry, cond, conditionKey, "", finalize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -139,7 +139,7 @@ func (s3a *S3ApiServer) routedMultipartFinalize(owner pb.ServerAddress, bucket,
|
||||
routeKey = s3a.objectRouteKey(bucket, object)
|
||||
}
|
||||
conditionKey, condition := uploadExistsCondition(s3a.genUploadsFolder(bucket) + "/" + uploadID)
|
||||
resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, []*filer_pb.ObjectMutation{
|
||||
resp, err := s3a.routedPut(owner, routeKey, s3a.toFilerPath(bucket, object), versionDir+"/"+versionFileName, versionEntry, condition, conditionKey, "", []*filer_pb.ObjectMutation{
|
||||
s3a.removeUploadDirMutation(bucket, uploadID),
|
||||
s3a.latestPointerRecompute(bucket, object, useInvertedFormat, "", true),
|
||||
})
|
||||
|
||||
@@ -28,6 +28,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/iam/policy"
|
||||
"github.com/seaweedfs/seaweedfs/weed/iam/sts"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
|
||||
@@ -675,6 +676,79 @@ func (s3a *S3ApiServer) recheckPolicyWithObjectEntry(r *http.Request, bucket, ob
|
||||
return errCode
|
||||
}
|
||||
|
||||
// recheckCopySourcePolicy repeats the source object's policy check once its
|
||||
// entry is fetched, so conditions on the object's existing tags apply to a
|
||||
// CopyObject / UploadPartCopy / RenameObject source the same way they do to a
|
||||
// plain GET. The synthesized GET keeps the destination's request shape (PUT,
|
||||
// partNumber, ...) from resolving a different action.
|
||||
func (s3a *S3ApiServer) recheckCopySourcePolicy(r *http.Request, bucket, object, versionId string, objectEntry map[string][]byte, handlerName string) s3err.ErrorCode {
|
||||
keyReq := objectKeyPolicyRequest(r, http.MethodGet, bucket, object, versionId)
|
||||
return s3a.recheckPolicyWithObjectEntry(keyReq, bucket, object, string(ACTION_READ), objectEntry, handlerName)
|
||||
}
|
||||
|
||||
// recheckTagConditionedPolicyWithLookup repeats policy evaluation with the
|
||||
// object's entry in hand so s3:ExistingObjectTag conditions apply to tagging
|
||||
// writes that did not otherwise fetch the entry. It only does the extra
|
||||
// lookup when the bucket has a policy; entry misses return ErrNone because
|
||||
// the handler reports its own NoSuchKey.
|
||||
func (s3a *S3ApiServer) recheckTagConditionedPolicyWithLookup(r *http.Request, bucket, object, action, handlerName string) s3err.ErrorCode {
|
||||
if s3a.policyEngine == nil || !s3a.policyEngine.HasPolicyForBucket(bucket) {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s", s3a.bucketDir(bucket), object))
|
||||
dir, name := target.DirAndName()
|
||||
entry, err := s3a.getEntry(dir, name)
|
||||
if err != nil {
|
||||
// The tag conditions cannot be evaluated without the entry; failing
|
||||
// open would let a tag-gated Deny slip through a transient miss.
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
glog.Errorf("%s: recheck lookup for %s/%s failed: %v", handlerName, bucket, object, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
if entry == nil {
|
||||
return s3err.ErrNone
|
||||
}
|
||||
return s3a.recheckPolicyWithObjectEntry(r, bucket, object, action, entry.Extended, handlerName)
|
||||
}
|
||||
|
||||
// recheckPolicyForKeyWithLookup re-decides a first-phase deny on one key for a
|
||||
// synthesized method, resolving the entry the way the read path does (versions
|
||||
// included) so s3:ExistingObjectTag conditions see the object's real tags.
|
||||
// Unlike the deny-side rechecks, an indeterminate result does not grant: the
|
||||
// caller reports the original verdict either way, so only a statement that
|
||||
// matches with tags in hand can turn a deny into an allow, and a missed lookup
|
||||
// never widens access or reveals whether the key exists.
|
||||
func (s3a *S3ApiServer) recheckPolicyForKeyWithLookup(r *http.Request, method, bucket, object, versionId, action, handlerName string) s3err.ErrorCode {
|
||||
if s3a.policyEngine == nil || !s3a.policyEngine.HasPolicyForBucket(bucket) {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
versioningState, err := s3a.getVersioningState(bucket)
|
||||
if err != nil {
|
||||
glog.Errorf("%s: recheck versioning state for %s: %v", handlerName, bucket, err)
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
entry, err := s3a.resolveCopySourceEntry(bucket, object, versionId, versioningState)
|
||||
entry = prefixObjectSource(entry)
|
||||
if err != nil || entry == nil {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
keyReq := objectKeyPolicyRequest(r, method, bucket, object, versionId)
|
||||
errCode, checked := s3a.checkPolicyWithEntry(keyReq, bucket, object, action, "", entry.Extended)
|
||||
if errCode != s3err.ErrNone || !checked {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
// A bucket-policy Allow never overrides an explicit Deny in the caller's
|
||||
// own attached or session policies — the first phase applied that veto,
|
||||
// so the tag-aware recheck must too.
|
||||
identity, _ := GetIdentityFromContext(r).(*Identity)
|
||||
if s3a.iam != nil && s3a.iam.isActionExplicitlyDeniedByApplicablePolicies(keyReq, identity, Action(action), bucket, object) {
|
||||
return s3err.ErrAccessDenied
|
||||
}
|
||||
return s3err.ErrNone
|
||||
}
|
||||
|
||||
// classifyDomainNames classifies domains into path-style and virtual-host style domains.
|
||||
// A domain is considered path-style if:
|
||||
// 1. It contains a dot (has subdomains)
|
||||
|
||||
@@ -238,20 +238,12 @@ func (h *S3TablesHandler) getAccountID(r *http.Request) string {
|
||||
// Prefer stable claims from JWT/STS identities. Only "sub" is guaranteed durable per OIDC;
|
||||
// preferred_username is ergonomic but can rotate and may orphan ownership data, while email
|
||||
// is explicitly excluded to avoid storing PII in metadata.
|
||||
claimsField := val.FieldByName("Claims")
|
||||
if claimsField.IsValid() && claimsField.Kind() == reflect.Map && !claimsField.IsNil() && claimsField.Type().Key().Kind() == reflect.String {
|
||||
for _, claimKey := range []string{"sub", "preferred_username"} {
|
||||
claimVal := claimsField.MapIndex(reflect.ValueOf(claimKey))
|
||||
if !claimVal.IsValid() {
|
||||
continue
|
||||
}
|
||||
if claimVal.Kind() == reflect.Interface && !claimVal.IsNil() {
|
||||
claimVal = claimVal.Elem()
|
||||
}
|
||||
if claimVal.Kind() == reflect.String {
|
||||
if principal := normalizePrincipalID(claimVal.String()); principal != "" {
|
||||
return principal
|
||||
}
|
||||
if claims := identityClaimsMap(val); claims != nil {
|
||||
// aws:userid is the issuer-namespaced (sub, iss) hash; a bare
|
||||
// sub or preferred_username collides across providers.
|
||||
for _, claimKey := range []string{"aws:userid", "sub", "preferred_username"} {
|
||||
if principal := normalizePrincipalID(claimString(claims, claimKey)); principal != "" {
|
||||
return principal
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -291,6 +283,46 @@ func (h *S3TablesHandler) getAccountID(r *http.Request) string {
|
||||
return h.accountID
|
||||
}
|
||||
|
||||
// identityClaimsMap extracts the Claims map from an identity value, using
|
||||
// reflection to avoid an import cycle with the s3api identity type.
|
||||
func identityClaimsMap(val reflect.Value) map[string]interface{} {
|
||||
claimsField := val.FieldByName("Claims")
|
||||
if !claimsField.IsValid() || claimsField.Kind() != reflect.Map || claimsField.IsNil() || claimsField.Type().Key().Kind() != reflect.String {
|
||||
return nil
|
||||
}
|
||||
claims, ok := claimsField.Interface().(map[string]interface{})
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return claims
|
||||
}
|
||||
|
||||
func claimString(claims map[string]interface{}, key string) string {
|
||||
v, ok := claims[key]
|
||||
if !ok || v == nil {
|
||||
return ""
|
||||
}
|
||||
str, _ := v.(string)
|
||||
return str
|
||||
}
|
||||
|
||||
// trustedIssuers returns the OIDC issuers the auth layer asserted into the
|
||||
// identity's claims; they re-anchor legacy owner checks in checkPermission.
|
||||
func (h *S3TablesHandler) trustedIssuers(r *http.Request) []string {
|
||||
identityRaw := s3_constants.GetIdentityFromContext(r)
|
||||
if identityRaw == nil {
|
||||
return nil
|
||||
}
|
||||
val := reflect.ValueOf(identityRaw)
|
||||
if val.Kind() == reflect.Ptr {
|
||||
val = val.Elem()
|
||||
}
|
||||
if val.Kind() != reflect.Struct {
|
||||
return nil
|
||||
}
|
||||
return strings.Fields(claimString(identityClaimsMap(val), "seaweed:oidc-issuers"))
|
||||
}
|
||||
|
||||
// normalizePrincipalID collapses ARN and identity strings to a key that is stable within a single account.
|
||||
// WARNING: this assumes identity names are unique per account; distinct principals such as
|
||||
// arn:aws:iam::111:user/alice and arn:aws:iam::222:user/alice will both normalize to "alice".
|
||||
|
||||
@@ -65,6 +65,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
|
||||
owner = DefaultAccountID
|
||||
}
|
||||
if !CheckPermissionWithContext("CreateTableBucket", principal, owner, "", "", &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}) {
|
||||
|
||||
@@ -70,6 +70,7 @@ func (h *S3TablesHandler) handleGetTableBucket(w http.ResponseWriter, r *http.Re
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("GetTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -192,6 +193,7 @@ func (h *S3TablesHandler) handleListTableBuckets(w http.ResponseWriter, r *http.
|
||||
bucketARN := h.generateTableBucketARN(metadata.OwnerAccountID, entry.Entry.Name)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("GetTableBucket", accountID, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: entry.Entry.Name,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -296,6 +298,7 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("DeleteTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
|
||||
@@ -171,3 +171,18 @@ func TestGetAccountIDNormalizesAccountIDARN(t *testing.T) {
|
||||
|
||||
assert.Equal(t, "bob", h.getAccountID(req), "expected ARN account ID to be normalized to the suffix")
|
||||
}
|
||||
|
||||
func TestGetAccountIDPrefersIssuerScopedUserID(t *testing.T) {
|
||||
h := NewS3TablesHandler()
|
||||
id := &testIdentity{
|
||||
Claims: map[string]interface{}{
|
||||
"aws:userid": "issuerA-hash",
|
||||
"sub": "alice-sub",
|
||||
"preferred_username": "alice",
|
||||
},
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), id))
|
||||
|
||||
assert.Equal(t, "issuerA-hash", h.getAccountID(req), "aws:userid should win over a bare sub")
|
||||
}
|
||||
|
||||
@@ -349,6 +349,7 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien
|
||||
bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
|
||||
principal := h.getAccountID(r)
|
||||
if !CheckPermissionWithContext(operation, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: getIdentityActions(r),
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -385,6 +386,7 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient
|
||||
tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName)
|
||||
principal := h.getAccountID(r)
|
||||
if !CheckPermissionWithContext(operation, principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
|
||||
@@ -114,6 +114,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
|
||||
identityActions := getIdentityActions(r)
|
||||
glog.Infof("S3Tables: CreateNamespace permission check - principal=%s, owner=%s, actions=%v", principal, bucketMetadata.OwnerAccountID, identityActions)
|
||||
if !CheckPermissionWithContext("CreateNamespace", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
@@ -255,6 +256,7 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("GetNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
@@ -345,6 +347,7 @@ func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.R
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("UpdateNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
@@ -447,6 +450,7 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("ListNamespaces", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
@@ -631,6 +635,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("DeleteNamespace", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
|
||||
@@ -92,6 +92,7 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("PutTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, "", bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -170,6 +171,7 @@ func (h *S3TablesHandler) handleGetTableBucketPolicy(w http.ResponseWriter, r *h
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("GetTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, string(policy), bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -246,6 +248,7 @@ func (h *S3TablesHandler) handleDeleteTableBucketPolicy(w http.ResponseWriter, r
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("DeleteTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
@@ -345,6 +348,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("PutTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
@@ -453,6 +457,7 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("GetTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
@@ -543,6 +548,7 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("DeleteTablePolicy", principal, metadata.OwnerAccountID, bucketPolicy, tableARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
@@ -640,6 +646,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("TagResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
TableBucketTags: bucketTags,
|
||||
RequestTags: req.Tags,
|
||||
@@ -760,6 +767,7 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("ListTagsForResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tags,
|
||||
@@ -867,6 +875,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("UntagResource", principal, ownerAccountID, bucketPolicy, resourceARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
TableBucketTags: bucketTags,
|
||||
TagKeys: req.TagKeys,
|
||||
|
||||
@@ -419,26 +419,20 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
|
||||
tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespace+"/"+tableName)
|
||||
bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
|
||||
identityActions := getIdentityActions(r)
|
||||
tableAllowed := CheckPermissionWithContext("GetTable", accountID, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespace,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
bucketAllowed := CheckPermissionWithContext("GetTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespace,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
allowed := CheckTableOrBucketPermission("GetTable", accountID,
|
||||
metadata.OwnerAccountID, tablePolicy, tableARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespace,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
|
||||
if !tableAllowed && !bucketAllowed {
|
||||
if !allowed {
|
||||
h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName))
|
||||
return ErrAccessDenied
|
||||
}
|
||||
@@ -560,21 +554,16 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
|
||||
|
||||
bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
|
||||
identityActions := getIdentityActions(r)
|
||||
nsAllowed := CheckPermissionWithContext("ListTables", accountID, nsMeta.OwnerAccountID, namespacePolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
bucketAllowed := CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !nsAllowed && !bucketAllowed {
|
||||
if !CheckTableOrBucketPermission("ListTables", accountID,
|
||||
nsMeta.OwnerAccountID, namespacePolicy, bucketARN,
|
||||
bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}) {
|
||||
return ErrAccessDenied
|
||||
}
|
||||
|
||||
@@ -614,6 +603,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
|
||||
bucketARN := h.generateTableBucketARN(bucketMeta.OwnerAccountID, bucketName)
|
||||
identityActions := getIdentityActions(r)
|
||||
if !CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
@@ -950,25 +940,19 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
|
||||
bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
tableAllowed := CheckPermissionWithContext("DeleteTable", principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
bucketAllowed := CheckPermissionWithContext("DeleteTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !tableAllowed && !bucketAllowed {
|
||||
allowed := CheckTableOrBucketPermission("DeleteTable", principal,
|
||||
metadata.OwnerAccountID, tablePolicy, tableARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !allowed {
|
||||
if h.canReadEntry("GetTable", principal, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespaceName, tableName, bucketTags, tableTags, identityActions, r) {
|
||||
h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table")
|
||||
} else {
|
||||
@@ -1131,6 +1115,7 @@ func (h *S3TablesHandler) readTableMetadata(ctx context.Context, client filer_pb
|
||||
// cannot see the entry keeps the not-found veil.
|
||||
func (h *S3TablesHandler) canReadEntry(readOp, principal, tableOwner, bucketOwner, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespace, name string, bucketTags, tableTags map[string]string, identityActions []string, r *http.Request) bool {
|
||||
ctx := &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespace,
|
||||
TableName: name,
|
||||
@@ -1139,8 +1124,9 @@ func (h *S3TablesHandler) canReadEntry(readOp, principal, tableOwner, bucketOwne
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}
|
||||
return CheckPermissionWithContext(readOp, principal, tableOwner, tablePolicy, tableARN, ctx) ||
|
||||
CheckPermissionWithContext(readOp, principal, bucketOwner, bucketPolicy, bucketARN, ctx)
|
||||
return CheckTableOrBucketPermission(readOp, principal,
|
||||
tableOwner, tablePolicy, tableARN,
|
||||
bucketOwner, bucketPolicy, bucketARN, ctx)
|
||||
}
|
||||
|
||||
// handleRenameTable moves a table's catalog entry to a new namespace/name within
|
||||
@@ -1311,25 +1297,19 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
|
||||
bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
tableAllowed := CheckPermissionWithContext(kind.renameOp, principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: srcNamespace,
|
||||
TableName: srcName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
bucketAllowed := CheckPermissionWithContext(kind.renameOp, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: srcNamespace,
|
||||
TableName: srcName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !tableAllowed && !bucketAllowed {
|
||||
allowed := CheckTableOrBucketPermission(kind.renameOp, principal,
|
||||
metadata.OwnerAccountID, tablePolicy, tableARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: srcNamespace,
|
||||
TableName: srcName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !allowed {
|
||||
readTags := tableTags
|
||||
if !kind.tagsForRead {
|
||||
readTags = nil
|
||||
@@ -1384,23 +1364,17 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ
|
||||
// must also be allowed to create a table there (the source check alone lets a
|
||||
// caller move tables into namespaces they don't control). Denials report the
|
||||
// same not-found as a missing destination namespace.
|
||||
destNamespaceAllowed := CheckPermissionWithContext(kind.createOp, principal, destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: destNamespace,
|
||||
TableName: destName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
destBucketAllowed := CheckPermissionWithContext(kind.createOp, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: destNamespace,
|
||||
TableName: destName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
if !destNamespaceAllowed && !destBucketAllowed {
|
||||
if !CheckTableOrBucketPermission(kind.createOp, principal,
|
||||
destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: destNamespace,
|
||||
TableName: destName,
|
||||
TableBucketTags: bucketTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}) {
|
||||
h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", destNamespace))
|
||||
return NewAuthError(kind.renameOp, principal, "not authorized to create "+kind.noun+" in the destination namespace")
|
||||
}
|
||||
@@ -1595,26 +1569,20 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
|
||||
principal := h.getAccountID(r)
|
||||
identityActions := getIdentityActions(r)
|
||||
|
||||
tableAllowed := CheckPermissionWithContext("UpdateTable", principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
bucketAllowed := CheckPermissionWithContext("UpdateTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
allowed := CheckTableOrBucketPermission("UpdateTable", principal,
|
||||
metadata.OwnerAccountID, tablePolicy, tableARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
TableBucketTags: bucketTags,
|
||||
ResourceTags: tableTags,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
})
|
||||
|
||||
if !tableAllowed && !bucketAllowed {
|
||||
if !allowed {
|
||||
if h.canReadEntry("GetTable", principal, metadata.OwnerAccountID, bucketMetadata.OwnerAccountID, tablePolicy, bucketPolicy, tableARN, bucketARN, bucketName, namespaceName, tableName, bucketTags, tableTags, identityActions, r) {
|
||||
h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to update table")
|
||||
} else {
|
||||
@@ -1809,6 +1777,7 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re
|
||||
bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
|
||||
identityActions := getIdentityActions(r)
|
||||
policyContext := &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespaceName,
|
||||
TableName: tableName,
|
||||
@@ -1818,9 +1787,9 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}
|
||||
nsAllowed := CheckPermissionWithContext("CreateTable", accountID, namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, policyContext)
|
||||
bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, policyContext)
|
||||
if !nsAllowed && !bucketAllowed {
|
||||
if !CheckTableOrBucketPermission("CreateTable", accountID,
|
||||
namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN,
|
||||
bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, policyContext) {
|
||||
h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in this namespace")
|
||||
return nil, ErrAccessDenied
|
||||
}
|
||||
|
||||
@@ -635,6 +635,7 @@ func (h *S3TablesHandler) loadAuthContext(r *http.Request, filerClient FilerClie
|
||||
func (h *S3TablesHandler) authorizeViewOp(r *http.Request, operation, principal, resourceOwner, bucketOwner, resourcePolicy, bucketPolicy, resourceARN, bucketName, namespace, viewName string, bucketTags map[string]string) bool {
|
||||
identityActions := getIdentityActions(r)
|
||||
ctx := &PolicyContext{
|
||||
TrustedIssuers: h.trustedIssuers(r),
|
||||
TableBucketName: bucketName,
|
||||
Namespace: namespace,
|
||||
TableName: viewName,
|
||||
@@ -642,10 +643,10 @@ func (h *S3TablesHandler) authorizeViewOp(r *http.Request, operation, principal,
|
||||
IdentityActions: identityActions,
|
||||
DefaultAllow: h.defaultAllowFor(r),
|
||||
}
|
||||
resourceAllowed := CheckPermissionWithContext(operation, principal, resourceOwner, resourcePolicy, resourceARN, ctx)
|
||||
bucketARN := h.generateTableBucketARN(bucketOwner, bucketName)
|
||||
bucketAllowed := CheckPermissionWithContext(operation, principal, bucketOwner, bucketPolicy, bucketARN, ctx)
|
||||
return resourceAllowed || bucketAllowed
|
||||
return CheckTableOrBucketPermission(operation, principal,
|
||||
resourceOwner, resourcePolicy, resourceARN,
|
||||
bucketOwner, bucketPolicy, bucketARN, ctx)
|
||||
}
|
||||
|
||||
// singleClient adapts a bound filer client to the FilerClient interface so the
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/iam/sts"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
|
||||
@@ -88,6 +89,11 @@ type PolicyContext struct {
|
||||
KMSKeyArn string
|
||||
StorageClass string
|
||||
DefaultAllow bool
|
||||
// TrustedIssuers lists the OIDC issuers the deployment is configured to
|
||||
// trust, used to re-anchor owners/principals written before issuer-scoped
|
||||
// account IDs. A bare subject is only resolvable when exactly one issuer
|
||||
// is configured; more than one leaves it ambiguous, so it must not match.
|
||||
TrustedIssuers []string
|
||||
}
|
||||
|
||||
// CheckPermissionWithContext checks permission with optional resource and condition context.
|
||||
@@ -107,10 +113,67 @@ func CheckPermissionWithContext(operation, principal, owner, resourcePolicy, res
|
||||
return checkPermission(operation, principal, owner, resourcePolicy, resourceARN, ctx)
|
||||
}
|
||||
|
||||
// CheckTableOrBucketPermission combines a table or namespace policy with its
|
||||
// table bucket's policy into one decision: a matching Deny on either policy
|
||||
// vetoes the request even when the other policy would allow it.
|
||||
func CheckTableOrBucketPermission(operation, principal, resourceOwner, resourcePolicy, resourceARN, bucketOwner, bucketPolicy, bucketARN string, ctx *PolicyContext) bool {
|
||||
if principal == "" {
|
||||
return false
|
||||
}
|
||||
if principal == s3_constants.AccountAdminId {
|
||||
return true
|
||||
}
|
||||
if matchesDenyStatement(operation, principal, resourcePolicy, resourceARN, ctx) ||
|
||||
matchesDenyStatement(operation, principal, bucketPolicy, bucketARN, ctx) {
|
||||
return false
|
||||
}
|
||||
// Each side fails closed when its owner is unknown, matching
|
||||
// CheckPermissionWithContext; an ownerless side contributes nothing.
|
||||
resourceOK := resourceOwner != "" &&
|
||||
checkPermission(operation, principal, resourceOwner, resourcePolicy, resourceARN, ctx)
|
||||
bucketOK := bucketOwner != "" &&
|
||||
checkPermission(operation, principal, bucketOwner, bucketPolicy, bucketARN, ctx)
|
||||
return resourceOK || bucketOK
|
||||
}
|
||||
|
||||
// matchesDenyStatement reports whether the JSON policy carries a statement
|
||||
// matching this request with Effect Deny. Malformed policies return false;
|
||||
// the caller's normal evaluation still denies them.
|
||||
func matchesDenyStatement(operation, principal, resourcePolicy, resourceARN string, ctx *PolicyContext) bool {
|
||||
if resourcePolicy == "" {
|
||||
return false
|
||||
}
|
||||
fullAction := operation
|
||||
if !strings.Contains(operation, ":") {
|
||||
fullAction = "s3tables:" + operation
|
||||
}
|
||||
var policy PolicyDocument
|
||||
if err := json.Unmarshal([]byte(resourcePolicy), &policy); err != nil {
|
||||
return false
|
||||
}
|
||||
for _, stmt := range policy.Statement {
|
||||
if stmt.Effect != "Deny" ||
|
||||
!matchesCaller(stmt.Principal, principal, ctx) ||
|
||||
!matchesAction(stmt.Action, fullAction) ||
|
||||
(resourceARN != "" && !matchesResource(stmt.Resource, resourceARN)) ||
|
||||
!matchesConditions(stmt.Condition, ctx) {
|
||||
continue
|
||||
}
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func checkPermission(operation, principal, owner, resourcePolicy, resourceARN string, ctx *PolicyContext) bool {
|
||||
|
||||
// Owner always has permission
|
||||
if principal == owner {
|
||||
// An explicit Deny in the resource policy vetoes owner and identity
|
||||
// grants as well as allows within the same document.
|
||||
if matchesDenyStatement(operation, principal, resourcePolicy, resourceARN, ctx) {
|
||||
return false
|
||||
}
|
||||
|
||||
// Owner always has permission.
|
||||
if principal == owner || matchesLegacySubject(owner, principal, ctx) {
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -145,7 +208,7 @@ func checkPermission(operation, principal, owner, resourcePolicy, resourceARN st
|
||||
|
||||
for _, stmt := range policy.Statement {
|
||||
// Check if principal matches
|
||||
if !matchesPrincipal(stmt.Principal, principal) {
|
||||
if !matchesCaller(stmt.Principal, principal, ctx) {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -225,6 +288,30 @@ func hasAdminAction(actions []string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// matchesCaller matches a statement's Principal against the caller's scoped
|
||||
// principal or, for policies written before issuer scoping, a bare subject
|
||||
// that resolves to it under the deployment's single trusted issuer.
|
||||
func matchesCaller(principalSpec interface{}, principal string, ctx *PolicyContext) bool {
|
||||
if matchesPrincipal(principalSpec, principal) {
|
||||
return true
|
||||
}
|
||||
legacy, ok := principalSpec.(string)
|
||||
return ok && matchesLegacySubject(legacy, principal, ctx)
|
||||
}
|
||||
|
||||
// matchesLegacySubject reports whether a bare subject stored before issuer
|
||||
// scoping belongs to the caller. Re-anchoring is only sound when exactly one
|
||||
// OIDC issuer is configured: the stored subject is hashed under that issuer
|
||||
// and compared to the caller's scoped principal. With multiple issuers a bare
|
||||
// subject is ambiguous — matching it under the caller's own issuer would let
|
||||
// a same-sub identity from any configured issuer claim it.
|
||||
func matchesLegacySubject(legacy, principal string, ctx *PolicyContext) bool {
|
||||
if ctx == nil || len(ctx.TrustedIssuers) != 1 {
|
||||
return false
|
||||
}
|
||||
return sts.ComputeParentUser(legacy, ctx.TrustedIssuers[0]) == principal
|
||||
}
|
||||
|
||||
// matchesPrincipal checks if the principal matches the statement's principal
|
||||
func matchesPrincipal(principalSpec interface{}, principal string) bool {
|
||||
if principalSpec == nil {
|
||||
|
||||
@@ -3,6 +3,8 @@ package s3tables
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/iam/sts"
|
||||
)
|
||||
|
||||
func TestMatchesActionPattern(t *testing.T) {
|
||||
@@ -251,3 +253,107 @@ func TestCheckPermissionWithDefaultAllow(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckTableOrBucketPermissionDenyVeto(t *testing.T) {
|
||||
tablePolicy := `{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Deny",
|
||||
"Principal": "*",
|
||||
"Action": "s3tables:DeleteTable",
|
||||
"Resource": "*"
|
||||
}]
|
||||
}`
|
||||
bucketAllow := `{
|
||||
"Version": "2012-10-17",
|
||||
"Statement": [{
|
||||
"Effect": "Allow",
|
||||
"Principal": "*",
|
||||
"Action": "s3tables:*",
|
||||
"Resource": "*"
|
||||
}]
|
||||
}`
|
||||
tableARN := "arn:aws:s3tables:us-east-1:acct:bucket/b/table/ns/t"
|
||||
bucketARN := "arn:aws:s3tables:us-east-1:acct:bucket/b"
|
||||
ctx := &PolicyContext{TableBucketName: "b", Namespace: "ns", TableName: "t"}
|
||||
|
||||
// A Deny on the table policy vetoes the bucket Allow.
|
||||
if CheckTableOrBucketPermission("DeleteTable", "user123", "owner", tablePolicy, tableARN, "owner", bucketAllow, bucketARN, ctx) {
|
||||
t.Error("table Deny should veto bucket Allow")
|
||||
}
|
||||
|
||||
// The same Deny in the bucket policy vetoes a table Allow.
|
||||
if CheckTableOrBucketPermission("DeleteTable", "user123", "owner", bucketAllow, tableARN, "owner", tablePolicy, bucketARN, ctx) {
|
||||
t.Error("bucket Deny should veto table Allow")
|
||||
}
|
||||
|
||||
// A table Deny also vetoes ownership of the table.
|
||||
if CheckTableOrBucketPermission("DeleteTable", "owner", "owner", tablePolicy, tableARN, "owner", "", bucketARN, ctx) {
|
||||
t.Error("table Deny should veto the owner grant")
|
||||
}
|
||||
|
||||
// No Deny anywhere: either policy may allow.
|
||||
if !CheckTableOrBucketPermission("DeleteTable", "user123", "other-owner", "", tableARN, "owner", bucketAllow, bucketARN, ctx) {
|
||||
t.Error("bucket Allow should grant when no policy denies")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckTableOrBucketPermissionOwnerlessFailsClosed(t *testing.T) {
|
||||
ctx := &PolicyContext{
|
||||
IdentityActions: []string{"s3tables:*"},
|
||||
DefaultAllow: true,
|
||||
}
|
||||
allowPolicy := `{"Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`
|
||||
|
||||
if CheckTableOrBucketPermission("GetTable", "user1",
|
||||
"", "", "arn:aws:s3tables:t",
|
||||
"", "", "arn:aws:s3tables:b", ctx) {
|
||||
t.Error("identity actions and default allow must not reach an ownerless resource")
|
||||
}
|
||||
|
||||
// With no identity or default grants in play, an ownerless resource
|
||||
// side contributes nothing even when its own policy allows.
|
||||
plainCtx := &PolicyContext{}
|
||||
if CheckTableOrBucketPermission("GetTable", "user1",
|
||||
"", allowPolicy, "arn:aws:s3tables:t",
|
||||
"owner", "", "arn:aws:s3tables:b", plainCtx) {
|
||||
t.Error("an ownerless resource side must not grant even via an explicit allow")
|
||||
}
|
||||
|
||||
if !CheckTableOrBucketPermission("GetTable", "user1",
|
||||
"", "", "arn:aws:s3tables:t",
|
||||
"owner", allowPolicy, "arn:aws:s3tables:b", plainCtx) {
|
||||
t.Error("the bucket side can still grant when its owner is set")
|
||||
}
|
||||
}
|
||||
|
||||
// Owners written before issuer scoping hold a bare subject. Re-anchoring it
|
||||
// is only unambiguous when exactly one OIDC issuer is configured; under two
|
||||
// or more issuers a bare subject could belong to either, so it must not
|
||||
// match anyone.
|
||||
func TestCheckPermissionLegacyOwnerScopedToIssuer(t *testing.T) {
|
||||
owner := "alice-sub"
|
||||
principal := sts.ComputeParentUser(owner, "corp-idp")
|
||||
|
||||
if !CheckPermissionWithContext("GetTable", principal, owner, "", "arn", &PolicyContext{
|
||||
TrustedIssuers: []string{"corp-idp"},
|
||||
}) {
|
||||
t.Error("single-issuer deployments must keep recognizing the legacy owner")
|
||||
}
|
||||
|
||||
if CheckPermissionWithContext("GetTable", principal, owner, "", "arn", &PolicyContext{
|
||||
TrustedIssuers: []string{"corp-idp", "evil-idp"},
|
||||
}) {
|
||||
t.Error("a bare subject is ambiguous across multiple issuers and must not match")
|
||||
}
|
||||
|
||||
if CheckPermissionWithContext("GetTable", sts.ComputeParentUser(owner, "evil-idp"), owner, "", "arn", &PolicyContext{
|
||||
TrustedIssuers: []string{"corp-idp", "evil-idp"},
|
||||
}) {
|
||||
t.Error("a same-sub caller from another issuer must not claim the legacy owner")
|
||||
}
|
||||
|
||||
if CheckPermissionWithContext("GetTable", principal, owner, "", "arn", &PolicyContext{}) {
|
||||
t.Error("legacy owner matching must not engage without a trusted issuer")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/util"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
@@ -343,6 +344,8 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
|
||||
ConditionKey: req.ConditionKey,
|
||||
RouteKey: req.RouteKey,
|
||||
IsMoved: true,
|
||||
RequestId: req.RequestId,
|
||||
ConditionAltKey: req.ConditionAltKey,
|
||||
}
|
||||
var resp *filer_pb.ObjectTransactionResponse
|
||||
handled, err := fs.forwardToWriteOwner(ctx, req.RouteKey, func(owner pb.ServerAddress) error {
|
||||
@@ -365,6 +368,28 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
|
||||
pathLock := fs.entryLockTable.AcquireLock("ObjectTransaction", lockPath, util.ExclusiveLock)
|
||||
defer fs.entryLockTable.ReleaseLock(lockPath, pathLock)
|
||||
|
||||
// Under the same lock the first application committed with, a failover
|
||||
// replay of the same request replays its recorded outcome instead of
|
||||
// re-applying mutations over whatever superseded it.
|
||||
if resp, found := fs.recentObjectTxn(req.RequestId, lockPath); found {
|
||||
glog.V(2).InfofCtx(ctx, "ObjectTransaction %s: replaying recorded outcome for request %s", req.LockKey, req.RequestId)
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
resp := &filer_pb.ObjectTransactionResponse{}
|
||||
// Outcomes are replayed rather than re-run: a mutation error may follow a
|
||||
// partial apply, and a retry that re-runs could write the stale mutation
|
||||
// over whatever superseded it, while replaying the recorded failure at
|
||||
// least matches what the first attempt reported. The one exception is the
|
||||
// condition lookup below: it fails before any mutation runs, so the retry
|
||||
// must re-evaluate rather than replay an empty outcome.
|
||||
record := true
|
||||
defer func() {
|
||||
if record {
|
||||
fs.rememberObjectTxn(req.RequestId, lockPath, resp)
|
||||
}
|
||||
}()
|
||||
|
||||
if conditionIsSet(req.Condition) {
|
||||
// The condition is evaluated against condition_key when set (e.g. a
|
||||
// version entry whose WORM guards gate the delete), while the lock stays
|
||||
@@ -374,29 +399,151 @@ func (fs *FilerServer) ObjectTransaction(ctx context.Context, req *filer_pb.Obje
|
||||
conditionPath = util.FullPath(req.ConditionKey)
|
||||
}
|
||||
current, findErr := fs.filer.FindEntry(ctx, conditionPath)
|
||||
if findErr == nil && current != nil && current.IsDirectory() && req.ConditionAltKey != "" {
|
||||
// A directory is not the object: implicit directories can share a
|
||||
// path with an S3 key, so treat it as an absent slot. Only the
|
||||
// alt-key callers want object-slot semantics; an explicit
|
||||
// condition_key may legitimately name a directory.
|
||||
current, findErr = nil, filer_pb.ErrNotFound
|
||||
}
|
||||
// The caller can name a fallback evaluated when the primary is absent —
|
||||
// a suspended-versioning write checks "the null slot if present, else
|
||||
// the latest version" atomically under this lock. condition_alt_key is
|
||||
// the object's .versions directory, and the latest it resolves to is
|
||||
// read from the pointer maintained by the same lock, so a concurrent
|
||||
// version delete cannot move which entry the condition sees.
|
||||
if findErr == filer_pb.ErrNotFound && req.ConditionAltKey != "" {
|
||||
current, conditionPath, findErr = fs.latestVersionedConditionEntry(ctx, util.FullPath(req.ConditionAltKey))
|
||||
}
|
||||
if findErr != nil && findErr != filer_pb.ErrNotFound {
|
||||
return &filer_pb.ObjectTransactionResponse{}, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, findErr)
|
||||
record = false // a lookup error decides nothing; a retry must re-run, not replay
|
||||
return nil, fmt.Errorf("ObjectTransaction condition %s: %w", conditionPath, findErr)
|
||||
}
|
||||
if findErr == filer_pb.ErrNotFound {
|
||||
current = nil
|
||||
}
|
||||
if !writeConditionSatisfied(req.Condition, current) {
|
||||
glog.V(3).InfofCtx(ctx, "ObjectTransaction %s: precondition failed", conditionPath)
|
||||
return &filer_pb.ObjectTransactionResponse{
|
||||
Error: "precondition failed",
|
||||
ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED,
|
||||
}, nil
|
||||
resp.Error = "precondition failed"
|
||||
resp.ErrorCode = filer_pb.FilerError_PRECONDITION_FAILED
|
||||
return resp, nil
|
||||
}
|
||||
}
|
||||
|
||||
for i, m := range req.Mutations {
|
||||
if err := fs.applyObjectMutation(ctx, m, req.IsFromOtherCluster, req.Signatures); err != nil {
|
||||
glog.V(2).InfofCtx(ctx, "ObjectTransaction %s mutation %d (%v): %v", lockPath, i, m.Type, err)
|
||||
return &filer_pb.ObjectTransactionResponse{Error: fmt.Sprintf("mutation %d: %v", i, err)}, nil
|
||||
resp.Error = fmt.Sprintf("mutation %d: %v", i, err)
|
||||
return resp, nil
|
||||
}
|
||||
}
|
||||
|
||||
return &filer_pb.ObjectTransactionResponse{}, nil
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
const (
|
||||
recentObjectTxnTTL = 2 * time.Minute
|
||||
recentObjectTxnSweepCap = 10000
|
||||
recentObjectTxnSweepPeriod = 30 * time.Second
|
||||
)
|
||||
|
||||
type recentObjectTxn struct {
|
||||
resp *filer_pb.ObjectTransactionResponse
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// recentObjectTxnKey scopes an idempotency token to the transaction's lock
|
||||
// path, so a token reused on a different object cannot replay this outcome.
|
||||
func recentObjectTxnKey(requestID string, lockPath util.FullPath) string {
|
||||
return requestID + "\x00" + string(lockPath)
|
||||
}
|
||||
|
||||
// recentObjectTxn returns the recorded outcome of an already-applied request,
|
||||
// so a failover replay of the same ObjectTransaction does not re-apply it.
|
||||
func (fs *FilerServer) recentObjectTxn(requestID string, lockPath util.FullPath) (*filer_pb.ObjectTransactionResponse, bool) {
|
||||
if requestID == "" {
|
||||
return nil, false
|
||||
}
|
||||
fs.recentObjectTxnsMu.Lock()
|
||||
defer fs.recentObjectTxnsMu.Unlock()
|
||||
|
||||
key := recentObjectTxnKey(requestID, lockPath)
|
||||
txn, found := fs.recentObjectTxns[key]
|
||||
if !found {
|
||||
return nil, false
|
||||
}
|
||||
if time.Now().After(txn.expiresAt) {
|
||||
delete(fs.recentObjectTxns, key)
|
||||
return nil, false
|
||||
}
|
||||
return txn.resp, true
|
||||
}
|
||||
|
||||
func (fs *FilerServer) rememberObjectTxn(requestID string, lockPath util.FullPath, resp *filer_pb.ObjectTransactionResponse) {
|
||||
if requestID == "" || resp == nil {
|
||||
return
|
||||
}
|
||||
fs.recentObjectTxnsMu.Lock()
|
||||
defer fs.recentObjectTxnsMu.Unlock()
|
||||
if fs.recentObjectTxns == nil {
|
||||
fs.recentObjectTxns = make(map[string]recentObjectTxn)
|
||||
}
|
||||
// A live record is never evicted early: a replay that misses re-applies the
|
||||
// transaction. The sweep is amortized so a hot cache does not scan on every
|
||||
// write; between sweeps the map may exceed the cap, still bounded by the
|
||||
// insert rate over the TTL.
|
||||
now := time.Now()
|
||||
if len(fs.recentObjectTxns) >= recentObjectTxnSweepCap {
|
||||
if now.After(fs.recentObjectTxnsSweepAt) {
|
||||
fs.recentObjectTxnsSweepAt = now.Add(recentObjectTxnSweepPeriod)
|
||||
for id, txn := range fs.recentObjectTxns {
|
||||
if now.After(txn.expiresAt) {
|
||||
delete(fs.recentObjectTxns, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(fs.recentObjectTxns) >= recentObjectTxnSweepCap {
|
||||
// Full of unexpired records: drop the new one rather than evict
|
||||
// an outcome a replay may still need.
|
||||
return
|
||||
}
|
||||
}
|
||||
fs.recentObjectTxns[recentObjectTxnKey(requestID, lockPath)] = recentObjectTxn{
|
||||
resp: resp,
|
||||
expiresAt: now.Add(recentObjectTxnTTL),
|
||||
}
|
||||
}
|
||||
|
||||
// latestVersionedConditionEntry resolves the object a conditional write should
|
||||
// be checked against inside a .versions directory: the entry the latest-version
|
||||
// pointer names, or nothing when the pointer names a delete marker — a
|
||||
// tombstone is not a current version. A missing pointer or a pointer to a file
|
||||
// that is gone means the metadata cannot answer the question, which reports an
|
||||
// error rather than guessing at either branch.
|
||||
func (fs *FilerServer) latestVersionedConditionEntry(ctx context.Context, versionsDir util.FullPath) (*filer.Entry, util.FullPath, error) {
|
||||
versionsEntry, err := fs.filer.FindEntry(ctx, versionsDir)
|
||||
if err != nil {
|
||||
return nil, versionsDir, err
|
||||
}
|
||||
if versionsEntry == nil {
|
||||
return nil, versionsDir, filer_pb.ErrNotFound
|
||||
}
|
||||
latestName := string(versionsEntry.Extended[s3_constants.ExtLatestVersionFileNameKey])
|
||||
if latestName == "" {
|
||||
return nil, versionsDir, fmt.Errorf("%s has no latest-version pointer", versionsDir)
|
||||
}
|
||||
latestPath := util.NewFullPath(string(versionsDir), latestName)
|
||||
entry, err := fs.filer.FindEntry(ctx, latestPath)
|
||||
if err == filer_pb.ErrNotFound {
|
||||
return nil, latestPath, fmt.Errorf("latest-version pointer %s names a missing entry", latestPath)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, latestPath, err
|
||||
}
|
||||
if string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" {
|
||||
return nil, latestPath, filer_pb.ErrNotFound
|
||||
}
|
||||
return entry, latestPath, nil
|
||||
}
|
||||
|
||||
// ObjectTransactionBatch applies several object transactions in one round trip,
|
||||
|
||||
@@ -2,8 +2,11 @@ package weed_server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -852,3 +855,243 @@ func TestObjectTransactionForwardsToOwner(t *testing.T) {
|
||||
t.Errorf("non-owner must forward, not apply locally")
|
||||
}
|
||||
}
|
||||
|
||||
// A failover replay carries the same request_id. Under the path lock the owner
|
||||
// replays the recorded outcome instead of re-applying the mutations over a
|
||||
// newer entry.
|
||||
func TestObjectTransactionRequestIdDedupesReplay(t *testing.T) {
|
||||
now := time.Unix(1700000000, 0)
|
||||
fs, store := newTxnTestServer(t, nil)
|
||||
|
||||
put := func(etag string) *filer_pb.ObjectMutation {
|
||||
return &filer_pb.ObjectMutation{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: "/buckets/bkt",
|
||||
Entry: &filer_pb.Entry{
|
||||
Name: "obj",
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 1},
|
||||
Extended: map[string][]byte{"etag": []byte(etag)},
|
||||
},
|
||||
}
|
||||
}
|
||||
txn := func(requestID, etag string) *filer_pb.ObjectTransactionRequest {
|
||||
return &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: "/buckets/bkt/obj",
|
||||
RequestId: requestID,
|
||||
Mutations: []*filer_pb.ObjectMutation{put(etag)},
|
||||
}
|
||||
}
|
||||
|
||||
respA, err := fs.ObjectTransaction(context.Background(), txn("req-a", "A"))
|
||||
if err != nil || respA.Error != "" {
|
||||
t.Fatalf("first put: err=%v resp=%q", err, respA.Error)
|
||||
}
|
||||
|
||||
// a newer writer lands the same path
|
||||
respB, err := fs.ObjectTransaction(context.Background(), txn("req-b", "B"))
|
||||
if err != nil || respB.Error != "" {
|
||||
t.Fatalf("second put: err=%v resp=%q", err, respB.Error)
|
||||
}
|
||||
|
||||
// the replayed request replays A's outcome; the stored entry stays B
|
||||
respReplay, err := fs.ObjectTransaction(context.Background(), txn("req-a", "A"))
|
||||
if err != nil || respReplay.Error != "" {
|
||||
t.Fatalf("replayed put: err=%v resp=%q", err, respReplay.Error)
|
||||
}
|
||||
entry := store.entries["/buckets/bkt/obj"]
|
||||
if entry == nil {
|
||||
t.Fatal("entry missing after replay")
|
||||
}
|
||||
if got := string(entry.Extended["etag"]); got != "B" {
|
||||
t.Fatalf("replayed transaction re-applied the stale entry: etag = %q, want B", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A recorded application-level failure replays to the retried request even
|
||||
// after the world changes enough for the condition to pass, and its mutation
|
||||
// still does not run.
|
||||
func TestObjectTransactionRequestIdReplaysRecordedError(t *testing.T) {
|
||||
now := time.Unix(1700000000, 0)
|
||||
fs, store := newTxnTestServer(t, nil)
|
||||
|
||||
req := &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: "/buckets/bkt/obj",
|
||||
RequestId: "req-fail",
|
||||
Condition: &filer_pb.WriteCondition{
|
||||
Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_EXISTS}},
|
||||
},
|
||||
Mutations: []*filer_pb.ObjectMutation{{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: "/buckets/bkt",
|
||||
Entry: &filer_pb.Entry{
|
||||
Name: "obj",
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 1},
|
||||
Extended: map[string][]byte{"etag": []byte("stale")},
|
||||
},
|
||||
}},
|
||||
}
|
||||
|
||||
resp, err := fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("precondition: err=%v code=%v", err, resp.GetErrorCode())
|
||||
}
|
||||
|
||||
// another writer lands the object the precondition wanted
|
||||
other := &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: "/buckets/bkt/obj",
|
||||
RequestId: "req-other",
|
||||
Mutations: []*filer_pb.ObjectMutation{{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: "/buckets/bkt",
|
||||
Entry: &filer_pb.Entry{
|
||||
Name: "obj",
|
||||
Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 2},
|
||||
Extended: map[string][]byte{"etag": []byte("live")},
|
||||
},
|
||||
}},
|
||||
}
|
||||
respOther, err := fs.ObjectTransaction(context.Background(), other)
|
||||
if err != nil || respOther.Error != "" {
|
||||
t.Fatalf("other put: err=%v resp=%q", err, respOther.Error)
|
||||
}
|
||||
|
||||
// the retry replays the recorded failure instead of re-evaluating and
|
||||
// overwriting the live entry
|
||||
resp2, err := fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil || resp2.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("replayed precondition: err=%v code=%v", err, resp2.GetErrorCode())
|
||||
}
|
||||
if got := string(store.entries["/buckets/bkt/obj"].Extended["etag"]); got != "live" {
|
||||
t.Fatalf("replayed transaction overwrote the live entry: etag = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// At capacity the cache drops new recordings rather than evicting a live
|
||||
// outcome a replay may still need.
|
||||
func TestObjectTransactionReplayCacheDropsNewAtCap(t *testing.T) {
|
||||
fs, _ := newTxnTestServer(t, nil)
|
||||
path := util.FullPath("/buckets/bkt/obj")
|
||||
resp := &filer_pb.ObjectTransactionResponse{}
|
||||
|
||||
for i := 0; i < recentObjectTxnSweepCap; i++ {
|
||||
fs.rememberObjectTxn(fmt.Sprintf("req-%d", i), path, resp)
|
||||
}
|
||||
fs.rememberObjectTxn("req-overflow", path, resp)
|
||||
|
||||
if _, found := fs.recentObjectTxn("req-0", path); !found {
|
||||
t.Fatal("a live record was evicted to admit a new one")
|
||||
}
|
||||
if _, found := fs.recentObjectTxn("req-overflow", path); found {
|
||||
t.Fatal("overflow record was recorded instead of dropped")
|
||||
}
|
||||
if len(fs.recentObjectTxns) > recentObjectTxnSweepCap {
|
||||
t.Fatalf("cache exceeded cap: %d", len(fs.recentObjectTxns))
|
||||
}
|
||||
}
|
||||
|
||||
// A suspended-versioning conditional write falls back to the latest version
|
||||
// under the transaction lock: a real latest satisfies existence checks, and a
|
||||
// delete marker reads as absent.
|
||||
func TestObjectTransactionConditionFallbackResolvesLatestPointer(t *testing.T) {
|
||||
now := time.Unix(1700000000, 0)
|
||||
seed := func(latestName, marker string) map[string]*filer.Entry {
|
||||
entries := map[string]*filer.Entry{
|
||||
"/buckets/bkt/obj/.versions": {
|
||||
Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755},
|
||||
Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte(latestName)},
|
||||
},
|
||||
}
|
||||
ext := map[string][]byte{s3_constants.ExtETagKey: []byte("abc")}
|
||||
if marker != "" {
|
||||
ext[s3_constants.ExtDeleteMarkerKey] = []byte(marker)
|
||||
}
|
||||
entries["/buckets/bkt/obj/.versions/"+latestName] = &filer.Entry{
|
||||
Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644},
|
||||
Extended: ext,
|
||||
}
|
||||
return entries
|
||||
}
|
||||
req := func() *filer_pb.ObjectTransactionRequest {
|
||||
return &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: "/buckets/bkt/obj",
|
||||
ConditionAltKey: "/buckets/bkt/obj/.versions",
|
||||
Condition: &filer_pb.WriteCondition{
|
||||
Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}},
|
||||
},
|
||||
Mutations: []*filer_pb.ObjectMutation{{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: "/buckets/bkt",
|
||||
Entry: &filer_pb.Entry{Name: "obj", Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 9}},
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
fs, store := newTxnTestServer(t, seed("v_live", ""))
|
||||
resp, err := fs.ObjectTransaction(context.Background(), req())
|
||||
if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("latest version present: err=%v code=%v, want precondition failed", err, resp.GetErrorCode())
|
||||
}
|
||||
|
||||
fs, store = newTxnTestServer(t, seed("v_gone", "true"))
|
||||
resp, err = fs.ObjectTransaction(context.Background(), req())
|
||||
if err != nil || resp.Error != "" {
|
||||
t.Fatalf("delete-marker latest: err=%v resp=%q, want applied", err, resp.Error)
|
||||
}
|
||||
if store.entries["/buckets/bkt/obj"] == nil {
|
||||
t.Fatal("mutation did not apply over a delete-marker latest")
|
||||
}
|
||||
}
|
||||
|
||||
// A bare directory at the object path — left over when /bkt/obj/child was
|
||||
// created — is not the object, so the condition must fall back to the
|
||||
// .versions resolution instead of counting the directory as an existing key.
|
||||
func TestObjectTransactionConditionIgnoresDirectory(t *testing.T) {
|
||||
now := time.Unix(1700000000, 0)
|
||||
seed := func(marker string) map[string]*filer.Entry {
|
||||
return map[string]*filer.Entry{
|
||||
"/buckets/bkt/obj": {
|
||||
Attr: filer.Attr{Inode: 1, Mtime: now, Crtime: now, Mode: 0755 | os.ModeDir},
|
||||
},
|
||||
"/buckets/bkt/obj/.versions": {
|
||||
Attr: filer.Attr{Inode: 2, Mtime: now, Crtime: now, Mode: 0755},
|
||||
Extended: map[string][]byte{s3_constants.ExtLatestVersionFileNameKey: []byte("v1")},
|
||||
},
|
||||
"/buckets/bkt/obj/.versions/v1": {
|
||||
Attr: filer.Attr{Inode: 3, Mtime: now, Crtime: now, Mode: 0644},
|
||||
Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("abc"), s3_constants.ExtDeleteMarkerKey: []byte(marker)},
|
||||
},
|
||||
}
|
||||
}
|
||||
req := &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: "/buckets/bkt/obj",
|
||||
ConditionAltKey: "/buckets/bkt/obj/.versions",
|
||||
Condition: &filer_pb.WriteCondition{
|
||||
Clauses: []*filer_pb.WriteCondition_Clause{{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}},
|
||||
},
|
||||
Mutations: []*filer_pb.ObjectMutation{{
|
||||
Type: filer_pb.ObjectMutation_PUT,
|
||||
Directory: "/buckets/bkt",
|
||||
Entry: &filer_pb.Entry{Name: "obj", Attributes: &filer_pb.FuseAttributes{Mtime: now.Unix(), FileMode: 0644, Inode: 9}},
|
||||
}},
|
||||
}
|
||||
|
||||
fs, _ := newTxnTestServer(t, seed("true"))
|
||||
resp, err := fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("directory at key with delete-marker latest: %v", err)
|
||||
}
|
||||
// The condition must pass so the write reaches the type-conflict path;
|
||||
// the directory itself rejects the PUT, which is the caller's business.
|
||||
if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("condition must not fail on a bare directory: resp=%q", resp.Error)
|
||||
}
|
||||
if !strings.Contains(resp.Error, "is a directory") {
|
||||
t.Fatalf("want the dir/file type-conflict error, got %q", resp.Error)
|
||||
}
|
||||
|
||||
fs, _ = newTxnTestServer(t, seed(""))
|
||||
resp, err = fs.ObjectTransaction(context.Background(), req)
|
||||
if err != nil || resp.ErrorCode != filer_pb.FilerError_PRECONDITION_FAILED {
|
||||
t.Fatalf("directory at key with live latest: err=%v code=%v, want precondition failed", err, resp.GetErrorCode())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -142,6 +142,13 @@ type FilerServer struct {
|
||||
recentCopyRequestsMu sync.Mutex
|
||||
recentCopyRequests map[string]recentCopyRequest
|
||||
|
||||
// outcomes of recently applied object transactions keyed by request_id, so
|
||||
// a failover replay of the same transaction replays its outcome instead of
|
||||
// re-applying its mutations over a newer entry
|
||||
recentObjectTxnsMu sync.Mutex
|
||||
recentObjectTxns map[string]recentObjectTxn
|
||||
recentObjectTxnsSweepAt time.Time
|
||||
|
||||
// credential manager for IAM operations
|
||||
CredentialManager *credential.CredentialManager
|
||||
|
||||
@@ -236,6 +243,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
|
||||
subscribers: make(map[int32]*metadataSubscriber),
|
||||
inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
|
||||
recentCopyRequests: make(map[string]recentCopyRequest),
|
||||
recentObjectTxns: make(map[string]recentObjectTxn),
|
||||
CredentialManager: option.CredentialManager,
|
||||
entryLockTable: util.NewLockTable[util.FullPath](),
|
||||
posixLocks: posixlock.NewManager(),
|
||||
|
||||
@@ -96,27 +96,9 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
|
||||
waitChan := make(chan struct{}, *concurrency)
|
||||
defer close(waitChan)
|
||||
var wg sync.WaitGroup
|
||||
var mu sync.Mutex
|
||||
var loadErr error
|
||||
getErr := func() error {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return loadErr
|
||||
}
|
||||
setErr := func(e error) {
|
||||
mu.Lock()
|
||||
if loadErr == nil {
|
||||
loadErr = e
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
|
||||
for {
|
||||
if _, err := io.ReadFull(dst, sizeBuf); err != nil {
|
||||
wg.Wait()
|
||||
if loadErr := getErr(); loadErr != nil {
|
||||
return loadErr
|
||||
}
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
}
|
||||
@@ -128,13 +110,11 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
|
||||
data := make([]byte, int(size))
|
||||
|
||||
if _, err := io.ReadFull(dst, data); err != nil {
|
||||
wg.Wait()
|
||||
return err
|
||||
}
|
||||
|
||||
fullEntry := &filer_pb.FullEntry{}
|
||||
if err = proto.Unmarshal(data, fullEntry); err != nil {
|
||||
wg.Wait()
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -159,9 +139,6 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
|
||||
fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x")
|
||||
if fullEntry.Entry.IsDirectory {
|
||||
wg.Wait()
|
||||
if loadErr := getErr(); loadErr != nil {
|
||||
return loadErr
|
||||
}
|
||||
if errEntry := filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{
|
||||
Directory: fullEntry.Dir,
|
||||
Entry: fullEntry.Entry,
|
||||
@@ -177,14 +154,13 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
|
||||
Directory: entry.Dir,
|
||||
Entry: entry.Entry,
|
||||
}); errEntry != nil {
|
||||
setErr(errEntry)
|
||||
err = errEntry
|
||||
}
|
||||
defer wg.Done()
|
||||
<-waitChan
|
||||
}(fullEntry)
|
||||
if loadErr := getErr(); loadErr != nil {
|
||||
wg.Wait()
|
||||
return loadErr
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fileCount++
|
||||
}
|
||||
|
||||
@@ -149,13 +149,16 @@ func markVolumeReplicaWritable(ctx context.Context, grpcDialOption grpc.DialOpti
|
||||
return markVolumeWritable(ctx, grpcDialOption, volumeId, location.ServerAddress(), writable, persist)
|
||||
}
|
||||
|
||||
func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) error {
|
||||
// markVolumeReplicasWritable marks each replica and returns the ones that
|
||||
// took the mark, so a caller can roll back or report a partial operation.
|
||||
func markVolumeReplicasWritable(ctx context.Context, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, locations []wdclient.Location, writable, persist bool) (marked []wdclient.Location, err error) {
|
||||
for _, location := range locations {
|
||||
if err := markVolumeReplicaWritable(ctx, grpcDialOption, volumeId, location, writable, persist); err != nil {
|
||||
return err
|
||||
if err = markVolumeReplicaWritable(ctx, grpcDialOption, volumeId, location, writable, persist); err != nil {
|
||||
return
|
||||
}
|
||||
marked = append(marked, location)
|
||||
}
|
||||
return nil
|
||||
return
|
||||
}
|
||||
|
||||
// replicateVolumeToServer copies a volume from sourceAddress to targetAddress via the VolumeCopy gRPC stream.
|
||||
|
||||
@@ -9,10 +9,12 @@ import (
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/operation"
|
||||
"github.com/seaweedfs/seaweedfs/weed/operation/volume_move"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -112,7 +114,7 @@ func (c *commandVolumeTierCompact) Do(args []string, commandEnv *CommandEnv, wri
|
||||
|
||||
var failedCount int
|
||||
for _, rv := range remoteVolumes {
|
||||
if err = doVolumeTierCompact(commandEnv, writer, rv, *garbageThreshold, *concurrency); err != nil {
|
||||
if err = doVolumeTierCompact(commandEnv, writer, topologyInfo, rv, *garbageThreshold, *concurrency); err != nil {
|
||||
fmt.Fprintf(writer, "error compacting volume %d: %v\n", rv.vid, err)
|
||||
failedCount++
|
||||
}
|
||||
@@ -199,11 +201,49 @@ func collectRemoteVolumesWithInfo(topoInfo *master_pb.TopologyInfo, collectionPa
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error {
|
||||
// collectVolumeTierCompactLocations returns the volume's tiered replicas first
|
||||
// followed by any local ones. Every tiered replica references the same remote
|
||||
// object, so the source is always a tiered replica and the last tiered replica
|
||||
// doubles as the one that can retire the old object after the resync.
|
||||
func collectVolumeTierCompactLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string) (tiered, local []wdclient.Location) {
|
||||
eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) {
|
||||
for _, disk := range dn.DiskInfos {
|
||||
for _, vi := range disk.VolumeInfos {
|
||||
if needle.VolumeId(vi.Id) != vid || vi.Collection != collection {
|
||||
continue
|
||||
}
|
||||
loc := wdclient.Location{
|
||||
Url: dn.Id,
|
||||
PublicUrl: dn.Id,
|
||||
GrpcPort: int(dn.GrpcPort),
|
||||
DataCenter: string(dc),
|
||||
}
|
||||
if vi.RemoteStorageName != "" {
|
||||
tiered = append(tiered, loc)
|
||||
} else {
|
||||
local = append(local, loc)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, topoInfo *master_pb.TopologyInfo, rv remoteVolumeInfo, garbageThreshold float64, concurrency int) error {
|
||||
grpcDialOption := commandEnv.option.GrpcDialOption
|
||||
ctx := context.Background()
|
||||
|
||||
tieredLocations, localLocations := collectVolumeTierCompactLocations(topoInfo, rv.vid, rv.collection)
|
||||
if len(tieredLocations) == 0 {
|
||||
return fmt.Errorf("volume %d has no remote replica", rv.vid)
|
||||
}
|
||||
source := tieredLocations[0].ServerAddress()
|
||||
// all replicas of the volume: tiered ones share one remote object, and any
|
||||
// local copy is re-pointed at the new object during the resync below
|
||||
locations := append(append([]wdclient.Location{}, tieredLocations...), localLocations...)
|
||||
|
||||
// step 1: check garbage level
|
||||
garbageRatio, err := checkVolumeGarbage(grpcDialOption, rv.vid, rv.serverAddress)
|
||||
garbageRatio, err := checkVolumeGarbage(grpcDialOption, rv.vid, source)
|
||||
if err != nil {
|
||||
return fmt.Errorf("check garbage for volume %d: %v", rv.vid, err)
|
||||
}
|
||||
@@ -216,43 +256,209 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu
|
||||
|
||||
fmt.Fprintf(writer, "volume %d garbage ratio %.4f, starting compaction...\n", rv.vid, garbageRatio)
|
||||
|
||||
// step 2: download .dat from remote to local
|
||||
// this deletes the remote file and reloads the volume as local, then re-uploads below
|
||||
fmt.Fprintf(writer, " downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName)
|
||||
err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false, concurrency)
|
||||
// step 2: freeze every replica. The mark is persisted because the download
|
||||
// remounts the volume into a fresh in-memory state, and a local .dat with no
|
||||
// persisted readonly flag would load writable while its siblings stay readonly.
|
||||
frozen, err := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, locations, false, true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("download volume %d from remote: %v", rv.vid, err)
|
||||
// nothing changed yet: thaw the replicas already marked
|
||||
if _, thawErr := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, frozen, true, false); thawErr != nil {
|
||||
return fmt.Errorf("mark volume %d readonly: %v (thawing the %d marked replicas also failed: %v)",
|
||||
rv.vid, err, len(frozen), thawErr)
|
||||
}
|
||||
return fmt.Errorf("mark volume %d readonly: %v", rv.vid, err)
|
||||
}
|
||||
// Past this point a failure keeps every replica frozen on purpose: a replica
|
||||
// mid-transition may hold divergent data, and thawing it would expose that.
|
||||
// Re-running volume.tier.compact completes the resync and releases the marks.
|
||||
stillFrozen := func(stageErr error) error {
|
||||
return fmt.Errorf("%v (volume %d replicas stay readonly; rerun volume.tier.compact to resume)", stageErr, rv.vid)
|
||||
}
|
||||
|
||||
// step 3: compact the local volume
|
||||
fmt.Fprintf(writer, " compacting volume %d...\n", rv.vid)
|
||||
err = compactVolumeOnServer(grpcDialOption, writer, rv.vid, rv.serverAddress)
|
||||
// step 3: download .dat from remote to local; the shared object stays while
|
||||
// sibling replicas still reference it
|
||||
fmt.Fprintf(writer, " downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName)
|
||||
err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, source, len(tieredLocations) > 1, concurrency)
|
||||
if err != nil {
|
||||
// compaction failed, but volume is now local without remote reference
|
||||
// upload the uncompacted volume back to restore cloud tier state
|
||||
fmt.Fprintf(writer, " compaction failed: %v\n", err)
|
||||
return stillFrozen(fmt.Errorf("download volume %d from remote: %v", rv.vid, err))
|
||||
}
|
||||
|
||||
// step 4: compact the local volume; on failure re-upload what is there so
|
||||
// the replica returns to the tier and the resync below can still proceed
|
||||
compactErr := compactVolumeOnServer(grpcDialOption, writer, rv.vid, source)
|
||||
if compactErr != nil {
|
||||
fmt.Fprintf(writer, " compaction failed: %v\n", compactErr)
|
||||
fmt.Fprintf(writer, " re-uploading volume %d to %s without compaction...\n", rv.vid, rv.remoteStorageName)
|
||||
uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
|
||||
if uploadErr != nil {
|
||||
return fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local",
|
||||
err, uploadErr, rv.vid)
|
||||
}
|
||||
|
||||
// step 5: upload the (possibly uncompacted) volume back to remote under a new key
|
||||
fmt.Fprintf(writer, " uploading volume %d to %s...\n", rv.vid, rv.remoteStorageName)
|
||||
uploadErr := uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, source, rv.remoteStorageName, false, concurrency)
|
||||
if uploadErr != nil {
|
||||
if compactErr != nil {
|
||||
return stillFrozen(fmt.Errorf("compaction failed (%v) and re-upload also failed (%v), volume %d remains local",
|
||||
compactErr, uploadErr, rv.vid))
|
||||
}
|
||||
return stillFrozen(fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)",
|
||||
rv.vid, rv.remoteStorageName, uploadErr))
|
||||
}
|
||||
|
||||
// step 6: push the new .idx/.vif to every sibling so all replicas reference
|
||||
// the new remote object; the last tiered replica goes last so the old shared
|
||||
// object still has a holder until then. Replicas stay frozen until every
|
||||
// copy lands, so a delete cannot slip past a sibling's index snapshot; a
|
||||
// copied readonly mark is inert on a remote-backed volume anyway.
|
||||
syncErr := syncVolumeTierCompactReplicas(ctx, grpcDialOption, writer, rv, tieredLocations, localLocations)
|
||||
if compactErr != nil {
|
||||
if syncErr != nil {
|
||||
return stillFrozen(fmt.Errorf("compaction failed (%v), volume %d re-uploaded without compaction but resync failed: %v",
|
||||
compactErr, rv.vid, syncErr))
|
||||
}
|
||||
return fmt.Errorf("compaction failed (%v), volume %d re-uploaded to %s without compaction",
|
||||
err, rv.vid, rv.remoteStorageName)
|
||||
compactErr, rv.vid, rv.remoteStorageName)
|
||||
}
|
||||
if syncErr != nil {
|
||||
return stillFrozen(syncErr)
|
||||
}
|
||||
|
||||
// step 4: upload compacted volume back to remote
|
||||
fmt.Fprintf(writer, " uploading compacted volume %d to %s...\n", rv.vid, rv.remoteStorageName)
|
||||
err = uploadDatToRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, rv.remoteStorageName, false, concurrency)
|
||||
// step 7: unfreeze every replica, clearing the persisted readonly mark on
|
||||
// each. Remote volumes refuse writes regardless; this re-enables deletes.
|
||||
thawed, err := markVolumeReplicasWritable(ctx, grpcDialOption, rv.vid, locations, true, false)
|
||||
if err != nil {
|
||||
return fmt.Errorf("upload compacted volume %d to %s: %v (volume remains local with compacted data)",
|
||||
rv.vid, rv.remoteStorageName, err)
|
||||
return fmt.Errorf("unfreeze volume %d replicas: %v (%d of %d released)", rv.vid, err, len(thawed), len(locations))
|
||||
}
|
||||
|
||||
fmt.Fprintf(writer, "volume %d compacted and uploaded to %s successfully\n", rv.vid, rv.remoteStorageName)
|
||||
return nil
|
||||
}
|
||||
|
||||
// planTierCompactResync orders the sibling resyncs. The last tiered replica is
|
||||
// held out for last: the resync retires a remote object only once its final
|
||||
// holder is re-pointed, so processing the old object's expected last holder at
|
||||
// the end keeps it alive for every sibling still referencing it. A
|
||||
// single-replica volume has no such holder: its download already removed the
|
||||
// object.
|
||||
func planTierCompactResync(tieredLocations, localLocations []wdclient.Location) (order, deleter []wdclient.Location) {
|
||||
rest := append(append([]wdclient.Location{}, tieredLocations[1:]...), localLocations...)
|
||||
if len(tieredLocations) > 1 {
|
||||
last := tieredLocations[len(tieredLocations)-1]
|
||||
for _, loc := range rest {
|
||||
if loc.ServerAddress() == last.ServerAddress() {
|
||||
deleter = append(deleter, loc)
|
||||
continue
|
||||
}
|
||||
order = append(order, loc)
|
||||
}
|
||||
return
|
||||
}
|
||||
return rest, nil
|
||||
}
|
||||
|
||||
func syncVolumeTierCompactReplicas(ctx context.Context, grpcDialOption grpc.DialOption, writer io.Writer, rv remoteVolumeInfo, tieredLocations, localLocations []wdclient.Location) error {
|
||||
source := tieredLocations[0].ServerAddress()
|
||||
mover := volume_move.NewMover(grpcDialOption)
|
||||
|
||||
resync := func(loc wdclient.Location) error {
|
||||
fmt.Fprintf(writer, " replicating volume %d metadata from %s to %s\n", rv.vid, tieredLocations[0].Url, loc.Url)
|
||||
return replicateVolumeToServer(ctx, grpcDialOption, writer, rv.vid, source, loc.ServerAddress(), "", 0)
|
||||
}
|
||||
|
||||
order, deleter := planTierCompactResync(tieredLocations, localLocations)
|
||||
|
||||
// Every replica's current remote objects, so a resynced-over replica can
|
||||
// retire an object once it is the last holder — covers the shared object
|
||||
// and any object a previous interrupted run left referenced. The source
|
||||
// counts as a holder of the new object so a replica already sharing it is
|
||||
// never mistaken for the last holder.
|
||||
allLocations := append(append([]wdclient.Location{}, tieredLocations...), localLocations...)
|
||||
holders, readErr := readRemoteObjectHolders(ctx, grpcDialOption, rv.vid, allLocations)
|
||||
if readErr != nil {
|
||||
// Fall back to the explicit last-replica deleter when a replica's
|
||||
// remote files cannot be read.
|
||||
for _, loc := range order {
|
||||
if err := resync(loc); err != nil {
|
||||
return fmt.Errorf("replicate volume %d from %s to %s: %v", rv.vid, tieredLocations[0].Url, loc.Url, err)
|
||||
}
|
||||
}
|
||||
if len(deleter) == 0 {
|
||||
return nil
|
||||
}
|
||||
fmt.Fprintf(writer, " removing old remote copy of volume %d on %s\n", rv.vid, deleter[0].Url)
|
||||
if err := mover.DeleteVolume(ctx, rv.vid, deleter[0].ServerAddress(), false, false, false); err != nil {
|
||||
return fmt.Errorf("delete old copy of volume %d on %s: %v", rv.vid, deleter[0].Url, err)
|
||||
}
|
||||
return resync(deleter[0])
|
||||
}
|
||||
|
||||
for _, loc := range append(append([]wdclient.Location{}, order...), deleter...) {
|
||||
files := holders.files[loc.ServerAddress()]
|
||||
soleHolder := len(files) > 0
|
||||
for _, f := range files {
|
||||
if holders.count[remoteFileKey(f)] > 1 {
|
||||
soleHolder = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if soleHolder {
|
||||
// This replica is the only holder of its remote objects: removing
|
||||
// its copy retires them, and the resync restores the replica on the
|
||||
// new object.
|
||||
fmt.Fprintf(writer, " removing old remote copy of volume %d on %s\n", rv.vid, loc.Url)
|
||||
if err := mover.DeleteVolume(ctx, rv.vid, loc.ServerAddress(), false, false, false); err != nil {
|
||||
return fmt.Errorf("delete old copy of volume %d on %s: %v", rv.vid, loc.Url, err)
|
||||
}
|
||||
}
|
||||
for _, f := range files {
|
||||
holders.count[remoteFileKey(f)]--
|
||||
}
|
||||
if err := resync(loc); err != nil {
|
||||
return fmt.Errorf("replicate volume %d from %s to %s: %v", rv.vid, tieredLocations[0].Url, loc.Url, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// remoteObjectHolders reads every replica's remote file list and counts how
|
||||
// many replicas reference each remote object.
|
||||
type remoteObjectHolders struct {
|
||||
files map[pb.ServerAddress][]*volume_server_pb.RemoteFile
|
||||
count map[string]int
|
||||
}
|
||||
|
||||
func remoteFileKey(f *volume_server_pb.RemoteFile) string {
|
||||
return f.BackendType + "/" + f.BackendId + "/" + f.Key
|
||||
}
|
||||
|
||||
func readRemoteObjectHolders(ctx context.Context, grpcDialOption grpc.DialOption, vid needle.VolumeId, locations []wdclient.Location) (*remoteObjectHolders, error) {
|
||||
h := &remoteObjectHolders{
|
||||
files: make(map[pb.ServerAddress][]*volume_server_pb.RemoteFile),
|
||||
count: make(map[string]int),
|
||||
}
|
||||
for _, loc := range locations {
|
||||
addr := loc.ServerAddress()
|
||||
err := operation.WithVolumeServerClient(false, addr, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
|
||||
resp, reqErr := client.ReadVolumeFileStatus(ctx, &volume_server_pb.ReadVolumeFileStatusRequest{
|
||||
VolumeId: uint32(vid),
|
||||
})
|
||||
if reqErr != nil {
|
||||
return reqErr
|
||||
}
|
||||
if resp == nil || resp.VolumeInfo == nil {
|
||||
return fmt.Errorf("volume %d status has no remote file info", vid)
|
||||
}
|
||||
h.files[addr] = resp.VolumeInfo.GetFiles()
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, f := range h.files[addr] {
|
||||
h.count[remoteFileKey(f)]++
|
||||
}
|
||||
}
|
||||
return h, nil
|
||||
}
|
||||
|
||||
func checkVolumeGarbage(grpcDialOption grpc.DialOption, vid needle.VolumeId, server pb.ServerAddress) (float64, error) {
|
||||
var garbageRatio float64
|
||||
err := operation.WithVolumeServerClient(false, server, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
@@ -253,3 +254,89 @@ func TestCollectRemoteVolumesWithInfoInvalidPattern(t *testing.T) {
|
||||
_, err := collectRemoteVolumesWithInfo(topo, "[invalid")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func buildTestTopologyWithReplicas() *master_pb.TopologyInfo {
|
||||
mkDn := func(id string, volumes ...*master_pb.VolumeInformationMessage) *master_pb.DataNodeInfo {
|
||||
return &master_pb.DataNodeInfo{
|
||||
Id: id,
|
||||
GrpcPort: 18080,
|
||||
DiskInfos: map[string]*master_pb.DiskInfo{
|
||||
"": {VolumeInfos: volumes},
|
||||
},
|
||||
}
|
||||
}
|
||||
remote := func(vid uint32) *master_pb.VolumeInformationMessage {
|
||||
return &master_pb.VolumeInformationMessage{
|
||||
Id: vid,
|
||||
Collection: "col1",
|
||||
RemoteStorageName: "s3.default",
|
||||
RemoteStorageKey: "col1/vid.dat",
|
||||
}
|
||||
}
|
||||
local := func(vid uint32) *master_pb.VolumeInformationMessage {
|
||||
return &master_pb.VolumeInformationMessage{Id: vid, Collection: "col1"}
|
||||
}
|
||||
return &master_pb.TopologyInfo{
|
||||
DataCenterInfos: []*master_pb.DataCenterInfo{
|
||||
{
|
||||
RackInfos: []*master_pb.RackInfo{
|
||||
{
|
||||
DataNodeInfos: []*master_pb.DataNodeInfo{
|
||||
mkDn("n1:8080", remote(1)),
|
||||
mkDn("n2:8080", remote(1), local(2)),
|
||||
mkDn("n3:8080", remote(1)),
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestCollectVolumeTierCompactLocations(t *testing.T) {
|
||||
topo := buildTestTopologyWithReplicas()
|
||||
|
||||
tiered, local := collectVolumeTierCompactLocations(topo, needle.VolumeId(1), "col1")
|
||||
assert.Equal(t, 3, len(tiered))
|
||||
assert.Equal(t, 0, len(local))
|
||||
|
||||
tiered, local = collectVolumeTierCompactLocations(topo, needle.VolumeId(2), "col1")
|
||||
assert.Equal(t, 0, len(tiered))
|
||||
assert.Equal(t, 1, len(local))
|
||||
|
||||
// a different collection or volume id matches nothing
|
||||
tiered, local = collectVolumeTierCompactLocations(topo, needle.VolumeId(1), "other")
|
||||
assert.Equal(t, 0, len(tiered))
|
||||
assert.Equal(t, 0, len(local))
|
||||
}
|
||||
|
||||
func TestPlanTierCompactResync(t *testing.T) {
|
||||
loc := func(url string) wdclient.Location {
|
||||
return wdclient.Location{Url: url, GrpcPort: 1}
|
||||
}
|
||||
urls := func(locs []wdclient.Location) (out []string) {
|
||||
for _, l := range locs {
|
||||
out = append(out, l.Url)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// single replica: nothing to resync, nothing to delete (its download
|
||||
// already removed the object)
|
||||
order, deleter := planTierCompactResync([]wdclient.Location{loc("a")}, nil)
|
||||
assert.Empty(t, order)
|
||||
assert.Empty(t, deleter)
|
||||
|
||||
// two tiered replicas: the sibling is the last holder and goes last
|
||||
order, deleter = planTierCompactResync([]wdclient.Location{loc("a"), loc("b")}, nil)
|
||||
assert.Empty(t, order)
|
||||
assert.Equal(t, []string{"b"}, urls(deleter))
|
||||
|
||||
// three tiered replicas plus one local copy: middle replicas resync in
|
||||
// order, the last tiered replica deletes and is resynced last
|
||||
order, deleter = planTierCompactResync(
|
||||
[]wdclient.Location{loc("a"), loc("b"), loc("c")},
|
||||
[]wdclient.Location{loc("d")})
|
||||
assert.Equal(t, []string{"b", "d"}, urls(order))
|
||||
assert.Equal(t, []string{"c"}, urls(deleter))
|
||||
}
|
||||
|
||||
@@ -322,7 +322,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i
|
||||
}
|
||||
|
||||
// mark all replicas as read only
|
||||
if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil {
|
||||
if _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, false, false); err != nil {
|
||||
return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
|
||||
}
|
||||
newAddress := pb.NewServerAddressFromDataNode(dst.dataNode)
|
||||
@@ -337,7 +337,7 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i
|
||||
// — reopening the replicas beside that copy would fork the volume.
|
||||
if !errors.Is(moveErr, volume_move.ErrSourceKeptReadonly) {
|
||||
// mark all replicas as writable
|
||||
if err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil {
|
||||
if _, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, locations, true, false); err != nil {
|
||||
glog.Errorf("mark volume %d as writable on %s: %v", vid, locations[0].Url, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -153,7 +153,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str
|
||||
return fmt.Errorf("volume %d not found in collection %s", vid, collection)
|
||||
}
|
||||
|
||||
err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false)
|
||||
_, err = markVolumeReplicasWritable(context.Background(), commandEnv.option.GrpcDialOption, vid, existingLocations, false, false)
|
||||
if err != nil {
|
||||
return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, existingLocations[0].Url, err)
|
||||
}
|
||||
|
||||
@@ -36,6 +36,7 @@ type localDirBackend struct {
|
||||
|
||||
mu sync.Mutex
|
||||
deletes []string // history of DeleteFile keys, for assertions
|
||||
failDel bool // when set, DeleteFile reports an error without deleting
|
||||
}
|
||||
|
||||
func newLocalDirBackend(t *testing.T) *localDirBackend {
|
||||
@@ -97,11 +98,20 @@ func (b *localDirBackend) DownloadFile(fileName string, key string, fn func(prog
|
||||
|
||||
func (b *localDirBackend) DeleteFile(key string) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.deletes = append(b.deletes, key)
|
||||
b.mu.Unlock()
|
||||
if b.failDel {
|
||||
return fmt.Errorf("injected remote delete failure for %s", key)
|
||||
}
|
||||
return os.Remove(filepath.Join(b.root, key))
|
||||
}
|
||||
|
||||
func (b *localDirBackend) failDeletes(fail bool) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.failDel = fail
|
||||
}
|
||||
|
||||
func (b *localDirBackend) deleteHistory() []string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
@@ -369,6 +379,34 @@ func TestRemoteTier_RealDelete_RemovesRemoteObject(t *testing.T) {
|
||||
require.Equal(t, []string{key}, b.deleteHistory())
|
||||
}
|
||||
|
||||
// TestRemoteTier_FailedRemoteDelete_KeepsLocalVolume locks in the failure
|
||||
// ordering: a backend delete error must abort Destroy before local files are
|
||||
// removed, leaving the registered replica fully intact and retryable.
|
||||
func TestRemoteTier_FailedRemoteDelete_KeepsLocalVolume(t *testing.T) {
|
||||
b := newLocalDirBackend(t)
|
||||
registerTestBackend(t, b)
|
||||
|
||||
dir := t.TempDir()
|
||||
const vid = needle.VolumeId(34)
|
||||
_, key := tierUpVolume(t, dir, vid, b)
|
||||
require.True(t, b.objectExists(key))
|
||||
|
||||
v := reloadVolume(t, dir, vid)
|
||||
require.True(t, v.HasRemoteFile())
|
||||
|
||||
b.failDeletes(true)
|
||||
err := v.Destroy(false, false, false)
|
||||
require.Error(t, err, "Destroy must surface the remote delete failure")
|
||||
|
||||
require.True(t, util.FileExists(filepath.Join(dir, "34.idx")), "local .idx must survive a failed remote delete")
|
||||
require.True(t, util.FileExists(filepath.Join(dir, "34.vif")), "local .vif must survive a failed remote delete")
|
||||
require.True(t, b.objectExists(key), "remote object must survive a failed remote delete")
|
||||
|
||||
b.failDeletes(false)
|
||||
require.NoError(t, v.Destroy(false, false, false), "retry must succeed once the backend recovers")
|
||||
require.False(t, b.objectExists(key))
|
||||
}
|
||||
|
||||
// TestRemoteTier_Vacuum_DoesNotDeleteRemote runs the compact paths against a
|
||||
// remote-tier volume and asserts the safety property: regardless of whether
|
||||
// compact succeeds, errors, or no-ops, it must not delete the cloud object.
|
||||
|
||||
@@ -156,15 +156,19 @@ func (v *Volume) destroyLocked(onlyEmpty bool, onlyGarbage bool, keepRemoteData
|
||||
err = fmt.Errorf("volume %d is compacting", v.Id)
|
||||
return
|
||||
}
|
||||
v.stopWorker()
|
||||
if !keepRemoteData {
|
||||
storageName, storageKey := v.RemoteStorageNameKey()
|
||||
if v.HasRemoteFile() && storageName != "" && storageKey != "" {
|
||||
if backendStorage, found := backend.BackendStorages[storageName]; found {
|
||||
backendStorage.DeleteFile(storageKey)
|
||||
storageName, storageKey := v.RemoteStorageNameKey()
|
||||
if !keepRemoteData && v.HasRemoteFile() && storageName != "" && storageKey != "" {
|
||||
if backendStorage, found := backend.BackendStorages[storageName]; found {
|
||||
if remoteErr := backendStorage.DeleteFile(storageKey); remoteErr != nil {
|
||||
// Stop short of local cleanup: a failed remote delete must
|
||||
// not strand a registered replica with no local files.
|
||||
v.isCompactionInProgress.Store(false)
|
||||
err = fmt.Errorf("volume %d remote object %s/%s delete failed: %w", v.Id, storageName, storageKey, remoteErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
v.stopWorker()
|
||||
// A regular volume and an EC volume for the same id share <base>.vif. When
|
||||
// EC artefacts coexist on this disk (e.g. shards distributed onto a source
|
||||
// replica before it is deleted), keep the .vif so removing the regular
|
||||
|
||||
Reference in New Issue
Block a user