TrueCloudLab/frostfs-s3-gw
17
3
Fork 20
Code Issues 28 Pull requests 7 Projects Releases 31 Packages 1 Activity Actions

bugfix/split_tree #437

Merged
alexvanin merged 2 commits from dkirillov/frostfs-s3-gw:bugfix/split_tree into support/v0.30 2024-09-04 19:51:14 +00:00
Conversation 1 Commits 2 Files changed 5 +155 -78
5 changed files with 155 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
api/layer/cors.go
View file

@ -49,17 +49,19 @@ func (n *Layer) PutBucketCORS(ctx context.Context, p *PutCORSParams) error {
return fmt.Errorf("put system object: %w", err)
}
objIDToDelete, err := n.treeService.PutBucketCORS(ctx, p.BktInfo, objID)
objIDsToDelete, err := n.treeService.PutBucketCORS(ctx, p.BktInfo, objID)
objIDToDeleteNotFound := errorsStd.Is(err, ErrNoNodeToRemove)
if err != nil && !objIDToDeleteNotFound {
return err
}
if !objIDToDeleteNotFound {
if err = n.objectDelete(ctx, p.BktInfo, objIDToDelete); err != nil {
n.reqLogger(ctx).Error(logs.CouldntDeleteCorsObject, zap.Error(err),
zap.String("cnrID", p.BktInfo.CID.EncodeToString()),
zap.String("objID", objIDToDelete.EncodeToString()))
for _, id := range objIDsToDelete {
if err = n.objectDelete(ctx, p.BktInfo, id); err != nil {
n.reqLogger(ctx).Error(logs.CouldntDeleteCorsObject, zap.Error(err),
zap.String("cnrID", p.BktInfo.CID.EncodeToString()),
zap.String("objID", id.EncodeToString()))
}
}
}
@ -81,14 +83,16 @@ func (n *Layer) GetBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) (*d
}
func (n *Layer) DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) error {
objID, err := n.treeService.DeleteBucketCORS(ctx, bktInfo)
objIDs, err := n.treeService.DeleteBucketCORS(ctx, bktInfo)
objIDNotFound := errorsStd.Is(err, ErrNoNodeToRemove)
if err != nil && !objIDNotFound {
return err
}
if !objIDNotFound {
if err = n.objectDelete(ctx, bktInfo, objID); err != nil {
return err
for _, id := range objIDs {
if err = n.objectDelete(ctx, bktInfo, id); err != nil {
return err
}
}
}

6
api/layer/tree_mock.go
View file

@ -124,7 +124,7 @@ func (t *TreeServiceMock) GetBucketCORS(_ context.Context, bktInfo *data.BucketI
return node.OID, nil
}
func (t *TreeServiceMock) PutBucketCORS(_ context.Context, bktInfo *data.BucketInfo, objID oid.ID) (oid.ID, error) {
func (t *TreeServiceMock) PutBucketCORS(_ context.Context, bktInfo *data.BucketInfo, objID oid.ID) ([]oid.ID, error) {
systemMap, ok := t.system[bktInfo.CID.EncodeToString()]
if !ok {
systemMap = make(map[string]*data.BaseNodeVersion)
@ -136,10 +136,10 @@ func (t *TreeServiceMock) PutBucketCORS(_ context.Context, bktInfo *data.BucketI
t.system[bktInfo.CID.EncodeToString()] = systemMap
return oid.ID{}, ErrNoNodeToRemove
return nil, ErrNoNodeToRemove
}
func (t *TreeServiceMock) DeleteBucketCORS(context.Context, *data.BucketInfo) (oid.ID, error) {
func (t *TreeServiceMock) DeleteBucketCORS(context.Context, *data.BucketInfo) ([]oid.ID, error) {
panic("implement me")
}

8
api/layer/tree_service.go
View file

@ -25,13 +25,13 @@ type TreeService interface {
// PutBucketCORS puts a node to a system tree and returns objectID of a previous cors config which must be deleted in FrostFS.
//
// If object id to remove is not found returns ErrNoNodeToRemove error.
PutBucketCORS(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) (oid.ID, error)
// If object ids to remove is not found returns ErrNoNodeToRemove error.
PutBucketCORS(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) ([]oid.ID, error)
// DeleteBucketCORS removes a node from a system tree and returns objID which must be deleted in FrostFS.
//
// If object id to remove is not found returns ErrNoNodeToRemove error.
DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) (oid.ID, error)
// If object ids to remove is not found returns ErrNoNodeToRemove error.
DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) ([]oid.ID, error)
GetObjectTagging(ctx context.Context, bktInfo *data.BucketInfo, objVersion *data.NodeVersion) (map[string]string, error)
PutObjectTagging(ctx context.Context, bktInfo *data.BucketInfo, objVersion *data.NodeVersion, tagSet map[string]string) error

12
internal/logs/logs.go
View file

@ -142,11 +142,13 @@ const (
CouldntCacheSubject = "couldn't cache subject info"
UserGroupsListIsEmpty = "user groups list is empty, subject not found"
CouldntCacheUserKey = "couldn't cache user key"
FoundSeveralBucketCorsNodes = "found several bucket cors nodes, latest be used"
FoundSeveralObjectTaggingNodes = "found several object tagging nodes, latest be used"
FoundSeveralBucketTaggingNodes = "found several bucket tagging nodes, latest be used"
FoundSeveralBucketSettingsNodes = "found several bucket settings nodes, latest be used"
ObjectTaggingNodeHasMultipleIDs = "object tagging node has multiple ids"
BucketTaggingNodeHasMultipleIDs = "bucket tagging node has multiple ids"
BucketSettingsNodeHasMultipleIDs = "bucket settings node has multiple ids"
BucketCORSNodeHasMultipleIDs = "bucket cors node has multiple ids"
SystemNodeHasMultipleIDs = "system node has multiple ids"
FailedToRemoveOldSystemNode = "failed to remove old system node"
UnexpectedMultiNodeIDsInSubTreeMultiParts = "unexpected multi node ids in sub tree multi parts"
FoundSeveralSystemNodes = "found several system nodes, latest be used"
FoundSeveralSystemNodes = "found several system nodes"
FailedToParsePartInfo = "failed to parse part info"
)

187
pkg/service/tree/tree.go
View file

@ -53,6 +53,11 @@ type (
Meta map[string]string
}
multiSystemNode struct {
// the first element is latest
nodes []*treeNode
}
GetNodesParams struct {
BktInfo *data.BucketInfo
TreeID string
@ -268,6 +273,45 @@ func newNodeVersionFromTreeNode(log *zap.Logger, filePath string, treeNode *tree
return version, nil
}
func newMultiNode(nodes []NodeResponse) (*multiSystemNode, error) {
var (
err error
index int
maxTimestamp uint64
)
if len(nodes) == 0 {
return nil, errors.New("multi node must have at least one node")
}
treeNodes := make([]*treeNode, len(nodes))
for i, node := range nodes {
if treeNodes[i], err = newTreeNode(node); err != nil {
return nil, fmt.Errorf("parse system node response: %w", err)
}
if timestamp := getMaxTimestamp(node); timestamp > maxTimestamp {
index = i
maxTimestamp = timestamp
}
}
treeNodes[0], treeNodes[index] = treeNodes[index], treeNodes[0]
return &multiSystemNode{
nodes: treeNodes,
}, nil
}
func (m *multiSystemNode) Latest() *treeNode {
return m.nodes[0]
}
func (m *multiSystemNode) Old() []*treeNode {
return m.nodes[1:]
}
func newMultipartInfoFromTreeNode(log *zap.Logger, filePath string, treeNode *treeNode) (*data.MultipartInfo, error) {
uploadID, _ := treeNode.Get(uploadIDKV)
if uploadID == "" {
@ -394,11 +438,13 @@ func newPartInfo(node NodeResponse) (*data.PartInfo, error) {
}
func (c *Tree) GetSettingsNode(ctx context.Context, bktInfo *data.BucketInfo) (*data.BucketSettings, error) {
node, err := c.getSystemNode(ctx, bktInfo, []string{settingsFileName})
multiNode, err := c.getSystemNode(ctx, bktInfo, settingsFileName)
if err != nil {
return nil, fmt.Errorf("couldn't get node: %w", err)
}
node := multiNode.Latest()
settings := &data.BucketSettings{Versioning: data.VersioningUnversioned}
if versioningValue, ok := node.Get(versioningKV); ok {
settings.Versioning = versioningValue
@ -422,7 +468,7 @@ func (c *Tree) GetSettingsNode(ctx context.Context, bktInfo *data.BucketInfo) (*
}
func (c *Tree) PutSettingsNode(ctx context.Context, bktInfo *data.BucketInfo, settings *data.BucketSettings) error {
node, err := c.getSystemNode(ctx, bktInfo, []string{settingsFileName})
multiNode, err := c.getSystemNode(ctx, bktInfo, settingsFileName)
isErrNotFound := errors.Is(err, layer.ErrNodeNotFound)
if err != nil && !isErrNotFound {
return fmt.Errorf("couldn't get node: %w", err)
@ -435,28 +481,35 @@ func (c *Tree) PutSettingsNode(ctx context.Context, bktInfo *data.BucketInfo, se
return err
}
ind := node.GetLatestNodeIndex()
if node.IsSplit() {
c.reqLogger(ctx).Warn(logs.FoundSeveralBucketSettingsNodes)
latest := multiNode.Latest()
ind := latest.GetLatestNodeIndex()
if latest.IsSplit() {
c.reqLogger(ctx).Error(logs.BucketSettingsNodeHasMultipleIDs, zap.Uint64s("ids", latest.ID))
}
return c.service.MoveNode(ctx, bktInfo, systemTree, node.ID[ind], 0, meta)
if err = c.service.MoveNode(ctx, bktInfo, systemTree, latest.ID[ind], 0, meta); err != nil {
return fmt.Errorf("move settings node: %w", err)
}
c.cleanOldNodes(ctx, multiNode.Old(), bktInfo)
return nil
}
func (c *Tree) GetBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) (oid.ID, error) {
node, err := c.getSystemNode(ctx, bktInfo, []string{corsFilename})
node, err := c.getSystemNode(ctx, bktInfo, corsFilename)
if err != nil {
return oid.ID{}, err
}
return node.ObjID, nil
return node.Latest().ObjID, nil
}
func (c *Tree) PutBucketCORS(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) (oid.ID, error) {
node, err := c.getSystemNode(ctx, bktInfo, []string{corsFilename})
func (c *Tree) PutBucketCORS(ctx context.Context, bktInfo *data.BucketInfo, objID oid.ID) ([]oid.ID, error) {
multiNode, err := c.getSystemNode(ctx, bktInfo, corsFilename)
isErrNotFound := errors.Is(err, layer.ErrNodeNotFound)
if err != nil && !isErrNotFound {
return oid.ID{}, fmt.Errorf("couldn't get node: %w", err)
return nil, fmt.Errorf("couldn't get node: %w", err)
}
meta := make(map[string]string)
@ -465,35 +518,64 @@ func (c *Tree) PutBucketCORS(ctx context.Context, bktInfo *data.BucketInfo, objI
if isErrNotFound {
if _, err = c.service.AddNode(ctx, bktInfo, systemTree, 0, meta); err != nil {
return oid.ID{}, err
return nil, err
}
return oid.ID{}, layer.ErrNoNodeToRemove
return nil, layer.ErrNoNodeToRemove
}
ind := node.GetLatestNodeIndex()
if node.IsSplit() {
c.reqLogger(ctx).Warn(logs.FoundSeveralBucketCorsNodes)
latest := multiNode.Latest()
ind := latest.GetLatestNodeIndex()
if latest.IsSplit() {
c.reqLogger(ctx).Error(logs.BucketCORSNodeHasMultipleIDs)
}
return node.ObjID, c.service.MoveNode(ctx, bktInfo, systemTree, node.ID[ind], 0, meta)
if err = c.service.MoveNode(ctx, bktInfo, systemTree, latest.ID[ind], 0, meta); err != nil {
return nil, fmt.Errorf("move cors node: %w", err)
}
objToDelete := make([]oid.ID, 1, len(multiNode.nodes))
objToDelete[0] = latest.ObjID
objToDelete = append(objToDelete, c.cleanOldNodes(ctx, multiNode.Old(), bktInfo)...)
return objToDelete, nil
}
func (c *Tree) DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) (oid.ID, error) {
node, err := c.getSystemNode(ctx, bktInfo, []string{corsFilename})
if err != nil && !errors.Is(err, layer.ErrNodeNotFound) {
return oid.ID{}, err
func (c *Tree) DeleteBucketCORS(ctx context.Context, bktInfo *data.BucketInfo) ([]oid.ID, error) {
multiNode, err := c.getSystemNode(ctx, bktInfo, corsFilename)
isErrNotFound := errors.Is(err, layer.ErrNodeNotFound)
if err != nil && !isErrNotFound {
return nil, err
}
if node != nil {
if isErrNotFound {
return nil, layer.ErrNoNodeToRemove
}
objToDelete := c.cleanOldNodes(ctx, multiNode.nodes, bktInfo)
if len(objToDelete) != len(multiNode.nodes) {
return nil, fmt.Errorf("clean old cors nodes: %w", err)
}
return objToDelete, nil
}
func (c *Tree) cleanOldNodes(ctx context.Context, nodes []*treeNode, bktInfo *data.BucketInfo) []oid.ID {
res := make([]oid.ID, 0, len(nodes))
for _, node := range nodes {
ind := node.GetLatestNodeIndex()
if node.IsSplit() {
c.reqLogger(ctx).Warn(logs.FoundSeveralBucketCorsNodes)
c.reqLogger(ctx).Error(logs.SystemNodeHasMultipleIDs, zap.String("FileName", node.Meta[FileNameKey]), zap.Uint64s("ids", node.ID))
}
if err := c.service.RemoveNode(ctx, bktInfo, systemTree, node.ID[ind]); err != nil {
c.reqLogger(ctx).Warn(logs.FailedToRemoveOldSystemNode, zap.String("FileName", node.Meta[FileNameKey]), zap.Uint64("id", node.ID[ind]))
} else {
res = append(res, node.ObjID)
}
return node.ObjID, c.service.RemoveNode(ctx, bktInfo, systemTree, node.ID[ind])
}
return oid.ID{}, layer.ErrNoNodeToRemove
return res
}
func (c *Tree) GetObjectTagging(ctx context.Context, bktInfo *data.BucketInfo, objVersion *data.NodeVersion) (map[string]string, error) {
@ -541,7 +623,7 @@ func (c *Tree) PutObjectTagging(ctx context.Context, bktInfo *data.BucketInfo, o
ind := tagNode.GetLatestNodeIndex()
if tagNode.IsSplit() {
c.reqLogger(ctx).Warn(logs.FoundSeveralObjectTaggingNodes)
c.reqLogger(ctx).Error(logs.ObjectTaggingNodeHasMultipleIDs)
}
return c.service.MoveNode(ctx, bktInfo, versionTree, tagNode.ID[ind], objVersion.ID, treeTagSet)
@ -552,14 +634,14 @@ func (c *Tree) DeleteObjectTagging(ctx context.Context, bktInfo *data.BucketInfo
}
func (c *Tree) GetBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) (map[string]string, error) {
node, err := c.getSystemNode(ctx, bktInfo, []string{bucketTaggingFilename})
multiNode, err := c.getSystemNode(ctx, bktInfo, bucketTaggingFilename)
if err != nil {
return nil, err
}
tags := make(map[string]string)
for key, val := range node.Meta {
for key, val := range multiNode.Latest().Meta {
if strings.HasPrefix(key, userDefinedTagPrefix) {
tags[strings.TrimPrefix(key, userDefinedTagPrefix)] = val
}
@ -569,7 +651,7 @@ func (c *Tree) GetBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) (
}
func (c *Tree) PutBucketTagging(ctx context.Context, bktInfo *data.BucketInfo, tagSet map[string]string) error {
node, err := c.getSystemNode(ctx, bktInfo, []string{bucketTaggingFilename})
multiNode, err := c.getSystemNode(ctx, bktInfo, bucketTaggingFilename)
isErrNotFound := errors.Is(err, layer.ErrNodeNotFound)
if err != nil && !isErrNotFound {
return fmt.Errorf("couldn't get node: %w", err)
@ -587,12 +669,19 @@ func (c *Tree) PutBucketTagging(ctx context.Context, bktInfo *data.BucketInfo, t
return err
}
ind := node.GetLatestNodeIndex()
if node.IsSplit() {
c.reqLogger(ctx).Warn(logs.FoundSeveralBucketTaggingNodes)
latest := multiNode.Latest()
ind := latest.GetLatestNodeIndex()
if latest.IsSplit() {
c.reqLogger(ctx).Error(logs.BucketTaggingNodeHasMultipleIDs, zap.Uint64s("ids", latest.ID))
}
return c.service.MoveNode(ctx, bktInfo, systemTree, node.ID[ind], 0, treeTagSet)
if err = c.service.MoveNode(ctx, bktInfo, systemTree, latest.ID[ind], 0, treeTagSet); err != nil {
return fmt.Errorf("move bucket tagging node: %w", err)
}
c.cleanOldNodes(ctx, multiNode.Old(), bktInfo)
return nil
}
func (c *Tree) DeleteBucketTagging(ctx context.Context, bktInfo *data.BucketInfo) error {
@ -615,6 +704,8 @@ func (c *Tree) getTreeNodes(ctx context.Context, bktInfo *data.BucketInfo, nodeI
return nil, err
}
// consider using map[string][]*treeNode
// to be able to remove unused node, that can be added during split
treeNodes := make(map[string]*treeNode, len(keys))
for _, s := range subtree {
@ -689,26 +780,6 @@ func getLatestVersionNode(nodes []NodeResponse) (NodeResponse, error) {
return nodes[targetIndexNode], nil
}
func getLatestNode(nodes []NodeResponse) NodeResponse {
if len(nodes) == 0 {
return nil
}
var (
index int
maxTimestamp uint64
)
for i, node := range nodes {
if timestamp := getMaxTimestamp(node); timestamp > maxTimestamp {
index = i
maxTimestamp = timestamp
}
}
return nodes[index]
}
func getMaxTimestamp(node NodeResponse) uint64 {
var maxTimestamp uint64
@ -1558,11 +1629,11 @@ func metaFromMultipart(info *data.MultipartInfo, fileName string) map[string]str
return info.Meta
}
func (c *Tree) getSystemNode(ctx context.Context, bktInfo *data.BucketInfo, path []string) (*treeNode, error) {
func (c *Tree) getSystemNode(ctx context.Context, bktInfo *data.BucketInfo, name string) (*multiSystemNode, error) {
p := &GetNodesParams{
BktInfo: bktInfo,
TreeID: systemTree,
Path: path,
Path: []string{name},
LatestOnly: false,
AllAttrs: true,
}
@ -1577,10 +1648,10 @@ func (c *Tree) getSystemNode(ctx context.Context, bktInfo *data.BucketInfo, path
return nil, layer.ErrNodeNotFound
}
if len(nodes) != 1 {
c.reqLogger(ctx).Warn(logs.FoundSeveralSystemNodes, zap.String("path", strings.Join(path, "/")))
c.reqLogger(ctx).Warn(logs.FoundSeveralSystemNodes, zap.String("name", name))
}
return newTreeNode(getLatestNode(nodes))
return newMultiNode(nodes)
}
func filterMultipartNodes(nodes []NodeResponse) []NodeResponse {