refactor: Migrate from legacy storage implementations to disk.Store and tiered.Store - #658
Open
Anton-Kalpakchiev wants to merge 17 commits into
Open
Anton-Kalpakchiev wants to merge 17 commits into
Anton-Kalpakchiev wants to merge 17 commits into
Conversation
Anton-Kalpakchiev
force-pushed
the
replace-ca-store
branch
3 times, most recently
from
August 17, 2026 20:13
6977955 to
74672a3
Compare
thijmv
reviewed
Aug 18, 2026
Comment on lines
+145
to
+147
| if os.IsExist(err) { | ||
| return nil | ||
| } |
Collaborator
There was a problem hiding this comment.
What if the write to the file fails, and the file is stuck in an incomplete state indefinitely?
Collaborator
Author
There was a problem hiding this comment.
I've been thinking about this. I'll add an async job to cleanup stale leaked blobs. Check this commit description for more reasoning to do it.
| if err == nil { | ||
| blobSize = fi.Size() | ||
| blobSize, err := s.tieredStore.ScopeComplete().Stat(d.Hex()) | ||
| if err != nil { |
Collaborator
There was a problem hiding this comment.
Can we log and return an error when the stat fails?
Collaborator
Author
There was a problem hiding this comment.
i'll add a log, but we shouldn't return an error, as pointed out by the comment:
// Don't fail the commit if replication fails - blob is still uploaded.
The approach to fail-open when local write succeeds, but not replication was there before this commit. I don't want to change that behavior.
Collaborator
There was a problem hiding this comment.
I see, but do we see any reasons why this Stat would fail if the blob has just been uploaded?
Anton-Kalpakchiev
force-pushed
the
replace-ca-store
branch
2 times, most recently
from
August 18, 2026 22:52
029b85c to
7387489
Compare
Comment on lines
+85
to
88
| if len(nameParts) < numShards { | ||
| p.log.With("path", path).Error( | ||
| "invariant violation - cannot reboot blob, as its path is shallower than the configured shard length - failing open by not rebooting it") | ||
| return nil |
Collaborator
There was a problem hiding this comment.
Should this be
len(nameParts) <= numShards ?
| return fmt.Errorf("get digest: %s", err) | ||
| } | ||
| if err := u.cas.CreateCacheFile(d.Hex(), bytes.NewReader(content)); err != nil { | ||
| f, err := u.store.Create(d.Hex(), 1) |
Collaborator
There was a problem hiding this comment.
Why are we only passing 1 here?
Can you create a constant for the value? This appears as a magic number
Comment on lines
+87
to
+96
| err := u.store.RenameKey(uid, d.Hex()) | ||
| if errors.Is(err, os.ErrNotExist) { | ||
| log.With("digest", d.Hex(), "uid", uid).Warn("File not found during commit") | ||
| return handler.ErrorStatus(http.StatusNotFound) | ||
| } | ||
| if errors.Is(err, os.ErrExist) { | ||
| log.With("digest", d.Hex(), "uid", uid).Warn("File is already complete and in store") | ||
| return handler.ErrorStatus(http.StatusConflict) | ||
| } | ||
| err = u.store.MarkComplete(d.Hex()) |
Collaborator
There was a problem hiding this comment.
Could the cleaner be added in the next commit, accidentally clearing a blob that is renamed but not marked complete?
Collaborator
Author
There was a problem hiding this comment.
It only cleans blobs not touched for 5+ minutes that are also incomplete. If it starts cleaning blobs that should not be cleaned (we will detect that through the logs), then we can increase the 5min TTI to reduce false positives.
| uid = uuid.Generate().String() | ||
| f, err := u.store.Create(d.Hex(), size) | ||
| if errors.Is(err, os.ErrExist) { | ||
| if _, ok := u.store.Has(d.Hex()); ok { |
Collaborator
There was a problem hiding this comment.
Why are we not using ScopeComplete().Has() here? Like in patch?
thijmv
reviewed
Aug 20, 2026
| @@ -0,0 +1,75 @@ | |||
| package disk | |||
Collaborator
There was a problem hiding this comment.
For the tiered store, the main path is writing blobs to memory. Blobs can only be incomplete on disk if they are directly written to disk because memory is full. I therefore think we should look into a way to clean up the stale memory writes, as that will happen more frequently.
| }() | ||
| } | ||
|
|
||
| func (s *store) close() { |
Collaborator
There was a problem hiding this comment.
Can we call this when cleaning up the store to avoid a dangling goroutine?
|
|
||
| func (t *testTransferer) Upload(namespace string, d core.Digest, blob store.FileReader) error { | ||
| return t.cas.CreateCacheFile(d.Hex(), blob) | ||
| // Upload caches blob under d, tolerating re-uploads of an already-cached |
Collaborator
There was a problem hiding this comment.
nit: Consider updating the comment to state that re-uploading an already-cached digest is a no-op, as the current wording suggests it would start overriding it.
Currently, the disk.Store assumes that all its use cases will be to store blobs of varying sizes, thus it assumes 1. that its capacity will be in bytes 2. that the blob's client-provided size passed in Create will be in bytes 3. and thus, that when rebooting a blob from disk after a crash, that it can use the blob's real size, as the store operates in bytes. However, one of the use cases that disk.Store will replace is SimpleStore used in build-index, where all files are the same small size (a tag) and thus build-index runs an LRU cache capped at the number of entries, not the total size of all entries, i.e. an unweighted LRU cache. To support that use case, disk.Store must remove its assumption that its size and capacity are all in bytes and accept that their unit might be `number of entries`, i.e. each call to Create will pass `size == 1` and Capacity will represent the total number of entries the store can hold. This commit makes the necessary changes to support the use case: - Ensure that when rebooting a blob, we always look at the client- provided size (we currently look at the blob's real size sometimes) - Change all `sizeBytes` and `capacityBytes` variable names in Store to just `size` and `capacity`. Same applies for metric/log field names - Add a comment in the `Config` struct's `Capacity` field to explain how the store supports both use cases. Also explain how the Capacity limit is soft and not hard, due to Linux semantics. - Add a test for the new behavior. - A few small improvements here and there.
build-index does not properly reboot image tags from disk due to a bug in disk.Store. Blobs whose keys have "/" in their name (e.g. image tags, which build-index uses as keys for the disk.Store) would not be rebooted, as the "/" in their name would get interpreted as a path separator, instead of a part of the blob's key, which the pather didn't account for. This commit fixes that and adds a test that catches the bug.
… disk.Store and tiered.Store Replaces all usage of CAStore and SimpleStore with usage of disk.Store and tiered.Store. I had to migrate origin and build-index in the same commit, as they share some libraries (e.g. writeback) which use CAStore. The next commits will include agent and proxy. Small changes were necessary to support the migration. Most notably: - origin's API to upload blobs now stores each file as it's being uploaded in disk.Store, using its digest as key, whereas before it was the randomly- generated uid that was used. There are some subtle changes in the API's behavior due to this: 1) now multiple clients cannot upload the same blob at the same time. Previously, they were allowed to fully upload the blob, but they would race for success at the commit phase of the API. Now, they race at the start phase instead. 2) if for some reason, clients of the API stop halfway-through, the file will be leaked. I will add a metric to observe and alert on such cases in the next commit. I am thinking of adding a TTL for incomplete blobs, after which they get cleaned, to ensure that they don't get leaked to disk. - add Close method to tiered.Store used for testing. - change the closers.Close log from Debug to Warn level so we can flag any bugs after rollout instead of silently dropping them - delete the forceCleanupHandler in origin, as it is no longer needed (forceCleanupHandlerV2 replaced it) - add fixtures for disk.Store and tiered.Store - base.yaml files had to have their configs changed
Replaces all usage of CADownloadStore with disk.Store. - changed base.yaml as default config has changed. - we no longer use a client-side interface in torrent.go, as we no longer need to mock the store for testing purposes. - made a few improvements to the test code (e.g. instead of panicing on error, pass in `t *testing.T` and assert that there's no error) - removed some dead code in storage driver
TLDR - proxy needs this API to function correctly. When an image is uploaded to proxy, the client does not provided the digest of each blob before sending it. Instead, proxy first caches the blob locally on disk, after which it calculates its digest by parsing it. This means that the `store.Create` call cannot provide the blob's digest as key. To circumvent that, I am adding a RenameKey API that allows for a blob's key to be renamed after it's added to the store. Thus, proxy will 1) generate a random key, 2) store the blob on disk, 3) calculate its digest, 4) rename its key to the digest.
While Kraken code itself never calls Size() after Close() is called, we use a library to expose the Docker Registry V2 API and we sometimes pass disk.File to that library, after which it may call Size() after Close(). This breaks the whole image push flow, as PATCH requests when uploading a blob need to know how far they've gotten with pushing, so they call Size(), after which they pass the resulting offset to the next PATCH request, which uses it to know which bytes to send to the server. Currently, Size() returns 0 if Close() is called beforehand, even if 0 is not the right offset, causing errors for the push flow.
- Replace the legacy CAStore with disk.Store in proxy. Behavior should be unchanged. - Since the disk.Store does not support metadatas with "/" in their name (also called a "suffix"), I refactored the `hashStateMetadata` to use a different suffix. This change is container within the `hashStateMetadata` implementation and does not change any APIs/ contracts/etc. There was even a TODO to make this exact change, which has been there for years. - We keep the same behavior as CAStore to drop incomplete blobs on reboot. There are 2 non-trivial parts of the migration: 1. Leaked blobs on disk - if a proxy client stops the upload mid-way due to a crash or something else, the blob is leaked on disk. CAStore had an async job that cleaned up such leaked files, but disk.Store doesn't. I will add an async job that cleans up in a follow up commit. 2. disk.Store's intent was to provide a capacity in GB for how many blobs should be stored and enforce it through an LRU. Thus, the Create API it exposes takes the blob's size as a param to reserve that capacity upfront and evict other blobs, if needed. However, when Kraken clients upload blobs to proxy, they are not required to provide the size of the blob beforehand. We cannot change this constraint, as it's a part of the Docker Registry V2 API spec. Therefore, we don't know how big the blob will end up being until it's uploaded fully to proxy. I see 2 main ways to get around this: 1) set capacity to be the number of blobs at once in proxy and do `store.Create(key, size==1)` in all calls. 2) assume each blob is some big capacity like 10GB upfront and then change its actual size after the upload is finished. This will require a new API in disk store like `ChangeSize(key string)`. 1) is easier to implement, but since blobs can differ wildly in their sizes, we can only approximate how much disk proxy will end up using, which if not done well can lead to either disk exhaustion OR overreservation and thus underutilization. 2) is harder to implement but since we correct our estimate after the upload is done, the actual capacity set by the config will be respected very closely. I decided to go for the simpler approach 1) for now and see if we need to iterate to 2). Let me know if you agree.
- add missing closers.Close() calls after calling store.Open() or store.Create() - replace os.IsNotFound(err) with errors.Is(err, os.ErrNotFound) - add a WARN log in disk.Store and memory.Store when Delete() deletes a blob banned from eviction - improve some error wrapping messages - add log for when upload to origin succeeds but replicating the writeback task fails - revert the upload behavior to origin to what it was like before the migration to disk.Store. Namely, now a randomly generated uid is used as the key of an in-progress blob being uploaded. Before, I had changed that behavior to instead just directly use the digest, but that has some nuisanced when multiple clients try to upload at the same time so I decided to avoid changing that subtle functionality.
Due to client misuse, subtle bugs, or crashes at the wrong time, it's possible that blobs get leaked in the disk store as incomplete. Since the store's LRU logic only considers complete blobs as evictable, this means the incomplete blobs never get deleted, leaking them. To prevent that, I'm adding an async worker that runs every 10mins and delete incomplete blobs that have not been modified for 5+ mins (both values are configurable through the yaml config). Logs are emitted when we collect leaked files so we can later debug why the files got leaked.
- Make origin use tiered.Store instead of disk.Store upon blob replication. There are 2 reasons to make the change: 1. correctness - tiered.Store uses disk.Store internally with some complex logic to sometimes keep blobs in both and eventually flush data from mem to disk without leaks/undefined behavior. When clients try operating on the same blob through both, undefined behavior can occur, as now the invariants that tiered.Store depends on may no longer be true. For example, tiered.Store always assumes if a blob is incomplete in memory, it can't be on disk yet, as blobs are flushed to disk only after completeness. But if the client manually adds the blob to disk.Store while the blob is in the tiered.Store but incomplete (and thus only in memory.Store), then the blob is duplicated in disk.Store and mem.Store in different states, causing tiered.Store to start behaving incorrectly. This actually happened while testing on staging, so I'm ensuring that all blob usage uses EITHER disk.Store XOR tiered.Store (this commit) to prevent the bug. Also added a comment to document the invariant. 2. performance - when an agent requests a blob from an origin, it will most of the time request it from all origin replicas that own that blob. Thus, as soon as 1 origin gets that requests and downloads the blob from GCS, it replicates the blob to the other origins. This replication right now goes to disk, while the first origin to be hit by the request downloads to tiered.Store. This makes no sense, as all 3 origins do the same thing - seed the blob to agents. Thus, all 3 of them should try downloading to memory first (more performant), and thus use tiered.Store. - Fix a bug where an `if err != nil` check was missing. - Add a few clarifying comments here and there
- Before the CAStore -> tiered.Store/disk.Store migration, when a blob was out of scope, os.ErrNotExist was returned. Now, to allow clients to react more intelligently on blob being out of scope, `store.ErrOutOfScope`` is returned instead. While the change itself is completely ok, when I migrated from CAStore to tiered/disk store, I did not correctly change all `if errors.Is(err, os.ErrNotExist)` checks to `if errors.Is(err, os.ErrNotExist) || if errors.Is(err, store.ErrOutOfScope)` , causing subtle issues. This commit fixes them. - Add Warn logs when MarkComplete is called on an already complete blob as while ok, such usage might indicate incorrect use by clients. - correct a small bug where I forgot to do `.ScopeComplete` in uploader.go - don't emit an error log when the uploader returns StatusConflict ( this is another regression from the CAStore -> disk/tiered store migration) - simplify code here and there by removing unnecessary `store.Has` calls.
Instead of doing `log.Errorf("foo: %w", err)`, we now do
`log.With("error", err).Error("foo")`. This allows for `groupBy message`
functionality when looking at logs from a UI, as now the message is the
same between logs.
While Refresher deduplicates parallel download requests for the same blob, it does not deduplicate sequential requests for the same blob, i.e. download request A comes in and succeeds and download request B comes in afterward. Before, this didn't use to be an issue, as the CAStore would always create a tmp file where the blob would be downloaded, after which it would mark it as complete - if marking as complete failed due to the blob already being present, the tmp blob would be discarded and the error would be ignored. Now, since tiered.Store just tries to do store.Create, it fails-hard on os.ErrExist, instead of swallowing it. This commit fixes that. Also, while correct, CAStore's approach meant that there were redundant downloads from GCS to disk, wasting disk IO - Kraken's bottleneck. The new approach will now prevent that from happening.
Anton-Kalpakchiev
force-pushed
the
replace-ca-store
branch
from
September 17, 2026 13:29
b123998 to
cec444e
Compare
This commit improves Kraken's ability to prevent blob leaks on disk and in mem by: 1. tuning the disk leak collector to have a 1h TTI instead of 5m. This is necessary, as under peak load, it is possible for files not to make progress for a while. 1h should be a more sensible default compared. 2. Preventing leaks by deleting files from disk when we are already 100% certain that a blob will be leaked. For instance, if we create a file, write some data to it, then the writing fails, we already know this file will be leaked, so instead of waiting for the leak collector to delete it 1h later, we can delete it straight away. I've added an `Abort` API that to be reused across the repo for this purpose.
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Now that
disk.Storeandtiered.Storeare implemented, we can finally move away from using CAStore, SimpleStore, etc. and deprecate them. This PR implements the migration with a subsequent PR to follow to delete all the no longer necessary legacy code.For more context, check #633.
Please review commit by commit, as there will be a few extra necessary changes in this PR!