Stream optional interfaces

README.md

@@ -1,4 +1,3 @@
-
 # Simpleblob
 
 [![Go Reference](https://pkg.go.dev/badge/github.com/PowerDNS/simpleblob.svg)](https://pkg.go.dev/github.com/PowerDNS/simpleblob)
@@ -10,6 +9,9 @@ Simpleblob is a Go module that simplifies the storage of arbitrary data by key f
 - `fs`: File storage (one file per blob)
 - `memory`: Memory storage (for tests)
 
+
+## Usage
+
 The interface implemented by the backends is:
 
 ```go
@@ -38,16 +40,33 @@ type Storage struct {
 }
 ```
 
+
+### `io` interfaces
+
+Reading from or writing to a blob directly can be done using the `NewReader` and `NewWriter` functions.
+
+```go
+func NewReader(ctx context.Context, storage Interface, blobName string) (io.ReadCloser, error)
+func NewWriter(ctx context.Context, storage Interface, blobName string) (io.WriteCloser, error)
+```
+
+The returned ReadCloser or WriteCloser is an optimized implementation if the backend being used implements the `StreamReader` or `StreamWriter` interfaces.
+If not, a convenience wrapper for the storage is returned.
+
+| Backend | StreamReader | StreamWriter |
+| --- | --- | --- |
+| S3 | ✔ | ✔ |
+| Filesystem | ✔ | ✔ |
+| Memory | ✖ | ✖ |
+
+
 ## Limitations
 
 The interface currently does not support streaming of large blobs. In the future we may provide this by implementing `fs.FS` in the backend for reading, and a similar interface for writing new blobs.
 
+
 ## API Stability
 
 We support the last two stable Go versions, currently 1.17 and 1.18.
 
 From a API consumer point of view, we do not plan any backward incompatible changes before a v1.0.
-
-For storage backends, any future extensions most likely be added with optional interface, similar to the `fs.FS` design. Utility functions that return a compatible implementation will be used for backends that do not implement the interface, when possible.
-
-

backends/fs/atomic.go

@@ -0,0 +1,90 @@
+package fs
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+)
+
+// createAtomic creates a new File. The given fpath is the file path of the final destination.
+func createAtomic(fpath string) (*atomicFile, error) {
+	fpath, err := filepath.Abs(fpath)
+	if err != nil {
+		return nil, fmt.Errorf("absolute path for atomic file %q: %w", fpath, err)
+	}
+	// Using the PID under the assumption that the same program will not be writing to
+	// the same path at the same time. An overwrite later on retry is desired, if
+	// not cleaned properly.
+	tmp := fmt.Sprintf("%s.%d%s", fpath, os.Getpid(), ignoreSuffix)
+	file, err := os.Create(tmp)
+	if err != nil {
+		return nil, fmt.Errorf("create atomic file %q: %w", fpath, err)
+	}
+	return &atomicFile{
+		file: file,
+		path: fpath,
+		tmp:  tmp,
+	}, nil
+}
+
+// atomicFile implements an io.WriteCloser that writes to a temp file and moves it
+// atomically into place on Close.
+type atomicFile struct {
+	file *os.File // The underlying file being written to.
+	path string   // The final path of the file.
+	tmp  string   // The path of the file during write.
+}
+
+// Write implements io.Writer
+func (f *atomicFile) Write(data []byte) (int, error) {
+	return f.file.Write(data)
+}
+
+// Clean aborts the creation of the file if called before Close. If called
+// after Close, it does nothing. This makes it useful in a defer.
+func (f *atomicFile) Clean() {
+	_ = f.file.Close()
+	_ = os.Remove(f.tmp)
+}
+
+// Close closes the temp file and moves it to the final destination.
+func (f *atomicFile) Close() error {
+	var err error
+	defer func() {
+		if err != nil {
+			// The rename did not happen and we're left with
+			// the temporary file hanging.
+			_ = os.Remove(f.tmp)
+		}
+	}()
+
+	// Some of the file content may still be cached by the OS,
+	// and is not guaranteed to be written to physical device on close.
+	// Behaviour is inconsistent across devices and C standard libraries.
+	// Syncing file AND its parent directory (here) ensure this.
+	// See fsync(2) and open(2).
+	if err = f.file.Sync(); err != nil {
+		_ = f.file.Close()
+		return err
+	}
+	if err = f.file.Close(); err != nil {
+		return err
+	}
+
+	// Move into place
+	if err = os.Rename(f.tmp, f.path); err != nil {
+		return err
+	}
+
+	var dir *os.File
+	dir, err = os.Open(filepath.Dir(f.path))
+	if err != nil {
+		return err
+	}
+	err = dir.Sync()
+	if err != nil {
+		_ = dir.Close()
+		return err
+	}
+	return dir.Close()
+}

backends/fs/fs.go

@@ -11,6 +11,10 @@ import (
 	"github.com/PowerDNS/simpleblob"
 )
 
+// ignoreSuffix is the suffix to use internally
+// to hide a file from (*Backend).List.
+const ignoreSuffix = ".tmp"
+
 // Options describes the storage options for the fs backend
 type Options struct {
 	RootPath string `yaml:"root_path"`
@@ -71,7 +75,7 @@ func (b *Backend) Store(ctx context.Context, name string, data []byte) error {
 		return os.ErrPermission
 	}
 	fullPath := filepath.Join(b.rootPath, name)
-	tmpPath := fullPath + ".tmp" // ignored by List()
+	tmpPath := fullPath + ignoreSuffix // ignored by List()
 	if err := writeFile(tmpPath, data); err != nil {
 		return err
 	}
@@ -100,7 +104,7 @@ func allowedName(name string) bool {
 	if strings.HasPrefix(name, ".") {
 		return false
 	}
-	if strings.HasSuffix(name, ".tmp") {
+	if strings.HasSuffix(name, ignoreSuffix) {
 		return false // used for our temp files when writing
 	}
 	return true

backends/fs/stream.go

@@ -0,0 +1,32 @@
+package fs
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+)
+
+// NewReader provides an optimized way to read from named file.
+func (b *Backend) NewReader(ctx context.Context, name string) (io.ReadCloser, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if !allowedName(name) {
+		return nil, os.ErrPermission
+	}
+	fullPath := filepath.Join(b.rootPath, name)
+	return os.Open(fullPath)
+}
+
+// NewWriter provides an optimized way to write to a file.
+func (b *Backend) NewWriter(ctx context.Context, name string) (io.WriteCloser, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	if !allowedName(name) {
+		return nil, os.ErrPermission
+	}
+	fullPath := filepath.Join(b.rootPath, name)
+	return createAtomic(fullPath)
+}

backends/s3/marker.go

@@ -18,6 +18,7 @@ func (b *Backend) setMarker(ctx context.Context, name, etag string, isDel bool)
 	}
 	nanos := time.Now().UnixNano()
 	s := fmt.Sprintf("%s:%s:%d:%v", name, etag, nanos, isDel)
+	// Here, we're not using Store because markerName already has the global prefix.
 	_, err := b.doStore(ctx, b.markerName, []byte(s))
 	if err != nil {
 		return err

backends/s3/s3.go

@@ -87,6 +87,10 @@ type Options struct {
 	// DisableContentMd5 defines whether to disable sending the Content-MD5 header
 	DisableContentMd5 bool `yaml:"disable_send_content_md5"`
 
+	// NumMinioThreads defines the number of threads that Minio uses for its workers.
+	// It defaults to the using the default value defined by the Minio client.
+	NumMinioThreads uint `yaml:"num_minio_threads"`
+
 	// TLS allows customising the TLS configuration
 	// See https://github.com/PowerDNS/go-tlsconfig for the available options
 	TLS tlsconfig.Config `yaml:"tls"`
@@ -152,6 +156,8 @@ func (b *Backend) List(ctx context.Context, prefix string) (blobList simpleblob.
 		return b.doList(ctx, combinedPrefix)
 	}
 
+	// Using Load, that will itself prepend the global prefix to the marker name.
+	// So we're using the raw marker name here.
 	m, err := b.Load(ctx, UpdateMarkerFilename)
 	exists := !errors.Is(err, os.ErrNotExist)
 	if err != nil && exists {
@@ -229,25 +235,44 @@ func (b *Backend) doList(ctx context.Context, prefix string) (simpleblob.BlobLis
 // Load retrieves the content of the object identified by name from S3 Bucket
 // configured in b.
 func (b *Backend) Load(ctx context.Context, name string) ([]byte, error) {
-	// Prepend global prefix
 	name = b.prependGlobalPrefix(name)
 
+	r, err := b.doLoadReader(ctx, name)
+	if err != nil {
+		return nil, err
+	}
+	defer r.Close()
+
+	p, err := io.ReadAll(r)
+	if err = convertMinioError(err, false); err != nil {
+		return nil, err
+	}
+	return p, nil
+}
+
+func (b *Backend) doLoadReader(ctx context.Context, name string) (io.ReadCloser, error) {
 	metricCalls.WithLabelValues("load").Inc()
 	metricLastCallTimestamp.WithLabelValues("load").SetToCurrentTime()
 
 	obj, err := b.client.GetObject(ctx, b.opt.Bucket, name, minio.GetObjectOptions{})
 	if err = convertMinioError(err, false); err != nil {
+		metricCallErrors.WithLabelValues("load").Inc()
 		return nil, err
-	} else if obj == nil {
+	}
+	if obj == nil {
 		return nil, os.ErrNotExist
 	}
-	defer obj.Close()
-
-	p, err := io.ReadAll(obj)
+	info, err := obj.Stat()
 	if err = convertMinioError(err, false); err != nil {
+		metricCallErrors.WithLabelValues("load").Inc()
 		return nil, err
 	}
-	return p, nil
+	if info.Key == "" {
+		// minio will return an object with empty fields when name
+		// is not present in bucket.
+		return nil, os.ErrNotExist
+	}
+	return obj, nil
 }
 
 // Store sets the content of the object identified by name to the content
@@ -263,18 +288,25 @@ func (b *Backend) Store(ctx context.Context, name string, data []byte) error {
 	return b.setMarker(ctx, name, info.ETag, false)
 }
 
+// doStore is a convenience wrapper around doStoreReader.
 func (b *Backend) doStore(ctx context.Context, name string, data []byte) (minio.UploadInfo, error) {
+	return b.doStoreReader(ctx, name, bytes.NewReader(data), int64(len(data)))
+}
+
+// doStoreReader stores data with key name in S3, using r as a source for data.
+// The value of size may be -1, in case the size is not known.
+func (b *Backend) doStoreReader(ctx context.Context, name string, r io.Reader, size int64) (minio.UploadInfo, error) {
 	metricCalls.WithLabelValues("store").Inc()
 	metricLastCallTimestamp.WithLabelValues("store").SetToCurrentTime()
 
 	putObjectOptions := minio.PutObjectOptions{
-		NumThreads: 3,
-	}
-	if !b.opt.DisableContentMd5 {
-		putObjectOptions.SendContentMd5 = true
+		NumThreads:     b.opt.NumMinioThreads,
+		SendContentMd5: !b.opt.DisableContentMd5,
 	}
 
-	info, err := b.client.PutObject(ctx, b.opt.Bucket, name, bytes.NewReader(data), int64(len(data)), putObjectOptions)
+	// minio accepts size == -1, meaning the size is unknown.
+	info, err := b.client.PutObject(ctx, b.opt.Bucket, name, r, size, putObjectOptions)
+	err = convertMinioError(err, false)
 	if err != nil {
 		metricCallErrors.WithLabelValues("store").Inc()
 	}
@@ -371,9 +403,9 @@ func New(ctx context.Context, opt Options) (*Backend, error) {
 	case "https":
 		useSSL = true
 	case "":
-		return nil, fmt.Errorf("no scheme provided for endpoint URL '%s', use http or https.", opt.EndpointURL)
+		return nil, fmt.Errorf("no scheme provided for endpoint URL %q, use http or https", opt.EndpointURL)
 	default:
-		return nil, fmt.Errorf("unsupported scheme for S3: '%s', use http or https.", u.Scheme)
+		return nil, fmt.Errorf("unsupported scheme for S3: %q, use http or https", u.Scheme)
 	}
 
 	creds := credentials.NewStaticV4(opt.AccessKey, opt.SecretKey, "")