Support for stdin streams (kopia#862)

* Add StreamingFile interface
* unit test for virtualfs
* CLI: Snapshot create support for stdin sources
* Uploader support for fs.StreamingFile
* End to end test for stdin source snapshot
* upload test to improve coverage

Author: pavannd1

cli/command_snapshot_create.go

@@ -2,12 +2,15 @@ package cli
 
 import (
 	"context"
+	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/pkg/errors"
 
+	"github.com/kopia/kopia/fs"
+	"github.com/kopia/kopia/fs/virtualfs"
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/snapshot"
 	"github.com/kopia/kopia/snapshot/policy"
@@ -22,7 +25,7 @@ const (
 var (
 	snapshotCreateCommand = snapshotCommands.Command("create", "Creates a snapshot of local directory or file.").Default()
 
-	snapshotCreateSources                 = snapshotCreateCommand.Arg("source", "Files or directories to create snapshot(s) of.").ExistingFilesOrDirs()
+	snapshotCreateSources                 = snapshotCreateCommand.Arg("source", "Files or directories to create snapshot(s) of.").Strings()
 	snapshotCreateAll                     = snapshotCreateCommand.Flag("all", "Create snapshots for files or directories previously backed up by this user on this computer").Bool()
 	snapshotCreateCheckpointUploadLimitMB = snapshotCreateCommand.Flag("upload-limit-mb", "Stop the backup process after the specified amount of data (in MB) has been uploaded.").PlaceHolder("MB").Default("0").Int64()
 	snapshotCreateCheckpointInterval      = snapshotCreateCommand.Flag("checkpoint-interval", "Frequency for creating periodic checkpoint.").Duration()
@@ -34,6 +37,7 @@ var (
 	snapshotCreateEndTime                 = snapshotCreateCommand.Flag("end-time", "Override snapshot end timestamp.").String()
 	snapshotCreateForceEnableActions      = snapshotCreateCommand.Flag("force-enable-actions", "Enable snapshot actions even if globally disabled on this client").Hidden().Bool()
 	snapshotCreateForceDisableActions     = snapshotCreateCommand.Flag("force-disable-actions", "Disable snapshot actions even if globally enabled on this client").Hidden().Bool()
+	snapshotCreateStdinFileName           = snapshotCreateCommand.Flag("stdin-file", "File path to be used for stdin data snapshot.").String()
 )
 
 func runSnapshotCommand(ctx context.Context, rep repo.RepositoryWriter) error {
@@ -163,9 +167,24 @@ func startTimeAfterEndTime(startTime, endTime time.Time) bool {
 func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *snapshotfs.Uploader, sourceInfo snapshot.SourceInfo) error {
 	log(ctx).Infof("Snapshotting %v ...", sourceInfo)
 
-	localEntry, err := getLocalFSEntry(ctx, sourceInfo.Path)
-	if err != nil {
-		return errors.Wrap(err, "unable to get local filesystem entry")
+	var (
+		err       error
+		fsEntry   fs.Entry
+		setManual bool
+	)
+
+	if *snapshotCreateStdinFileName != "" {
+		// stdin source will be snapshotted using a virtual static root directory with a single streaming file entry
+		// Create a new static directory with the given name and add a streaming file entry with os.Stdin reader
+		fsEntry = virtualfs.NewStaticDirectory(sourceInfo.Path, fs.Entries{
+			virtualfs.StreamingFileFromReader(*snapshotCreateStdinFileName, os.Stdin),
+		})
+		setManual = true
+	} else {
+		fsEntry, err = getLocalFSEntry(ctx, sourceInfo.Path)
+		if err != nil {
+			return errors.Wrap(err, "unable to get local filesystem entry")
+		}
 	}
 
 	previous, err := findPreviousSnapshotManifest(ctx, rep, sourceInfo, nil)
@@ -180,7 +199,7 @@ func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *sna
 
 	log(ctx).Debugf("uploading %v using %v previous manifests", sourceInfo, len(previous))
 
-	manifest, err := u.Upload(ctx, localEntry, policyTree, sourceInfo, previous...)
+	manifest, err := u.Upload(ctx, fsEntry, policyTree, sourceInfo, previous...)
 	if err != nil {
 		// fail-fast uploads will fail here without recording a manifest, other uploads will
 		// possibly fail later.
@@ -218,6 +237,12 @@ func snapshotSingleSource(ctx context.Context, rep repo.RepositoryWriter, u *sna
 		return errors.Wrap(err, "unable to apply retention policy")
 	}
 
+	if setManual {
+		if err = policy.SetManual(ctx, rep, sourceInfo); err != nil {
+			return errors.Wrap(err, "unable to set manual field in scheduling policy for source")
+		}
+	}
+
 	if ferr := rep.Flush(ctx); ferr != nil {
 		return errors.Wrap(ferr, "flush error")
 	}

fs/entry.go

@@ -50,6 +50,12 @@ type File interface {
 	Open(ctx context.Context) (Reader, error)
 }
 
+// StreamingFile represents an entry that is a stream.
+type StreamingFile interface {
+	Entry
+	GetReader(ctx context.Context) (io.Reader, error)
+}
+
 // Directory represents contents of a directory.
 type Directory interface {
 	Entry

fs/virtualfs/virtualfs.go

@@ -0,0 +1,129 @@
+// Package virtualfs implements an in-memory abstraction of fs.Directory and fs.StreamingFile.
+package virtualfs
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"time"
+
+	"github.com/kopia/kopia/fs"
+)
+
+const (
+	defaultPermissions os.FileMode = 0777
+)
+
+// virtualEntry is an in-memory implementation of a directory entry.
+type virtualEntry struct {
+	name    string
+	mode    os.FileMode
+	size    int64
+	modTime time.Time
+	owner   fs.OwnerInfo
+	device  fs.DeviceInfo
+}
+
+func (e *virtualEntry) Name() string {
+	return e.name
+}
+
+func (e *virtualEntry) IsDir() bool {
+	return e.mode.IsDir()
+}
+
+func (e *virtualEntry) Mode() os.FileMode {
+	return e.mode
+}
+
+func (e *virtualEntry) ModTime() time.Time {
+	return e.modTime
+}
+
+func (e *virtualEntry) Size() int64 {
+	return e.size
+}
+
+func (e *virtualEntry) Sys() interface{} {
+	return nil
+}
+
+func (e *virtualEntry) Owner() fs.OwnerInfo {
+	return e.owner
+}
+
+func (e *virtualEntry) Device() fs.DeviceInfo {
+	return e.device
+}
+
+func (e *virtualEntry) LocalFilesystemPath() string {
+	return ""
+}
+
+// staticDirectory is an in-memory implementation of fs.Directory.
+type staticDirectory struct {
+	virtualEntry
+	entries fs.Entries
+}
+
+// Child gets the named child of a directory.
+func (sd *staticDirectory) Child(ctx context.Context, name string) (fs.Entry, error) {
+	return fs.ReadDirAndFindChild(ctx, sd, name)
+}
+
+// Readdir gets the contents of a directory.
+func (sd *staticDirectory) Readdir(ctx context.Context) (fs.Entries, error) {
+	return append(fs.Entries(nil), sd.entries...), nil
+}
+
+// NewStaticDirectory returns a virtual static directory.
+func NewStaticDirectory(name string, entries fs.Entries) fs.Directory {
+	return &staticDirectory{
+		virtualEntry: virtualEntry{
+			name: name,
+			mode: defaultPermissions | os.ModeDir,
+		},
+		entries: entries,
+	}
+}
+
+// virtualFile is an implementation of fs.StreamingFile with an io.Reader.
+type virtualFile struct {
+	virtualEntry
+	reader io.Reader
+}
+
+var errReaderAlreadyUsed = errors.New("cannot use streaming file reader more than once")
+
+// GetReader returns the streaming file's reader.
+// Note: Caller of this function has to ensure concurrency safety.
+// The file's reader is set to nil after the first call.
+func (vf *virtualFile) GetReader(ctx context.Context) (io.Reader, error) {
+	if vf.reader == nil {
+		return nil, errReaderAlreadyUsed
+	}
+
+	// reader must be fetched only once
+	ret := vf.reader
+	vf.reader = nil
+
+	return ret, nil
+}
+
+// StreamingFileFromReader returns a streaming file with given name and reader.
+func StreamingFileFromReader(name string, reader io.Reader) fs.StreamingFile {
+	return &virtualFile{
+		virtualEntry: virtualEntry{
+			name: name,
+			mode: defaultPermissions,
+		},
+		reader: reader,
+	}
+}
+
+var (
+	_ fs.Directory     = &staticDirectory{}
+	_ fs.StreamingFile = &virtualFile{}
+	_ fs.Entry         = &virtualEntry{}
+)

fs/virtualfs/virtualfs_test.go

@@ -0,0 +1,111 @@
+package virtualfs
+
+import (
+	"context"
+	"errors"
+	"os"
+	"reflect"
+	"testing"
+
+	"github.com/kopia/kopia/fs"
+)
+
+func TestStreamingFile(t *testing.T) {
+	// Create a temporary file with test data
+	content := []byte("Temporary file content")
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("error creating pipe file: %v", err)
+	}
+
+	if _, err = w.Write(content); err != nil {
+		t.Fatalf("error writing to pipe file: %v", err)
+	}
+
+	w.Close()
+
+	filename := "stream-file"
+	f := StreamingFileFromReader(filename, r)
+
+	rootDir := NewStaticDirectory("root", fs.Entries{f})
+
+	e, err := rootDir.Child(context.TODO(), filename)
+	if err != nil {
+		t.Fatalf("error getting child entry: %v", err)
+	}
+
+	if e.Name() != filename {
+		t.Fatalf("did not get expected filename: (actual) %v != %v (expected)", e.Name(), filename)
+	}
+
+	entries, err := rootDir.Readdir(context.TODO())
+	if err != nil {
+		t.Fatalf("error getting dir entries %v", err)
+	}
+
+	if len(entries) == 0 {
+		t.Errorf("expected directory with 1 entry, got %v", rootDir)
+	}
+
+	// Read and compare data
+	reader, err := f.GetReader(context.TODO())
+	if err != nil {
+		t.Fatalf("error getting streaming file reader: %v", err)
+	}
+
+	result := make([]byte, len(content))
+
+	if _, err = reader.Read(result); err != nil {
+		t.Fatalf("error reading streaming file: %v", err)
+	}
+
+	if !reflect.DeepEqual(result, content) {
+		t.Fatalf("did not get expected file content: (actual) %v != %v (expected)", result, content)
+	}
+}
+
+func TestStreamingFileGetReader(t *testing.T) {
+	// Create a temporary file with test data
+	content := []byte("Temporary file content")
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("error creating pipe file: %v", err)
+	}
+
+	if _, err = w.Write(content); err != nil {
+		t.Fatalf("error writing to pipe file: %v", err)
+	}
+
+	w.Close()
+
+	filename := "stream-file"
+	f := StreamingFileFromReader(filename, r)
+
+	// Read and compare data
+	reader, err := f.GetReader(context.TODO())
+	if err != nil {
+		t.Fatalf("error getting streaming file reader: %v", err)
+	}
+
+	result := make([]byte, len(content))
+
+	if _, err = reader.Read(result); err != nil {
+		t.Fatalf("error reading streaming file: %v", err)
+	}
+
+	if !reflect.DeepEqual(result, content) {
+		t.Fatalf("did not get expected file content: (actual) %v != %v (expected)", result, content)
+	}
+
+	// Second call to GetReader must fail
+	_, err = f.GetReader(context.TODO())
+	if err == nil {
+		t.Fatal("expected error, got none")
+	}
+
+	if !errors.Is(err, errReaderAlreadyUsed) {
+		t.Fatalf("did not get expected error: (actual) %v != %v (expected)", err, errReaderAlreadyUsed)
+	}
+}