diff --git a/Makefile b/Makefile index 4a2fbcc..a719205 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ PROTO_FILES := $(wildcard proto/sisyphus/v1/*.proto) $(wildcard proto/sisyphus/node/v1/*.proto) -COVERED := ./apps/...,./packages/ai/...,./packages/geo/...,./packages/hardware/...,./packages/identity/...,./packages/job-model/...,./packages/kubo/...,./packages/nodedb/...,./packages/runtime/...,./packages/sealed/...,./packages/storage/... +COVERED := ./apps/...,./packages/ai/...,./packages/geo/...,./packages/hardware/...,./packages/identity/...,./packages/job-model/...,./packages/kubo/...,./packages/nodedb/...,./packages/runtime/...,./packages/s3/...,./packages/sealed/...,./packages/storage/... .PHONY: build test cover demo proto tools fmt diff --git a/apps/sisyphusd/README.md b/apps/sisyphusd/README.md index 8f95da1..6ae00ad 100644 --- a/apps/sisyphusd/README.md +++ b/apps/sisyphusd/README.md @@ -236,6 +236,22 @@ A coordinator also collects every `--gc-interval` (default one hour), and with ` A job's result must be stored by the workload's aggregation step to be kept; a blob a task stored is treated as intermediate unless the aggregation stores it again. +### In a bucket + +A coordinator can keep the pool's stored data in a bucket of an object store that speaks S3, instead of on its own disk: Ceph, MinIO, SeaweedFS, Storj, Wasabi, AWS. + +```sh +printf '%s\n%s\n' "$ACCESS_KEY" "$SECRET_KEY" > s3.keys && chmod 600 s3.keys +bin/sisyphusd run --s3-endpoint https://s3.eu-central-1.wasabisys.com --s3-bucket my-pool --s3-credentials s3.keys +``` + +- **The bucket must exist**; the node checks that it can reach it and stops if not. `--s3-prefix` puts the pool's objects under a name of their own, to share a bucket; `--s3-region` is for stores that have regions. +- **Nothing else changes.** A file has the same CID wherever its bytes are, so jobs, workers and clients need know nothing of it. Workers still fetch from the coordinator and each other; only the coordinator talks to the bucket, and the credentials never leave it. +- **Each block is an object**, of up to 256 KiB, under `blocks/`. What is kept and for how long is still the node's to decide: pins are in its data directory, and `blob gc` deletes from the bucket what nothing is keeping. +- **It is slower than a disk** when the bucket is far away: reading a gigabyte is four thousand requests, one after another. Nothing is cached on the coordinator. Use it for durability and room, and a bucket near the coordinator. +- It cannot be combined with `--kubo`, which is another answer to where the data is kept. +- Tried against SeaweedFS's S3 gateway. The others named speak the same dialect and have not been tried. + ## What each machine has A worker looks at the machine it runs on and tells its coordinator what it finds: the processor, how much memory, and any NVIDIA graphics cards. `sisyphusd nodes` shows it. diff --git a/apps/sisyphusd/cli_test.go b/apps/sisyphusd/cli_test.go index ac8ce40..4c4d445 100644 --- a/apps/sisyphusd/cli_test.go +++ b/apps/sisyphusd/cli_test.go @@ -29,10 +29,17 @@ func TestMain(m *testing.M) { // Nothing a test does may land in the real user's data directory. os.Unsetenv("XDG_DATA_HOME") code := m.Run() + for _, stop := range stopAtExit { + stop() + } os.RemoveAll(home) os.Exit(code) } +// stopAtExit is what to stop when the tests are over: whatever a test +// started that outlives it, such as a container shared between tests. +var stopAtExit []func() + // dataDirs records, for each address a test node listens on, the data // directory holding that node's key. Commands sent to the address are run // from that directory, which makes them the node's owner. diff --git a/apps/sisyphusd/daemon.go b/apps/sisyphusd/daemon.go index 503477b..06981ab 100644 --- a/apps/sisyphusd/daemon.go +++ b/apps/sisyphusd/daemon.go @@ -46,6 +46,7 @@ import ( nodepb "github.com/sisyphus-network/Sisyphus/packages/protocol/sisyphus/node/v1" pb "github.com/sisyphus-network/Sisyphus/packages/protocol/sisyphus/v1" "github.com/sisyphus-network/Sisyphus/packages/runtime" + "github.com/sisyphus-network/Sisyphus/packages/s3" "github.com/sisyphus-network/Sisyphus/packages/sealed" "github.com/sisyphus-network/Sisyphus/packages/storage" ) @@ -79,6 +80,11 @@ func runDaemon(ctx context.Context, args []string) error { gcInterval := fs.Duration("gc-interval", time.Hour, "coordinator role: how often to delete stored data nothing is keeping; 0 never does") maxStore := fs.Uint64("max-store-bytes", 0, "coordinator role: refuse uploads once stored data uses this much disk; 0 means no limit") maxCache := fs.Uint64("max-cache-bytes", 0, "worker-only node: evict the least recently used cached blobs once the cache uses this much disk; 0 means no limit") + s3Endpoint := fs.String("s3-endpoint", "", "coordinator: keep stored data in a bucket of the S3-style object store at this address, such as https://s3.wasabisys.com or http://127.0.0.1:9000, instead of on this machine's disk") + s3Bucket := fs.String("s3-bucket", "", "the bucket, which must exist") + s3Prefix := fs.String("s3-prefix", "", "put this before the name of every object, to share the bucket with other things") + s3Region := fs.String("s3-region", "", "the bucket's region, for stores that have them") + s3Credentials := fs.String("s3-credentials", "", "a file with the access key on its first line and the secret key on its second; they stay on this node") useKubo := fs.Bool("kubo", false, "keep stored data in a Kubo (IPFS) daemon that this node starts and runs alongside itself, on a private network with the rest of its pool; needs the ipfs program installed, and for a worker, a coordinator that uses it too") swarmPort := fs.Int("swarm-port", 0, "coordinator role with --kubo: TCP port to open so that members' Kubo daemons can connect to this node's directly, which is faster; 0 opens none, and they reach it through --listen") syncCache := fs.Bool("sync-cache", false, "worker-only node: wait for the disk when caching a blob; slower, but the cache then survives a power cut without downloading again") @@ -128,6 +134,16 @@ func runDaemon(ctx context.Context, args []string) error { if host, _, err := net.SplitHostPort(*webListen); *webListen != "" && (err != nil || !net.ParseIP(host).IsLoopback()) { return errors.New("--web-listen must be a loopback address such as 127.0.0.1:50052: the local API is for this machine only, its web pages included") } + if *s3Endpoint != "" { + switch { + case !isCoordinator: + return errors.New("--s3-endpoint is for a node that coordinates a pool: it is the pool's stored data that is kept there") + case *useKubo: + return errors.New("--s3-endpoint and --kubo each say where stored data is kept: choose one") + case *s3Bucket == "": + return errors.New("--s3-endpoint needs --s3-bucket, the bucket to keep stored data in") + } + } if isCoordinator && *invitation != "" { return errors.New("--join is for worker-only nodes") } @@ -243,6 +259,15 @@ func runDaemon(ctx context.Context, args []string) error { // stopped. var store *storage.Store switch { + case *s3Endpoint != "": + bucket, err := openBucket(ctx, *s3Endpoint, *s3Bucket, *s3Prefix, *s3Region, *s3Credentials) + if err != nil { + return err + } + log.Info("stored data is kept in a bucket", "endpoint", *s3Endpoint, "bucket", *s3Bucket) + if store, err = storage.OpenBucket(bucket, filepath.Join(*dataDir, "s3-pins")); err != nil { + return fmt.Errorf("%w (nodes sharing a machine each need their own --data-dir)", err) + } case isCoordinator && *useKubo: store, err = storage.OpenKubo(sidecar, filepath.Join(*dataDir, "kubo-pins")) case isCoordinator: @@ -756,6 +781,25 @@ func withTrust(peers []*nodepb.Peer, takes []string, worksFor func(id string) bo return peers } +// openBucket opens the bucket a node keeps its stored data in, with the +// credentials in a file: the access key on the first line and the secret +// key on the second. +func openBucket(ctx context.Context, endpoint, bucket, prefix, region, credentialsFile string) (*s3.Client, error) { + cfg := s3.Config{Endpoint: endpoint, Bucket: bucket, Prefix: prefix, Region: region} + if credentialsFile != "" { + keys, err := os.ReadFile(credentialsFile) + if err != nil { + return nil, fmt.Errorf("read the bucket's credentials: %w", err) + } + lines := strings.Fields(string(keys)) + if len(lines) != 2 { + return nil, fmt.Errorf("%s should hold the access key on its first line and the secret key on its second", credentialsFile) + } + cfg.AccessKey, cfg.SecretKey = lines[0], lines[1] + } + return s3.Open(ctx, cfg) +} + // countryOf says which country an address is registered in. It is the // table the daemon carries; tests put another here. var countryOf = geo.Country diff --git a/apps/sisyphusd/s3_test.go b/apps/sisyphusd/s3_test.go new file mode 100644 index 0000000..43f6aa3 --- /dev/null +++ b/apps/sisyphusd/s3_test.go @@ -0,0 +1,156 @@ +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + + "github.com/sisyphus-network/Sisyphus/packages/s3" +) + +// objectStore is a real object store, SeaweedFS with its S3 gateway in a +// container, started the first time a test wants it. +var objectStore struct { + once sync.Once + endpoint string + err error +} + +const ( + objectStoreAccess = "sisyphus-test" + objectStoreSecret = "sisyphus-test-secret" +) + +// bucketFor starts the object store if need be, makes a bucket in it for a +// test, and returns where it is and the bucket's name. A machine with no +// Docker skips the test, unless told that it must have it. +func bucketFor(t *testing.T) (endpoint, bucket string) { + t.Helper() + objectStore.once.Do(func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + objectStore.err = err + return + } + port := lis.Addr().(*net.TCPAddr).Port + lis.Close() + name := fmt.Sprintf("sisyphus-daemon-s3-test-%d", os.Getpid()) + out, err := exec.Command("docker", "run", "-d", "--rm", "--name", name, "-p", fmt.Sprintf("127.0.0.1:%d:8333", port), + "-e", "AWS_ACCESS_KEY_ID="+objectStoreAccess, "-e", "AWS_SECRET_ACCESS_KEY="+objectStoreSecret, + "chrislusf/seaweedfs:latest", "server", "-s3", "-dir=/data").CombinedOutput() + if err != nil { + objectStore.err = fmt.Errorf("start the object store: %v: %s", err, bytes.TrimSpace(out)) + return + } + stopAtExit = append(stopAtExit, func() { exec.Command("docker", "rm", "-f", name).Run() }) + at := fmt.Sprintf("http://127.0.0.1:%d", port) + for deadline := time.Now().Add(90 * time.Second); time.Now().Before(deadline); time.Sleep(250 * time.Millisecond) { + if res, err := http.Get(at); err == nil { + res.Body.Close() + objectStore.endpoint = at + return + } + } + objectStore.err = errors.New("the object store did not start in time") + }) + if objectStore.err != nil { + if os.Getenv("SISYPHUS_REQUIRE_DOCKER") != "" { + t.Fatalf("no object store to test against, and SISYPHUS_REQUIRE_DOCKER is set: %v", objectStore.err) + } + t.Skipf("no object store to test against: %v", objectStore.err) + } + api, err := minio.New(strings.TrimPrefix(objectStore.endpoint, "http://"), &minio.Options{Creds: credentials.NewStaticV4(objectStoreAccess, objectStoreSecret, "")}) + if err != nil { + t.Fatal(err) + } + bucket = fmt.Sprintf("test-%d", time.Now().UnixNano()) + // The gateway answers before it is ready to make buckets. + for deadline := time.Now().Add(60 * time.Second); ; time.Sleep(500 * time.Millisecond) { + if err = api.MakeBucket(context.Background(), bucket, minio.MakeBucketOptions{}); err == nil { + return objectStore.endpoint, bucket + } + if time.Now().After(deadline) { + t.Fatalf("make bucket: %v", err) + } + } +} + +func TestACoordinatorKeepsItsStoredDataInABucket(t *testing.T) { + endpoint, bucket := bucketFor(t) + dataDir := t.TempDir() + keys := filepath.Join(t.TempDir(), "keys") + os.WriteFile(keys, []byte(objectStoreAccess+"\n"+objectStoreSecret+"\n"), 0o600) + addr := freeAddr(t) + args := []string{"--data-dir", dataDir, "--listen", addr, "--name", "rig", "--slots", "2", + "--s3-endpoint", endpoint, "--s3-bucket", bucket, "--s3-prefix", "pool-one", "--s3-credentials", keys} + stop := startDaemon(t, args...) + waitForOutput(t, "rig", "nodes", "--addr", addr) + + input := filepath.Join(t.TempDir(), "words.txt") + os.WriteFile(input, []byte(strings.Repeat("the boulder rolls and the boulder rolls again\n", 2000)), 0o600) + cid := strings.TrimSpace(mustCLI(t, "blob", "put", "--addr", addr, input)) + out := mustCLI(t, "job", "submit", "--addr", addr, "--workload", "wordcount", "--params", `{"input":"`+cid+`"}`, "--tasks", "2") + if !strings.Contains(out, `"words":16000`) { + t.Fatalf("a job over data kept in a bucket:\n%s", out) + } + // The data is in the bucket, under the prefix, and nowhere on this + // machine but for the record of what is kept. + client, err := s3.Open(context.Background(), s3.Config{Endpoint: endpoint, Bucket: bucket, Prefix: "pool-one", AccessKey: objectStoreAccess, SecretKey: objectStoreSecret}) + if err != nil { + t.Fatal(err) + } + blocks := 0 + client.List(context.Background(), func(name string, _ int64) bool { + if strings.HasPrefix(name, "blocks/") { + blocks++ + } + return true + }) + if blocks < 2 { + t.Errorf("the bucket holds %d blocks", blocks) + } + if _, err := os.Stat(filepath.Join(dataDir, "blobs")); !os.IsNotExist(err) { + t.Errorf("a store on this machine's disk beside the bucket: %v", err) + } + // It is all still there for a coordinator started again. + stop() + startDaemon(t, args...) + waitForOutput(t, "rig", "nodes", "--addr", addr) + if got := mustCLI(t, "blob", "stat", "--addr", addr, cid); !strings.Contains(got, "92000") { + t.Errorf("the file after a restart: %s", got) + } + + oneLine := filepath.Join(t.TempDir(), "keys") + os.WriteFile(oneLine, []byte(objectStoreAccess+"\n"), 0o600) + inUse := []string{"--data-dir", dataDir, "--s3-endpoint", endpoint, "--s3-bucket", bucket, "--s3-prefix", "pool-one", "--s3-credentials", keys} + for _, tt := range []struct { + name string + args []string + want string + }{ + {"no bucket", []string{"--data-dir", t.TempDir(), "--s3-endpoint", endpoint}, "needs --s3-bucket"}, + {"with Kubo too", []string{"--data-dir", t.TempDir(), "--s3-endpoint", endpoint, "--s3-bucket", bucket, "--kubo"}, "choose one"}, + {"on a node with no pool", []string{"--data-dir", t.TempDir(), "--role", "worker", "--coordinator", "127.0.0.1:1", "--s3-endpoint", endpoint, "--s3-bucket", bucket}, "is for a node that coordinates a pool"}, + {"credentials that are not there", []string{"--data-dir", t.TempDir(), "--s3-endpoint", endpoint, "--s3-bucket", bucket, "--s3-credentials", filepath.Join(t.TempDir(), "absent")}, "read the bucket's credentials"}, + {"credentials of one line", []string{"--data-dir", t.TempDir(), "--s3-endpoint", endpoint, "--s3-bucket", bucket, "--s3-credentials", oneLine}, "the access key on its first line and the secret key on its second"}, + {"a bucket that is not there", []string{"--data-dir", t.TempDir(), "--s3-endpoint", endpoint, "--s3-bucket", "no-such-bucket-here", "--s3-credentials", keys}, "there is no bucket"}, + {"a data directory in use", inUse, "in use by another process"}, + } { + if _, err := cli(t, append([]string{"run", "--listen", freeAddr(t)}, tt.args...)...); err == nil || !strings.Contains(err.Error(), tt.want) { + t.Errorf("--s3-endpoint %s: %v, want %q", tt.name, err, tt.want) + } + } +} diff --git a/docs/development.md b/docs/development.md index ebab44a..fb7a176 100644 --- a/docs/development.md +++ b/docs/development.md @@ -32,6 +32,7 @@ CI runs formatting, `go vet`, the tests under the race detector, `make cover` an | `packages/job-model` | Job and task state machines, and which blobs a job consumed and produced. No I/O, no locks. | | `packages/nodedb` | The node's SQLite database, where a coordinator keeps its jobs, its members and the invitations it has issued. Schema changes are new numbered files in `migrations/`, never edits to old ones. | | `packages/runtime` | The `Workload` interface, the built-in workloads, and the recorder that notes what a workload reads and writes. | +| `packages/s3` | A bucket in an S3-style object store, for a store that keeps its blocks there. Its tests run against a real one, SeaweedFS in a container. | | `packages/sealed` | Encrypting a private job's blobs: the format, sealing and seekable unsealing. | | `packages/storage` | The blob store: IPFS-compatible import, pins, garbage collection, verification. | | `apps/sisyphusd/coordinator` | Scheduling, retries, aggregation, worker connections, pinning a job's data. | diff --git a/go.mod b/go.mod index 3fb7cd2..3810c92 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/ipfs/go-ipld-format v0.6.4 github.com/libp2p/go-libp2p v0.50.0 github.com/libp2p/go-libp2p-kad-dht v0.42.2 + github.com/minio/minio-go/v7 v7.3.0 github.com/modelcontextprotocol/go-sdk v1.8.0 github.com/multiformats/go-multiaddr v0.16.1 github.com/multiformats/go-multihash v0.2.3 @@ -63,7 +64,9 @@ require ( github.com/ipld/go-ipld-prime v0.24.0 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect + github.com/klauspost/compress v1.19.2 // indirect github.com/klauspost/cpuid/v2 v2.4.0 // indirect + github.com/klauspost/crc32 v1.3.0 // indirect github.com/koron/go-ssdp v0.9.1 // indirect github.com/libp2p/go-buffer-pool v0.1.0 // indirect github.com/libp2p/go-cidranger v1.1.0 // indirect @@ -82,6 +85,8 @@ require ( github.com/miekg/dns v1.1.73 // indirect github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b // indirect github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect + github.com/minio/crc64nvme v1.1.1 // indirect + github.com/minio/md5-simd v1.1.2 // indirect github.com/minio/sha256-simd v1.0.1 // indirect github.com/mr-tron/base58 v1.3.0 // indirect github.com/multiformats/go-base32 v0.1.0 // indirect @@ -96,6 +101,7 @@ require ( github.com/ncruces/go-strftime v1.0.0 // indirect github.com/pb33f/ordered-map/v2 v2.3.1 // indirect github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect + github.com/philhofer/fwd v1.2.0 // indirect github.com/pion/datachannel v1.5.10 // indirect github.com/pion/dtls/v3 v3.1.4 // indirect github.com/pion/ice/v4 v4.0.10 // indirect @@ -122,6 +128,7 @@ require ( github.com/quic-go/quic-go v0.62.0 // indirect github.com/quic-go/webtransport-go v0.13.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/rs/xid v1.6.0 // indirect github.com/segmentio/asm v1.1.3 // indirect github.com/segmentio/encoding v0.5.4 // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect @@ -130,10 +137,12 @@ require ( github.com/tidwall/match v1.1.1 // indirect github.com/tidwall/pretty v1.2.1 // indirect github.com/tidwall/sjson v1.2.5 // indirect + github.com/tinylib/msgp v1.6.4 // indirect github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f // indirect github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect github.com/wlynxg/anet v0.0.5 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect + github.com/zeebo/xxh3 v1.1.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel v1.46.0 // indirect go.opentelemetry.io/otel/metric v1.46.0 // indirect @@ -142,6 +151,7 @@ require ( go.uber.org/mock v0.6.0 // indirect go.uber.org/multierr v1.11.0 // indirect go.uber.org/zap v1.28.0 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.2 // indirect golang.org/x/crypto v0.57.0 // indirect golang.org/x/exp v0.0.0-20260908205506-85c1c2202aba // indirect @@ -157,6 +167,7 @@ require ( gonum.org/v1/gonum v0.17.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260819154853-08b0e4226688 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260819154853-08b0e4226688 // indirect + gopkg.in/ini.v1 v1.67.3 // indirect lukechampine.com/blake3 v1.4.1 // indirect modernc.org/libc v1.77.1 // indirect modernc.org/mathutil v1.7.1 // indirect diff --git a/go.sum b/go.sum index 6e9ee80..39fd575 100644 --- a/go.sum +++ b/go.sum @@ -31,6 +31,7 @@ github.com/crackcomm/go-gitignore v0.0.0-20241020182519-7843d2ba8fdf/go.mod h1:p github.com/cskr/pubsub v1.0.2 h1:vlOzMhl6PFn60gRlTQQsIfVwaPB/B/8MziK8FhEPt/0= github.com/cskr/pubsub v1.0.2/go.mod h1:/8MzYXk/NJAz782G8RPkFzXTZVu63VotefPnR9TIRis= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c h1:pFUpOrbxDR6AkioZ1ySsx5yxlDQZ8stG2b88gTPxgJU= github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c/go.mod h1:6UhI8N9EjYm1c2odKpFpAYeR8dsBeM7PtzQhRgxRr9U= github.com/decred/dcrd/crypto/blake256 v1.1.0 h1:zPMNGQCm0g4QTY27fOCorQW7EryeQ/U0x++OzVrdms8= @@ -138,8 +139,13 @@ github.com/jbenet/go-temp-err-catcher v0.1.0 h1:zpb3ZH6wIE8Shj2sKS+khgRvf7T7RABo github.com/jbenet/go-temp-err-catcher v0.1.0/go.mod h1:0kJRvmDZXNMIiJirNPEYfhpPwbGVtZVWC34vc5WLsDk= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= +github.com/klauspost/crc32 v1.3.0 h1:sSmTt3gUt81RP655XGZPElI0PelVTZ6YwCRnPSupoFM= +github.com/klauspost/crc32 v1.3.0/go.mod h1:D7kQaZhnkX/Y0tstFGf8VUzv2UofNGqCjnC3zdHB0Hw= github.com/koron/go-ssdp v0.9.1 h1:zvxbAAuJftJIZ8Jh8mda+LI7V92hYZf/sKprmOxpxwA= github.com/koron/go-ssdp v0.9.1/go.mod h1:C43c047jWkDaeg9YuZlSh/QGqOieuWV6dbhWi/jcaLk= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= @@ -195,6 +201,12 @@ github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b/go.mod h1:lxPUiZwKo github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc h1:PTfri+PuQmWDqERdnNMiD9ZejrlswWrCpBEZgWOiTrc= github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc/go.mod h1:cGKTAVKx4SxOuR/czcZ/E2RSJ3sfHs8FpHhQ5CWMf9s= github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ= +github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI= +github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg= +github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= +github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= +github.com/minio/minio-go/v7 v7.3.0 h1:HM4pFCSQq/TK+j0/zmorSh5ddh81iDgRgU0BG0Vz/YU= +github.com/minio/minio-go/v7 v7.3.0/go.mod h1:KUPWdecEO1LWyUz+sTGXAuf2jZHrPh5fCsRH86QbPfk= github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM= github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM= github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8= @@ -233,6 +245,8 @@ github.com/pb33f/ordered-map/v2 v2.3.1 h1:5319HDO0aw4DA4gzi+zv4FXU9UlSs3xGZ40wcP github.com/pb33f/ordered-map/v2 v2.3.1/go.mod h1:qxFQgd0PkVUtOMCkTapqotNgzRhMPL7VvaHKbd1HnmQ= github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 h1:onHthvaw9LFnH4t2DcNVpwGmV9E1BkGknEliJkfwQj0= github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58/go.mod h1:DXv8WO4yhMYhSNPKjeNKa5WY9YCIEBRbNzFFPJbWO6Y= +github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= +github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= github.com/pion/datachannel v1.5.10 h1:ly0Q26K1i6ZkGf42W7D4hQYR90pZwzFOjTq5AuCKk4o= github.com/pion/datachannel v1.5.10/go.mod h1:p/jJfC9arb29W7WrxyKbepTU20CFgyx5oLo8Rs4Py/M= github.com/pion/dtls/v3 v3.1.4 h1:QhvtMflMfu9Kf0RcDC5BJBle4caPskByrKQR6uuYqpY= @@ -290,6 +304,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= +github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc= github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg= github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= @@ -303,7 +319,14 @@ github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2 github.com/standard-webhooks/standard-webhooks/libraries v0.0.1 h1:uOfcYT+3QungH6tIGSVCR/Y3KJmgJiHcojJbMTPDZAI= github.com/standard-webhooks/standard-webhooks/libraries v0.0.1/go.mod h1:L1MQhA6x4dn9r007T033lsaZMv9EmBAdXyU/+EF40fo= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE= @@ -318,6 +341,8 @@ github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= +github.com/tinylib/msgp v1.6.4 h1:mOwYbyYDLPj35mkA2BjjYejgJk9BuHxDdvRnb6v2ZcQ= +github.com/tinylib/msgp v1.6.4/go.mod h1:RSp0LW9oSxFut3KzESt5Voq4GVWyS+PSulT77roAqEA= github.com/warpfork/go-testmark v0.12.1 h1:rMgCpJfwy1sJ50x0M0NgyphxYYPMOODIJHhsXyEHU0s= github.com/warpfork/go-testmark v0.12.1/go.mod h1:kHwy7wfvGSPh1rQJYKayD4AbtNaeyZdcGi9tNJTaa5Y= github.com/whyrusleeping/chunker v0.0.0-20181014151217-fe64bd25879f h1:jQa4QT2UP9WYv2nzyawpKMOCl+Z/jW7djv2/J50lj9E= @@ -328,6 +353,10 @@ github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA= github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4= github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4= +github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ= +github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/otel v1.46.0 h1:FHt5/CDyVxi/8IM1CH7VE/rRgq3kLHa2mSTVMO8AWyc= @@ -419,9 +448,13 @@ google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/ini.v1 v1.67.3 h1:iM9Lhz5MRSGhHVGGwCuzG9KO8PoirCXj/m/qTmOJJQw= +gopkg.in/ini.v1 v1.67.3/go.mod h1:x/cyOwCgZqOkJoDIJ3c1KNHMo10+nLGAhh+kn3Zizss= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= lukechampine.com/blake3 v1.4.1 h1:I3Smz7gso8w4/TunLKec6K2fn+kyKtDxr/xcQEN84Wg= lukechampine.com/blake3 v1.4.1/go.mod h1:QFosUxmjB8mnrWFSNwKmvxHpfY72bmD2tQ0kBMM3kwo= modernc.org/cc/v4 v4.29.7 h1:q+NXGJ0bK3b4TXFYQQVr9pYETGnmwFWkrUzJnMya/Tg= diff --git a/packages/s3/s3.go b/packages/s3/s3.go new file mode 100644 index 0000000..5690e2d --- /dev/null +++ b/packages/s3/s3.go @@ -0,0 +1,125 @@ +// Package s3 reaches a bucket in an object store that speaks Amazon's S3 +// dialect, which most do: Ceph's gateway, MinIO, SeaweedFS, Storj's +// gateway, Wasabi, and AWS itself. +package s3 + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/url" + "strings" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + + "github.com/sisyphus-network/Sisyphus/packages/storage" +) + +// Config says which bucket, where, and as whom. +type Config struct { + // Endpoint is where the object store is, such as + // https://s3.eu-central-1.wasabisys.com or http://127.0.0.1:9000. One + // with no scheme is taken to be https. + Endpoint string + Bucket string + // Prefix, if set, goes before the name of every object, so that a + // bucket can be shared with other things. + Prefix string + // Region is the region the bucket is in, for stores that have them. + Region string + // AccessKey and SecretKey are the credentials. They stay on this node. + AccessKey, SecretKey string +} + +// Client is a bucket. It is a storage.Bucket. +type Client struct { + api *minio.Client + bucket string + prefix string +} + +// Open returns the bucket cfg describes, having checked that it is there +// and can be reached with the credentials given. +func Open(ctx context.Context, cfg Config) (*Client, error) { + endpoint := cfg.Endpoint + if !strings.Contains(endpoint, "://") { + endpoint = "https://" + endpoint + } + at, err := url.Parse(endpoint) + if err != nil || at.Host == "" { + return nil, fmt.Errorf("object store address %q is not one", cfg.Endpoint) + } + if cfg.Bucket == "" { + return nil, fmt.Errorf("no bucket was named") + } + api, err := minio.New(at.Host, &minio.Options{ + Creds: credentials.NewStaticV4(cfg.AccessKey, cfg.SecretKey, ""), + Secure: at.Scheme == "https", + Region: cfg.Region, + }) + if err != nil { + return nil, fmt.Errorf("object store at %s: %w", cfg.Endpoint, err) + } + found, err := api.BucketExists(ctx, cfg.Bucket) + if err != nil { + return nil, fmt.Errorf("reach bucket %s at %s: %w", cfg.Bucket, cfg.Endpoint, err) + } + if !found { + return nil, fmt.Errorf("there is no bucket %s at %s: make it first, this does not", cfg.Bucket, cfg.Endpoint) + } + prefix := strings.Trim(cfg.Prefix, "/") + if prefix != "" { + prefix += "/" + } + return &Client{api: api, bucket: cfg.Bucket, prefix: prefix}, nil +} + +func (c *Client) Put(ctx context.Context, name string, data []byte) error { + _, err := c.api.PutObject(ctx, c.bucket, c.prefix+name, bytes.NewReader(data), int64(len(data)), minio.PutObjectOptions{ContentType: "application/octet-stream"}) + return err +} + +func (c *Client) Get(ctx context.Context, name string) ([]byte, error) { + // Opening an object asks for nothing: what is wrong with it is found + // on reading it. + object, _ := c.api.GetObject(ctx, c.bucket, c.prefix+name, minio.GetObjectOptions{}) + defer object.Close() + data, err := io.ReadAll(object) + return data, absent(err) +} + +func (c *Client) Size(ctx context.Context, name string) (int, error) { + info, err := c.api.StatObject(ctx, c.bucket, c.prefix+name, minio.StatObjectOptions{}) + return int(info.Size), absent(err) +} + +func (c *Client) Remove(ctx context.Context, name string) error { + return c.api.RemoveObject(ctx, c.bucket, c.prefix+name, minio.RemoveObjectOptions{}) +} + +func (c *Client) List(ctx context.Context, visit func(name string, size int64) bool) error { + // Stopping the listing early needs its context ended, or it goes on + // listing to nobody. + ctx, stop := context.WithCancel(ctx) + defer stop() + for object := range c.api.ListObjects(ctx, c.bucket, minio.ListObjectsOptions{Prefix: c.prefix, Recursive: true}) { + if object.Err != nil { + return object.Err + } + if !visit(strings.TrimPrefix(object.Key, c.prefix), object.Size) { + return nil + } + } + return nil +} + +// absent turns the object store's "not there" into the one a store knows. +func absent(err error) error { + if minio.ToErrorResponse(err).StatusCode == http.StatusNotFound { + return storage.ErrNoObject + } + return err +} diff --git a/packages/s3/s3_test.go b/packages/s3/s3_test.go new file mode 100644 index 0000000..85ff7c5 --- /dev/null +++ b/packages/s3/s3_test.go @@ -0,0 +1,209 @@ +package s3 + +import ( + "bytes" + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" + + "github.com/sisyphus-network/Sisyphus/packages/storage" +) + +// objectStore is a real object store for the tests to use: SeaweedFS with +// its S3 gateway, in a container, started once and stopped when the tests +// are over. +var objectStore struct { + once sync.Once + endpoint string + err error +} + +const ( + storeImage = "chrislusf/seaweedfs:latest" + accessKey = "sisyphus-test" + secretKey = "sisyphus-test-secret" +) + +func TestMain(m *testing.M) { + code := m.Run() + if objectStore.endpoint != "" { + exec.Command("docker", "rm", "-f", containerName()).Run() + } + os.Exit(code) +} + +func containerName() string { return fmt.Sprintf("sisyphus-s3-test-%d", os.Getpid()) } + +// endpoint returns where the object store is, starting it if need be. A +// machine with no Docker skips the test, unless told that it must have it. +func endpoint(t *testing.T) string { + t.Helper() + objectStore.once.Do(func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + objectStore.err = err + return + } + port := lis.Addr().(*net.TCPAddr).Port + lis.Close() + out, err := exec.Command("docker", "run", "-d", "--rm", "--name", containerName(), "-p", fmt.Sprintf("127.0.0.1:%d:8333", port), + "-e", "AWS_ACCESS_KEY_ID="+accessKey, "-e", "AWS_SECRET_ACCESS_KEY="+secretKey, + storeImage, "server", "-s3", "-dir=/data").CombinedOutput() + if err != nil { + objectStore.err = fmt.Errorf("start %s: %v: %s", storeImage, err, bytes.TrimSpace(out)) + return + } + at := fmt.Sprintf("http://127.0.0.1:%d", port) + for deadline := time.Now().Add(90 * time.Second); time.Now().Before(deadline); time.Sleep(250 * time.Millisecond) { + if res, err := http.Get(at); err == nil { + res.Body.Close() + objectStore.endpoint = at + return + } + } + objectStore.err = errors.New("the object store did not start in time") + }) + if objectStore.err != nil { + if os.Getenv("SISYPHUS_REQUIRE_DOCKER") != "" { + t.Fatalf("no object store to test against, and SISYPHUS_REQUIRE_DOCKER is set: %v", objectStore.err) + } + t.Skipf("no object store to test against: %v", objectStore.err) + } + return objectStore.endpoint +} + +// newBucket makes a bucket of its own for a test and returns its name. +func newBucket(t *testing.T) string { + t.Helper() + at := endpoint(t) + api, err := minio.New(strings.TrimPrefix(at, "http://"), &minio.Options{Creds: credentials.NewStaticV4(accessKey, secretKey, "")}) + if err != nil { + t.Fatal(err) + } + name := fmt.Sprintf("test-%d", time.Now().UnixNano()) + // The gateway answers before it is ready to make buckets. + for deadline := time.Now().Add(60 * time.Second); ; time.Sleep(500 * time.Millisecond) { + if err = api.MakeBucket(context.Background(), name, minio.MakeBucketOptions{}); err == nil { + return name + } + if time.Now().After(deadline) { + t.Fatalf("make bucket: %v", err) + } + } +} + +func TestBlocksAreKeptInARealObjectStore(t *testing.T) { + ctx := context.Background() + bucket := newBucket(t) + client, err := Open(ctx, Config{Endpoint: endpoint(t), Bucket: bucket, Prefix: "/pool-one/", AccessKey: accessKey, SecretKey: secretKey}) + if err != nil { + t.Fatal(err) + } + if err := client.Put(ctx, "blocks/a", []byte("the boulder")); err != nil { + t.Fatal(err) + } + if err := client.Put(ctx, "blocks/b", []byte("rolls")); err != nil { + t.Fatal(err) + } + if got, err := client.Get(ctx, "blocks/a"); err != nil || string(got) != "the boulder" { + t.Errorf("Get = %q, %v", got, err) + } + if size, err := client.Size(ctx, "blocks/b"); err != nil || size != 5 { + t.Errorf("Size = %d, %v", size, err) + } + // What is not there is said to be not there, in the store's own words. + if _, err := client.Get(ctx, "blocks/absent"); !errors.Is(err, storage.ErrNoObject) { + t.Errorf("Get of nothing: %v", err) + } + if _, err := client.Size(ctx, "blocks/absent"); !errors.Is(err, storage.ErrNoObject) { + t.Errorf("Size of nothing: %v", err) + } + // Another prefix in the same bucket is another store's, and unseen. + other, err := Open(ctx, Config{Endpoint: endpoint(t), Bucket: bucket, AccessKey: accessKey, SecretKey: secretKey}) + if err != nil { + t.Fatal(err) + } + if err := other.Put(ctx, "elsewhere", []byte("x")); err != nil { + t.Fatal(err) + } + listed := map[string]int64{} + if err := client.List(ctx, func(name string, size int64) bool { listed[name] = size; return true }); err != nil || len(listed) != 2 || listed["blocks/a"] != 11 || listed["blocks/b"] != 5 { + t.Errorf("List = %v, %v", listed, err) + } + // A listing can be stopped part way. + seen := 0 + if err := client.List(ctx, func(string, int64) bool { seen++; return false }); err != nil || seen != 1 { + t.Errorf("a listing stopped at once saw %d, %v", seen, err) + } + if err := client.Remove(ctx, "blocks/a"); err != nil { + t.Fatal(err) + } + if _, err := client.Size(ctx, "blocks/a"); !errors.Is(err, storage.ErrNoObject) { + t.Errorf("Size of what was removed: %v", err) + } + + // And a whole store over it: a blob put, read back and verified. + store, err := storage.OpenBucket(client, t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer store.Close() + blob := bytes.Repeat([]byte("one must imagine Sisyphus happy. "), 20_000) + c, err := store.Put(ctx, bytes.NewReader(blob)) + if err != nil { + t.Fatal(err) + } + if err := store.Verify(ctx, c); err != nil { + t.Errorf("Verify: %v", err) + } + if size, err := store.Size(ctx); err != nil || size < uint64(len(blob)) { + t.Errorf("Size = %d, %v", size, err) + } +} + +func TestABucketThatCannotBeUsedIsSaidSo(t *testing.T) { + ctx := context.Background() + at := endpoint(t) + bucket := newBucket(t) + soon, done := context.WithTimeout(ctx, 3*time.Second) + defer done() + for name, tt := range map[string]struct { + cfg Config + want string + }{ + "an address that is none": {Config{Endpoint: "http://", Bucket: bucket}, "is not one"}, + "an address that cannot be read": {Config{Endpoint: "http://a b", Bucket: bucket}, "is not one"}, + "no bucket named": {Config{Endpoint: at}, "no bucket was named"}, + "a host that is no host": {Config{Endpoint: "http://-not_a_host-:9", Bucket: bucket}, "object store at"}, + "a bucket that is not there": {Config{Endpoint: at, Bucket: "no-such-bucket-here", AccessKey: accessKey, SecretKey: secretKey}, "there is no bucket no-such-bucket-here"}, + "nothing at the address": {Config{Endpoint: "127.0.0.1:1", Bucket: bucket}, "reach bucket"}, + } { + if _, err := Open(soon, tt.cfg); err == nil || !strings.Contains(err.Error(), tt.want) { + t.Errorf("%s: %v, want %q", name, err, tt.want) + } + } + // A listing that fails part way is a failure, not a short list. + client, err := Open(ctx, Config{Endpoint: at, Bucket: bucket, AccessKey: accessKey, SecretKey: secretKey}) + if err != nil { + t.Fatal(err) + } + gone, cancel := context.WithCancel(ctx) + cancel() + if err := client.List(gone, func(string, int64) bool { return true }); err == nil { + t.Error("a listing that could not be made was taken for an empty one") + } + if err := client.Put(gone, "x", []byte("x")); err == nil { + t.Error("a put that could not be made was taken for done") + } +} diff --git a/packages/storage/s3.go b/packages/storage/s3.go new file mode 100644 index 0000000..874fbfd --- /dev/null +++ b/packages/storage/s3.go @@ -0,0 +1,162 @@ +package storage + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/gofrs/flock" + blocks "github.com/ipfs/go-block-format" + "github.com/ipfs/go-cid" + ipld "github.com/ipfs/go-ipld-format" +) + +// Bucket is the part of an S3-style object store that a store keeps its +// blocks in: one object for each block, named by the block's CID. An +// s3.Client is one. +type Bucket interface { + Put(ctx context.Context, name string, data []byte) error + // Get returns an object, and Size how long it is; both return + // ErrNoObject for one that is not there. + Get(ctx context.Context, name string) ([]byte, error) + Size(ctx context.Context, name string) (int, error) + Remove(ctx context.Context, name string) error + // List calls visit with the name and size of every object, until it + // returns false. + List(ctx context.Context, visit func(name string, size int64) bool) error +} + +// ErrNoObject is what a Bucket returns for an object it does not hold. +var ErrNoObject = errors.New("no such object") + +// OpenBucket opens a durable store whose blocks are kept as objects in a +// bucket, for a pool that runs or rents object storage: Ceph, MinIO, +// SeaweedFS, Storj, Wasabi, AWS. A blob has the same CID wherever its bytes +// are, so nothing else need know. Pins are this store's own, kept in dir as +// OpenLocal keeps them. +// +// A block is an object and reading one is a request, so a store in a bucket +// far away is slower than a disk: a blob of a gigabyte is four thousand +// blocks. Nothing is kept on this machine but the pins. +func OpenBucket(bucket Bucket, dir string) (*Store, error) { + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, fmt.Errorf("open blob store %s: %w", dir, err) + } + lock := flock.New(filepath.Join(dir, "LOCK")) + locked, err := lock.TryLock() + if err != nil { + return nil, fmt.Errorf("open blob store %s: %w", dir, err) + } + if !locked { + return nil, fmt.Errorf("open blob store %s: in use by another process", dir) + } + held := bucketBlocks{bucket} + store := newStoreOver(held, false) + store.size = held.size + store.close = func() error { return nil } // the bucket is its owner's to close + store.lock = lock + store.pinFile = filepath.Join(dir, "pins.json") + if store.pins, err = loadPins(store.pinFile); err != nil { + store.Close() + return nil, fmt.Errorf("open blob store %s: %w", dir, err) + } + return store, nil +} + +// bucketBlocks keeps blocks as objects in a bucket. +type bucketBlocks struct { + bucket Bucket +} + +// blockPrefix begins the name of every object that is a block, so that a +// bucket may hold other things beside them and they are left alone. +const blockPrefix = "blocks/" + +func objectName(c cid.Cid) string { return blockPrefix + c.String() } + +func (b bucketBlocks) Put(ctx context.Context, block blocks.Block) error { + return b.bucket.Put(ctx, objectName(block.Cid()), block.RawData()) +} + +func (b bucketBlocks) PutMany(ctx context.Context, many []blocks.Block) error { + for _, block := range many { + if err := b.Put(ctx, block); err != nil { + return err + } + } + return nil +} + +func (b bucketBlocks) Get(ctx context.Context, c cid.Cid) (blocks.Block, error) { + data, err := b.bucket.Get(ctx, objectName(c)) + if err != nil { + return nil, b.translate(err, c) + } + return blocks.NewBlockWithCid(data, c) +} + +func (b bucketBlocks) GetSize(ctx context.Context, c cid.Cid) (int, error) { + size, err := b.bucket.Size(ctx, objectName(c)) + return size, b.translate(err, c) +} + +func (b bucketBlocks) Has(ctx context.Context, c cid.Cid) (bool, error) { + _, err := b.bucket.Size(ctx, objectName(c)) + if errors.Is(err, ErrNoObject) { + return false, nil + } + return err == nil, err +} + +func (b bucketBlocks) DeleteBlock(ctx context.Context, c cid.Cid) error { + return b.bucket.Remove(ctx, objectName(c)) +} + +// AllKeysChan lists the blocks in the bucket. The listing is read to its +// end before anything is handed on, so that a failure part-way is reported +// and does not pass for a short list, which a garbage collection would act +// on. +func (b bucketBlocks) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { + var all []cid.Cid + err := b.bucket.List(ctx, func(name string, _ int64) bool { + if id, is := strings.CutPrefix(name, blockPrefix); is { + if c, err := cid.Decode(id); err == nil { + all = append(all, c) + } + } + return true + }) + if err != nil { + return nil, err + } + keys := make(chan cid.Cid, len(all)) + for _, c := range all { + keys <- c + } + close(keys) + return keys, nil +} + +// size adds up the blocks in the bucket. +func (b bucketBlocks) size(ctx context.Context) (uint64, error) { + var total uint64 + err := b.bucket.List(ctx, func(name string, size int64) bool { + if strings.HasPrefix(name, blockPrefix) { + total += uint64(size) + } + return true + }) + return total, err +} + +// translate turns the bucket's "not there" into the one the rest of the +// store recognises. +func (bucketBlocks) translate(err error, c cid.Cid) error { + if errors.Is(err, ErrNoObject) { + return ipld.ErrNotFound{Cid: c} + } + return err +} diff --git a/packages/storage/s3_test.go b/packages/storage/s3_test.go new file mode 100644 index 0000000..3f37858 --- /dev/null +++ b/packages/storage/s3_test.go @@ -0,0 +1,209 @@ +package storage + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "path/filepath" + "sort" + "testing" + "time" + + blocks "github.com/ipfs/go-block-format" +) + +// memoryBucket is a bucket kept in memory, which can be made to fail. +type memoryBucket struct { + objects map[string][]byte + fail error +} + +var errBucket = errors.New("the object store is down") + +func newBucket() *memoryBucket { return &memoryBucket{objects: map[string][]byte{}} } + +func (b *memoryBucket) Put(_ context.Context, name string, data []byte) error { + if b.fail != nil { + return b.fail + } + b.objects[name] = bytes.Clone(data) + return nil +} + +func (b *memoryBucket) Get(_ context.Context, name string) ([]byte, error) { + data, held := b.objects[name] + if b.fail != nil { + return nil, b.fail + } + if !held { + return nil, ErrNoObject + } + return data, nil +} + +func (b *memoryBucket) Size(ctx context.Context, name string) (int, error) { + data, err := b.Get(ctx, name) + return len(data), err +} + +func (b *memoryBucket) Remove(_ context.Context, name string) error { + delete(b.objects, name) + return b.fail +} + +func (b *memoryBucket) List(_ context.Context, visit func(string, int64) bool) error { + if b.fail != nil { + return b.fail + } + names := make([]string, 0, len(b.objects)) + for name := range b.objects { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + if !visit(name, int64(len(b.objects[name]))) { + break + } + } + return nil +} + +func openBucket(t *testing.T, bucket Bucket, dir string) *Store { + t.Helper() + s, err := OpenBucket(bucket, dir) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + return s +} + +func TestAStoreInABucketKeepsEachBlockAsAnObject(t *testing.T) { + bucket := newBucket() + // The bucket may hold other things, which are not the store's. + bucket.objects["notes.txt"] = []byte("someone else's") + bucket.objects["blocks/not-a-cid"] = []byte("nor is this a block") + dir := t.TempDir() + s := openBucket(t, bucket, dir) + + a := put(t, s, blobA) + blob, err := s.Open(ctx, a) + if err != nil { + t.Fatal(err) + } + read, _ := io.ReadAll(blob) + blob.Close() + if !bytes.Equal(read, blobA) || blob.Size() != uint64(len(blobA)) { + t.Fatalf("read back %d bytes of %d", len(read), len(blobA)) + } + // 700,000 bytes is three blocks and a node that lists them. + if len(bucket.objects) != 6 { + t.Errorf("the bucket holds %d objects", len(bucket.objects)) + } + if held, err := s.Has(ctx, a); err != nil || !held { + t.Errorf("Has = %v, %v", held, err) + } + if err := s.Verify(ctx, a); err != nil { + t.Errorf("Verify: %v", err) + } + // Its size is that of its blocks, not of what else is in the bucket. + if size, err := s.Size(ctx); err != nil || size < uint64(len(blobA)) || size > uint64(len(blobA))+1000 { + t.Errorf("Size = %d, %v", size, err) + } + + // Pins and collection work as on any store, and are kept beside it: + // what is pinned stays, what is not goes, and what is not the store's + // is left alone. + b := put(t, s, blobB) + if err := s.Pin(ctx, "someone", time.Time{}, a); err != nil { + t.Fatal(err) + } + collected, err := s.GC(ctx, afterGrace()) + if err != nil || collected.Blocks == 0 { + t.Fatalf("GC = %+v, %v", collected, err) + } + if held, _ := s.Has(ctx, b); held { + t.Error("an unpinned blob outlived the collection") + } + if _, err := s.Open(ctx, b); !errors.Is(err, ErrNotFound) { + t.Errorf("opening what was collected: %v", err) + } + if held, _ := s.Has(ctx, a); !held || string(bucket.objects["notes.txt"]) != "someone else's" { + t.Error("the collection took what it should have left") + } + s.Close() + again := openBucket(t, bucket, dir) + if pins := again.Pins(); len(pins) != 1 || pins[0].CID != a { + t.Errorf("pins after reopening: %v", pins) + } +} + +func TestAStoreInABucketReportsABucketThatFails(t *testing.T) { + bucket := newBucket() + s := openBucket(t, bucket, t.TempDir()) + c := put(t, s, blobA) + bucket.fail = errBucket + if _, err := s.Put(ctx, bytes.NewReader([]byte("more"))); !errors.Is(err, errBucket) { + t.Errorf("Put: %v", err) + } + if _, err := s.Open(ctx, c); !errors.Is(err, errBucket) { + t.Errorf("Open: %v", err) + } + if _, err := s.Has(ctx, c); !errors.Is(err, errBucket) { + t.Errorf("Has: %v", err) + } + if _, err := s.GC(ctx, afterGrace()); !errors.Is(err, errBucket) { + t.Errorf("GC: %v", err) + } + if _, err := s.Size(ctx); !errors.Is(err, errBucket) { + t.Errorf("Size: %v", err) + } + held := bucketBlocks{bucket} + if _, err := held.GetSize(ctx, c); !errors.Is(err, errBucket) { + t.Errorf("GetSize: %v", err) + } + bucket.fail = nil + if size, err := held.GetSize(ctx, c); err != nil || size == 0 { + t.Errorf("GetSize = %d, %v", size, err) + } + // Several blocks at once are stored one after another, and stop at + // the first that cannot be. + block, _ := held.Get(ctx, c) + if err := held.PutMany(ctx, []blocks.Block{block, block}); err != nil { + t.Errorf("PutMany: %v", err) + } + bucket.fail = errBucket + if err := held.PutMany(ctx, []blocks.Block{block}); !errors.Is(err, errBucket) { + t.Errorf("PutMany into a bucket that fails: %v", err) + } +} + +func TestOpeningAStoreInABucketFails(t *testing.T) { + bucket := newBucket() + file := filepath.Join(t.TempDir(), "a-file") + os.WriteFile(file, nil, 0o600) + if _, err := OpenBucket(bucket, filepath.Join(file, "store")); err == nil { + t.Error("opened a store beneath a regular file") + } + lockIsDir := t.TempDir() + os.Mkdir(filepath.Join(lockIsDir, "LOCK"), 0o700) + if _, err := OpenBucket(bucket, lockIsDir); err == nil { + t.Error("opened a store whose lock file cannot be created") + } + dir := t.TempDir() + first := openBucket(t, bucket, dir) + if second, err := OpenBucket(bucket, dir); err == nil { + second.Close() + t.Error("opened a store directory that is already open") + } + first.Close() + badPins := t.TempDir() + os.WriteFile(filepath.Join(badPins, "pins.json"), []byte("not json"), 0o600) + if _, err := OpenBucket(bucket, badPins); err == nil { + t.Error("opened a store whose pins cannot be read") + } + os.Remove(filepath.Join(badPins, "pins.json")) + openBucket(t, bucket, badPins) +}