From 1525f7ec2a43602194aa54cb7cb8a537faeab602 Mon Sep 17 00:00:00 2001 From: jkoberg Date: Fri, 31 May 2024 18:15:54 +0200 Subject: [PATCH] feat(ocis): use only one channel for backup command Signed-off-by: jkoberg --- .drone.star | 1 - ocis/pkg/backup/backup.go | 91 +++++++++++------------ ocis/pkg/backup/backup_test.go | 19 ++--- ocis/pkg/backup/provider.go | 130 ++++++++++++++++----------------- 4 files changed, 113 insertions(+), 128 deletions(-) diff --git a/.drone.star b/.drone.star index 4b5e36b63c..cdb55be3ea 100644 --- a/.drone.star +++ b/.drone.star @@ -1989,7 +1989,6 @@ def ocisServer(storage, accounts_hash_difficulty = 4, volumes = [], depends_on = "commands": [ "%s init --insecure true" % ocis_bin, "cat $OCIS_CONFIG_DIR/ocis.yaml", - "%s backup consistency -p /var/lib/ocis/storage/users" % ocis_bin, ] + (wrapper_commands), "volumes": volumes, "depends_on": depends_on, diff --git a/ocis/pkg/backup/backup.go b/ocis/pkg/backup/backup.go index 37a57c71e8..27186a6f66 100644 --- a/ocis/pkg/backup/backup.go +++ b/ocis/pkg/backup/backup.go @@ -69,14 +69,53 @@ func CheckProviderConsistency(storagepath string, lbs ListBlobstore) error { } c := NewConsistency() - c.GatherData(p.Nodes, p.Links, p.Blobs, p.Quit) + c.GatherData(p.Events) return c.PrintResults(storagepath) } // GatherData gathers and evaluates data produced by the DataProvider -func (c *Consistency) GatherData(nodes <-chan NodeData, links <-chan LinkData, blobs <-chan BlobData, quit <-chan struct{}) { - c.gatherData(nodes, links, blobs, quit) +func (c *Consistency) GatherData(events <-chan interface{}) { + for ev := range events { + switch d := ev.(type) { + case NodeData: + // does it have inconsistencies? + if len(d.Inconsistencies) != 0 { + c.Nodes[d.NodePath] = append(c.Nodes[d.NodePath], d.Inconsistencies...) + } + // is it linked? + if _, ok := c.LinkedNodes[d.NodePath]; ok { + deleteInconsistency(c.LinkedNodes, d.NodePath) + } else if d.RequiresSymlink && c.Nodes[d.NodePath] == nil { + c.Nodes[d.NodePath] = []Inconsistency{} + } + // does it have a blob? + if d.BlobPath != "" { + if _, ok := c.Blobs[d.BlobPath]; ok { + deleteInconsistency(c.Blobs, d.BlobPath) + } else { + c.BlobReferences[d.BlobPath] = []Inconsistency{} + c.blobToNode[d.BlobPath] = d.NodePath + } + } + case LinkData: + // does it have a node? + if _, ok := c.Nodes[d.NodePath]; ok { + deleteInconsistency(c.Nodes, d.NodePath) + } else { + c.LinkedNodes[d.NodePath] = []Inconsistency{} + c.nodeToLink[d.NodePath] = d.LinkPath + } + case BlobData: + // does it have a reference? + if _, ok := c.BlobReferences[d.BlobPath]; ok { + deleteInconsistency(c.BlobReferences, d.BlobPath) + } else { + c.Blobs[d.BlobPath] = []Inconsistency{} + } + + } + } for n := range c.Nodes { if len(c.Nodes[n]) == 0 { @@ -94,52 +133,6 @@ func (c *Consistency) GatherData(nodes <-chan NodeData, links <-chan LinkData, b } } -func (c *Consistency) gatherData(nodes <-chan NodeData, links <-chan LinkData, blobs <-chan BlobData, quit <-chan struct{}) { - for { - select { - case n := <-nodes: - // does it have inconsistencies? - if len(n.Inconsistencies) != 0 { - c.Nodes[n.NodePath] = append(c.Nodes[n.NodePath], n.Inconsistencies...) - } - // is it linked? - if _, ok := c.LinkedNodes[n.NodePath]; ok { - deleteInconsistency(c.LinkedNodes, n.NodePath) - } else if n.RequiresSymlink { - c.Nodes[n.NodePath] = c.Nodes[n.NodePath] - } - // does it have a blob? - if n.BlobPath != "" { - if _, ok := c.Blobs[n.BlobPath]; ok { - deleteInconsistency(c.Blobs, n.BlobPath) - } else { - c.BlobReferences[n.BlobPath] = []Inconsistency{} - c.blobToNode[n.BlobPath] = n.NodePath - } - } - case l := <-links: - // does it have a node? - if _, ok := c.Nodes[l.NodePath]; ok { - deleteInconsistency(c.Nodes, l.NodePath) - } else { - c.LinkedNodes[l.NodePath] = []Inconsistency{} - c.nodeToLink[l.NodePath] = l.LinkPath - } - case b := <-blobs: - // does it have a reference? - if _, ok := c.BlobReferences[b.BlobPath]; ok { - deleteInconsistency(c.BlobReferences, b.BlobPath) - } else { - c.Blobs[b.BlobPath] = []Inconsistency{} - } - case <-quit: - return - - } - } - -} - // PrintResults prints the results of the evaluation func (c *Consistency) PrintResults(discpath string) error { if len(c.Nodes) != 0 { diff --git a/ocis/pkg/backup/backup_test.go b/ocis/pkg/backup/backup_test.go index 6aff566f37..21e20842d7 100644 --- a/ocis/pkg/backup/backup_test.go +++ b/ocis/pkg/backup/backup_test.go @@ -90,31 +90,24 @@ func TestGatherData(t *testing.T) { } for _, tc := range testcases { - nodes := make(chan backup.NodeData) - links := make(chan backup.LinkData) - blobs := make(chan backup.BlobData) - quit := make(chan struct{}) + events := make(chan interface{}) go func() { for _, ev := range tc.Events { switch e := ev.(type) { case backup.NodeData: - nodes <- e + events <- e case backup.LinkData: - links <- e + events <- e case backup.BlobData: - blobs <- e + events <- e } } - quit <- struct{}{} - close(nodes) - close(links) - close(blobs) - close(quit) + close(events) }() c := backup.NewConsistency() - c.GatherData(nodes, links, blobs, quit) + c.GatherData(events) require.Equal(t, tc.Expected.Nodes, c.Nodes) require.Equal(t, tc.Expected.LinkedNodes, c.LinkedNodes) diff --git a/ocis/pkg/backup/provider.go b/ocis/pkg/backup/provider.go index 7f3e98529a..b9af5c8158 100644 --- a/ocis/pkg/backup/provider.go +++ b/ocis/pkg/backup/provider.go @@ -21,10 +21,7 @@ type ListBlobstore interface { // DataProvider provides data for the consistency check type DataProvider struct { - Nodes chan NodeData - Links chan LinkData - Blobs chan BlobData - Quit chan struct{} + Events chan interface{} fsys fs.FS discpath string @@ -53,10 +50,7 @@ type BlobData struct { // NewProvider creates a new DataProvider object func NewProvider(fsys fs.FS, discpath string, lbs ListBlobstore) *DataProvider { return &DataProvider{ - Nodes: make(chan NodeData), - Links: make(chan LinkData), - Blobs: make(chan BlobData), - Quit: make(chan struct{}), + Events: make(chan interface{}), fsys: fsys, discpath: discpath, @@ -82,47 +76,7 @@ func (dp *DataProvider) ProduceData() error { wg.Add(1) go func() { for _, d := range dirs { - entries, err := fs.ReadDir(dp.fsys, d) - if err != nil { - fmt.Println("error reading dir", err) - continue - } - - if len(entries) == 0 { - fmt.Println("empty dir", filepath.Join(dp.discpath, d)) - continue - } - - for _, e := range entries { - switch { - case e.IsDir(): - ls, err := fs.ReadDir(dp.fsys, filepath.Join(d, e.Name())) - if err != nil { - fmt.Println("error reading dir", err) - continue - } - for _, l := range ls { - linkpath := filepath.Join(dp.discpath, d, e.Name(), l.Name()) - - r, _ := os.Readlink(linkpath) - nodePath := filepath.Join(dp.discpath, d, e.Name(), r) - dp.Links <- LinkData{LinkPath: linkpath, NodePath: nodePath} - } - fallthrough - case filepath.Ext(e.Name()) == "" || _versionRegex.MatchString(e.Name()) || _trashRegex.MatchString(e.Name()): - np := filepath.Join(dp.discpath, d, e.Name()) - var inc []Inconsistency - if !dp.filesExist(filepath.Join(d, e.Name())) { - inc = append(inc, InconsistencyFilesMissing) - } - bp, i := dp.getBlobPath(filepath.Join(d, e.Name())) - if i != "" { - inc = append(inc, i) - } - - dp.Nodes <- NodeData{NodePath: np, BlobPath: bp, RequiresSymlink: requiresSymlink(np), Inconsistencies: inc} - } - } + dp.evaluateNodeDir(d) } wg.Done() }() @@ -130,16 +84,7 @@ func (dp *DataProvider) ProduceData() error { // crawl trash wg.Add(1) go func() { - linkpaths, err := fs.Glob(dp.fsys, "spaces/*/*/trash/*/*/*/*/*") - if err != nil { - fmt.Println("error reading trash", err) - } - for _, l := range linkpaths { - linkpath := filepath.Join(dp.discpath, l) - r, _ := os.Readlink(linkpath) - p := filepath.Join(dp.discpath, l, "..", r) - dp.Links <- LinkData{LinkPath: linkpath, NodePath: p} - } + dp.evaluateTrashDir() wg.Done() }() @@ -152,7 +97,7 @@ func (dp *DataProvider) ProduceData() error { } for _, bn := range bs { - dp.Blobs <- BlobData{BlobPath: dp.lbs.Path(bn)} + dp.Events <- BlobData{BlobPath: dp.lbs.Path(bn)} } wg.Done() }() @@ -187,6 +132,65 @@ func (dp *DataProvider) getBlobPath(path string) (string, Inconsistency) { return "", "" } +func (dp *DataProvider) evaluateNodeDir(d string) { + // d is something like spaces/a8/e5d981-41e4-4468-b532-258d5fb457d3/nodes/2d/08/8d/24 + // we could have multiple nodes under this, but we are only interested in one file per node - the one with "" extension + entries, err := fs.ReadDir(dp.fsys, d) + if err != nil { + fmt.Println("error reading dir", err) + return + } + + if len(entries) == 0 { + fmt.Println("empty dir", filepath.Join(dp.discpath, d)) + return + } + + for _, e := range entries { + switch { + case e.IsDir(): + ls, err := fs.ReadDir(dp.fsys, filepath.Join(d, e.Name())) + if err != nil { + fmt.Println("error reading dir", err) + continue + } + for _, l := range ls { + linkpath := filepath.Join(dp.discpath, d, e.Name(), l.Name()) + + r, _ := os.Readlink(linkpath) + nodePath := filepath.Join(dp.discpath, d, e.Name(), r) + dp.Events <- LinkData{LinkPath: linkpath, NodePath: nodePath} + } + fallthrough + case filepath.Ext(e.Name()) == "" || _versionRegex.MatchString(e.Name()) || _trashRegex.MatchString(e.Name()): + np := filepath.Join(dp.discpath, d, e.Name()) + var inc []Inconsistency + if !dp.filesExist(filepath.Join(d, e.Name())) { + inc = append(inc, InconsistencyFilesMissing) + } + bp, i := dp.getBlobPath(filepath.Join(d, e.Name())) + if i != "" { + inc = append(inc, i) + } + + dp.Events <- NodeData{NodePath: np, BlobPath: bp, RequiresSymlink: requiresSymlink(np), Inconsistencies: inc} + } + } +} + +func (dp *DataProvider) evaluateTrashDir() { + linkpaths, err := fs.Glob(dp.fsys, "spaces/*/*/trash/*/*/*/*/*") + if err != nil { + fmt.Println("error reading trash", err) + } + for _, l := range linkpaths { + linkpath := filepath.Join(dp.discpath, l) + r, _ := os.Readlink(linkpath) + p := filepath.Join(dp.discpath, l, "..", r) + dp.Events <- LinkData{LinkPath: linkpath, NodePath: p} + } +} + func (dp *DataProvider) filesExist(path string) bool { check := func(p string) bool { _, err := fs.Stat(dp.fsys, p) @@ -196,11 +200,7 @@ func (dp *DataProvider) filesExist(path string) bool { } func (dp *DataProvider) quit() { - dp.Quit <- struct{}{} - close(dp.Nodes) - close(dp.Links) - close(dp.Blobs) - close(dp.Quit) + close(dp.Events) } func requiresSymlink(path string) bool {