Proposal: Uncommitted GC

[N-o-Z]

Closes #1933

[itaiad200]

I think you're on to something, great work! See my comments inline.

[itaiad200]

What is a Branch Object and what is a Repository Object? I think we need terminology section.

[N-o-Z]

Rephrased - let me know if it clarifies it - or that I should add a terminology section

[itaiad200]

Will a standalone lakeFS storageNamespace structure look different than a scaled lakeFS? What if the container restarts, should it use the same unique prefix?

[N-o-Z]

Effectively it will look the same.
Each lakeFS instance will maintain it's own partitions appending a unique suffix to the partition name, in the following manner:
<sortable_descending_serialized_uid>_<lakefs_instance_uid>
Each instance will manage an exclusive partition - on container restart the lakeFS instance will created a new partition according to the above conventions.
The current idea is to track partition size in memory - this means a restarted instance does not have context on the existing partitions, effectively leaving partitions partially allocated. If you think this is not sufficient we might want to think of an additional solution on how to track the partitions

[N-o-Z]

Updated document

[itaiad200]

Is it a hard bound? Might be hard to implement.

[N-o-Z]

As mentioned above this is an upper bound, which might not be fulfilled in the case of the instance restart / shutdown

[arielshaqed]

It still requires serializing all uploads.

[itaiad200]

Per branch? Where?

[N-o-Z]

Per branch - updated

[itaiad200]

Noice

[itaiad200]

Suggested change

	2. Read addresses from branch's new commits (all commits up to the last GC run commit id) -> `lakeFS DF`
	2. Read addresses from branch's new commits (all commits from to the last GC run commit id) -> `lakeFS DF`

[N-o-Z]

Since log commits returns commits from newest to last, shouldn't it be "down to"?

[itaiad200]

Suggested change

	1. Read all objects on branch path up to the previous run's last read partition (can be done in parallel by 'partition') -> `Branch DF`
	1. Read all objects on branch path up from the previous run's last read partition (can be done in parallel by 'partition') -> `Branch DF`

[N-o-Z]

It's actually down to the previous run's last read partition, since we are utilizing the adapter's list property to retrieve the entries sorted alphabetically

[ozkatz]

Great ideas @N-o-Z! This is a great direction to explore.
I'm wondering:

1. Why are we scoping each "partition" with a branch? Isn't it simpler to simply have all partitions live under the storage namespace directly?
2. The concept of partitions is really nice! Perhaps we can even start with a simpler implementation that isn't incremental? because partitions are more-or-less bounded in size, you've also solved prior design's limitation with object listing that couldn't be parallelized! do a prefix listing once, then have the executors divvy up these prefixes (with little to no skew since they are all pretty much the same size!)

still O(n) but at least it doesn't have to execute serially..

[itaidavid]

What's DF?

[N-o-Z]

Data Frame - since eventually this design is designated to be implemented over a Spark job, I found using this terminology the most sensible

[talSofer]

How about using committedDF? It is more self explanatory

[itaidavid]

Create a new what? Prefix (i.e. unique prefix?)

[N-o-Z]

Rephrased - let me know if it's clearer

[itaidavid]

It is. Thanks

[N-o-Z]

Great ideas @N-o-Z! This is a great direction to explore. I'm wondering:

1. Why are we scoping each "partition" with a branch? Isn't it simpler to simply have all partitions live under the storage namespace directly?
2. The concept of partitions is really nice! Perhaps we can even start with a simpler implementation that isn't incremental? because partitions are more-or-less bounded in size, you've also solved prior design's limitation with object listing that couldn't be parallelized! do a prefix listing once, then have the executors divvy up these prefixes (with little to no skew since they are all pretty much the same size!)

still O(n) but at least it doesn't have to execute serially..

Thank you @ozkatz

1. scoping the "partitions" to branches allows us to scope our views of the committed and uncommitted objects to the branch level as well. It in fact allows us to run the GC process on the branch level and enables the concurrency. If we created the partitions on the repository scope (namespace), we would have to go over all committed and uncommitted data for the entire namespace in order to ascertain which objects can be deleted.
2. We can definitely start with the clean flow, and add the optimized flow later.
   Though we need to take several things into mind:
   1. Even when parallelized, testing S3 listing of a path with ~500,000 objects on a databrick notebook running on the given configuration took around 25 seconds:
      
      taking for example a repo with around 1 billion objects (not so far fetched as I understand), even if it is perfectly partitioned into bulks of 500,000 objects, the entire listing might take an unreasonable amount of time
   2. While relying only on the clean flow this becomes is a continually growing problem, both the bucket size and commits are unbounded and eventually we will reach an unmanageable size

