Immutable tiered storage

config/config.go

@@ -7,6 +7,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/treeverse/lakefs/pyramid"
+
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
@@ -28,6 +30,10 @@ const (
 	DefaultBlockStoreS3StreamingChunkSize    = 2 << 19         // 1MiB by default per chunk
 	DefaultBlockStoreS3StreamingChunkTimeout = time.Second * 1 // or 1 seconds, whatever comes first
 
+	DefaultDiskAllocatedBytes     = 1 * 1024 * 1024 * 1024
+	DefaultDiskBaseDir            = "~/lakefs/metadata"
+	DefaultDiskBlockStoragePrefix = "_lakefs"
+
 	DefaultBlockStoreGSS3Endpoint = "https://storage.googleapis.com"
 
 	DefaultAuthCacheEnabled = true
@@ -88,6 +94,10 @@ func setDefaults() {
 	viper.SetDefault("blockstore.s3.streaming_chunk_timeout", DefaultBlockStoreS3StreamingChunkTimeout)
 	viper.SetDefault("blockstore.s3.max_retries", DefaultS3MaxRetries)
 
+	viper.SetDefault("disk.allocated_bytes", DefaultDiskAllocatedBytes)
+	viper.SetDefault("disk.base_dir", DefaultDiskBaseDir)
+	viper.SetDefault("disk.block_storage_prefix", DefaultDiskBlockStoragePrefix)
+
 	viper.SetDefault("gateways.s3.domain_name", DefaultS3GatewayDomainName)
 	viper.SetDefault("gateways.s3.region", DefaultS3GatewayRegion)
 
@@ -107,6 +117,14 @@ func (c *Config) GetDatabaseParams() dbparams.Database {
 	}
 }
 
+func (c *Config) GetLocalDiskParams() pyramid.Params {
+	return pyramid.Params{
+		AllocatedBytes:     viper.GetInt64("disk.allocated_bytes"),
+		BaseDir:            viper.GetString("disk.base_dir"),
+		BlockStoragePrefix: viper.GetString("disk.block_storage_prefix"),
+	}
+}
+
 func (c *Config) GetCatalogerType() string {
 	return viper.GetString("cataloger.type")
 }

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/containerd/continuity v0.0.0-20200710164510-efbc4488d8fe // indirect
 	github.com/cznic/mathutil v0.0.0-20180504122225-ca4c9f2c1369
 	github.com/davecgh/go-spew v1.1.1
+	github.com/dgraph-io/ristretto v0.0.3
 	github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654 // indirect
 	github.com/dlmiddlecote/sqlstats v1.0.1
 	github.com/georgysavva/scany v0.2.6
@@ -29,6 +30,7 @@ require (
 	github.com/gofrs/uuid v3.3.0+incompatible // indirect
 	github.com/golang-migrate/migrate/v4 v4.12.2
 	github.com/golang/protobuf v1.4.2
+	github.com/golang/snappy v0.0.2-0.20190904063534-ff6b7dc882cf // indirect
 	github.com/golangci/golangci-lint v1.30.0
 	github.com/google/uuid v1.1.1
 	github.com/gopherjs/gopherjs v0.0.0-20200217142428-fce0ec30dd00 // indirect

go.sum

@@ -57,6 +57,7 @@ github.com/Microsoft/go-winio v0.4.14 h1:+hMXMk01us9KgxGb7ftKQt2Xpf5hH/yky+TDA+q
 github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jBhyzoq1bpyYA=
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
+github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/OpenPeeDeeP/depguard v1.0.1 h1:VlW4R6jmBIv3/u1JNlawEvJMM4J+dPORPaZasQee8Us=
 github.com/OpenPeeDeeP/depguard v1.0.1/go.mod h1:xsIw86fROiiwelg+jB2uM9PiKihMMmUx/1V+TNhjQvM=
@@ -191,8 +192,12 @@ github.com/denis-tingajkin/go-header v0.3.1 h1:ymEpSiFjeItCy1FOP+x0M2KdCELdEAHUs
 github.com/denis-tingajkin/go-header v0.3.1/go.mod h1:sq/2IxMhaZX+RRcgHfCRx/m0M5na0fBt4/CRe7Lrji0=
 github.com/denisenkom/go-mssqldb v0.0.0-20191124224453-732737034ffd/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/denisenkom/go-mssqldb v0.0.0-20200620013148-b91950f658ec/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
+github.com/dgraph-io/ristretto v0.0.3 h1:jh22xisGBjrEVnRZ1DVTpBVQm0Xndu8sMl0CWDzSIBI=
+github.com/dgraph-io/ristretto v0.0.3/go.mod h1:KPxhHT9ZxKefz+PCeOGsrHpl1qZ7i70dGTu2u+Ahh6E=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
+github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/dgryski/go-gk v0.0.0-20140819190930-201884a44051 h1:ByJUvQYyTtNNCVfYNM48q6uYUT4fAlN0wNmd3th4BSo=
 github.com/dgryski/go-gk v0.0.0-20140819190930-201884a44051/go.mod h1:qm+vckxRlDt0aOla0RYJJVeqHZlWfOm2UIxHaqPB46E=
 github.com/dgryski/go-gk v0.0.0-20200319235926-a69029f61654 h1:XOPLOMn/zT4jIgxfxSsoXPxkrzz0FaCHwp33x5POJ+Q=
@@ -449,6 +454,8 @@ github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8l
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.2-0.20190904063534-ff6b7dc882cf h1:gFVkHXmVAhEbxZVDln5V9GKrLaluNoFHDbrZwAWZgws=
+github.com/golang/snappy v0.0.2-0.20190904063534-ff6b7dc882cf/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golangci/check v0.0.0-20180506172741-cfe4005ccda2 h1:23T5iq8rbUYlhpt5DB4XJkc6BU31uODLD1o1gKvZmD0=
 github.com/golangci/check v0.0.0-20180506172741-cfe4005ccda2/go.mod h1:k9Qvh+8juN+UKMCS/3jFtGICgW8O96FVaZsaxdzDkR4=
 github.com/golangci/dupl v0.0.0-20180902072040-3e9179ac440a h1:w8hkcTqaFpzKqonE9uMCefW1WDie15eSP/4MssdenaM=
@@ -600,7 +607,6 @@ github.com/jackc/pgconn v1.5.0/go.mod h1:QeD3lBfpTFe8WUnPZWN5KY/mB8FGMIYRdd8P8Jr
 github.com/jackc/pgconn v1.5.1-0.20200601181101-fa742c524853/go.mod h1:QeD3lBfpTFe8WUnPZWN5KY/mB8FGMIYRdd8P8Jr0fAI=
 github.com/jackc/pgconn v1.6.4 h1:S7T6cx5o2OqmxdHaXLH1ZeD1SbI8jBznyYE9Ec0RCQ8=
 github.com/jackc/pgconn v1.6.4/go.mod h1:w2pne1C2tZgP+TvjqLpOigGzNqjBgQW9dUw/4Chex78=
-github.com/jackc/pgconn v1.7.2 h1:195tt17jkjy+FrFlY0pgyrul5kRLb7BGXY3JTrNxeXU=
 github.com/jackc/pgerrcode v0.0.0-20190803225404-afa3381909a6 h1:geJ1mgTGd0WQo67wEd+H4OjFG5uA2e3cEBz9D5+pftU=
 github.com/jackc/pgerrcode v0.0.0-20190803225404-afa3381909a6/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=
@@ -1025,6 +1031,7 @@ github.com/sonatard/noctx v0.0.1/go.mod h1:9D2D/EoULe8Yy2joDHJj7bv3sZoq9AaSb8B4l
 github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
 github.com/sourcegraph/go-diff v0.5.3 h1:lhIKJ2nXLZZ+AfbHpYxTn0pXpNTTui0DX7DO3xeb1Zs=
 github.com/sourcegraph/go-diff v0.5.3/go.mod h1:v9JDtjCE4HHHCZGId75rg8gkKKa98RVjBcBGsVmMmak=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.2.1/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=

pyramid/eviction.go

@@ -0,0 +1,49 @@
+package pyramid
+
+import (
+	"fmt"
+
+	"github.com/dgraph-io/ristretto"
+)
+
+type lruSizeEviction struct {
+	cache *ristretto.Cache
+}
+
+const (
+	// Per ristretto, this is the optimized (static) buffer items
+	bufferItems = 64
+)
+
+func newLRUSizeEviction(capacity, estimatedFileBytes int64, evict func(rPath relativePath)) (*lruSizeEviction, error) {
+	cache, err := ristretto.NewCache(&ristretto.Config{
+		// Per ristretto, this is the optimized counters num
+		NumCounters: int64(10.0 * float64(capacity) / float64(estimatedFileBytes)),
+		MaxCost:     capacity,
+		Metrics:     false,
+		OnEvict:     onEvict(evict),
+		BufferItems: bufferItems,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("creating cache: %w", err)
+	}
+	return &lruSizeEviction{
+		cache: cache,
+	}, nil
+}
+
+func onEvict(evict func(rPath relativePath)) func(uint64, uint64, interface{}, int64) {
+	return func(_, _ uint64, value interface{}, _ int64) {
+		evict(value.(relativePath))
+	}
+}
+
+func (am *lruSizeEviction) touch(rPath relativePath) {
+	am.cache.Get(string(rPath))
+}
+
+func (am *lruSizeEviction) store(rPath relativePath, filesize int64) bool {
+	// must store the path as value since the returned key is the hash,
+	// not the actual key.
+	return am.cache.Set(string(rPath), rPath, filesize)
+}

pyramid/file.go

@@ -0,0 +1,62 @@
+package pyramid
+
+import (
+	"errors"
+	"os"
+)
+
+// File is pyramid wrapper for os.file that triggers pyramid hooks for file actions.
+type File struct {
+	fh       *os.File
+	eviction *lruSizeEviction
+
+	readOnly bool
+	rPath    relativePath
+
+	store func() error
+}
+
+var ErrReadOnlyFile = errors.New("file is read-only")
+
+func (f *File) Read(p []byte) (n int, err error) {
+	if f.readOnly {
+		// file is being written, eviction policies don't apply to it
+		f.eviction.touch(f.rPath)
+	}
+	return f.fh.Read(p)
+}
+
+func (f *File) ReadAt(p []byte, off int64) (n int, err error) {
+	if f.readOnly {
+		// file is being written, eviction policies don't apply to it
+		f.eviction.touch(f.rPath)
+	}
+	return f.fh.ReadAt(p, off)
+}
+
+func (f *File) Write(p []byte) (n int, err error) {
+	if f.readOnly {
+		return 0, ErrReadOnlyFile
+	}
+
+	return f.fh.Write(p)
+}
+
+func (f *File) Stat() (os.FileInfo, error) {
+	return f.fh.Stat()
+}
+
+func (f *File) Sync() error {
+	return f.fh.Sync()
+}
+
+func (f *File) Close() error {
+	if err := f.fh.Close(); err != nil {
+		return err
+	}
+
+	if f.store == nil {
+		return nil
+	}
+	return f.store()
+}

pyramid/params.go

@@ -0,0 +1,15 @@
+package pyramid
+
+// Params is pyramid.FS params that are identical for all file-systems
+// in a single lakeFS instance.
+type Params struct {
+	// AllocatedBytes is the disk size in bytes that lakeFS is allowed to use.
+	AllocatedBytes int64
+
+	// BaseDir is the local directory where lakeFS app is storing the files.
+	BaseDir string
+
+	// BlockStoragePrefix is the prefix prepended to lakeFS metadata files in
+	// the blockstore.
+	BlockStoragePrefix string
+}

pyramid/pyramid.go

@@ -0,0 +1,19 @@
+package pyramid
+
+// FS is pyramid abstraction of filesystem where the persistent storage-layer is the block storage.
+// Files on the local disk are transient and might be cleaned up by the eviction policy.
+// File structure under a namespace and namespace itself are flat (no directories).
+type FS interface {
+	// Store adds the file from the originalPath to the FS. It uploads the file to the
+	// block-storage and to the localpath.
+	// Once completed, it will not be available from the original path.
+	Store(namespace, originalPath, filename string) error
+
+	// Create creates a new file in the FS.
+	// It will only be persistent after the returned file is closed.
+	Create(namespace, filename string) (*File, error)
+
+	// Open finds the referenced file and returns the file descriptor.
+	// If file isn't in the local disk, it is fetched from the block storage.
+	Open(namespace, filename string) (*File, error)
+}