restic/archiver.go

763 lines
18 KiB
Go
Raw Normal View History

2014-12-05 20:45:49 +00:00
package restic
2014-09-23 20:39:12 +00:00
import (
"crypto/sha256"
"encoding/json"
"fmt"
"io"
2014-09-23 20:39:12 +00:00
"os"
2015-02-21 13:23:49 +00:00
"path/filepath"
2015-03-02 13:48:47 +00:00
"sort"
2014-11-16 21:50:20 +00:00
"sync"
2015-10-12 21:59:17 +00:00
"time"
2014-09-23 20:39:12 +00:00
2015-07-07 22:55:58 +00:00
"github.com/restic/chunker"
2014-12-05 20:45:49 +00:00
"github.com/restic/restic/backend"
2015-01-14 21:08:48 +00:00
"github.com/restic/restic/debug"
"github.com/restic/restic/pack"
2015-02-15 13:44:54 +00:00
"github.com/restic/restic/pipe"
"github.com/restic/restic/repository"
"github.com/juju/errors"
2014-09-23 20:39:12 +00:00
)
2014-11-16 21:50:20 +00:00
const (
2015-05-02 14:39:24 +00:00
maxConcurrentBlobs = 32
maxConcurrency = 10
2014-11-16 21:50:20 +00:00
)
2015-04-30 01:41:51 +00:00
var archiverAbortOnAllErrors = func(str string, fi os.FileInfo, err error) error { return err }
var archiverAllowAllFiles = func(string, os.FileInfo) bool { return true }
2015-05-02 13:51:40 +00:00
// Archiver is used to backup a set of directories.
2014-09-23 20:39:12 +00:00
type Archiver struct {
2015-05-09 21:59:58 +00:00
repo *repository.Repository
2014-09-23 20:39:12 +00:00
2014-11-22 21:05:39 +00:00
blobToken chan struct{}
2014-11-16 21:50:20 +00:00
Error func(dir string, fi os.FileInfo, err error) error
SelectFilter pipe.SelectFunc
Excludes []string
2014-09-23 20:39:12 +00:00
}
2015-05-02 13:51:40 +00:00
// NewArchiver returns a new archiver.
2015-05-09 21:59:58 +00:00
func NewArchiver(repo *repository.Repository) *Archiver {
2014-11-16 21:50:20 +00:00
arch := &Archiver{
2015-05-09 11:32:52 +00:00
repo: repo,
2014-11-22 21:05:39 +00:00
blobToken: make(chan struct{}, maxConcurrentBlobs),
2014-11-16 21:50:20 +00:00
}
2014-11-22 21:05:39 +00:00
for i := 0; i < maxConcurrentBlobs; i++ {
arch.blobToken <- struct{}{}
}
2015-04-30 01:41:51 +00:00
arch.Error = archiverAbortOnAllErrors
arch.SelectFilter = archiverAllowAllFiles
2014-09-23 20:39:12 +00:00
2015-04-30 01:41:51 +00:00
return arch
}
2015-05-09 11:35:55 +00:00
// Save stores a blob read from rd in the repository.
func (arch *Archiver) Save(t pack.BlobType, id backend.ID, length uint, rd io.Reader) error {
2015-01-14 21:08:48 +00:00
debug.Log("Archiver.Save", "Save(%v, %v)\n", t, id.Str())
// test if this blob is already known
2015-05-09 11:32:52 +00:00
if arch.repo.Index().Has(id) {
debug.Log("Archiver.Save", "(%v, %v) already saved\n", t, id.Str())
return nil
}
2014-09-23 20:39:12 +00:00
// otherwise save blob
err := arch.repo.SaveFrom(t, &id, length, rd)
if err != nil {
debug.Log("Archiver.Save", "Save(%v, %v): error %v\n", t, id.Str(), err)
return err
2014-09-23 20:39:12 +00:00
}
debug.Log("Archiver.Save", "Save(%v, %v): new blob\n", t, id.Str())
return nil
2014-09-23 20:39:12 +00:00
}
2015-05-09 11:35:55 +00:00
// SaveTreeJSON stores a tree in the repository.
func (arch *Archiver) SaveTreeJSON(item interface{}) (backend.ID, error) {
data, err := json.Marshal(item)
2014-09-23 20:39:12 +00:00
if err != nil {
return backend.ID{}, err
2014-09-23 20:39:12 +00:00
}
2015-04-30 01:41:51 +00:00
data = append(data, '\n')
2014-09-23 20:39:12 +00:00
// check if tree has been saved before
id := backend.Hash(data)
2015-10-12 20:34:12 +00:00
if arch.repo.Index().IsInFlight(id) || arch.repo.Index().Has(id) {
return id, nil
}
2014-09-23 20:39:12 +00:00
2015-05-09 11:32:52 +00:00
return arch.repo.SaveJSON(pack.Tree, item)
2014-09-23 20:39:12 +00:00
}
func (arch *Archiver) reloadFileIfChanged(node *Node, file *os.File) (*Node, error) {
fi, err := file.Stat()
if err != nil {
return nil, err
}
if fi.ModTime() == node.ModTime {
return node, nil
}
err = arch.Error(node.path, fi, errors.New("file has changed"))
if err != nil {
return nil, err
}
node, err = NodeFromFileInfo(node.path, fi)
if err != nil {
debug.Log("Archiver.SaveFile", "NodeFromFileInfo returned error for %v: %v", node.path, err)
return nil, err
}
return node, nil
}
type saveResult struct {
id backend.ID
bytes uint64
}
2014-11-22 21:05:39 +00:00
func (arch *Archiver) saveChunk(chunk *chunker.Chunk, p *Progress, token struct{}, file *os.File, resultChannel chan<- saveResult) {
hash := chunk.Digest
id := backend.ID{}
copy(id[:], hash)
err := arch.Save(pack.Data, id, chunk.Length, chunk.Reader(file))
// TODO handle error
if err != nil {
panic(err)
}
p.Report(Stat{Bytes: uint64(chunk.Length)})
arch.blobToken <- token
resultChannel <- saveResult{id: id, bytes: uint64(chunk.Length)}
}
2014-11-22 21:05:39 +00:00
func waitForResults(resultChannels [](<-chan saveResult)) ([]saveResult, error) {
results := []saveResult{}
for _, ch := range resultChannels {
results = append(results, <-ch)
}
if len(results) != len(resultChannels) {
return nil, fmt.Errorf("chunker returned %v chunks, but only %v blobs saved", len(resultChannels), len(results))
2014-09-23 20:39:12 +00:00
}
return results, nil
}
func updateNodeContent(node *Node, results []saveResult) error {
debug.Log("Archiver.Save", "checking size for file %s", node.path)
var bytes uint64
node.Content = make([]backend.ID, len(results))
for i, b := range results {
node.Content[i] = b.id
bytes += b.bytes
debug.Log("Archiver.Save", " adding blob %s, %d bytes", b.id.Str(), b.bytes)
}
if bytes != node.Size {
return fmt.Errorf("errors saving node %q: saved %d bytes, wanted %d bytes", node.path, bytes, node.Size)
2014-09-23 20:39:12 +00:00
}
debug.Log("Archiver.SaveFile", "SaveFile(%q): %v blobs\n", node.path, len(results))
return nil
2014-09-23 20:39:12 +00:00
}
// SaveFile stores the content of the file on the backend as a Blob by calling
// Save for each chunk.
func (arch *Archiver) SaveFile(p *Progress, node *Node) error {
file, err := node.OpenForReading()
defer file.Close()
if err != nil {
return err
}
node, err = arch.reloadFileIfChanged(node, file)
if err != nil {
return err
}
2015-05-09 11:32:52 +00:00
chnker := chunker.New(file, arch.repo.Config.ChunkerPolynomial, sha256.New())
resultChannels := [](<-chan saveResult){}
for {
chunk, err := chnker.Next()
if err == io.EOF {
break
}
if err != nil {
2015-05-02 12:34:33 +00:00
return errors.Annotate(err, "SaveFile() chunker.Next()")
}
resCh := make(chan saveResult, 1)
go arch.saveChunk(chunk, p, <-arch.blobToken, file, resCh)
resultChannels = append(resultChannels, resCh)
}
results, err := waitForResults(resultChannels)
if err != nil {
return err
}
err = updateNodeContent(node, results)
return err
}
2015-03-02 13:48:47 +00:00
func (arch *Archiver) fileWorker(wg *sync.WaitGroup, p *Progress, done <-chan struct{}, entCh <-chan pipe.Entry) {
2015-03-07 10:53:32 +00:00
defer func() {
debug.Log("Archiver.fileWorker", "done")
wg.Done()
}()
2015-03-02 13:48:47 +00:00
for {
select {
case e, ok := <-entCh:
if !ok {
// channel is closed
return
}
2015-03-07 10:53:32 +00:00
debug.Log("Archiver.fileWorker", "got job %v", e)
2015-03-15 11:20:30 +00:00
// check for errors
if e.Error() != nil {
debug.Log("Archiver.fileWorker", "job %v has errors: %v", e.Path(), e.Error())
// TODO: integrate error reporting
fmt.Fprintf(os.Stderr, "error for %v: %v\n", e.Path(), e.Error())
// ignore this file
e.Result() <- nil
p.Report(Stat{Errors: 1})
continue
2015-03-15 11:20:30 +00:00
}
2015-03-07 10:53:32 +00:00
node, err := NodeFromFileInfo(e.Fullpath(), e.Info())
2015-03-02 13:48:47 +00:00
if err != nil {
2015-03-21 13:43:33 +00:00
// TODO: integrate error reporting
debug.Log("Archiver.fileWorker", "NodeFromFileInfo returned error for %v: %v", node.path, err)
e.Result() <- nil
p.Report(Stat{Errors: 1})
2015-03-21 13:43:33 +00:00
continue
2015-03-02 13:48:47 +00:00
}
2015-03-07 10:53:32 +00:00
// try to use old node, if present
if e.Node != nil {
debug.Log("Archiver.fileWorker", " %v use old data", e.Path())
oldNode := e.Node.(*Node)
// check if all content is still available in the repository
contentMissing := false
for _, blob := range oldNode.blobs {
2015-05-17 18:51:32 +00:00
if ok, err := arch.repo.Backend().Test(backend.Data, blob.Storage.String()); !ok || err != nil {
2015-03-07 10:53:32 +00:00
debug.Log("Archiver.fileWorker", " %v not using old data, %v (%v) is missing", e.Path(), blob.ID.Str(), blob.Storage.Str())
contentMissing = true
break
}
}
if !contentMissing {
node.Content = oldNode.Content
node.blobs = oldNode.blobs
debug.Log("Archiver.fileWorker", " %v content is complete", e.Path())
}
} else {
debug.Log("Archiver.fileWorker", " %v no old data", e.Path())
}
// otherwise read file normally
if node.Type == "file" && len(node.Content) == 0 {
debug.Log("Archiver.fileWorker", " read and save %v, content: %v", e.Path(), node.Content)
err = arch.SaveFile(p, node)
2015-03-02 13:48:47 +00:00
if err != nil {
// TODO: integrate error reporting
fmt.Fprintf(os.Stderr, "error for %v: %v\n", node.path, err)
// ignore this file
e.Result() <- nil
p.Report(Stat{Errors: 1})
continue
2015-03-02 13:48:47 +00:00
}
2015-03-07 10:53:32 +00:00
} else {
// report old data size
p.Report(Stat{Bytes: node.Size})
2015-03-02 13:48:47 +00:00
}
2015-03-07 10:53:32 +00:00
debug.Log("Archiver.fileWorker", " processed %v, %d/%d blobs", e.Path(), len(node.Content), len(node.blobs))
e.Result() <- node
2015-03-02 13:48:47 +00:00
p.Report(Stat{Files: 1})
case <-done:
// pipeline was cancelled
return
}
}
2015-03-02 13:48:47 +00:00
}
2015-03-02 13:48:47 +00:00
func (arch *Archiver) dirWorker(wg *sync.WaitGroup, p *Progress, done <-chan struct{}, dirCh <-chan pipe.Dir) {
2015-03-07 10:53:32 +00:00
defer func() {
debug.Log("Archiver.dirWorker", "done")
wg.Done()
}()
2015-03-02 13:48:47 +00:00
for {
select {
case dir, ok := <-dirCh:
if !ok {
// channel is closed
return
}
2015-03-07 10:53:32 +00:00
debug.Log("Archiver.dirWorker", "save dir %v\n", dir.Path())
2015-03-02 13:48:47 +00:00
tree := NewTree()
2015-02-15 13:44:54 +00:00
2015-03-02 13:48:47 +00:00
// wait for all content
for _, ch := range dir.Entries {
res := <-ch
// if we get a nil pointer here, an error has happened while
// processing this entry. Ignore it for now.
if res == nil {
continue
}
// else insert node
node := res.(*Node)
2015-03-02 13:48:47 +00:00
tree.Insert(node)
2015-02-15 13:44:54 +00:00
2015-03-02 13:48:47 +00:00
if node.Type == "dir" {
2015-03-07 10:53:32 +00:00
debug.Log("Archiver.dirWorker", "got tree node for %s: %v", node.path, node.blobs)
if node.Subtree.IsNull() {
panic("invalid null subtree ID")
}
2015-02-15 13:44:54 +00:00
}
2015-03-02 13:48:47 +00:00
}
2015-02-15 13:44:54 +00:00
var (
node *Node
err error
)
if dir.Path() == "" {
// if this is the top-level dir, only create a stub node
node = &Node{}
} else {
2015-04-25 19:40:42 +00:00
// else create node from path and fi
node, err = NodeFromFileInfo(dir.Path(), dir.Info())
if err != nil {
node.Error = err.Error()
dir.Result() <- node
continue
}
2015-03-02 13:48:47 +00:00
}
id, err := arch.SaveTreeJSON(tree)
2015-03-02 13:48:47 +00:00
if err != nil {
panic(err)
2015-02-15 13:44:54 +00:00
}
debug.Log("Archiver.dirWorker", "save tree for %s: %v", dir.Path(), id.Str())
if id.IsNull() {
panic("invalid null subtree ID return from SaveTreeJSON()")
}
2015-03-02 13:48:47 +00:00
node.Subtree = &id
2015-03-02 13:48:47 +00:00
2015-03-07 10:53:32 +00:00
dir.Result() <- node
2015-04-25 19:40:42 +00:00
if dir.Path() != "" {
p.Report(Stat{Dirs: 1})
}
2015-03-02 13:48:47 +00:00
case <-done:
// pipeline was cancelled
return
2015-02-15 13:44:54 +00:00
}
}
2015-03-02 13:48:47 +00:00
}
2015-02-15 13:44:54 +00:00
2015-05-02 13:31:31 +00:00
type archivePipe struct {
2015-03-07 10:53:32 +00:00
Old <-chan WalkTreeJob
New <-chan pipe.Job
}
func copyJobs(done <-chan struct{}, in <-chan pipe.Job, out chan<- pipe.Job) {
var (
2015-05-02 14:36:48 +00:00
// disable sending on the outCh until we received a job
outCh chan<- pipe.Job
// enable receiving from in
inCh = in
job pipe.Job
ok bool
2015-03-07 10:53:32 +00:00
)
2015-05-02 14:36:48 +00:00
2015-03-07 10:53:32 +00:00
for {
select {
case <-done:
return
2015-05-02 14:36:48 +00:00
case job, ok = <-inCh:
2015-03-07 10:53:32 +00:00
if !ok {
2015-05-02 14:36:48 +00:00
// input channel closed, we're done
debug.Log("copyJobs", "input channel closed, we're done")
2015-03-07 10:53:32 +00:00
return
}
2015-05-02 14:36:48 +00:00
inCh = nil
outCh = out
case outCh <- job:
outCh = nil
inCh = in
2015-03-07 10:53:32 +00:00
}
}
}
type archiveJob struct {
hasOld bool
old WalkTreeJob
new pipe.Job
}
2015-05-02 13:31:31 +00:00
func (a *archivePipe) compare(done <-chan struct{}, out chan<- pipe.Job) {
2015-03-07 10:53:32 +00:00
defer func() {
close(out)
debug.Log("ArchivePipe.compare", "done")
}()
debug.Log("ArchivePipe.compare", "start")
var (
loadOld, loadNew bool = true, true
ok bool
oldJob WalkTreeJob
newJob pipe.Job
)
for {
if loadOld {
oldJob, ok = <-a.Old
// if the old channel is closed, just pass through the new jobs
if !ok {
debug.Log("ArchivePipe.compare", "old channel is closed, copy from new channel")
// handle remaining newJob
if !loadNew {
out <- archiveJob{new: newJob}.Copy()
}
copyJobs(done, a.New, out)
return
}
loadOld = false
}
if loadNew {
newJob, ok = <-a.New
// if the new channel is closed, there are no more files in the current snapshot, return
if !ok {
debug.Log("ArchivePipe.compare", "new channel is closed, we're done")
return
}
loadNew = false
}
debug.Log("ArchivePipe.compare", "old job: %v", oldJob.Path)
debug.Log("ArchivePipe.compare", "new job: %v", newJob.Path())
// at this point we have received an old job as well as a new job, compare paths
file1 := oldJob.Path
file2 := newJob.Path()
dir1 := filepath.Dir(file1)
dir2 := filepath.Dir(file2)
if file1 == file2 {
debug.Log("ArchivePipe.compare", " same filename %q", file1)
// send job
out <- archiveJob{hasOld: true, old: oldJob, new: newJob}.Copy()
loadOld = true
loadNew = true
continue
} else if dir1 < dir2 {
debug.Log("ArchivePipe.compare", " %q < %q, file %q added", dir1, dir2, file2)
// file is new, send new job and load new
loadNew = true
out <- archiveJob{new: newJob}.Copy()
continue
2015-03-09 21:56:23 +00:00
} else if dir1 == dir2 {
if file1 < file2 {
debug.Log("ArchivePipe.compare", " %q < %q, file %q removed", file1, file2, file1)
// file has been removed, load new old
loadOld = true
continue
} else {
debug.Log("ArchivePipe.compare", " %q > %q, file %q added", file1, file2, file2)
// file is new, send new job and load new
loadNew = true
out <- archiveJob{new: newJob}.Copy()
continue
}
2015-03-07 10:53:32 +00:00
}
debug.Log("ArchivePipe.compare", " %q > %q, file %q removed", file1, file2, file1)
// file has been removed, throw away old job and load new
loadOld = true
}
}
func (j archiveJob) Copy() pipe.Job {
if !j.hasOld {
return j.new
}
// handle files
if isRegularFile(j.new.Info()) {
2015-03-07 10:53:32 +00:00
debug.Log("archiveJob.Copy", " job %v is file", j.new.Path())
// if type has changed, return new job directly
if j.old.Node == nil {
return j.new
}
// if file is newer, return the new job
if j.old.Node.isNewer(j.new.Fullpath(), j.new.Info()) {
debug.Log("archiveJob.Copy", " job %v is newer", j.new.Path())
return j.new
}
debug.Log("archiveJob.Copy", " job %v add old data", j.new.Path())
// otherwise annotate job with old data
e := j.new.(pipe.Entry)
e.Node = j.old.Node
return e
}
// dirs and other types are just returned
return j.new
}
2015-10-12 21:59:17 +00:00
const saveIndexTime = 30 * time.Second
// saveIndexes regularly queries the master index for full indexes and saves them.
func (arch *Archiver) saveIndexes(wg *sync.WaitGroup, done <-chan struct{}) {
defer wg.Done()
ticker := time.NewTicker(saveIndexTime)
defer ticker.Stop()
for {
select {
case <-done:
return
case <-ticker.C:
debug.Log("Archiver.saveIndexes", "saving full indexes")
err := arch.repo.SaveFullIndex()
if err != nil {
debug.Log("Archiver.saveIndexes", "save indexes returned an error: %v", err)
fmt.Fprintf(os.Stderr, "error saving preliminary index: %v\n", err)
}
}
}
}
2015-05-02 13:51:40 +00:00
// Snapshot creates a snapshot of the given paths. If parentID is set, this is
// used to compare the files to the ones archived at the time this snapshot was
// taken.
func (arch *Archiver) Snapshot(p *Progress, paths []string, parentID *backend.ID) (*Snapshot, backend.ID, error) {
2015-03-02 13:48:47 +00:00
debug.Log("Archiver.Snapshot", "start for %v", paths)
2015-02-15 13:44:54 +00:00
debug.RunHook("Archiver.Snapshot", nil)
2015-03-02 13:48:47 +00:00
sort.Strings(paths)
2015-02-15 13:44:54 +00:00
2015-03-07 10:53:32 +00:00
// signal the whole pipeline to stop
done := make(chan struct{})
var err error
2015-03-02 13:48:47 +00:00
p.Start()
defer p.Done()
2015-02-15 13:44:54 +00:00
2015-03-07 10:53:32 +00:00
// create new snapshot
2015-03-02 13:48:47 +00:00
sn, err := NewSnapshot(paths)
if err != nil {
return nil, backend.ID{}, err
2015-02-15 13:44:54 +00:00
}
sn.Excludes = arch.Excludes
2015-02-15 13:44:54 +00:00
2015-05-02 13:31:31 +00:00
jobs := archivePipe{}
2015-03-02 13:48:47 +00:00
2015-03-07 10:53:32 +00:00
// use parent snapshot (if some was given)
2015-05-02 13:51:40 +00:00
if parentID != nil {
sn.Parent = parentID
2015-03-07 10:53:32 +00:00
// load parent snapshot
parent, err := LoadSnapshot(arch.repo, *parentID)
2015-03-07 10:53:32 +00:00
if err != nil {
return nil, backend.ID{}, err
2015-03-07 10:53:32 +00:00
}
// start walker on old tree
ch := make(chan WalkTreeJob)
go WalkTree(arch.repo, *parent.Tree, done, ch)
2015-03-07 10:53:32 +00:00
jobs.Old = ch
} else {
// use closed channel
ch := make(chan WalkTreeJob)
close(ch)
jobs.Old = ch
}
// start walker
pipeCh := make(chan pipe.Job)
resCh := make(chan pipe.Result, 1)
go func() {
err := pipe.Walk(paths, arch.SelectFilter, done, pipeCh, resCh)
2015-03-07 10:53:32 +00:00
if err != nil {
debug.Log("Archiver.Snapshot", "pipe.Walk returned error %v", err)
return
}
debug.Log("Archiver.Snapshot", "pipe.Walk done")
}()
jobs.New = pipeCh
2015-03-02 13:48:47 +00:00
2015-03-07 10:53:32 +00:00
ch := make(chan pipe.Job)
go jobs.compare(done, ch)
2015-03-02 13:48:47 +00:00
2015-02-15 13:44:54 +00:00
var wg sync.WaitGroup
2015-03-02 13:48:47 +00:00
entCh := make(chan pipe.Entry)
dirCh := make(chan pipe.Dir)
// split
wg.Add(1)
go func() {
2015-03-07 10:53:32 +00:00
pipe.Split(ch, dirCh, entCh)
debug.Log("Archiver.Snapshot", "split done")
2015-03-02 13:48:47 +00:00
close(dirCh)
close(entCh)
wg.Done()
}()
// run workers
2015-02-15 13:44:54 +00:00
for i := 0; i < maxConcurrency; i++ {
wg.Add(2)
2015-03-02 13:48:47 +00:00
go arch.fileWorker(&wg, p, done, entCh)
go arch.dirWorker(&wg, p, done, dirCh)
2015-02-15 13:44:54 +00:00
}
2015-10-12 21:59:17 +00:00
// run index saver
var wgIndexSaver sync.WaitGroup
stopIndexSaver := make(chan struct{})
wgIndexSaver.Add(1)
go arch.saveIndexes(&wgIndexSaver, stopIndexSaver)
2015-02-15 13:44:54 +00:00
// wait for all workers to terminate
2015-03-02 13:48:47 +00:00
debug.Log("Archiver.Snapshot", "wait for workers")
2015-02-15 13:44:54 +00:00
wg.Wait()
2015-10-12 21:59:17 +00:00
// stop index saver
close(stopIndexSaver)
wgIndexSaver.Wait()
2015-03-02 13:48:47 +00:00
debug.Log("Archiver.Snapshot", "workers terminated")
2015-02-15 13:44:54 +00:00
// receive the top-level tree
root := (<-resCh).(*Node)
debug.Log("Archiver.Snapshot", "root node received: %v", root.Subtree.Str())
sn.Tree = root.Subtree
2014-11-23 21:26:01 +00:00
2014-09-23 20:39:12 +00:00
// save snapshot
2015-05-09 11:32:52 +00:00
id, err := arch.repo.SaveJSONUnpacked(backend.Snapshot, sn)
2014-09-23 20:39:12 +00:00
if err != nil {
return nil, backend.ID{}, err
2014-09-23 20:39:12 +00:00
}
2015-03-09 21:56:44 +00:00
// store ID in snapshot struct
sn.id = &id
debug.Log("Archiver.Snapshot", "saved snapshot %v", id.Str())
2015-03-09 21:56:44 +00:00
2015-05-09 11:35:55 +00:00
// flush repository
2015-05-09 11:32:52 +00:00
err = arch.repo.Flush()
if err != nil {
return nil, backend.ID{}, err
}
2015-03-09 21:26:39 +00:00
// save index
2015-10-12 20:34:12 +00:00
err = arch.repo.SaveIndex()
2015-03-09 21:26:39 +00:00
if err != nil {
debug.Log("Archiver.Snapshot", "error saving index: %v", err)
return nil, backend.ID{}, err
2015-03-09 21:26:39 +00:00
}
2015-10-12 20:34:12 +00:00
debug.Log("Archiver.Snapshot", "saved indexes")
return sn, id, nil
2014-09-23 20:39:12 +00:00
}
2015-02-21 13:23:49 +00:00
func isRegularFile(fi os.FileInfo) bool {
2015-03-07 10:53:32 +00:00
if fi == nil {
return false
}
2015-02-21 13:23:49 +00:00
return fi.Mode()&(os.ModeType|os.ModeCharDevice) == 0
}
2015-05-02 13:51:40 +00:00
// Scan traverses the dirs to collect Stat information while emitting progress
// information with p.
func Scan(dirs []string, filter pipe.SelectFunc, p *Progress) (Stat, error) {
2015-02-21 13:23:49 +00:00
p.Start()
defer p.Done()
var stat Stat
2015-03-02 13:48:47 +00:00
for _, dir := range dirs {
2015-03-15 11:20:30 +00:00
debug.Log("Scan", "Start for %v", dir)
2015-03-02 13:48:47 +00:00
err := filepath.Walk(dir, func(str string, fi os.FileInfo, err error) error {
2015-03-21 13:43:33 +00:00
// TODO: integrate error reporting
if err != nil {
fmt.Fprintf(os.Stderr, "error for %v: %v\n", str, err)
return nil
}
if fi == nil {
fmt.Fprintf(os.Stderr, "error for %v: FileInfo is nil\n", str)
return nil
}
if !filter(str, fi) {
debug.Log("Scan.Walk", "path %v excluded", str)
if fi.IsDir() {
return filepath.SkipDir
}
return nil
}
2015-03-02 13:48:47 +00:00
s := Stat{}
if fi.IsDir() {
2015-03-02 13:48:47 +00:00
s.Dirs++
} else {
s.Files++
if isRegularFile(fi) {
s.Bytes += uint64(fi.Size())
}
2015-03-02 13:48:47 +00:00
}
2015-02-21 13:23:49 +00:00
2015-03-02 13:48:47 +00:00
p.Report(s)
stat.Add(s)
2015-02-21 13:23:49 +00:00
2015-03-02 13:48:47 +00:00
// TODO: handle error?
return nil
})
2015-03-15 11:20:30 +00:00
debug.Log("Scan", "Done for %v, err: %v", dir, err)
2015-03-02 13:48:47 +00:00
if err != nil {
return Stat{}, err
}
}
2015-02-21 13:23:49 +00:00
2015-03-02 13:48:47 +00:00
return stat, nil
2015-02-21 13:23:49 +00:00
}