[itaidavid]

If I understand correctly, it`s a part of our current GC, right? Deleted-uncommited and expired-committed will be deleted in the same run?

[N-o-Z]

I believe it should be incorporated into the same ecosystem. Whether it is the same job, or a different one is to be considered. I'm not sure deciding this now is a requirement for this proposal

[talSofer]

@N-o-Z let's make sure we do discuss this detail before settling on a final design. I'm fine with separating the algorithmic discussion from job architecture but there are trade offs we need to consider here.

[itaidavid]

So, this is kinda "fake" commit with all uncomitted-undeleted objects, correct?
Pretty cool

[N-o-Z]

Credit to @itaiad200 😄

[itaidavid]

Can we improve it by limiting GC cycles (time limit or number of objects to handle)?

[N-o-Z]

This will be a bit difficult IMO. Since object creation date is loosely coupled with commit date it might be very complicated to achieve

[arielshaqed]

This is an ops issue. I might prefer occasional long runs, or common short runs.

[N-o-Z]

Rephrased

[itaidavid]

Thank you @N-o-Z. If I understood correctly you suggest to treat all uncommitted (undeleted) objects as a retained commit, and delete all objects that are not a part of this or any other reatined commit, correct?
What I mainly miss in this document is a descitption of the approach, before the dive into detail - some story of the solution behind the details. I think it will help understanding the solution and will make it an eaiser read.

[arielshaqed]

Thanks! It looks really exciting, but I didn't entirely understand what "partitions" are.

[arielshaqed]

I don't understand why the race here (between token valid and entry creation) is not a problem: what if the instance verifies a valid token, then the token becomes invalid, and then the entry is created?

In particular, "time interval" is often a race -- particularly in difficult conditions when cluster members have poor time sync.

[johnnyaug]

Yes, time sync introduces a challenge here. What if GC thinks a token has expired while the server thinks it's valid?

[arielshaqed]

It's not a commit, it's "just" a metarange.

[arielshaqed]

What are the advantages of doing this over directly providing an API to list the branch objects? If the goal is "merely" to get a static listing, could we instead provide an API to push a new staging token and return the previous list of staging tokens? Then the GC Spark job could use another (new) API call list all objects on that list of staging tokens.

Otherwise we end up with a strange unused metarange. 🤷🏽

[N-o-Z]

This metarange is used not only for the current GC run, but used later in the next run as a reference to the current state of the branch and enables an important optimization. Afterwards it can be safely removed.
Also I don't think the sealed tokens provide us with a static state - changes can still occur on the branch which can modify the sealed tokens state.

[arielshaqed]

I don't think that that works: these are all addresses to keep, they might occur only on another branch (e.g. if copied from another branch by staging them to this one).

[arielshaqed]

Can we have a slower (one-time) run to cleanup old repositories, once?

[N-o-Z]

I think we won't have any option but to do that :). I'm just not sure this should be a part of this proposal. WDYT?

[arielshaqed]

In a separate proposal is great, just let's have it :-)

[N-o-Z]

Added a discovery task:
#4382

[arielshaqed]

Where do we store this "run ID"?

[N-o-Z]

Added:

8. Finally, save the current run's GC commit the last read partition and newest commit id in a designated location on the branch path

[arielshaqed]

It still requires serializing all uploads.

[arielshaqed]

Not sure what these partitions are: Staging tokens? Are different instances allowed to scan and use other instances' partitions?

Can you add some motivation for this? It raises some questions.

• How "ascending" does this need to be? Would a timestamp, which is not really ascending but generally satisfies this, be good enough?
• How "unique" does this need to be?

[johnnyaug]

What kind of race? Is this the token thing?

[N-o-Z]

By not allowing StageObject on addresses inside the repo namespace, we prevent a race where an un-referenced address is being staged using StageObject while it is being deleted by GC

[johnnyaug]

Yes, time sync introduces a challenge here. What if GC thinks a token has expired while the server thinks it's valid?

[johnnyaug]

Note that today, the storage namespace is tied to a repository - and not the installation.
So partitioning according to repository is irrelevant.

[N-o-Z]

You are right - but we need to take into consideration scenarios where a repository is deleted and created again with the same name. Although it has the same name, the unique identifier is different and we need to take that into consideration.
Perhaps you have a better suggestion on how to do this?

[johnnyaug]

Are you referring to cleaning up the data of deleted repositories? If so, I think we can relax this requirement.

I don't necessarily object to a "global storage namespace": I think it's a great idea. It's just that we may want to avoid it because it's too big at this point. If we choose to go this way, it should be mentioned explicitly in this doc (and have a design of its own when the time comes).

[itaiad200]

Even today, you can't* create a repository in a storage namespace that serves as a storage namespace for some other repo. So it doesn't matter if the repo is deleted, created by an installation that is no longer active, etc, you can't recreate the repo in the same storage namespace unless it was cleaned.

• Repo creation checks for the dummy object in the root of the namespace. You can argue that it's not safe enough, but I think we can rely on it for the time being.

[johnnyaug]

I didn't understand what we are trying to mitigate here.

[johnnyaug]

I think partitioning according to branch is a brilliant idea, but I'm wondering whether it's premature optimization. It does introduce some complexity and I'm not sure we have evidence that it will benefit us in the real world.

[N-o-Z]

This allows us to scope the GC process only to the commits and staging area that are related to the branch. Otherwise, we will have to read all the information from all commits and all staging areas to understand what we need to delete.

[johnnyaug]

I don't understand if this is required for correctness or just an optimization.

[johnnyaug]

Is this a listing on the object store? In what sense is it parallel?

[N-o-Z]

Since the objects on a given branch a divided into partition prefixes we can use workers to list objects by partition and aggregate the result

[johnnyaug]

Why is the instance ID a part of the partition?

[N-o-Z]

Each instance writes only to its own partition under the branch prefix. This allows the lakeFS instance to track the amount of files uploaded to this partition and decide when to create a new partition

[nopcoder]

Comments in the body - think this is the way, just need to extend the description to how we handle delete of uncommitted data also above the branch level. (branch delete and repository delete).

[nopcoder]

Suggest explaining the idea before the required changes to implement it. Or at least per section explain the reasoning on why we suggest the change so the implementor can understand the goal and not the how.

[talSofer]

+1, I'm missing context to understand the suggested changes. It will add clarity if we explain the overall flow at a high level, possibly with a diagram, before diving into the separation between lakeFS server changes and Spark client changes.
Also, it will be useful to describe the idea behind the solution - e.g. the input for the algorithm is committed, uncommitted, objects ever written to lakefs, the algorithm aims to find objects that are in 3 and not in 1 or 2.

[nopcoder]

Need to explain the constraint and/or if we are going to remove/change any current functionality or just enforce existing one.

[nopcoder]

Is this item a requirement from lakefs? or just explaining that the result of the previous requirement.

[nopcoder]

Can you explain which copy object do you address - is it the option to do so using the S3 gateway? a new OpenAPI that we need to add to address the fact that we can't stage object from staging?
In case this is S3 only - we need to explain how we do the same using the OpenAPI

[nopcoder]

Suggested change

	Uncommitted data which is no longer referenced (due to branch deletion, reset branch etc.) is not being deleted by lakeFS.
	Uncommitted data which is no longer referenced (due to any staged data deletion or override, reset branch etc.) is not being deleted by lakeFS.

[nopcoder]

Consider moving item two to be first - as it will require active tracking using storage, when we have storage we can also enforce any validation while using the link without embedding information on the path itself.

[nopcoder]

• Prefer to have a new folder for each request in order to have history, support parallel request, prevent data delete or keeping old files around. And/or we should keep a state of the current request, progress, completion and errors.
• A risk here will be that a single instance is responsible to extract all the staging information. The request can take a lot of time, it is handled by single instance that can go down and we will need to start from scratch.
• From the work Yoni did, we should consider write this data in a parquet format.

[arielshaqed]

Can we instead make this a paging call by passing the client some continuation token? I know that it will be slower for the client to read, but reducing load on the server may be more important. @lynnro314 has already done some research for how to speed up prepareGCCommits, I am not sure we want to open a new front with an API that may require similar work almost immediately.

[nopcoder]

Just double-checking because the design talked about branch level - this is one set of files per repository?

[nopcoder]

I think that the solution should include - how we delete uncommitted data at the 3 levels:

1. delete repository
2. delete branch
3. delete object

[nopcoder]

We should address create/delete/create of the same branch by using unique id also in the branch level.

[N-o-Z]

Thank you everyone for the most invaluable input! 🙏🏽
We've decided to try and modify this proposal to remove the use of branch prefixes and use a time based partition only.
Please hold off any new comments until I'm able to address the current ones and modify to proposal.
I'll update once I've made all the required changes - Thanks!

[talSofer]

@N-o-Z thanks for this great proposal!
I'm posting my comments since they are unrelated to the changes you are planning to the document.

[talSofer]

+1, I'm missing context to understand the suggested changes. It will add clarity if we explain the overall flow at a high level, possibly with a diagram, before diving into the separation between lakeFS server changes and Spark client changes.
Also, it will be useful to describe the idea behind the solution - e.g. the input for the algorithm is committed, uncommitted, objects ever written to lakefs, the algorithm aims to find objects that are in 3 and not in 1 or 2.

[talSofer]

How about using committedDF? It is more self explanatory

[talSofer]

@N-o-Z let's make sure we do discuss this detail before settling on a final design. I'm fine with separating the algorithmic discussion from job architecture but there are trade offs we need to consider here.

[talSofer]

here to add clarity I suggest adding the intent behind each step, i.e.

1. mark uncommitted data
2. get all committed addresses
3. get all uncommitted addresses
4. get all objects ever written to lakeFS on that path...
5. subtract committed from all objects on the branch
   etc...

[talSofer]

To do that, you may want to use the run ID concept the existing GC has. See https://github.com/treeverse/cloud-controlplane/blob/main/design/accepted/gc-with-run-id.md in this context. This is our plan for implementing incremental GC, and I suggest that we consider using the same concept.

[itaiad200]

I'm missing the actual requirement below. It is sort of an estimation of a single task of the uncommitted GC. How long will the entire GC run? What's the repo status (like number of commits, branches, objects to delete, etc.)

[N-o-Z]

Done

[Jonathan-Rosenberg]

An object that's in case 3 means that there's an active branch that holds the uncommitted object, or is it any uncommitted object under that repository ("dangling" or not)?

[N-o-Z]

An active branch that holds the uncommitted object.
We read the committed + uncommitted data from lakeFS via metaranges, so basically it's a static view of the repository (from lakeFS's point of view) in a specific point in time

[itaidavid]

How do we handle failures in such a long run - say, a crash after an hour of execution?
Does the entire process need to be rerun?
Maybe we can parallelize the GC itself, instead of just the listing?

[N-o-Z]

GC job will be parallelized, but we can't rely on failed runs. To get a correct picture of the namespace the process must run successfully and without errors.
Relying on partial data, and failed jobs will increase risk of deleting committed data - which is something that absolutely must not happen

[itaidavid]

Thanks @N-o-Z
Added a question regarding the impact of an error in such a long process, but other than that LGTM.

[itaiad200]

Correct me if I'm wrong @treeverse/ecosystem, but parquet outputs are multiple files (per column) and it's normally saved under a prefix of it's own for better separation, right?
So a better path would be _lakefs/gc/run_id/uncommitted/entries.parquet

[N-o-Z]

Moved proposal to "accepted" folder.
Please take the time to give final notes - before I merge this PR

[johnnyaug]

Why does it matter that the token is valid for a single use?

[N-o-Z]

See: #4438

[arielshaqed]

This adds an operational requirement for time synchronization between lakeFS instances that belong to the same cluster. I am afraid of adding these because they can be hard to achieve.

As just one example, consider a physical machine in an on-prem cluster that has just booted after a long while down: if its clock is ahead of real time, NTP will take a while to slip it back to the correct value, and in the meantime lakeFS will be up and happily destroying our assumptions :-(

[arielshaqed]

Also on AWS awslabs/amazon-eks-ami#249 is a recurring issue in AMIs, and a linked issue has someone with a 7 minute time skew on EKS.

Clocks should be synced, but I think we will exact too strong a penalty for failure here. We should add:

1. Some safety mechanism.

   For instance:

   • If when we scan we discover prefixes from the distant future (>1 minute?) then error out and mark some metric for alerting.
   • Create an additional marker object in every partition once an hour, and scan what the other cluster members put in theirs. This gives an estimate of time skew, which can be monitored.

2. Operational guidance (strongly worded documentation).

[itaiad200]

Found a bug :(

[itaiad200]

I think there's a bug.
I start preparing the uncommitted objects. During that time a Copy is happening on the same branch, making it a metadata operation. The copy copies from the path zz to the path aa. The preparation of the uncommitted started before aa was created, but by the time it reaches zz it was already deleted. Now I don't see neither of them in the output file and the physical object may be deleted by the GC.

[N-o-Z]

Thanks! Addressed in proposal

[itaiad200]

Thanks for addressing the comments, I added a few more but wish not to further block.

[itaiad200]

The timing here is critical - it MUST read the new table after it iterated on the staging area of all branches.

[N-o-Z]

Explained in PrepareUncommittedForGC - added additional note to emphasize

[itaiad200]

So it needs to maintain that new table too, right?

[N-o-Z]

Added

[itaiad200]

@N-o-Z Still relevant - If we have a way to logically do a rename with copy and delete, why do we need a non-atomic rename in the API?

[arielshaqed]

Thanks!

Really trying to avoid the namespace reorg, it sounds expensive :-(

[arielshaqed]

Can we use a larger instance, to see if we are blocked on network or on something else?

[N-o-Z]

Checked with an xlarge cluster and max workers 10.
Did some more tests and created a udf which reads a specific partition and returns the objects information.
For 1,000,000 objects divided equally into 100 partitions I was able to load the information into a DF within 30 seconds.
The results looks pretty promising - and I'm sure we can also improve on that.
I will update the proposal with final results this week

[arielshaqed]

Suggest not fixing a format for now: it is far from clear which format is best to use, but luckily this looks like an implementation detail that is easy to change.

[N-o-Z]

I agree this is an implementation detail and will remove the format specific details from the proposal

[arielshaqed]

This adds an operational requirement for time synchronization between lakeFS instances that belong to the same cluster. I am afraid of adding these because they can be hard to achieve.

As just one example, consider a physical machine in an on-prem cluster that has just booted after a long while down: if its clock is ahead of real time, NTP will take a while to slip it back to the correct value, and in the meantime lakeFS will be up and happily destroying our assumptions :-(

[arielshaqed]

Not sure that I understand. How does lakeFS know that GC actually ran?

[arielshaqed]

Copy operation? Not sure lakeFS can support atomic rename.

[N-o-Z]

We are tracking copy operations but the purpose is to avoid a race in rename.
Consider the following:

1. GC starts reading branch staging area
2. Rename operation create a copy of staged entry - which doesn't not get listed due to listing order
3. Rename operation deletes original entry before GC is able to scan it as part of the listing of staging area

As a result, the physical address will not be listed in uncommitted (or committed) and will be candidate for deletion.
Reading from the copy table - allows exempting copied objects from the delete list and avoid this race.

The copied entries lifespan can be minutes-hour or can be removed as part of a branch operation (commit/reset/delete branch)

In the next GC iteration - these addresses will either be in committed, still in staging or deleted and will be handled accordingly

[arielshaqed]

How long is this? The shorter it is, the greater the requirement for time sync between lakeFS instances. So it seems like it needs to be fairly large.

[arielshaqed]

Neat!

I'm only worried about the operational aspect of maintaining synchronized times across the cluster of lakeFS servers. But I believe that we can observe the time skew automatically, even as part of time-based partitioning!

Given that we're talking about deleting files, we should be very sure that we help operators control time sync. Ideally fail-shut and refuse to GC uncommitted or even work at all when time sync is not good enough. (I'd talking about minutes-level time sync here, of course. NTP gives you milliseconds-level time sync. But virtualization is notorious for screwing up time sync occasionally.)

[arielshaqed]

Also on AWS awslabs/amazon-eks-ami#249 is a recurring issue in AMIs, and a linked issue has someone with a 7 minute time skew on EKS.

Clocks should be synced, but I think we will exact too strong a penalty for failure here. We should add:

1. Some safety mechanism.

   For instance:

   • If when we scan we discover prefixes from the distant future (>1 minute?) then error out and mark some metric for alerting.
   • Create an additional marker object in every partition once an hour, and scan what the other cluster members put in theirs. This gives an estimate of time skew, which can be monitored.

2. Operational guidance (strongly worded documentation).