mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-14 07:48:41 +01:00
ensureDirectory looked the path up and, when it was missing, created it with a plain CreateEntry. A table create that lost the lookup race then replaced the directory another create had made, claimed and confirmed since, erasing that table's record, so a third create could claim the name and win as well. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1422 lines
55 KiB
Go
1422 lines
55 KiB
Go
package s3tables
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
var (
|
|
ErrAttributeNotFound = errors.New("attribute not found")
|
|
ErrConcurrentUpdate = errors.New("attribute changed concurrently")
|
|
ErrNameTaken = errors.New("catalog name is taken")
|
|
ErrLocationInUse = errors.New("location is in use by another table or view")
|
|
// ErrPublishedRecordChanged is the ErrConcurrentUpdate a create returns
|
|
// when the record it wrote was already visible, and another caller changed
|
|
// it — updated it, or renamed it away — before the create could back it
|
|
// out. Whatever the create staged for that record may still be in use.
|
|
ErrPublishedRecordChanged = fmt.Errorf("published record changed before the create could back it out: %w", ErrConcurrentUpdate)
|
|
)
|
|
|
|
// extendedAttributeUpdateAttempts bounds the read-modify-write retry loop in
|
|
// updateExtendedAttribute.
|
|
const extendedAttributeUpdateAttempts = 5
|
|
|
|
// Filer operations - Common functions for interacting with the filer
|
|
|
|
// createDirectory creates a new directory at the specified path
|
|
func (h *S3TablesHandler) createDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
|
|
return h.createDirectoryEntry(ctx, client, path, false, nil)
|
|
}
|
|
|
|
// createDirectoryExclusive creates a directory entry together with its
|
|
// extended attributes in one CreateEntry that fails with
|
|
// filer_pb.ErrEntryAlreadyExists when an entry is already there: nothing is
|
|
// replaced and no reader sees the directory without its attributes.
|
|
func (h *S3TablesHandler) createDirectoryExclusive(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, extended map[string][]byte) error {
|
|
return h.createDirectoryEntry(ctx, client, path, true, extended)
|
|
}
|
|
|
|
func (h *S3TablesHandler) createDirectoryEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, exclusive bool, extended map[string][]byte) error {
|
|
dir, name := splitPath(path)
|
|
now := time.Now().Unix()
|
|
return filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{
|
|
Directory: dir,
|
|
OExcl: exclusive,
|
|
Entry: &filer_pb.Entry{
|
|
Name: name,
|
|
IsDirectory: true,
|
|
Attributes: &filer_pb.FuseAttributes{
|
|
Mtime: now,
|
|
Crtime: now,
|
|
FileMode: uint32(0755 | os.ModeDir),
|
|
},
|
|
Extended: extended,
|
|
},
|
|
})
|
|
}
|
|
|
|
// ensureDirectory ensures a directory exists at the specified path. The create
|
|
// is exclusive: a directory another writer made since the lookup may already
|
|
// carry a record, which a plain create would replace with a bare directory.
|
|
func (h *S3TablesHandler) ensureDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
|
|
dir, name := splitPath(path)
|
|
_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
})
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
if err = h.createDirectoryExclusive(ctx, client, path, nil); errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
|
|
return nil
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// deleteEntryIfExists removes an entry if it exists, ignoring missing errors
|
|
func (h *S3TablesHandler) deleteEntryIfExists(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
|
|
dir, name := splitPath(path)
|
|
return filer_pb.DoRemove(ctx, client, dir, name, true, false, true, false, nil)
|
|
}
|
|
|
|
// mutateEntryExtended applies mutate to an entry's extended attributes and
|
|
// writes the entry back, asserting that nothing on it changed in between.
|
|
// UpdateEntry rewrites the whole entry from the snapshot read here, so an
|
|
// unconditional write would delete an attribute a concurrent writer created.
|
|
func (h *S3TablesHandler) mutateEntryExtended(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
mutate func(extended map[string][]byte) error,
|
|
) error {
|
|
return h.mutateEntry(ctx, client, path, func(entry *filer_pb.Entry) error {
|
|
return mutate(entry.Extended)
|
|
})
|
|
}
|
|
|
|
// mutateEntry is mutateEntryExtended for a mutation that also needs the entry
|
|
// itself, such as its type. The entry mutate inspects is the one written back,
|
|
// and the filer refuses to write a directory entry over a file, so a check on
|
|
// the entry's type holds at the write too.
|
|
func (h *S3TablesHandler) mutateEntry(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
mutate func(entry *filer_pb.Entry) error,
|
|
) error {
|
|
return h.mutateEntryAsserting(ctx, client, path, nil, mutate)
|
|
}
|
|
|
|
// mutateEntryAsserting is mutateEntry with the keys in absent asserted at the
|
|
// write too, as this package's own attributes are: one absent when the entry was
|
|
// read and written by someone else since fails the compare, instead of being
|
|
// written over with the entry as it was read.
|
|
func (h *S3TablesHandler) mutateEntryAsserting(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
absent []string,
|
|
mutate func(entry *filer_pb.Entry) error,
|
|
) error {
|
|
dir, name := splitPath(path)
|
|
|
|
for attempt := 0; attempt < extendedAttributeUpdateAttempts; attempt++ {
|
|
resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
entry := resp.Entry
|
|
expected := snapshotExtended(entry.Extended, absent...)
|
|
if entry.Extended == nil {
|
|
entry.Extended = make(map[string][]byte)
|
|
}
|
|
if err := mutate(entry); err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
|
|
Directory: dir,
|
|
Entry: entry,
|
|
ExpectedExtended: expected,
|
|
})
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
// The filer answers an entry removed since the lookup with
|
|
// codes.NotFound; report it as the lookup's own miss.
|
|
if status.Code(err) == codes.NotFound {
|
|
return fmt.Errorf("%w: %s", filer_pb.ErrNotFound, path)
|
|
}
|
|
if status.Code(err) != codes.FailedPrecondition {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
|
|
}
|
|
|
|
// setExtendedAttribute sets an extended attribute on an existing entry
|
|
func (h *S3TablesHandler) setExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string, data []byte) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
extended[key] = data
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// setExtendedAttributes sets multiple extended attributes on an existing entry
|
|
// in a single UpdateEntry, so callers don't leave the entry partially tagged.
|
|
func (h *S3TablesHandler) setExtendedAttributes(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
for key, data := range attrs {
|
|
extended[key] = data
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// removeExtendedAttributes deletes the given extended attributes from an entry,
|
|
// leaving the directory and its children intact.
|
|
func (h *S3TablesHandler) removeExtendedAttributes(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, keys ...string) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
for _, key := range keys {
|
|
delete(extended, key)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// removeExtendedAttributesIf drops the given attributes only while they still
|
|
// hold the values the caller captured. A rename copies attributes to the
|
|
// destination and then clears the source; without this, a write that landed in
|
|
// between is deleted here and never reaches the new name.
|
|
func (h *S3TablesHandler) removeExtendedAttributesIf(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
expected map[string][]byte,
|
|
keys ...string,
|
|
) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
for key, want := range expected {
|
|
if !bytes.Equal(extended[key], want) {
|
|
return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, path)
|
|
}
|
|
}
|
|
for _, key := range keys {
|
|
delete(extended, key)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// claimCatalogEntry writes a new catalog entry's marker attributes in one
|
|
// compare-checked update. Every catalog marker must still be absent: a
|
|
// delete's restore or a competing create that landed since the caller's check
|
|
// is a name conflict, not a record this write may overwrite — an overwrite
|
|
// would leave the loser's tags, policy, and maintenance settings on the new
|
|
// record. A record still marked by an unfinished create is the exception:
|
|
// invisible to every reader, it is adopted rather than refused. A rename
|
|
// destination is not adoptable — its writer still owes the record a publish
|
|
// or a keep, and adopting it strands the renamed record with neither name.
|
|
func (h *S3TablesHandler) claimCatalogEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
return claimExtended(extended, path, attrs)
|
|
})
|
|
}
|
|
|
|
// claimExtended is claimCatalogEntry's change to the attributes it read.
|
|
func claimExtended(extended map[string][]byte, path string, attrs map[string][]byte) error {
|
|
if string(extended[ExtendedKeyDeleting]) == createClaim {
|
|
// A marked record is absent to every reader, so it holds nothing a
|
|
// caller could have seen: the create that left it confirmed or lost
|
|
// the name already. Take the record over, dropping whatever the
|
|
// earlier write left, or an abandoned mark wedges the name forever.
|
|
for _, key := range s3tablesExtendedKeys {
|
|
delete(extended, key)
|
|
}
|
|
} else {
|
|
for _, key := range s3tablesExtendedKeys {
|
|
if _, ok := extended[key]; ok {
|
|
return fmt.Errorf("%w: %s set on %s", ErrNameTaken, key, path)
|
|
}
|
|
}
|
|
}
|
|
for key, data := range attrs {
|
|
extended[key] = data
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// retakeDirectory writes a create's attributes onto a directory that is already
|
|
// at path, the way an exclusive create would have: a leftover that holds no
|
|
// catalog attribute is taken over, and anything else — a file, or a directory
|
|
// that carries a record — is a taken name. The type is checked on the entry the
|
|
// write carries, so a file that replaces the directory cannot be tagged: the
|
|
// filer refuses to write the directory entry over it. It reports that refusal
|
|
// only as an opaque failure, so a failed write is followed by a look at what
|
|
// holds the name now: a file there is a taken name, not a server error.
|
|
func (h *S3TablesHandler) retakeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, attrs map[string][]byte) error {
|
|
err := h.mutateEntry(ctx, client, path, func(entry *filer_pb.Entry) error {
|
|
if !entry.IsDirectory {
|
|
return fmt.Errorf("%w: %s is not a directory", ErrNameTaken, path)
|
|
}
|
|
return claimExtended(entry.Extended, path, attrs)
|
|
})
|
|
if err == nil || errors.Is(err, ErrNameTaken) {
|
|
return err
|
|
}
|
|
if current, lookupErr := h.lookupEntry(ctx, client, path); lookupErr == nil && !current.IsDirectory {
|
|
return fmt.Errorf("%w: %s is not a directory", ErrNameTaken, path)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// claimCatalogRecord reserves a catalog record for an in-flight delete or
|
|
// rename under one compare-checked write: the entry must still hold exactly
|
|
// expected. A record claimed by another kind of operation refuses the claim,
|
|
// so the loser backs out before any data or destination entry is written;
|
|
// a retry of the same kind may adopt its own earlier claim.
|
|
func (h *S3TablesHandler) claimCatalogRecord(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
expected map[string][]byte,
|
|
claim string,
|
|
) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
for key, want := range expected {
|
|
if key == ExtendedKeyDeleting {
|
|
continue
|
|
}
|
|
if !bytes.Equal(extended[key], want) {
|
|
return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, path)
|
|
}
|
|
}
|
|
if cur, ok := extended[ExtendedKeyDeleting]; ok && string(cur) != claim {
|
|
return fmt.Errorf("%w: %s held on %s", ErrConcurrentUpdate, ExtendedKeyDeleting, path)
|
|
}
|
|
extended[ExtendedKeyDeleting] = []byte(claim)
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// releaseCatalogRecord lifts a claimCatalogRecord reservation while the entry
|
|
// still holds expected — which includes the claim itself — restoring the
|
|
// record so the failed operation can be retried.
|
|
func (h *S3TablesHandler) releaseCatalogRecord(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
expected map[string][]byte,
|
|
) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
for key, want := range expected {
|
|
if !bytes.Equal(extended[key], want) {
|
|
return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, path)
|
|
}
|
|
}
|
|
delete(extended, ExtendedKeyDeleting)
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// updateExtendedAttribute applies mutate to the current value of key and writes
|
|
// the result back. mutate runs again on each retry, so it always sees the
|
|
// current value.
|
|
func (h *S3TablesHandler) updateExtendedAttribute(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path, key string,
|
|
mutate func(current []byte) ([]byte, error),
|
|
) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
updated, err := mutate(extended[key])
|
|
if err != nil {
|
|
return err
|
|
}
|
|
extended[key] = updated
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// s3tablesExtendedKeys is every attribute this package stores on a bucket or
|
|
// table entry. A whole-entry write has to assert the absent ones too: a key
|
|
// missing from the precondition is a key a concurrent writer can create and
|
|
// this write will then delete.
|
|
var s3tablesExtendedKeys = []string{
|
|
ExtendedKeyTableBucket,
|
|
ExtendedKeyMetadata,
|
|
ExtendedKeyMetadataVersion,
|
|
ExtendedKeyPolicy,
|
|
ExtendedKeyTags,
|
|
ExtendedKeyMaintenance,
|
|
ExtendedKeyMaintenanceStatus,
|
|
ExtendedKeyEntryType,
|
|
ExtendedKeyDeleting,
|
|
ExtendedKeyMoving,
|
|
ExtendedKeyVectorPolicy,
|
|
}
|
|
|
|
// SnapshotExtended captures an entry's attributes as an UpdateEntry
|
|
// precondition, so a whole-entry write cannot silently revert or delete an
|
|
// attribute a concurrent writer touched. Attributes absent at read time are
|
|
// asserted absent, which fails the precondition if someone creates one first.
|
|
func SnapshotExtended(extended map[string][]byte, keys ...string) map[string][]byte {
|
|
return snapshotExtended(extended, keys...)
|
|
}
|
|
|
|
func snapshotExtended(extended map[string][]byte, keys ...string) map[string][]byte {
|
|
expected := make(map[string][]byte, len(extended)+len(s3tablesExtendedKeys))
|
|
for k, v := range extended {
|
|
expected[k] = v
|
|
}
|
|
for _, k := range s3tablesExtendedKeys {
|
|
if _, ok := expected[k]; !ok {
|
|
expected[k] = nil
|
|
}
|
|
}
|
|
for _, k := range keys {
|
|
if _, ok := expected[k]; !ok {
|
|
expected[k] = nil
|
|
}
|
|
}
|
|
return expected
|
|
}
|
|
|
|
// getExtendedAttribute gets an extended attribute from an entry
|
|
func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) ([]byte, error) {
|
|
dir, name := splitPath(path)
|
|
resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if resp.Entry.Extended == nil {
|
|
return nil, fmt.Errorf("%w: %s", ErrAttributeNotFound, key)
|
|
}
|
|
|
|
data, ok := resp.Entry.Extended[key]
|
|
if !ok {
|
|
return nil, fmt.Errorf("%w: %s", ErrAttributeNotFound, key)
|
|
}
|
|
|
|
return data, nil
|
|
}
|
|
|
|
// readNamespaceMetadataAttr returns the namespace-metadata xattr at
|
|
// namespacePath. A directory without it is not a namespace — a delete that kept
|
|
// the data under it leaves one — so its absence reads as a missing entry. So
|
|
// does a namespace still marked by its create: like a table's, it is absent to
|
|
// every reader until the create confirms it.
|
|
func (h *S3TablesHandler) readNamespaceMetadataAttr(ctx context.Context, client filer_pb.SeaweedFilerClient, namespacePath string) ([]byte, error) {
|
|
entry, err := h.lookupEntry(ctx, client, namespacePath)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
data, ok := entry.Extended[ExtendedKeyMetadata]
|
|
if !ok || isUnfinishedCreate(entry.Extended) {
|
|
return nil, fmt.Errorf("%w: %s", filer_pb.ErrNotFound, namespacePath)
|
|
}
|
|
return data, nil
|
|
}
|
|
|
|
// isFinishedNamespaceEntry is isNamespaceEntry for a namespace its create has
|
|
// confirmed, the only kind a reader or a delete may act on.
|
|
func isFinishedNamespaceEntry(entry *filer_pb.Entry) bool {
|
|
return isNamespaceEntry(entry) && !isUnfinishedCreate(entry.Extended)
|
|
}
|
|
|
|
// loadNamespaceMetadata resolves a namespace to its metadata. A directory that
|
|
// carries no namespace metadata is not a namespace, so a missing attribute
|
|
// reports the same absence as a missing entry and every caller tests one
|
|
// condition instead of forgetting the second.
|
|
func (h *S3TablesHandler) loadNamespaceMetadata(ctx context.Context, filerClient FilerClient, bucketName, namespaceName string) (*namespaceMetadata, error) {
|
|
var metadata namespaceMetadata
|
|
err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
data, err := h.readNamespaceMetadataAttr(ctx, client, GetNamespacePath(bucketName, namespaceName))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := json.Unmarshal(data, &metadata); err != nil {
|
|
return fmt.Errorf("failed to unmarshal namespace metadata: %w", err)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &metadata, nil
|
|
}
|
|
|
|
// readTableMetadataAttr returns the table-metadata xattr at tablePath. A
|
|
// directory without it is not a table, so its absence reads as a missing entry.
|
|
func (h *S3TablesHandler) readTableMetadataAttr(ctx context.Context, client filer_pb.SeaweedFilerClient, tablePath string) ([]byte, error) {
|
|
data, err := h.getExtendedAttribute(ctx, client, tablePath, ExtendedKeyMetadata)
|
|
if errors.Is(err, ErrAttributeNotFound) {
|
|
return nil, fmt.Errorf("%w: %s", filer_pb.ErrNotFound, tablePath)
|
|
}
|
|
return data, err
|
|
}
|
|
|
|
// lookupEntry returns the filer entry at the given path.
|
|
func (h *S3TablesHandler) lookupEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) (*filer_pb.Entry, error) {
|
|
dir, name := splitPath(path)
|
|
resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return resp.Entry, nil
|
|
}
|
|
|
|
// deleteDirectory deletes a directory and all its contents
|
|
func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
|
|
dir, name := splitPath(path)
|
|
return filer_pb.DoRemove(ctx, client, dir, name, true, true, true, false, nil)
|
|
}
|
|
|
|
// deleteDirectoryIfMatches deletes a directory and all its contents only
|
|
// while the entry still carries the extended attributes the caller captured:
|
|
// the filer compares and removes in one operation under the path lock, so an
|
|
// attribute write that keeps Mtime or a delete+recreate in between refuses the
|
|
// delete with ErrConcurrentUpdate instead of losing the replacement. The raw
|
|
// RPC, not DoRemoveRequest: the filer answers a precondition delete on an
|
|
// entry that is already gone with FailedPrecondition carrying the not-found
|
|
// text, which that helper reads as a completed delete.
|
|
func (h *S3TablesHandler) deleteDirectoryIfMatches(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, expected *filer_pb.Entry) error {
|
|
dir, name := splitPath(path)
|
|
resp, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
IsDeleteData: true,
|
|
IsRecursive: true,
|
|
IgnoreRecursiveError: true,
|
|
ExpectedExtended: snapshotExtended(expected.Extended),
|
|
})
|
|
if status.Code(err) == codes.FailedPrecondition {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if resp != nil && resp.Error != "" {
|
|
return errors.New(resp.Error)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// errDirNotEmpty is a directory that holds something the caller meant to leave
|
|
// alone: a file in a leftover tree, or an entry the filer found in a directory
|
|
// it was asked to delete without its contents.
|
|
var errDirNotEmpty = errors.New("directory is not empty")
|
|
|
|
// deleteEmptyDirectoryIfMatches deletes a directory without its contents, only
|
|
// while it still carries the extended attributes the caller captured. The
|
|
// filer checks both under the path lock, in the one delete: a changed
|
|
// attribute is ErrConcurrentUpdate, and an entry that appeared in the
|
|
// directory is errDirNotEmpty, so neither a directory taken back since the
|
|
// caller looked at it nor a file written into it since is lost. The mark of an
|
|
// empty S3 object stored on the directory is asserted absent too, so one
|
|
// written since is kept the same way.
|
|
func (h *S3TablesHandler) deleteEmptyDirectoryIfMatches(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, expected *filer_pb.Entry) error {
|
|
dir, name := splitPath(path)
|
|
resp, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
IsDeleteData: true,
|
|
ExpectedExtended: snapshotExtended(expected.Extended, s3_constants.SeaweedFSPrefixObject),
|
|
})
|
|
if status.Code(err) == codes.FailedPrecondition {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if resp != nil && resp.Error != "" {
|
|
if err := filer.DeleteEntryError(resp.Error); !errors.Is(err, filer.ErrNonEmptyFolder) {
|
|
return err
|
|
}
|
|
return fmt.Errorf("%w: %s", errDirNotEmpty, path)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var (
|
|
errParentDeleting = errors.New("parent is being deleted")
|
|
errParentGone = errors.New("parent was deleted")
|
|
errParentCreating = errors.New("parent is still being created")
|
|
errNoDeleteClaim = errors.New("claim not held")
|
|
errBackOutFailed = errors.New("backing out the create failed")
|
|
errSourceGone = errors.New("rename source was deleted")
|
|
)
|
|
|
|
// claimForDelete writes a fresh token into ExtendedKeyDeleting, replacing any
|
|
// earlier claim, and returns it with the attributes the delete must still find
|
|
// when it removes the entry. The token makes a stale delete's removal fail.
|
|
func (h *S3TablesHandler) claimForDelete(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, isParent func(*filer_pb.Entry) bool) (string, map[string][]byte, error) {
|
|
claim := generateVersionToken()
|
|
var claimed map[string][]byte
|
|
err := h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
if !isParent(&filer_pb.Entry{Extended: extended}) {
|
|
return filer_pb.ErrNotFound
|
|
}
|
|
extended[ExtendedKeyDeleting] = []byte(claim)
|
|
claimed = snapshotExtended(extended)
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
return claim, claimed, nil
|
|
}
|
|
|
|
// tryReleaseDeleteClaim removes claim from path once, while path still holds
|
|
// it. A claim another delete replaced or lifted, or an entry that is gone,
|
|
// leaves nothing to release and is not an error.
|
|
func (h *S3TablesHandler) tryReleaseDeleteClaim(ctx context.Context, client filer_pb.SeaweedFilerClient, path, claim string) error {
|
|
err := h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
if string(extended[ExtendedKeyDeleting]) != claim {
|
|
return errNoDeleteClaim
|
|
}
|
|
delete(extended, ExtendedKeyDeleting)
|
|
return nil
|
|
})
|
|
if errors.Is(err, errNoDeleteClaim) || errors.Is(err, filer_pb.ErrNotFound) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// releaseDeleteClaim removes claim from path while path still holds it, even
|
|
// after the request was cancelled. A claim left behind refuses creates under a
|
|
// bucket or namespace and makes a leftover directory read as taken, so a failed
|
|
// release is retried, including one that lost to another write of the entry.
|
|
func (h *S3TablesHandler) releaseDeleteClaim(ctx context.Context, client filer_pb.SeaweedFilerClient, path, claim string) {
|
|
ctx = context.WithoutCancel(ctx)
|
|
var err error
|
|
for attempt := 0; attempt < extendedAttributeUpdateAttempts; attempt++ {
|
|
if attempt > 0 {
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
if err = h.tryReleaseDeleteClaim(ctx, client, path, claim); err == nil {
|
|
return
|
|
}
|
|
}
|
|
glog.Warningf("release delete claim on %s: %v", path, err)
|
|
}
|
|
|
|
// isTableBucketCreatedAt accepts only the table bucket created at createdAt. A
|
|
// bucket deleted and created again under the same name has a CreatedAt of its
|
|
// own, so a create authorized against the first reads the second as gone.
|
|
func isTableBucketCreatedAt(createdAt time.Time) func(*filer_pb.Entry) bool {
|
|
return func(entry *filer_pb.Entry) bool {
|
|
if !IsTableBucketEntry(entry) {
|
|
return false
|
|
}
|
|
var metadata tableBucketMetadata
|
|
if json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata) != nil {
|
|
return false
|
|
}
|
|
return metadata.CreatedAt.Equal(createdAt)
|
|
}
|
|
}
|
|
|
|
// isNamespaceCreatedAt is isTableBucketCreatedAt for a namespace its create has
|
|
// confirmed: one deleted and created again under the same name reads as the
|
|
// authorized one gone. A bucket created again holds only namespaces of its own,
|
|
// so a create bound to its namespace this way is bound to the bucket too.
|
|
func isNamespaceCreatedAt(createdAt time.Time) func(*filer_pb.Entry) bool {
|
|
return func(entry *filer_pb.Entry) bool {
|
|
if !isFinishedNamespaceEntry(entry) {
|
|
return false
|
|
}
|
|
var metadata namespaceMetadata
|
|
if json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata) != nil {
|
|
return false
|
|
}
|
|
return metadata.CreatedAt.Equal(createdAt)
|
|
}
|
|
}
|
|
|
|
// isNamespaceEntry reports whether entry carries a namespace's catalog record.
|
|
func isNamespaceEntry(entry *filer_pb.Entry) bool {
|
|
_, ok := entry.Extended[ExtendedKeyMetadata]
|
|
return ok
|
|
}
|
|
|
|
// checkParentOpen returns errParentGone if parentPath is missing or not a
|
|
// catalog entry, errParentCreating if the create that wrote it has not
|
|
// confirmed it yet, and errParentDeleting if a delete has claimed it.
|
|
func (h *S3TablesHandler) checkParentOpen(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool) error {
|
|
_, err := h.openParent(ctx, client, parentPath, isParent)
|
|
return err
|
|
}
|
|
|
|
// openParent is checkParentOpen returning the entry it accepted, so a caller
|
|
// can later ask whether the parent is still that same record.
|
|
func (h *S3TablesHandler) openParent(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool) (*filer_pb.Entry, error) {
|
|
entry, err := h.lookupEntry(ctx, client, parentPath)
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
return nil, errParentGone
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !isParent(entry) {
|
|
return nil, errParentGone
|
|
}
|
|
if isUnfinishedCreate(entry.Extended) {
|
|
return nil, errParentCreating
|
|
}
|
|
if _, claimed := entry.Extended[ExtendedKeyDeleting]; claimed {
|
|
return nil, errParentDeleting
|
|
}
|
|
return entry, nil
|
|
}
|
|
|
|
// parentStillOpen reports whether parentPath still holds the record parent
|
|
// captured. A delete claims, removes, or replaces the parent between the two
|
|
// reads; createdAt identifies the record since an update preserves it while a
|
|
// delete+recreate mints a new one, so an editable property changing in
|
|
// between does not read as a removal.
|
|
func (h *S3TablesHandler) parentStillOpen(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, parent *filer_pb.Entry) error {
|
|
current, err := h.openParent(ctx, client, parentPath, isParent)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var before, now namespaceMetadata
|
|
if json.Unmarshal(parent.Extended[ExtendedKeyMetadata], &before) != nil ||
|
|
json.Unmarshal(current.Extended[ExtendedKeyMetadata], &now) != nil ||
|
|
!now.CreatedAt.Equal(before.CreatedAt) {
|
|
return errParentGone
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// createClaim is the ExtendedKeyDeleting value of a catalog record whose
|
|
// create has written it but not yet confirmed its parent. renameDestClaim is
|
|
// the same not-yet-published mark on a rename's destination, with a value of
|
|
// its own so a create refuses it instead of adopting the record.
|
|
const (
|
|
createClaim = "create"
|
|
renameDestClaim = "rename-dest"
|
|
)
|
|
|
|
// isUnfinishedCreate reports whether extended belongs to a record still
|
|
// marked by the operation that wrote it — invisible to every reader.
|
|
func isUnfinishedCreate(extended map[string][]byte) bool {
|
|
switch string(extended[ExtendedKeyDeleting]) {
|
|
case createClaim, renameDestClaim:
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// mutateCatalogRecord applies mutate to a catalog record, refusing a record
|
|
// an in-flight operation claimed: a create's unfinished record is absent to
|
|
// every reader, and a record claimed by a delete or rename must not take a
|
|
// write the operation's compare then trips over.
|
|
func (h *S3TablesHandler) mutateCatalogRecord(
|
|
ctx context.Context,
|
|
client filer_pb.SeaweedFilerClient,
|
|
path string,
|
|
mutate func(extended map[string][]byte) error,
|
|
) error {
|
|
return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error {
|
|
if _, claimed := extended[ExtendedKeyDeleting]; !claimed {
|
|
if _, ok := extended[ExtendedKeyMetadata]; !ok {
|
|
return filer_pb.ErrNotFound
|
|
}
|
|
return mutate(extended)
|
|
}
|
|
if isUnfinishedCreate(extended) {
|
|
return filer_pb.ErrNotFound
|
|
}
|
|
return fmt.Errorf("%w: %s claimed on %s", ErrConcurrentUpdate, ExtendedKeyDeleting, path)
|
|
})
|
|
}
|
|
|
|
// requireTableCreatedAt refuses a write to the record at path unless it is the
|
|
// table created at createdAt, the one the caller was authorized against: a
|
|
// table deleted and created again under the name has a CreatedAt of its own.
|
|
// Only the creation time is compared, so the commits an Iceberg writer lands
|
|
// meanwhile do not turn a write of the table's policy, tags or maintenance
|
|
// settings into a conflict.
|
|
func requireTableCreatedAt(extended map[string][]byte, path string, createdAt time.Time) error {
|
|
var current tableMetadataInternal
|
|
if json.Unmarshal(extended[ExtendedKeyMetadata], ¤t) != nil || !current.CreatedAt.Equal(createdAt) {
|
|
return fmt.Errorf("%w: %s holds another table", ErrConcurrentUpdate, path)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// requireTableBucketCreatedAt is requireTableCreatedAt for a table bucket.
|
|
func requireTableBucketCreatedAt(extended map[string][]byte, path string, createdAt time.Time) error {
|
|
if !isTableBucketCreatedAt(createdAt)(&filer_pb.Entry{Extended: extended}) {
|
|
return fmt.Errorf("%w: %s holds another table bucket", ErrConcurrentUpdate, path)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// confirmParent re-checks the parent after a create wrote written onto
|
|
// childPath. If the parent is open, it clears the record's unfinished mark —
|
|
// then checks the parent once more, since a delete claiming it in between
|
|
// scans the still-marked record, skips it, and removes it under a create that
|
|
// already answered success. If the parent is gone or claimed at either check,
|
|
// it removes what the create wrote and returns why, or errBackOutFailed if
|
|
// the record stays.
|
|
func (h *S3TablesHandler) confirmParent(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, childPath string, written map[string][]byte, emptyDirs ...string) error {
|
|
return h.confirmChild(ctx, client, parentPath, isParent, true, childPath, written, emptyDirs)
|
|
}
|
|
|
|
// confirmParentOnce is confirmParent for a record that other creates build on,
|
|
// a namespace. Checking the parent again after the mark comes off would leave a
|
|
// window in which a child is created in the record and the check then removes
|
|
// the record under it, so the parent is checked once, before the mark comes
|
|
// off. That is enough while a delete of the parent counts the marked record: one
|
|
// claiming the parent after the check scans the record and is refused.
|
|
func (h *S3TablesHandler) confirmParentOnce(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, childPath string, written map[string][]byte, emptyDirs ...string) error {
|
|
return h.confirmChild(ctx, client, parentPath, isParent, false, childPath, written, emptyDirs)
|
|
}
|
|
|
|
func (h *S3TablesHandler) confirmChild(ctx context.Context, client filer_pb.SeaweedFilerClient, parentPath string, isParent func(*filer_pb.Entry) bool, recheck bool, childPath string, written map[string][]byte, emptyDirs []string) error {
|
|
parent, err := h.openParent(ctx, client, parentPath, isParent)
|
|
// mayBeVisible is whether the mark may have come off: it did, or a write
|
|
// taking it off was sent and failed without the filer saying it refused
|
|
// it. A failure before any such write — the lookup that precedes it, say —
|
|
// leaves the mark on, and the record as invisible as it was.
|
|
confirmed, mayBeVisible := false, false
|
|
if err == nil {
|
|
// The mark comes off only while the record still holds what this
|
|
// create wrote; a failed write backs the record out like a closed parent.
|
|
writes := &markRemovalWrites{SeaweedFilerClient: client}
|
|
err = h.removeExtendedAttributesIf(ctx, writes, childPath, written, ExtendedKeyDeleting)
|
|
mayBeVisible = err == nil || writes.uncertain
|
|
if err == nil {
|
|
confirmed = true
|
|
if recheck {
|
|
err = h.parentStillOpen(ctx, client, parentPath, isParent, parent)
|
|
}
|
|
}
|
|
}
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
err = h.backOutChild(ctx, client, childPath, written, confirmed, mayBeVisible, emptyDirs, err)
|
|
if errors.Is(err, errParentGone) {
|
|
h.removeLeftoverDirectory(context.WithoutCancel(ctx), client, parentPath)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// backOutChild removes what a create wrote onto childPath, after the create
|
|
// failed with err, and returns err, or why the record stays. mayBeVisible is
|
|
// whether readers may have seen the record: a record that moved on is then
|
|
// reported as ErrPublishedRecordChanged, since what was staged for it may
|
|
// still be in use.
|
|
func (h *S3TablesHandler) backOutChild(ctx context.Context, client filer_pb.SeaweedFilerClient, childPath string, written map[string][]byte, confirmed, mayBeVisible bool, emptyDirs []string, err error) error {
|
|
ctx = context.WithoutCancel(ctx)
|
|
expected := written
|
|
keys := make([]string, 0, len(written))
|
|
if confirmed {
|
|
// The confirmed record no longer carries the mark written expected.
|
|
expected = make(map[string][]byte, len(written))
|
|
for key, data := range written {
|
|
if key == ExtendedKeyDeleting {
|
|
continue
|
|
}
|
|
expected[key] = data
|
|
keys = append(keys, key)
|
|
}
|
|
} else {
|
|
for key := range written {
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
if backOutErr := h.removeExtendedAttributesIf(ctx, client, childPath, expected, keys...); backOutErr != nil && !errors.Is(backOutErr, filer_pb.ErrNotFound) {
|
|
glog.Warningf("back out %s: %v", childPath, backOutErr)
|
|
if errors.Is(backOutErr, ErrConcurrentUpdate) {
|
|
// The record moved past this create — adopted or rewritten — so
|
|
// there is nothing of its own left to remove. While it was still
|
|
// marked, only another create could take it over; once readers
|
|
// could see it, an update or a rename may have built on it.
|
|
if mayBeVisible {
|
|
return fmt.Errorf("%w: %s", ErrPublishedRecordChanged, childPath)
|
|
}
|
|
return fmt.Errorf("%w: %s changed during create", ErrConcurrentUpdate, childPath)
|
|
}
|
|
return fmt.Errorf("%w: %s: %v (%v)", errBackOutFailed, childPath, backOutErr, err)
|
|
}
|
|
for _, dir := range emptyDirs {
|
|
h.removeLeftoverDirectory(ctx, client, dir)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// markRemovalWrites passes a create's mark-removal calls through to the filer
|
|
// and records whether an UpdateEntry was sent whose outcome is not a refusal:
|
|
// it succeeded, or it failed in a way that does not say whether it landed. A
|
|
// compare the filer refused (FailedPrecondition) or an entry it no longer has
|
|
// (NotFound) changed nothing.
|
|
type markRemovalWrites struct {
|
|
filer_pb.SeaweedFilerClient
|
|
uncertain bool
|
|
}
|
|
|
|
func (w *markRemovalWrites) UpdateEntry(ctx context.Context, in *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
|
|
resp, err := w.SeaweedFilerClient.UpdateEntry(ctx, in, opts...)
|
|
if code := status.Code(err); code != codes.FailedPrecondition && code != codes.NotFound {
|
|
w.uncertain = true
|
|
}
|
|
return resp, err
|
|
}
|
|
|
|
// dirListPageSize is how many entries dirHoldsFiles reads per ListEntries call.
|
|
const dirListPageSize = 1024
|
|
|
|
// dirHoldsFiles reports whether any file sits under dirPath, at any depth. It
|
|
// walks breadth-first, pages through each directory, and stops at the first
|
|
// file it sees. Returning at that file abandons the page being read, so the
|
|
// listings run on a context of their own that ends with the call: otherwise
|
|
// the filer keeps producing that page until the whole request is over.
|
|
func (h *S3TablesHandler) dirHoldsFiles(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string) (bool, error) {
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
|
|
pending := []string{dirPath}
|
|
for len(pending) > 0 {
|
|
dir := pending[0]
|
|
pending = pending[1:]
|
|
|
|
lastFileName := ""
|
|
for {
|
|
resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
|
|
Directory: dir,
|
|
Limit: dirListPageSize,
|
|
StartFromFileName: lastFileName,
|
|
InclusiveStartFrom: lastFileName == "",
|
|
})
|
|
if err != nil {
|
|
if status.Code(err) == codes.NotFound {
|
|
// the directory is gone, so there is nothing under it to keep
|
|
break
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
received := false
|
|
for {
|
|
entry, err := resp.Recv()
|
|
if err != nil {
|
|
if err == io.EOF || status.Code(err) == codes.NotFound {
|
|
break
|
|
}
|
|
return false, err
|
|
}
|
|
if entry.Entry == nil {
|
|
continue
|
|
}
|
|
received = true
|
|
lastFileName = entry.Entry.Name
|
|
if !entry.Entry.IsDirectory {
|
|
return true, nil
|
|
}
|
|
pending = append(pending, path.Join(dir, entry.Entry.Name))
|
|
}
|
|
if !received {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
// eachDirEntry calls fn on the entries of dir page by page until fn returns
|
|
// false. A missing dir has none.
|
|
func eachDirEntry(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, fn func(*filer_pb.Entry) bool) error {
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
lastFileName := ""
|
|
for {
|
|
resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
|
|
Directory: dir,
|
|
Limit: dirListPageSize,
|
|
StartFromFileName: lastFileName,
|
|
InclusiveStartFrom: lastFileName == "",
|
|
})
|
|
if err != nil {
|
|
if status.Code(err) == codes.NotFound || errors.Is(err, filer_pb.ErrNotFound) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
received := false
|
|
for {
|
|
entry, err := resp.Recv()
|
|
if err != nil {
|
|
if err == io.EOF || status.Code(err) == codes.NotFound {
|
|
break
|
|
}
|
|
return err
|
|
}
|
|
if entry.Entry == nil {
|
|
continue
|
|
}
|
|
received = true
|
|
lastFileName = entry.Entry.Name
|
|
if !fn(entry.Entry) {
|
|
return nil
|
|
}
|
|
}
|
|
if !received {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
|
|
// CatalogRecordRoot returns the directory the catalog record at recordPath
|
|
// keeps its files under: the one its metadata location names inside the
|
|
// bucket, else the warehouse location assigned at create, else its own path.
|
|
func CatalogRecordRoot(bucketName, recordPath, metadataLocation, warehouseLocation string) string {
|
|
bucketPath := GetTableBucketPath(bucketName) + "/"
|
|
if dir := TableDataDirFromMetadataLocation(metadataLocation); strings.HasPrefix(dir, bucketPath) {
|
|
return dir
|
|
}
|
|
if dir := warehouseDir(warehouseLocation); strings.HasPrefix(dir, bucketPath) {
|
|
return dir
|
|
}
|
|
return recordPath
|
|
}
|
|
|
|
func warehouseDir(warehouseLocation string) string {
|
|
loc := strings.TrimPrefix(strings.TrimSuffix(warehouseLocation, "/"), "s3://")
|
|
if loc == "" {
|
|
return ""
|
|
}
|
|
return path.Join(TablesPath, loc)
|
|
}
|
|
|
|
// root returns the directory the record at recordPath keeps its files under.
|
|
func (m *tableMetadataInternal) root(bucketName, recordPath string) string {
|
|
return CatalogRecordRoot(bucketName, recordPath, m.MetadataLocation, m.WarehouseLocation)
|
|
}
|
|
|
|
// CatalogRecordRoots maps every table and view record of the bucket, in any
|
|
// namespace, to the directories it keeps its files under: the one its metadata
|
|
// gives it and, while an update moves it, the one it moves to. A renamed record
|
|
// keeps the directory of its old name.
|
|
func CatalogRecordRoots(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName string) (map[string][]string, error) {
|
|
var namespaces []string
|
|
if err := eachDirEntry(ctx, client, GetTableBucketPath(bucketName), func(entry *filer_pb.Entry) bool {
|
|
if entry.IsDirectory && !strings.HasPrefix(entry.Name, ".") {
|
|
namespaces = append(namespaces, entry.Name)
|
|
}
|
|
return true
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
roots := map[string][]string{}
|
|
for _, namespace := range namespaces {
|
|
namespacePath := GetNamespacePath(bucketName, namespace)
|
|
if err := eachDirEntry(ctx, client, namespacePath, func(entry *filer_pb.Entry) bool {
|
|
if _, ok := entry.Extended[ExtendedKeyMetadata]; !ok || !entry.IsDirectory {
|
|
return true
|
|
}
|
|
recordPath := path.Join(namespacePath, entry.Name)
|
|
for _, key := range []string{ExtendedKeyMetadata, ExtendedKeyMoving} {
|
|
data, ok := entry.Extended[key]
|
|
if !ok {
|
|
continue
|
|
}
|
|
var record struct {
|
|
MetadataLocation string `json:"metadataLocation"`
|
|
WarehouseLocation string `json:"warehouseLocation"`
|
|
}
|
|
_ = json.Unmarshal(data, &record)
|
|
roots[recordPath] = append(roots[recordPath], CatalogRecordRoot(bucketName, recordPath, record.MetadataLocation, record.WarehouseLocation))
|
|
}
|
|
return true
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return roots, nil
|
|
}
|
|
|
|
// PathsOverlap reports whether a and b are the same path or one holds the other.
|
|
func PathsOverlap(a, b string) bool {
|
|
return a == b || strings.HasPrefix(a, b+"/") || strings.HasPrefix(b, a+"/")
|
|
}
|
|
|
|
// RootClaimed reports whether a record other than selfPath has its entry or its
|
|
// files at root, above it or under it.
|
|
func RootClaimed(roots map[string][]string, selfPath, root string) bool {
|
|
for recordPath, others := range roots {
|
|
if recordPath == selfPath {
|
|
continue
|
|
}
|
|
if PathsOverlap(recordPath, root) {
|
|
return true
|
|
}
|
|
for _, other := range others {
|
|
if PathsOverlap(other, root) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// rootFilesShared reports whether a record other than selfPath keeps its files
|
|
// at root, above it or under it. Unlike RootClaimed it ignores where the other
|
|
// records themselves sit: a table created under a renamed table's old name has
|
|
// its entry on that directory and its files elsewhere.
|
|
func rootFilesShared(roots map[string][]string, selfPath, root string) bool {
|
|
for recordPath, others := range roots {
|
|
if recordPath == selfPath {
|
|
continue
|
|
}
|
|
for _, other := range others {
|
|
if PathsOverlap(other, root) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// checkLocationUnclaimed returns ErrLocationInUse when root, the one the record
|
|
// at recordPath would have, is claimed by another record.
|
|
func checkLocationUnclaimed(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, recordPath, root string) error {
|
|
roots, err := CatalogRecordRoots(ctx, client, bucketName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if RootClaimed(roots, recordPath, root) {
|
|
return ErrLocationInUse
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// claimCatalogEntryAt claims path like claimCatalogEntry, then checks its root
|
|
// again with the record written: of two creates racing for one root, at least
|
|
// one sees the other and backs out.
|
|
func (h *S3TablesHandler) claimCatalogEntryAt(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, root string, attrs map[string][]byte, emptyDirs ...string) error {
|
|
if err := h.claimCatalogEntry(ctx, client, path, attrs); err != nil {
|
|
return err
|
|
}
|
|
if err := checkLocationUnclaimed(ctx, client, bucketName, path, root); err != nil {
|
|
// The claim still carries its mark, so no reader has seen the record.
|
|
return h.backOutChild(ctx, client, path, attrs, false, false, emptyDirs, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// moveCatalogRecord applies write, which moves the record at path onto the
|
|
// root of moving, only once that root is reserved: moving is first stored
|
|
// beside the current metadata, so the record holds both roots while the new
|
|
// one is checked again. Of two writers racing for it at least one sees the
|
|
// other, and the old root stays held until the move lands. A move that backs
|
|
// out drops the reservation and leaves the metadata as it was.
|
|
func (h *S3TablesHandler) moveCatalogRecord(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName, path, root string, stored, moving []byte, write func(extended map[string][]byte) error) error {
|
|
if err := h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error {
|
|
if !bytes.Equal(extended[ExtendedKeyMetadata], stored) {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata)
|
|
}
|
|
if _, ok := extended[ExtendedKeyMoving]; ok {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMoving)
|
|
}
|
|
extended[ExtendedKeyMoving] = moving
|
|
return nil
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
err := checkLocationUnclaimed(ctx, client, bucketName, path, root)
|
|
if err == nil {
|
|
err = h.mutateCatalogRecord(ctx, client, path, func(extended map[string][]byte) error {
|
|
if !bytes.Equal(extended[ExtendedKeyMoving], moving) {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMoving)
|
|
}
|
|
return write(extended)
|
|
})
|
|
}
|
|
if err != nil {
|
|
if releaseErr := h.mutateCatalogRecord(context.WithoutCancel(ctx), client, path, func(extended map[string][]byte) error {
|
|
if !bytes.Equal(extended[ExtendedKeyMoving], moving) {
|
|
return ErrConcurrentUpdate
|
|
}
|
|
delete(extended, ExtendedKeyMoving)
|
|
return nil
|
|
}); releaseErr != nil && !errors.Is(releaseErr, ErrConcurrentUpdate) && !errors.Is(releaseErr, filer_pb.ErrNotFound) {
|
|
glog.Warningf("release move of %s: %v", path, releaseErr)
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// refuseClaimedLocation answers a request whose record would share its files
|
|
// with another table or view of the bucket.
|
|
func (h *S3TablesHandler) refuseClaimedLocation(w http.ResponseWriter, r *http.Request, filerClient FilerClient, bucketName, recordPath, root string) error {
|
|
err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
return checkLocationUnclaimed(r.Context(), client, bucketName, recordPath, root)
|
|
})
|
|
if errors.Is(err, ErrLocationInUse) {
|
|
h.writeError(w, http.StatusConflict, ErrCodeConflict, err.Error())
|
|
} else if err != nil {
|
|
h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to check location")
|
|
}
|
|
return err
|
|
}
|
|
|
|
// refuseUpdatedMetadataLocation answers an update whose metadata location the
|
|
// record at recordPath, whose stored metadata is current, may not take; see
|
|
// validateUpdatedMetadataLocation.
|
|
func (h *S3TablesHandler) refuseUpdatedMetadataLocation(w http.ResponseWriter, r *http.Request, filerClient FilerClient, bucketName, namespace, recordPath string, current *tableMetadataInternal, metadataLocation string) error {
|
|
var checkErr error
|
|
err := validateUpdatedMetadataLocation(metadataLocation, bucketName, namespace, recordPath, current, func(root string) (shared bool, err error) {
|
|
checkErr = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
|
roots, err := CatalogRecordRoots(r.Context(), client, bucketName)
|
|
shared = err == nil && rootFilesShared(roots, recordPath, root)
|
|
return err
|
|
})
|
|
return shared, checkErr
|
|
})
|
|
if checkErr != nil {
|
|
h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to check location")
|
|
} else if err != nil {
|
|
h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
|
|
}
|
|
return err
|
|
}
|
|
|
|
// claimTreeDirectory writes a fresh delete claim onto the directory at dirPath
|
|
// while it still carries expected, and returns the claim with the attributes it
|
|
// left. A claim seen in expected, one a delete that died left, is replaced like
|
|
// any other attribute; one written since fails the compare.
|
|
//
|
|
// The claim is what ties the later delete to this directory: the delete
|
|
// requires it, and only this directory carries it. A file or an S3 object at
|
|
// dirPath is refused with errDirNotEmpty, whether it was there when the entry
|
|
// was read or took the directory's place before the write, which the filer
|
|
// refuses since it never writes a directory entry over a file. One that takes
|
|
// the place after the write lacks the claim, so the delete is refused instead.
|
|
// An S3 object landing on the directory between the read and the write carries
|
|
// the prefix-object mark, which the write asserts absent, so the write fails
|
|
// instead of putting the bare directory back over the object.
|
|
func (h *S3TablesHandler) claimTreeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, expected map[string][]byte) (string, map[string][]byte, error) {
|
|
claim := generateVersionToken()
|
|
var claimed map[string][]byte
|
|
err := h.mutateEntryAsserting(ctx, client, dirPath, []string{s3_constants.SeaweedFSPrefixObject}, func(entry *filer_pb.Entry) error {
|
|
if !entry.IsDirectory || entry.IsDirectoryKeyObject() {
|
|
return fmt.Errorf("%w: %s is an object", errDirNotEmpty, dirPath)
|
|
}
|
|
for key, want := range expected {
|
|
if !bytes.Equal(entry.Extended[key], want) {
|
|
return fmt.Errorf("%w: %s changed on %s", ErrConcurrentUpdate, key, dirPath)
|
|
}
|
|
}
|
|
entry.Extended[ExtendedKeyDeleting] = []byte(claim)
|
|
claimed = snapshotExtended(entry.Extended)
|
|
return nil
|
|
})
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
return "", nil, fmt.Errorf("%w: %s", ErrConcurrentUpdate, dirPath)
|
|
}
|
|
if err != nil && !errors.Is(err, errDirNotEmpty) && !errors.Is(err, ErrConcurrentUpdate) {
|
|
// The filer reports the refused write over a file only as an opaque
|
|
// failure, so look at what holds the path now.
|
|
if current, lookupErr := h.lookupEntry(ctx, client, dirPath); lookupErr == nil && !current.IsDirectory {
|
|
return "", nil, fmt.Errorf("%w: %s is an object", errDirNotEmpty, dirPath)
|
|
}
|
|
}
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
return claim, claimed, nil
|
|
}
|
|
|
|
// removeTreeDirectory removes a directory the walk listed as listed, under a
|
|
// claim of its own, and lifts the claim if the delete is refused.
|
|
func (h *S3TablesHandler) removeTreeDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, listed *filer_pb.Entry) error {
|
|
claim, claimed, err := h.claimTreeDirectory(ctx, client, dirPath, snapshotExtended(listed.Extended))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := h.deleteEmptyDirectoryIfMatches(ctx, client, dirPath, &filer_pb.Entry{Extended: claimed}); err != nil {
|
|
h.releaseDeleteClaim(ctx, client, dirPath, claim)
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// removeEmptyTree removes dirPath and every directory under it, provided no
|
|
// file, no S3 object stored on a directory entry, and no catalog record sits
|
|
// anywhere in the tree, dirPath included. dirEntry is dirPath as the caller saw
|
|
// it.
|
|
//
|
|
// dirPath is claimed first, in a write that requires the attributes the caller
|
|
// saw: a directory taken back as a namespace since then fails it, before
|
|
// anything under it is touched, and the claim keeps it from being taken back
|
|
// while the tree goes. With dirPath no namespace, nothing under it can become a
|
|
// catalog record either, so a record the walk meets was there before and is
|
|
// refused, not removed.
|
|
//
|
|
// The walk lists the tree a page at a time and stops at the first file, object
|
|
// or record with errDirNotEmpty, before anything is removed. Directories then go
|
|
// deepest first, each claimed with the attributes it had when listed and
|
|
// deleted under that claim, which refuses a directory that holds an entry by
|
|
// now or was replaced by a file. dirPath goes last, still under its claim; any
|
|
// other way out lifts the claim.
|
|
func (h *S3TablesHandler) removeEmptyTree(ctx context.Context, client filer_pb.SeaweedFilerClient, dirPath string, dirEntry *filer_pb.Entry) (err error) {
|
|
claim, claimed, err := h.claimTreeDirectory(ctx, client, dirPath, snapshotExtended(dirEntry.Extended))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
if err != nil {
|
|
h.releaseDeleteClaim(ctx, client, dirPath, claim)
|
|
}
|
|
}()
|
|
|
|
type listedDir struct {
|
|
path string
|
|
entry *filer_pb.Entry
|
|
}
|
|
listed := []listedDir{{dirPath, &filer_pb.Entry{Extended: claimed}}}
|
|
|
|
walkCtx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
// A directory is listed after its parent, so removing in reverse order
|
|
// removes every directory after the ones under it.
|
|
for next := 0; next < len(listed); next++ {
|
|
dir := listed[next].path
|
|
lastFileName := ""
|
|
for {
|
|
resp, err := client.ListEntries(walkCtx, &filer_pb.ListEntriesRequest{
|
|
Directory: dir,
|
|
Limit: dirListPageSize,
|
|
StartFromFileName: lastFileName,
|
|
InclusiveStartFrom: lastFileName == "",
|
|
})
|
|
if err != nil {
|
|
if status.Code(err) == codes.NotFound {
|
|
break // gone, so its delete below reports the change
|
|
}
|
|
return err
|
|
}
|
|
|
|
received := false
|
|
for {
|
|
entry, err := resp.Recv()
|
|
if err != nil {
|
|
if err == io.EOF || status.Code(err) == codes.NotFound {
|
|
break
|
|
}
|
|
return err
|
|
}
|
|
if entry.Entry == nil {
|
|
continue
|
|
}
|
|
received = true
|
|
lastFileName = entry.Entry.Name
|
|
child := path.Join(dir, entry.Entry.Name)
|
|
if !entry.Entry.IsDirectory || entry.Entry.IsDirectoryKeyObject() {
|
|
return fmt.Errorf("%w: %s is an object", errDirNotEmpty, child)
|
|
}
|
|
if _, isRecord := entry.Entry.Extended[ExtendedKeyMetadata]; isRecord {
|
|
return fmt.Errorf("%w: %s is a catalog record", errDirNotEmpty, child)
|
|
}
|
|
listed = append(listed, listedDir{child, entry.Entry})
|
|
}
|
|
if !received {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
for i := len(listed) - 1; i > 0; i-- {
|
|
if err := h.removeTreeDirectory(ctx, client, listed[i].path, listed[i].entry); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return h.deleteEmptyDirectoryIfMatches(ctx, client, dirPath, &filer_pb.Entry{Extended: claimed})
|
|
}
|
|
|
|
// keepAsLeftover clears a namespace's catalog attributes and the delete's
|
|
// claim in one write, while the entry still holds claimed, keeping the
|
|
// directory and the data under it. The write compares against claimed so the
|
|
// delete's token still fails an ABA, and it drops the claim itself: a kept
|
|
// directory that carried it could never be claimed again. A directory removed
|
|
// since the scan is a concurrent change like any other the compare would
|
|
// catch, so its not-found answers ErrConcurrentUpdate rather than a filer
|
|
// failure.
|
|
func (h *S3TablesHandler) keepAsLeftover(ctx context.Context, client filer_pb.SeaweedFilerClient, path string, claimed map[string][]byte) error {
|
|
keys := make([]string, 0, len(s3tablesExtendedKeys))
|
|
for _, key := range s3tablesExtendedKeys {
|
|
if _, ok := claimed[key]; ok {
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
err := h.removeExtendedAttributesIf(ctx, client, path, claimed, keys...)
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
return fmt.Errorf("%w: %s", ErrConcurrentUpdate, path)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// removeLeftoverDirectory deletes path if it is an empty directory without
|
|
// attributes; the filer refuses the non-recursive delete otherwise.
|
|
func (h *S3TablesHandler) removeLeftoverDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) {
|
|
entry, err := h.lookupEntry(ctx, client, path)
|
|
if err != nil || !entry.IsDirectory || len(entry.Extended) > 0 {
|
|
return
|
|
}
|
|
dir, name := splitPath(path)
|
|
if _, err := filer_pb.DoRemoveRequest(ctx, client, &filer_pb.DeleteEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
IsDeleteData: true,
|
|
ExpectedExtended: snapshotExtended(entry.Extended),
|
|
}); err != nil {
|
|
glog.V(1).Infof("keep %s: %v", path, err)
|
|
}
|
|
}
|
|
|
|
// entryExists checks if an entry exists at the given path
|
|
func (h *S3TablesHandler) entryExists(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) bool {
|
|
dir, name := splitPath(path)
|
|
_, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
|
|
Directory: dir,
|
|
Name: name,
|
|
})
|
|
return err == nil
|
|
}
|