Add barebones caching nbd daemon

nbd/.gitignore

@@ -0,0 +1 @@
+cache

nbd/Makefile

@@ -5,3 +5,13 @@ GOTEST_TIMEOUT=5m
 .PHONY: tcpflow
 tcpflow:
 	$@ -i lo -cDg -X /dev/null host 127.0.0.189
+
+.PHONY: s3
+s3: ## Development S3 endpoint. No security!
+	docker run --rm \
+		-p "127.0.0.55:55555:9000" \
+		-e MINIO_ROOT_USER=access \
+		-e MINIO_ROOT_PASSWORD=secret123 \
+		-v "$$PWD/verity/testdata:/data/testdata:ro" \
+		-it "quay.io/minio/minio:RELEASE.2022-05-26T05-48-41Z" \
+		gateway nas /data

nbd/cmd/nbd/main.go

@@ -1,26 +1,38 @@
 package main
 
 import (
-	"context"
-	"log"
+	"encoding/json"
+	"fmt"
 	"os"
 
-	"github.com/sio/pond/nbd/server"
+	"github.com/sio/pond/nbd/daemon"
 )
 
+const config = `
+{
+	"s3": {
+		"endpoint": "http://127.0.0.55:55555",
+		"bucket": "testdata",
+		"access": "access",
+		"secret": "secret123"
+	},
+	"cache": {"dir": "./cache"},
+	"listen": [
+		{"network": "tcp", "address": "127.0.0.189:10809"}
+	]
+}
+`
+
 func main() {
-	exe, err := os.Executable()
+	var nbd daemon.Daemon
+	err := json.Unmarshal([]byte(config), &nbd)
 	if err != nil {
-		log.Fatal(err)
+		panic("default config: " + err.Error())
 	}
-	log.Println(exe)
-	s := server.New(context.Background(), func(name string) (server.Backend, error) {
-		log.Printf("exportFunc: client requested export name: %q\n", name)
-		return os.Open("Makefile")
-	})
-	go s.ListenShutdown()
-	e := s.Listen("tcp", "127.0.0.189:10809")
-	if e != nil {
-		log.Fatalf("Exit: %v", e)
+
+	err = nbd.Run()
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
 	}
 }

nbd/daemon/daemon.go

@@ -0,0 +1,99 @@
+package daemon
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path/filepath"
+	"sync"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/sio/pond/nbd/logger"
+	"github.com/sio/pond/nbd/s3"
+	"github.com/sio/pond/nbd/server"
+)
+
+type Daemon struct {
+	S3 struct {
+		Endpoint string
+		Bucket   string
+		Prefix   string
+		Access   string
+		Secret   string
+	}
+	Cache struct {
+		Dir string
+	}
+	Listen []struct {
+		Network string
+		Address string
+	}
+}
+
+func (d *Daemon) Run() error {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("shutting down the daemon"))
+	log := logger.FromContext(ctx)
+
+	// Exclusive lock on local cache directory
+	abs, err := filepath.Abs(d.Cache.Dir)
+	if err == nil {
+		d.Cache.Dir = abs
+	}
+	lock, err := Lock(filepath.Join(d.Cache.Dir, "lock"))
+	if err != nil {
+		return fmt.Errorf("acquire cache directory lock: %w", err)
+	}
+	defer func() {
+		err := lock.Close()
+		if err != nil {
+			log.Error("failed to release the lock file", "lock", lock, "error", err)
+		}
+	}()
+
+	// Cache object memoization
+	var (
+		volume   = make(map[string]server.Backend)
+		volumeMu sync.Mutex
+	)
+	export := func(name string) (server.Backend, error) {
+		volumeMu.Lock()
+		defer volumeMu.Unlock()
+
+		cache, found := volume[name]
+		if found {
+			return cache, nil
+		}
+		cache, err := s3.Open(
+			d.S3.Endpoint,
+			d.S3.Access,
+			d.S3.Secret,
+			d.S3.Bucket,
+			filepath.Join(d.S3.Prefix, name),
+			d.Cache.Dir,
+		)
+		if err != nil {
+			return nil, err
+		}
+		// TODO: clean up old cache artifacts when running low on disk space
+		volume[name] = cache
+		return cache, nil
+	}
+
+	// Launch NBD server
+	nbd := server.New(ctx, export)
+	go nbd.ListenShutdown()
+	var group errgroup.Group
+	for _, listener := range d.Listen {
+		listener := listener
+		group.Go(func() error {
+			err := nbd.Listen(listener.Network, listener.Address)
+			if err != nil {
+				log.Error("nbd listener failed", "listener", fmt.Sprintf("%s://%s", listener.Network, listener.Address), "error", err)
+			}
+			return err
+		})
+	}
+	return group.Wait()
+}

nbd/daemon/lockfile.go

@@ -0,0 +1,51 @@
+package daemon
+
+import (
+	"fmt"
+	"os"
+	"syscall"
+)
+
+type Lockfile struct {
+	f *os.File
+}
+
+func Lock(path string) (*Lockfile, error) {
+	file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		return nil, err
+	}
+	err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+	if err != nil {
+		_ = file.Close()
+		return nil, err
+	}
+	return &Lockfile{f: file}, nil
+}
+
+func (l *Lockfile) Close() error {
+	err := syscall.Flock(int(l.f.Fd()), syscall.LOCK_UN)
+	if err != nil {
+		return err
+	}
+	err = l.f.Close()
+	if err != nil {
+		return err
+	}
+	err = os.Remove(l.f.Name())
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (l *Lockfile) String() string {
+	var name string
+	if l.f != nil {
+		name = l.f.Name()
+	}
+	if len(name) == 0 {
+		name = "<nil>"
+	}
+	return fmt.Sprintf("<Lockfile: %s>", name)
+}

nbd/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/minio/minio-go/v7 v7.0.69
 	github.com/testcontainers/testcontainers-go v0.30.0
 	golang.org/x/crypto v0.19.0
+	golang.org/x/sync v0.3.0
 )
 
 require (