mirror/opencloud
1
0
Fork 0
mirror of https://github.com/opencloud-eu/opencloud.git synced 2026-07-19 12:04:09 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #9891 from kobergj/SpeedUpRevisionPurge

Browse Source 
Improve Revisions Purge
This commit is contained in:
kobergj
2024-08-23 11:11:19 +02:00
committed by GitHub
parent b8c4b86f7d e548f29b84
commit d66a82b470
4 changed files with 370 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
changelog/unreleased/improve-revisions-purge.md Normal file
View File

@@ -0,0 +1,5 @@
Enhancement: Improve revisions purge
The `revisions purge` command would time out on big spaces. We have improved performance by parallelizing the process.
https://github.com/owncloud/ocis/pull/9891

71
ocis/pkg/command/revisions.go
View File

@@ -5,6 +5,7 @@ import (
"fmt"
"path/filepath"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
ocisbs "github.com/cs3org/reva/v2/pkg/storage/fs/ocis/blobstore"
"github.com/cs3org/reva/v2/pkg/storage/fs/posix/lookup"
s3bs "github.com/cs3org/reva/v2/pkg/storage/fs/s3ng/blobstore"
@@ -19,7 +20,7 @@ import (
var (
// _nodesGlobPattern is the glob pattern to find all nodes
_nodesGlobPattern = "spaces/*/*/*/*/*/*/*/*"
_nodesGlobPattern = "spaces/*/*/nodes/"
)
// RevisionsCommand is the entrypoint for the revisions command.
@@ -30,7 +31,7 @@ func RevisionsCommand(cfg *config.Config) *cli.Command {
Subcommands: []*cli.Command{
PurgeRevisionsCommand(cfg),
},
Before: func(c *cli.Context) error {
Before: func(_ *cli.Context) error {
return configlog.ReturnError(parser.ParseConfig(cfg, true))
},
Action: func(_ *cli.Context) error {
@@ -74,6 +75,11 @@ func PurgeRevisionsCommand(cfg *config.Config) *cli.Command {
Aliases: []string{"r"},
Usage: "purge all revisions of this file/space. If not set, all revisions will be purged",
},
&cli.StringFlag{
Name: "glob-mechanism",
Usage: "the glob mechanism to find all nodes. Can be 'glob', 'list' or 'workers'. 'glob' uses globbing with a single worker. 'workers' spawns multiple go routines, accelatering the command drastically but causing high cpu and ram usage. 'list' looks for references by listing directories with multiple workers. Default is 'glob'",
Value: "glob",
},
},
Action: func(c *cli.Context) error {
basePath := c.String("basepath")
@@ -108,43 +114,72 @@ func PurgeRevisionsCommand(cfg *config.Config) *cli.Command {
return err
}
p, err := generatePath(basePath, c.String("resource-id"))
if err != nil {
fmt.Printf("❌ Error parsing resourceID: %s", err)
return err
var rid *provider.ResourceId
resid, err := storagespace.ParseID(c.String("resource-id"))
if err == nil {
rid = &resid
}
if err := revisions.PurgeRevisions(p, bs, c.Bool("dry-run"), c.Bool("verbose")); err != nil {
fmt.Printf("❌ Error purging revisions: %s", err)
return err
mechanism := c.String("glob-mechanism")
if rid.GetOpaqueId() != "" {
mechanism = "glob"
}
var ch <-chan string
switch mechanism {
default:
fallthrough
case "glob":
p := generatePath(basePath, rid)
if rid.GetOpaqueId() == "" {
p = filepath.Join(p, "*/*/*/*/*")
}
ch = revisions.Glob(p)
case "workers":
p := generatePath(basePath, rid)
ch = revisions.GlobWorkers(p, "/*", "/*/*/*/*")
case "list":
p := filepath.Join(basePath, "spaces")
if rid != nil {
p = generatePath(basePath, rid)
}
ch = revisions.List(p, 10)
}
files, blobs, revisions := revisions.PurgeRevisions(ch, bs, c.Bool("dry-run"), c.Bool("verbose"))
printResults(files, blobs, revisions, c.Bool("dry-run"))
return nil
},
}
}
func generatePath(basePath string, resourceID string) (string, error) {
if resourceID == "" {
return filepath.Join(basePath, _nodesGlobPattern), nil
func printResults(countFiles, countBlobs, countRevisions int, dryRun bool) {
switch {
case countFiles == 0 && countRevisions == 0 && countBlobs == 0:
fmt.Println("❎ No revisions found. Storage provider is clean.")
case !dryRun:
fmt.Printf("✅ Deleted %d revisions (%d files / %d blobs)\n", countRevisions, countFiles, countBlobs)
default:
fmt.Printf("👉 Would delete %d revisions (%d files / %d blobs)\n", countRevisions, countFiles, countBlobs)
}
}
rid, err := storagespace.ParseID(resourceID)
if err != nil {
return "", err
func generatePath(basePath string, rid *provider.ResourceId) string {
if rid == nil {
return filepath.Join(basePath, _nodesGlobPattern)
}
sid := lookup.Pathify(rid.GetSpaceId(), 1, 2)
if sid == "" {
sid = "*/*"
return ""
}
nid := lookup.Pathify(rid.GetOpaqueId(), 4, 2)
if nid == "" {
nid = "*/*/*/*/"
return filepath.Join(basePath, "spaces", sid, "nodes")
}
return filepath.Join(basePath, "spaces", sid, "nodes", nid+"*"), nil
return filepath.Join(basePath, "spaces", sid, "nodes", nid+"*")
}
func init() {

165
ocis/pkg/revisions/revisions.go
View File

@@ -2,12 +2,12 @@
package revisions
import (
"errors"
"fmt"
"os"
"path/filepath"
"regexp"
"strings"
"sync"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
"github.com/shamaton/msgpack/v2"
@@ -26,25 +26,121 @@ type DelBlobstore interface {
Delete(node *node.Node) error
}
// Glob uses globbing to find all revision nodes in a storage provider.
func Glob(pattern string) <-chan string {
ch := make(chan string)
go func() {
defer close(ch)
nodes, err := filepath.Glob(filepath.Join(pattern))
if err != nil {
fmt.Println("error globbing", pattern, err)
return
}
if len(nodes) == 0 {
fmt.Println("no nodes found. Double check storage path")
return
}
for _, n := range nodes {
if _versionRegex.MatchString(n) {
ch <- n
}
}
}()
return ch
}
// GlobWorkers uses multiple go routine to glob all revision nodes in a storage provider.
func GlobWorkers(pattern string, depth string, remainder string) <-chan string {
wg := sync.WaitGroup{}
ch := make(chan string)
go func() {
defer close(ch)
nodes, err := filepath.Glob(pattern + depth)
if err != nil {
fmt.Println("error globbing", pattern, err)
return
}
if len(nodes) == 0 {
fmt.Println("no nodes found. Double check storage path")
return
}
for _, node := range nodes {
wg.Add(1)
go func(node string) {
defer wg.Done()
nodes, err := filepath.Glob(node + remainder)
if err != nil {
fmt.Println("error globbing", node, err)
return
}
for _, n := range nodes {
if _versionRegex.MatchString(n) {
ch <- n
}
}
}(node)
}
wg.Wait()
}()
return ch
}
// Walk walks the storage provider to find all revision nodes.
func Walk(base string) <-chan string {
ch := make(chan string)
go func() {
defer close(ch)
err := filepath.Walk(base, func(path string, info os.FileInfo, err error) error {
if err != nil {
fmt.Println("error walking", base, err)
return err
}
if !_versionRegex.MatchString(info.Name()) {
return nil
}
ch <- path
return nil
})
if err != nil {
fmt.Println("error walking", base, err)
return
}
}()
return ch
}
// List uses directory listing to find all revision nodes in a storage provider.
func List(base string, workers int) <-chan string {
ch := make(chan string)
go func() {
defer close(ch)
if err := listFolder(base, ch, make(chan struct{}, workers)); err != nil {
fmt.Println("error listing", base, err)
return
}
}()
return ch
}
// PurgeRevisions removes all revisions from a storage provider.
func PurgeRevisions(pattern string, bs DelBlobstore, dryRun bool, verbose bool) error {
if verbose {
fmt.Println("Looking for nodes in", pattern)
}
nodes, err := filepath.Glob(pattern)
if err != nil {
return err
}
if len(nodes) == 0 {
return errors.New("no nodes found, double check storage path")
}
func PurgeRevisions(nodes <-chan string, bs DelBlobstore, dryRun, verbose bool) (int, int, int) {
countFiles := 0
countBlobs := 0
countRevisions := 0
for _, d := range nodes {
var err error
for d := range nodes {
if !_versionRegex.MatchString(d) {
continue
}
@@ -106,14 +202,37 @@ func PurgeRevisions(pattern string, bs DelBlobstore, dryRun bool, verbose bool)
}
}
switch {
case countFiles == 0 && countRevisions == 0 && countBlobs == 0:
fmt.Println("❎ No revisions found. Storage provider is clean.")
case !dryRun:
fmt.Printf("✅ Deleted %d revisions (%d files / %d blobs)\n", countRevisions, countFiles, countBlobs)
default:
fmt.Printf("👉 Would delete %d revisions (%d files / %d blobs)\n", countRevisions, countFiles, countBlobs)
return countFiles, countBlobs, countRevisions
}
func listFolder(path string, ch chan<- string, workers chan struct{}) error {
workers <- struct{}{}
wg := sync.WaitGroup{}
children, err := os.ReadDir(path)
if err != nil {
<-workers
return err
}
for _, child := range children {
if child.IsDir() {
wg.Add(1)
go func() {
defer wg.Done()
if err := listFolder(filepath.Join(path, child.Name()), ch, workers); err != nil {
fmt.Println("error listing", path, err)
}
}()
}
if _versionRegex.MatchString(child.Name()) {
ch <- filepath.Join(path, child.Name())
}
}
<-workers
wg.Wait()
return nil
}

170
ocis/pkg/revisions/revisions_test.go Normal file
View File

@@ -0,0 +1,170 @@
package revisions
import (
"fmt"
"io/fs"
"os"
"path/filepath"
"strconv"
"testing"
"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
"github.com/google/uuid"
"github.com/test-go/testify/require"
)
var (
_basePath = "/spaces/8f/638374-6ea8-4f0d-80c4-66d9b49830a5/nodes/"
)
// func TestInit(t *testing.T) {
// initialize(10, 2)
// defer os.RemoveAll("test_temp")
// }
func TestGlob30(t *testing.T) { test(t, 10, 2, glob) }
func TestGlob80(t *testing.T) { test(t, 20, 3, glob) }
func TestGlob250(t *testing.T) { test(t, 50, 4, glob) }
func TestGlob600(t *testing.T) { test(t, 100, 5, glob) }
func TestWalk30(t *testing.T) { test(t, 10, 2, walk) }
func TestWalk80(t *testing.T) { test(t, 20, 3, walk) }
func TestWalk250(t *testing.T) { test(t, 50, 4, walk) }
func TestWalk600(t *testing.T) { test(t, 100, 5, walk) }
func TestList30(t *testing.T) { test(t, 10, 2, list2) }
func TestList80(t *testing.T) { test(t, 20, 3, list10) }
func TestList250(t *testing.T) { test(t, 50, 4, list20) }
func TestList600(t *testing.T) { test(t, 100, 5, list2) }
func TestGlobWorkers30(t *testing.T) { test(t, 10, 2, globWorkersD1) }
func TestGlobWorkers80(t *testing.T) { test(t, 20, 3, globWorkersD2) }
func TestGlobWorkers250(t *testing.T) { test(t, 50, 4, globWorkersD4) }
func TestGlobWorkers600(t *testing.T) { test(t, 100, 5, globWorkersD2) }
func BenchmarkGlob30(b *testing.B) { benchmark(b, 10, 2, glob) }
func BenchmarkWalk30(b *testing.B) { benchmark(b, 10, 2, walk) }
func BenchmarkList30(b *testing.B) { benchmark(b, 10, 2, list2) }
func BenchmarkGlobWorkers30(b *testing.B) { benchmark(b, 10, 2, globWorkersD2) }
func BenchmarkGlob80(b *testing.B) { benchmark(b, 20, 3, glob) }
func BenchmarkWalk80(b *testing.B) { benchmark(b, 20, 3, walk) }
func BenchmarkList80(b *testing.B) { benchmark(b, 20, 3, list2) }
func BenchmarkGlobWorkers80(b *testing.B) { benchmark(b, 20, 3, globWorkersD2) }
func BenchmarkGlob250(b *testing.B) { benchmark(b, 50, 4, glob) }
func BenchmarkWalk250(b *testing.B) { benchmark(b, 50, 4, walk) }
func BenchmarkList250(b *testing.B) { benchmark(b, 50, 4, list2) }
func BenchmarkGlobWorkers250(b *testing.B) { benchmark(b, 50, 4, globWorkersD2) }
func BenchmarkGlobAT600(b *testing.B) { benchmark(b, 100, 5, glob) }
func BenchmarkWalkAT600(b *testing.B) { benchmark(b, 100, 5, walk) }
func BenchmarkList2AT600(b *testing.B) { benchmark(b, 100, 5, list2) }
func BenchmarkList10AT600(b *testing.B) { benchmark(b, 100, 5, list10) }
func BenchmarkList20AT600(b *testing.B) { benchmark(b, 100, 5, list20) }
func BenchmarkGlobWorkersD1AT600(b *testing.B) { benchmark(b, 100, 5, globWorkersD1) }
func BenchmarkGlobWorkersD2AT600(b *testing.B) { benchmark(b, 100, 5, globWorkersD2) }
func BenchmarkGlobWorkersD4AT600(b *testing.B) { benchmark(b, 100, 5, globWorkersD4) }
func BenchmarkGlobAT22000(b *testing.B) { benchmark(b, 2000, 10, glob) }
func BenchmarkWalkAT22000(b *testing.B) { benchmark(b, 2000, 10, walk) }
func BenchmarkList2AT22000(b *testing.B) { benchmark(b, 2000, 10, list2) }
func BenchmarkList10AT22000(b *testing.B) { benchmark(b, 2000, 10, list10) }
func BenchmarkList20AT22000(b *testing.B) { benchmark(b, 2000, 10, list20) }
func BenchmarkGlobWorkersD1AT22000(b *testing.B) { benchmark(b, 2000, 10, globWorkersD1) }
func BenchmarkGlobWorkersD2AT22000(b *testing.B) { benchmark(b, 2000, 10, globWorkersD2) }
func BenchmarkGlobWorkersD4AT22000(b *testing.B) { benchmark(b, 2000, 10, globWorkersD4) }
func BenchmarkGlob110000(b *testing.B) { benchmark(b, 10000, 10, glob) }
func BenchmarkWalk110000(b *testing.B) { benchmark(b, 10000, 10, walk) }
func BenchmarkList110000(b *testing.B) { benchmark(b, 10000, 10, list2) }
func BenchmarkGlobWorkers110000(b *testing.B) { benchmark(b, 10000, 10, globWorkersD2) }
func benchmark(b *testing.B, numNodes int, numRevisions int, f func(string) <-chan string) {
base := initialize(numNodes, numRevisions)
defer os.RemoveAll(base)
b.ResetTimer()
for i := 0; i < b.N; i++ {
ch := f(base)
PurgeRevisions(ch, nil, false, false)
}
b.StopTimer()
}
func test(t *testing.T, numNodes int, numRevisions int, f func(string) <-chan string) {
base := initialize(numNodes, numRevisions)
defer os.RemoveAll(base)
ch := f(base)
_, _, revisions := PurgeRevisions(ch, nil, false, false)
require.Equal(t, numNodes*numRevisions, revisions, "Deleted Revisions")
}
func glob(base string) <-chan string {
return Glob(base + _basePath + "*/*/*/*/*")
}
func walk(base string) <-chan string {
return Walk(base + _basePath)
}
func list2(base string) <-chan string {
return List(base+_basePath, 2)
}
func list10(base string) <-chan string {
return List(base+_basePath, 10)
}
func list20(base string) <-chan string {
return List(base+_basePath, 20)
}
func globWorkersD1(base string) <-chan string {
return GlobWorkers(base+_basePath, "*", "/*/*/*/*")
}
func globWorkersD2(base string) <-chan string {
return GlobWorkers(base+_basePath, "*/*", "/*/*/*")
}
func globWorkersD4(base string) <-chan string {
return GlobWorkers(base+_basePath, "*/*/*/*", "/*")
}
func initialize(numNodes int, numRevisions int) string {
base := "test_temp_" + uuid.New().String()
if err := os.Mkdir(base, os.ModePerm); err != nil {
fmt.Println("Error creating test_temp directory", err)
os.RemoveAll(base)
os.Exit(1)
}
// create base path
if err := os.MkdirAll(base+_basePath, fs.ModePerm); err != nil {
fmt.Println("Error creating base path", err)
os.RemoveAll(base)
os.Exit(1)
}
for i := 0; i < numNodes; i++ {
path := lookup.Pathify(uuid.New().String(), 4, 2)
dir := filepath.Dir(path)
if err := os.MkdirAll(base+_basePath+dir, fs.ModePerm); err != nil {
fmt.Println("Error creating test_temp directory", err)
os.RemoveAll(base)
os.Exit(1)
}
if _, err := os.Create(base + _basePath + path); err != nil {
fmt.Println("Error creating file", err)
os.RemoveAll(base)
os.Exit(1)
}
for i := 0; i < numRevisions; i++ {
os.Create(base + _basePath + path + ".REV.2024-05-22T07:32:53.89969" + strconv.Itoa(i) + "Z")
}
}
return base
}