GC & multitenancy #712
GC & multitenancy #712
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| c2: {ffs.APIID("ID3")}, | ||
| } | ||
| txn, _ := ds.NewTransaction(false) | ||
| err := migrateJobLogger(txn, cidOwners) |
There was a problem hiding this comment.
We run this sub-migration on the datastore.
| require.NoError(t, err) | ||
| require.NoError(t, txn.Commit()) | ||
|
|
||
| post(t, ds, "testdata/v1_JobLogger.post") |
There was a problem hiding this comment.
We assert that the end result of the datastore state is the same as described in v1_JobLogger.post.
This style of doing test on migrations is quite easy since most of it is creating the pre and post state files.
| "github.com/textileio/powergate/tests" | ||
| ) | ||
|
|
||
| func TestEmptyDatastore(t *testing.T) { |
There was a problem hiding this comment.
Here some tests of general Migrator library.
In this test we check the case of a new pristine Powergate. In this case, only the version will be set to the latest known version since doesn't make sense to run any migration.
| require.Equal(t, 1, v) | ||
| } | ||
|
|
||
| func TestNonEmptyDatastore(t *testing.T) { |
There was a problem hiding this comment.
Test the case of a Powergate already with data, but running the migration for the first time.
(e.g: all hosted or Hub powergate).
| require.Equal(t, 1, v) | ||
| } | ||
|
|
||
| func TestNoop(t *testing.T) { |
There was a problem hiding this comment.
Test the case of the migration running when the version is already the latest.. so most cases in general when Powergate is restarted or no new migrations exist.
| if currentVersion > targetVersion { | ||
| return fmt.Errorf("migrations are forward only, current version %d, target version %d", currentVersion, targetVersion) | ||
| } |
There was a problem hiding this comment.
Currently, this Migrator does forward only migrations. So check here for a situation in which the user might want to go back to an older tag of Powergate which isn't compatible with a newer datastore.
We can extend this lib to do forward and backward migrations. Now I did only forward since doesn't make sense for these big migrations of the PR to be rollbacked since that's impossible to do correctly.
We could define Migrations as rollback-able or not, and potentially allow backward migrations depending if the migration has a backward logic or not. That would be easy to do. Mentioning it just to consider for the future.
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
jsign
force-pushed
the
jsign/sfgc4
branch
7 times, most recently
from
54ad084
to
b28ac75
Compare
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
| @@ -4,7 +4,7 @@ RUN mkdir /app | |||
| WORKDIR /app | |||
|
|
|||
| COPY go.mod go.sum ./ | |||
| RUN go mod download | |||
| RUN go mod download -x | |||
There was a problem hiding this comment.
Since downloading deps can take some time, I prefer seeing detailed output about what's being donwloaded.
api/server/server.go
Outdated
| if err := runMigrations(conf); err != nil { | ||
| return nil, fmt.Errorf("running migrations: %s", err) | ||
| } |
There was a problem hiding this comment.
Before spinning internal components, we run migrations logic.
| l := joblogger.New(txndstr.Wrap(ds, "ffs/joblogger")) | ||
| l := joblogger.New(txndstr.Wrap(ds, "ffs/joblogger_v2")) |
There was a problem hiding this comment.
More about the need of suffixing migrated key namespaces soon.
| @@ -537,15 +537,19 @@ func (s *Server) Close() { | |||
| } | |||
| } | |||
|
|
|||
| func createDatastore(conf Config) (datastore.TxnDatastore, error) { | |||
| func createDatastore(conf Config, longTimeout bool) (datastore.TxnDatastore, error) { | |||
There was a problem hiding this comment.
We enable a new flag longTimeout which returns a datastore that has long timeouts for operations and transactions.
Basically, the datastore that Powergate uses for operating has false value. For migrations, we create use a true value to allow for bigger timeouts since we do long-running operations.
| repoPath, err := ioutil.TempDir("/tmp/powergate", ".powergate-*") | ||
| require.NoError(t, err) | ||
|
|
||
| repoPath := t.TempDir() |
There was a problem hiding this comment.
In Go 1.15, we can use (*testing.T).TempDir() which automatically cleans the folder when the test finishes, so no need to leave folders in tmp, or return "cls" funcs, and also deletes the folder even if the test panics.
We can use this in lots of places, just changed here.
|
|
||
| return owners, nil | ||
| } | ||
|
|
There was a problem hiding this comment.
All below v0 functions are helpers for the migration stuff, which resolves some important questions for migrations:
- Which iids exists.
- Which iids own a Cid
- Get the storage config of a Cid for a iid.
These methods should be agnostic of internal components, since if that is the case we could handcuff improving internal components in the future since migrations rely on them.
| require.Equal(t, 0, v) | ||
| } | ||
|
|
||
| func TestRealDataBadger(t *testing.T) { |
There was a problem hiding this comment.
I exported the go-datastore of one of our internal Powergate stacks locally in a Badger datastore.
And then just tested e2e with that real datastore that the migration runs correctly.
| func TestRealDataRemoteMongo(t *testing.T) { | ||
| t.SkipNow() |
There was a problem hiding this comment.
This test case was mostly to test the migration directly in copied datastores of many live hosted powergates.
Disabled the test since doesn't make sense for CI.
| func pre(t *testing.T, ds datastore.TxnDatastore, path string) { | ||
| t.Helper() | ||
| f, err := os.Open(path) | ||
| require.NoError(t, err) | ||
| defer func() { require.NoError(t, f.Close()) }() | ||
|
|
||
| s := bufio.NewScanner(f) | ||
| for s.Scan() { | ||
| parts := strings.SplitN(s.Text(), ",", 2) | ||
| err = ds.Put(datastore.NewKey(parts[0]), []byte(parts[1])) | ||
| require.NoError(t, err) | ||
| } | ||
| } | ||
|
|
||
| func post(t *testing.T, ds datastore.TxnDatastore, path string) { | ||
| t.Helper() | ||
|
|
||
| f, err := os.Open(path) | ||
| require.NoError(t, err) | ||
| defer func() { require.NoError(t, f.Close()) }() | ||
|
|
||
| current := map[string][]byte{} | ||
| q := query.Query{} | ||
| res, err := ds.Query(q) | ||
| require.NoError(t, err) | ||
| defer func() { require.NoError(t, res.Close()) }() | ||
| for r := range res.Next() { | ||
| require.NoError(t, r.Error) | ||
| current[r.Key] = r.Value | ||
| } | ||
|
|
||
| expected := map[string][]byte{} | ||
| s := bufio.NewScanner(f) | ||
| for s.Scan() { | ||
| parts := strings.SplitN(s.Text(), ",", 2) | ||
| expected[parts[0]] = []byte(parts[1]) | ||
| } | ||
|
|
||
| require.Equal(t, len(expected), len(current)) | ||
| for k1, v1 := range current { | ||
| v2, ok := expected[k1] | ||
| require.True(t, ok) | ||
| require.Equal(t, v2, v1) | ||
| } | ||
| } |
There was a problem hiding this comment.
pre is a function that reads a csv files with key,values as a initial state of a datastore for tests.
post is a function that asserts if a datastore has the same state as a csv file with key,values.
| @@ -0,0 +1,3 @@ | |||
| /ffs/joblogger_v2/ID1/QmPewMLNZEgnLxaenjo9Q5qwQwW3zHZ7Ac973UmeJ6VWHE/1602952162298722885,{"Cid":{"/":"QmPewMLNZEgnLxaenjo9Q5qwQwW3zHZ7Ac973UmeJ6VWHE"},"RetrievalID":"","Timestamp":1602952162298722885,"Jid":"ad6da2a0-5465-4275-a330-2537062765c8","Msg":"Deal 763168 with miner f022142 is active on-chain"} | |||
There was a problem hiding this comment.
Here the pre and post files for each test case.
A new datastore is populated with .pre file key-values, the migration is run, and then it's asserted if the datastore is exactly in the .post key-values state.
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
| @@ -34,8 +34,7 @@ type Store struct { | |||
| ds datastore.Datastore | |||
| watchers []watcher | |||
|
|
|||
| queued []ffs.StorageJob | |||
| executingCids map[cid.Cid]ffs.JobID | |||
There was a problem hiding this comment.
I deleted executingCids since now that we're multitenant I could simply use the executingJobs cache.
So one less cache to maintain.
| // We can already clean resumed started deals. | ||
| if err := s.sjs.RemoveStartedDeals(curr.APIID, curr.Cid); err != nil { | ||
| return ffs.ColdInfo{}, allErrors, fmt.Errorf("removing resumed started deals storage: %s", err) | ||
| } |
There was a problem hiding this comment.
Doing the /ffs/scheduler/sjstore/starteddeals migration with real data I realized that in some border cases we leaving some already handled started deals.
In general the resumed deals would been removed by the already existing RemoveStartedDeals that existed below. But if after resuming deals, we consider that we need to create new deals to ensure the replication factor and we have some early return for some reason, the resumed deals won't be removed.
| // For each cid owner, we create the same registry | ||
| // in the new key version. | ||
| for _, iid := range owners { | ||
| newKey := datastore.NewKey("/ffs/joblogger_v2/").ChildString(iid.String()).ChildString(cidStr).ChildString(timestampStr) |
There was a problem hiding this comment.
Here the detail about _v2 suffixes.
What's happening in many migrations is that we try doing something like:
- Original key:
/hello/world/<cid> - New key:
/hello/world/<iid>/<cid>
For doing the migration, we scan all original keys with Query{Prefix:"/hello/world"}.
Depending on the underlying way that the go-datastore implementation works, that can produce problems; and that was the case, since the new keys we're .Puting are also a result of the query itself.
That's to say, we start getting /hello/world/<cid> keys, but at the same time we're iterating that query we're inserting others /hello/world/<iid>/<cid> which will appear later in the query iteration and is not something we want (new keys satisfy the prefix from the query). We only want to iterate the original keys.
Not a big deal and is solved with the _v2 suffixing to avoid the iteration to consider the new keys.
In the Badger implementation of go-datastore this doesn't happen, since Badger has snapshot capabilities on the query execution and you won't see new keys that appear after the query was started. But in Mongo that isn't the case. It's just different ways that queries work in different databases.
| repoPath, err := ioutil.TempDir("/tmp/powergate", ".powergate-*") | ||
| require.NoError(t, err) | ||
|
|
||
| repoPath := t.TempDir() |
| if err := m.Ensure(); err != nil { | ||
| return fmt.Errorf("running migrations: %s", err) | ||
| } | ||
| log.Infof("Migrations ensured") |
There was a problem hiding this comment.
Beautiful, nice and simple (from here anyway).
| type TrackedStorageConfig struct { | ||
| IID ffs.APIID | ||
| StorageConfig ffs.StorageConfig | ||
| } | ||
|
|
||
| // New retruns a new Store. | ||
| func New(ds datastore.Datastore) (*Store, error) { | ||
| s := &Store{ | ||
| ds: ds, | ||
| repairables: map[cid.Cid]struct{}{}, | ||
| renewables: map[cid.Cid]struct{}{}, | ||
| } | ||
| if err := s.loadCaches(); err != nil { | ||
| return nil, fmt.Errorf("loading renewable/repairable caches: %s", err) | ||
| } | ||
| return s, nil | ||
| // TrackedCid contains tracked storage configs for a Cid. | ||
| type TrackedCid struct { | ||
| Cid cid.Cid | ||
| Tracked []TrackedStorageConfig | ||
| } |
There was a problem hiding this comment.
I guess I'm not familiar with the implementation of repair/renew, but just curious why storing a list per cid is better than using a instanceId/cid key with a StorageConfig value, avoiding that complexity you mentioned.
| @@ -163,6 +175,50 @@ func (s *Scheduler) Cancel(jid ffs.JobID) error { | |||
| return nil | |||
| } | |||
|
|
|||
| // GCStaged runs a unpinned garbage collection of stage-pins. | |||
| func (s *Scheduler) GCStaged(ctx context.Context) ([]cid.Cid, error) { | |||
| return s.gcStaged(ctx, 0) | |||
There was a problem hiding this comment.
Maybe I'm missing something here, but would expect 0 to be s.gc.StageGracePeriod
There was a problem hiding this comment.
Actually, this is on purpose. This API is for the manual GC call that I feel should clean all staged stuff (grace period 0).
I'd imagine that the user would like the manual option to be more direct clean.
| // For each cid owner, we create the same registry | ||
| // in the new key version. | ||
| for _, iid := range owners { | ||
| newKey := datastore.NewKey("/ffs/joblogger_v2/").ChildString(iid.String()).ChildString(cidStr).ChildString(timestampStr) |
|
Oh, meant to ask about |
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
Yeah, any external API calls to go-ipfs are out of the path of tracking. What we can do is create some other API to stage-pin a Cid. So after it does |
|
Yea that sounds like a good idea. |
Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
| options.Unixfs.Pin(false), | ||
| options.Unixfs.Pin(true), |
There was a problem hiding this comment.
So for staging folders we also pin now.
| _, err = d.client.StageCid(ctx, &userPb.StageCidRequest{Cid: pth.Cid().String()}) | ||
| if err != nil { | ||
| return "", fmt.Errorf("stage pinning cid: %s", err) | ||
| } | ||
|
|
There was a problem hiding this comment.
We call a new API to stage Cids instead of streams of data.
* rebase & squash Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * trackstore progress Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * more progress Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * extensive trackstore tests Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * fixes Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * revert ci change Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * lints Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * fix docs Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * start migrations Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * progress Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * migration progress Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * storageinfo migr Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * progress Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * work Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * migration tests and changes Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * pinstore migration tests Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * lints Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * punctuation Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * change Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * rename Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * rename again Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * forward only Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * lint Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * fix admin ctx Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * allow for non-transactional migrations & real migration test & fixes Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * resume deal fixes & migration Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * improve comment Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com> * cover the stage folder usecase Signed-off-by: Ignacio Hagopian <jsign.uy@gmail.com>
A TL;DR of this PR, more details below:
Details about GC safety
The main problem with having safe IPFS GC at any point in time is all the staging data that might not be actually have ran its corresponding job.
The solution to this is to allow HotStorage have two types of pins:
Now every time you
pow data stage, it gets pinned in the underlyinggo-ipfsnode. Internally, Powergate keeps track of which Cids are pinned in which way.For example, say that FFS instance IID1 stages some data with cid Cid1. From that moment Cid1 is staged in the IPFS node. Then we have two cases:
In case 1, when the Job ends up running we have two cases:
1.a - The config that was pushed had HotStorage enabled. In this case, Cid1 is now switched to be a "real pin" instead of a "staging pin".
2.a - The config that was pushed had HotStorage disabled. In this case, Cid1 gets unpinned from the go-ipfs node.
In summary for case 1, after the Job runs the staged Cid1 gets promoted to a "real pin" or gets unpinned. Working this way means that IPFS GC can run any point in time after the data was staged and it's always safe to do so, i.e: no needed data that a queued or executing Job need will be deleted by IFPS.
In case 2, we have a situation of pinned data that would never execute a Job thus, a priori, never gets unpinned without extra logic. To handle this case, Powergate now has a "Staging GC" cron that runs every X minutes (configurable via flags) to manage these cases.
How this GC works? It basically analyzes all staged Cids that:
Those Cids look OK to be unpinned and let be GCed. Saying it differently, if we have some data which is "Stage pinned" for a "long" time, and we don't have any scheduled (queued) Job or executing one, then we can assume these staged data is abandoned. It's important this check that the Cid isn't part of a Queued job, since if we have a long queue of pending Jobs there's the chance that staged data is staged for more than 1hr.
There're knobs via flags/env to configure:
--ffsgcinterval: Frequency in which the Powergate GC runs to unpin old/abandoned staged data. (15min by default)--ffsgcstagegraceperiod: How old staged data should be to be considered potentially unpinnable (1hr by default)I also included two extra admin CLI commands for GC related things:
pow admin data gcstaged: This is for doing a manual trigger of a Powergate GC unpinning run. (What gets automatically run every 15min). Might be useful if the user wants to disable automatic unpinning, and just want to run it manually (or via API) for some reason.pow admin data pinnedCids: This will dump all pinned Cids from Hot-Storage. Each Cid will say which FFS instances is pinning the data, and which kind of pin those are. e.g: we could see that Cid1 is pinned by FFS1 with a Staged Pin, and by FFS2 with a real pin. Just to have some visibility of why a Cid is pinned in IPFS, since if you see theipfs pin lsoutput you only see that Cid1 is pinned but you might not understand why it's the case.Exampleof
pinnedCidsoutput:Finally, as expected, if we have a Cid pinned by multiple FFS instances... only when all FFS instances unpin the Cid, it gets really unpinned from the go-ipfs node. Saying it differently, now HotStorage does reference counting to really know when a Cid can be unpinned from the IPFS node.
Details about multitenancy
Now we support different FFS instances storing the same Cid. This involved changes in multiple internal components of the scheduler:
Migrations library
A new
migrationpackage was created that keeps track of which version the underlying datastore is. It also registers differentMigrationfunctions that transition version X to X+1. These Migrations are run in datastore transaction, so all the logic executed in the migration is atomic so it can fail safely without leaving "half-baked" data transformations.Whenever Powergate runs, the first thing it does after initializing the underlying Datastore is ensuring the datastore version is equal to the latest version. If that isn't the case, it will run all the transformations to move the current version to the latest version.
The current version of the datastore is assumed to be in
/migration/version.We have to handle two special cases if Powergate recognizes that
/migration/versionkey doesn't exist:In both cases, the
/migration/versionkey doesn't exist. To distinguish both, the library checks if the datastore has at least 1 key. If there's at least 1 key in the datastore but/migration/versiondoesn't exist, then we're in the first case. If the datastore is completely empty, we're in the last case.The distinction is necessary since the first case should run the necessary migrations to move from version 0, to the latest version that the newest migration migrates to. If we're in the second case (Powergate run for the firs time), we avoid any migration logic and set
/migration/keyvalue directly to the latest version.As mentioned above, all existing Powergates are considered to be in version 0 now. In this PR there's a single migration function that has 4 sub-migrations:
/ffs/joblogger/<cid>/<timestamp>to/ffs/joblogger/<ffsid>/<cid>/<timestamp>. Since we can already have multiple FFS instances storing the same Cid, the migration automatically duplicates entries for all FFS instances that might be having the same Cid... so we don't leave any FFS instance "empty" on logs that it might already existed from its POV. (Hope makes sense, but I can extend if isn't clear).Most of the above logic is a bit confusing since migrations that involve transforming a non-multitenant to multitenant setup is a bit hairy. These transformations not only copy the data to a new key but also have to copy the data multiple times once for each FFS instance that makes sense.
All this is covering the full set of cases that might appear. Most probably 90% of the Powergate instances weren't pushing the same Cid in different FFS instances... but this is the correct way to do the migrations.