Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
First version of the immutable tiered storage
- Loading branch information
Showing
11 changed files
with
885 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| package pyramid | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| lru "github.com/treeverse/golang-lru" | ||
| "github.com/treeverse/golang-lru/simplelru" | ||
| ) | ||
|
|
||
| // eviction is an abstraction of the eviction control for easy testing | ||
| type eviction interface { | ||
| touch(rPath relativePath) | ||
| store(rPath relativePath, filesize int64) int | ||
| } | ||
|
|
||
| type lruSizeEviction struct { | ||
| cache simplelru.LRUCache | ||
| } | ||
|
|
||
| func newLRUSizeEviction(capacity int64, evict func(rPath relativePath)) (eviction, error) { | ||
| cache, err := lru.NewWithEvict(capacity, func(key interface{}, _ interface{}, _ int64) { | ||
| evict(key.(relativePath)) | ||
| }) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("creating cache: %w", err) | ||
| } | ||
| return &lruSizeEviction{ | ||
| cache: cache, | ||
| }, nil | ||
| } | ||
|
|
||
| func (am *lruSizeEviction) touch(rPath relativePath) { | ||
| // update last access time, value is meaningless | ||
| am.cache.Get(rPath) | ||
| } | ||
|
|
||
| func (am *lruSizeEviction) store(rPath relativePath, filesize int64) int { | ||
| return am.cache.Add(rPath, nil, filesize) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package pyramid | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "os" | ||
| ) | ||
|
|
||
| // File is pyramid wrapper for os.file that triggers pyramid hooks for file actions. | ||
| type File struct { | ||
| fh *os.File | ||
|
|
||
| closed bool | ||
| persisted bool | ||
| store func(string) error | ||
| } | ||
|
|
||
| func (f *File) Read(p []byte) (n int, err error) { | ||
| return f.fh.Read(p) | ||
| } | ||
|
|
||
| func (f *File) ReadAt(p []byte, off int64) (n int, err error) { | ||
| return f.fh.ReadAt(p, off) | ||
| } | ||
|
|
||
| func (f *File) Write(p []byte) (n int, err error) { | ||
| 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 { | ||
| f.closed = true | ||
| return f.fh.Close() | ||
| } | ||
|
|
||
| var ( | ||
| errAlreadyPersisted = fmt.Errorf("file is already persisted") | ||
| errFileNotClosed = fmt.Errorf("file isn't closed") | ||
| ) | ||
|
|
||
| // Store copies the closed file to all tiers of the pyramid. | ||
| func (f *File) Store(filename string) error { | ||
| if err := validateFilename(filename); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if f.persisted { | ||
| return errAlreadyPersisted | ||
| } | ||
| if !f.closed { | ||
| return errFileNotClosed | ||
| } | ||
|
|
||
| err := f.store(filename) | ||
| if err == nil { | ||
| f.persisted = true | ||
| } | ||
| return err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| package pyramid | ||
|
|
||
| import ( | ||
| "io/ioutil" | ||
| "os" | ||
| "path" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/google/uuid" | ||
| ) | ||
|
|
||
| func TestPyramidWriteFile(t *testing.T) { | ||
| filename := uuid.New().String() | ||
|
|
||
| fh, err := ioutil.TempFile("", filename) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| filepath := fh.Name() | ||
| defer os.Remove(filepath) | ||
|
|
||
| storeCalled := false | ||
| sut := File{ | ||
| fh: fh, | ||
| store: func(string) error { | ||
| storeCalled = true | ||
| return nil | ||
| }, | ||
| } | ||
|
|
||
| content := "some content to write to file" | ||
| n, err := sut.Write([]byte(content)) | ||
| require.Equal(t, len(content), n) | ||
| require.NoError(t, err) | ||
| require.NoError(t, sut.Sync()) | ||
|
|
||
| _, err = sut.Stat() | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, sut.Close()) | ||
| require.NoError(t, sut.Store(filename)) | ||
|
|
||
| require.True(t, storeCalled) | ||
| } | ||
|
|
||
| func TestWriteValidate(t *testing.T) { | ||
| filename := uuid.New().String() | ||
| fh, err := ioutil.TempFile("", filename) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| filepath := fh.Name() | ||
| defer os.Remove(filepath) | ||
|
|
||
| storeCalled := false | ||
|
|
||
| sut := File{ | ||
| fh: fh, | ||
| store: func(string) error { | ||
| storeCalled = true | ||
| return nil | ||
| }, | ||
| } | ||
|
|
||
| content := "some content to write to file" | ||
| n, err := sut.Write([]byte(content)) | ||
| require.Equal(t, len(content), n) | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, sut.Close()) | ||
| require.Error(t, sut.Store("workspace"+string(os.PathSeparator))) | ||
| require.False(t, storeCalled) | ||
|
|
||
| require.Error(t, sut.Close()) | ||
| require.NoError(t, sut.Store("validfilename")) | ||
| require.Error(t, sut.Store("validfilename")) | ||
| } | ||
|
|
||
| func TestPyramidReadFile(t *testing.T) { | ||
| filename := uuid.New().String() | ||
| filepath := path.Join("/tmp", filename) | ||
| content := "some content to write to file" | ||
| if err := ioutil.WriteFile(filepath, []byte(content), os.ModePerm); err != nil { | ||
| panic(err) | ||
| } | ||
| defer os.Remove(filepath) | ||
|
|
||
| mockEv := newMockEviction() | ||
|
|
||
| fh, err := os.Open(filepath) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| sut := ROFile{ | ||
| fh: fh, | ||
| eviction: mockEv, | ||
| rPath: relativePath(filename), | ||
| } | ||
|
|
||
| _, err = sut.Stat() | ||
| require.NoError(t, err) | ||
|
|
||
| bytes := make([]byte, len(content)) | ||
| n, err := sut.Read(bytes) | ||
| require.NoError(t, err) | ||
| require.Equal(t, len(content), n) | ||
| require.Equal(t, content, string(bytes)) | ||
| require.NoError(t, sut.Close()) | ||
|
|
||
| require.Equal(t, 2, mockEv.touchedTimes[relativePath(filename)]) | ||
| } | ||
|
|
||
| type mockEviction struct { | ||
| touchedTimes map[relativePath]int | ||
| } | ||
|
|
||
| func newMockEviction() *mockEviction { | ||
| return &mockEviction{ | ||
| touchedTimes: map[relativePath]int{}, | ||
| } | ||
| } | ||
|
|
||
| func (me *mockEviction) touch(rPath relativePath) { | ||
| me.touchedTimes[rPath]++ | ||
| } | ||
|
|
||
| func (me *mockEviction) store(_ relativePath, _ int64) int { | ||
| return 0 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| 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 { | ||
| // Create creates a new file in the FS. | ||
| // It will only be persistent after the returned file is stored. | ||
| Create(namespace string) (*File, error) | ||
|
|
||
| // Open finds the referenced file and returns its read-only File. | ||
| // If file isn't in the local disk, it is fetched from the block storage. | ||
| Open(namespace, filename string) (*ROFile, error) | ||
| } |
Oops, something went wrong.