Expire objects from S3 according to retention policy #309

arielshaqed · 2020-07-27T08:06:26Z

Add a new command lakefs expire. It should be run on the server in order to emit logs and to avoid holding a long-lived connection. When run:

Generate a list of entries to expire. Deduped entries only expire when all entries referring to the object expire.
Store list in temporary file (it may be too large for memory!)
For each affected S3 bucket create a batch job to tag objects with key "lakefs-expire" value "1".
(An S3 lifecycle rule should be configured that actually expires objects with that key on S3. lakefs diagnose verifies that there is such a lifecycle rule.)

Tested on my local-with-S3 lakefs instance.

Limitation: Object expiry on namespaced LakeFS repositories untested.

arielshaqed · 2020-07-27T09:33:11Z

Note well: retention needs an access key ID in order to retrieve the account ID. Currently we read it from blockstore.s3.credentials.access_key_id in the config file. This works nicely, except that it means we also must specify the access_secret_key there otherwise S3 configuration fails.

In a production setting I expect this field to be set.

Options:

It's OK because it will work in production.
Add a separate blockstore.s3.account_id field. This doubles some configuration, and makes production configuration a bit harder.
Special-case logic for blockstore.s3.credentials with an access key ID but without a secret key. This potentially behaves confusingly with Viper, which has multiple sources of configuration for different elements.
Add an optional separate access key ID field, say blockstore.s3.retension.access_key_id. Use it, but if missing use blockstore.s3.credentials.access_key_id instead. This probably works best in practice, but the logic driving it may be confusing.
Insert your good configuration here please!

ozkatz · 2020-07-28T11:42:27Z

catalog/cataloger_retention.go

-// QueryExpired returns ExpiryRows iterating over all objects to expire on repositoryName
-// according to policy to channel out.
-func (c *cataloger) QueryExpired(ctx context.Context, repositoryName string, policy *retention.Policy) (ExpiryRows, error) {
+func (c *cataloger) QueryExpired(ctx context.Context, repositoryName string, policy *Policy) (ExpiryRows, error) {
 	logger := logging.Default().WithContext(ctx).WithField("policy", *policy)


I think you might be looking for logging.FromContext(ctx) which would populate the logger returned with all context logging values?

SG, thanks!

ozkatz · 2020-07-28T11:44:59Z

catalog/cataloger_retention.go

+                             ON a.physical_address = b.physical_address)
+                            WHERE a.c = b.c)
+                    `,
+		expiryByEntriesQueryString,


why not use a builder?

No way to interpolate table names (because SQL or Postgres and placeholders). So the builder will buy us very little, I would still need %-interpolation. See e.g. this message on the PostgresQL mailing list.

ozkatz · 2020-07-28T11:49:31Z

catalog/cataloger_retention.go

 	if err != nil {
 		return nil, fmt.Errorf("running query: %w", err)
 	}
-	return &expiryRows{rows, repositoryName}, nil
+	var ret ExpiryRows = &expiryRows{rows: rows, RepositoryName: repositoryName}
+	return ret, nil


This could result in a huge list, exceeding available memory size.

I don't know we should address it now, but I'd probably go with pagination (using a very big page size) to make sure we don't end up with a crash loop

Per documentation and @nopcoder (not to mention implementation bugs I had when I returned unread rows from a transaction), rows is just an iterator. This is not in memory.

Regardless. @tzahij asked me to consider going back to using a table here, and also has concerns about the whole system as created on this series of PRs. We shall discuss f2f tomorrow.

ozkatz · 2020-07-28T11:51:43Z

catalog/cataloger_retention_test.go

 		},
 	})
 	if err != nil {
 		t.Fatalf("read all expiration records failed: %s", err)
 	}
 	resultByPhysicalAddress := make(map[string]*ExpireResult, len(allResults))
 	for _, result := range allResults {
+		t.Logf("Result: %+v", result)


I assume this was for debugging? do we still want to print it?

AFAIU test logs only show up on failures (by default). Let me know if you disagree or think it's too much, I'll happily remove.

ozkatz · 2020-07-28T11:55:45Z

cmd/lakefs/cmd/expire.go

+			logger.WithError(err).Fatal("cannot list repositories")
+		}
+
+		// TODO(ariels: fail on failure!


Separate PR: this is mainly called in buildS3Adapter, I want to understand better what counts as failure here.

ozkatz · 2020-07-28T12:07:17Z

config/config.go

+	return cfg
+}
+
+func GetAccount(awsConfig *aws.Config) (*string, error) {


any reason to return a pointer to a string here?

AWS likes to use pointers when serializing. Changing to be more Goish.

ozkatz · 2020-07-28T12:11:06Z

go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/lib/pq v1.7.0 // indirect
 	github.com/lunixbochs/vtclean v1.0.0 // indirect
 	github.com/manifoldco/promptui v0.7.0
+	github.com/matoous/go-nanoid v1.4.1


I don't see where we're using it in the PR?

Was using it, forgot to clean it. Thanks!

ozkatz · 2020-07-28T12:20:48Z

retention/expiry.go

+)
+
+// WriteExpiryResultsToSeekableReader returns a file-backed (Seeker) Reader holding the contents of expiryRows.
+func WriteExpiryResultsToSeekableReader(ctx context.Context, expiryRows catalog.ExpiryRows) (fileutil.RewindableReader, error) {


Usually easier to test (and compose) when we pass the Reader as a dependency instead of returning one.
it's nicer but def. not important.

I don't think it's that kind of Reader. This is a rewindable file abstraction: first you write it, then you transform it into a rewindable reader.

ozkatz · 2020-07-28T12:24:17Z

retention/expiry.go

+		Writer:    ret,
+		CsvWriter: csv.NewWriter(ret),
+	}
+	(*bw)[bucketName] = record


if I understand the rest of the code correctly, we only ever call GetWriter and iterate over bucket/writer pairs. so never really get or set values directly. Not sure type aliasing a map is the best API for that...

l.120 is random access to get the writer for a bucket. (We need this complexity because different repos could live on different buckets, and namespaces mean they could share them too)

ozkatz · 2020-07-28T12:25:18Z

retention/expiry.go

+		}
+		resetableReader, count, err := encodingData.Writer.StartReading()
+		if err != nil {
+			bucketLogger.WithError(err).Error("failed to start reading encoded CSVs; lose all bucket expiries")


all errors in this loop are swallowed?

Yes: we tried to start expiry and failed. We're not going to manage to expire them today. Best I can think of is log an error and hope tomorrow is another day. With other changes requested it does fail the expiry run -- but the objects did not get expired.
Monitoring will make it clearer that the repo is not being expired. Not sure what else I can do ("it's an ops problem").

arielshaqed

Thanks; PTAL!

arielshaqed · 2020-07-28T12:49:58Z

catalog/cataloger_retention.go

-// QueryExpired returns ExpiryRows iterating over all objects to expire on repositoryName
-// according to policy to channel out.
-func (c *cataloger) QueryExpired(ctx context.Context, repositoryName string, policy *retention.Policy) (ExpiryRows, error) {
+func (c *cataloger) QueryExpired(ctx context.Context, repositoryName string, policy *Policy) (ExpiryRows, error) {
 	logger := logging.Default().WithContext(ctx).WithField("policy", *policy)


SG, thanks!

arielshaqed · 2020-07-28T12:52:55Z

catalog/cataloger_retention.go

+                             ON a.physical_address = b.physical_address)
+                            WHERE a.c = b.c)
+                    `,
+		expiryByEntriesQueryString,


No way to interpolate table names (because SQL or Postgres and placeholders). So the builder will buy us very little, I would still need %-interpolation. See e.g. this message on the PostgresQL mailing list.

arielshaqed · 2020-07-28T12:58:10Z

catalog/cataloger_retention.go

 	if err != nil {
 		return nil, fmt.Errorf("running query: %w", err)
 	}
-	return &expiryRows{rows, repositoryName}, nil
+	var ret ExpiryRows = &expiryRows{rows: rows, RepositoryName: repositoryName}
+	return ret, nil


Per documentation and @nopcoder (not to mention implementation bugs I had when I returned unread rows from a transaction), rows is just an iterator. This is not in memory.

arielshaqed · 2020-07-28T12:59:47Z

catalog/cataloger_retention_test.go

 		},
 	})
 	if err != nil {
 		t.Fatalf("read all expiration records failed: %s", err)
 	}
 	resultByPhysicalAddress := make(map[string]*ExpireResult, len(allResults))
 	for _, result := range allResults {
+		t.Logf("Result: %+v", result)


AFAIU test logs only show up on failures (by default). Let me know if you disagree or think it's too much, I'll happily remove.

arielshaqed · 2020-07-28T13:01:56Z

cmd/lakefs/cmd/expire.go

+			logger.WithError(err).Fatal("cannot list repositories")
+		}
+
+		// TODO(ariels: fail on failure!


Separate PR: this is mainly called in buildS3Adapter, I want to understand better what counts as failure here.

arielshaqed · 2020-07-28T15:44:51Z

go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/lib/pq v1.7.0 // indirect
 	github.com/lunixbochs/vtclean v1.0.0 // indirect
 	github.com/manifoldco/promptui v0.7.0
+	github.com/matoous/go-nanoid v1.4.1


Was using it, forgot to clean it. Thanks!

arielshaqed · 2020-07-28T15:46:11Z

retention/expiry.go

+)
+
+// WriteExpiryResultsToSeekableReader returns a file-backed (Seeker) Reader holding the contents of expiryRows.
+func WriteExpiryResultsToSeekableReader(ctx context.Context, expiryRows catalog.ExpiryRows) (fileutil.RewindableReader, error) {


I don't think it's that kind of Reader. This is a rewindable file abstraction: first you write it, then you transform it into a rewindable reader.

arielshaqed · 2020-07-28T15:48:04Z

retention/expiry.go

+		Writer:    ret,
+		CsvWriter: csv.NewWriter(ret),
+	}
+	(*bw)[bucketName] = record


l.120 is random access to get the writer for a bucket. (We need this complexity because different repos could live on different buckets, and namespaces mean they could share them too)

arielshaqed · 2020-07-28T15:49:48Z

retention/expiry.go

+		}
+		resetableReader, count, err := encodingData.Writer.StartReading()
+		if err != nil {
+			bucketLogger.WithError(err).Error("failed to start reading encoded CSVs; lose all bucket expiries")


Yes: we tried to start expiry and failed. We're not going to manage to expire them today. Best I can think of is log an error and hope tomorrow is another day. With other changes requested it does fail the expiry run -- but the objects did not get expired.
Monitoring will make it clearer that the repo is not being expired. Not sure what else I can do ("it's an ops problem").

arielshaqed · 2020-07-29T06:09:45Z

cmd/lakefs/cmd/expire.go

+				continue
+			}
+
+			retention.ExpireOnS3(ctx, s3ControlClient, s3Client, cataloger, expiryReader, &expiryParams)


We do log the errors; there's not much more to do, but failing the process is a good idea!

arielshaqed · 2020-07-29T11:49:53Z

I plan to fix the race found by @tzahij under a separate PR, this one is large enough and the issue is not in any of the code on this one.

ozkatz

Great work on this!

Prevent dependency loops. They are *parsed* to API objects in retention, but catalog uses them and is at a lower level.

Operationally run this periodically (e.g. daily) on the lakeFS server (hence lakefs and not via the API). Adds these additional configuration variables in `blockstore.s3.retention`: - `role_arn`: ARN to use in batch tagging - `manifest_base_url`: S3 URL prefix (e.g. directory) to use to upload tagging manifest

Inter-branch copies can share the same dedupe ID, don't expire until they all agree about expiration. The object is available on all its branches until it expires from the last branch: retention is *not* a synchronous mechanism and applications are not allowed to rely on it occurring. - Manifests are CSV not JSON format (whoops) - Flush CSV encoder Tested with tiny expiration vs. S3 -- objects were tagged.

Find some trivial errors in the object, doesn't cost much to perform. Does *not* actually discover that no other fields are allowed in `Report` when setting `Enabled: false`.

arielshaqed · 2020-07-30T06:36:40Z

Thanks! Fixing numerous minor conflicts and pulling (unless one of the conflicts turns out to be non-minor).

- Repos with no retention should be skipped with no error - Use `logger.FromContext` - Remove unused go-nanoid

Tested by using IAM to fail CreateJob and seeing an appropriately FATAL report.

arielshaqed · 2020-07-30T06:44:52Z

@ozkatz I seem to have lost your approval due to the rebase changes (and/or the rebase itself). Dunno if that's our configuration or my mishandling of something git-(hub?)-ish.
Anyway, can you re-approve please?
THANKS!

…e-from-s3 Expire objects from S3 according to retention policy Former-commit-id: 3cc43a763b7ac551977418b922a5281109251eab

arielshaqed force-pushed the feature/retention-policy-expire-from-s3 branch from 8c60e01 to 9254611 Compare July 27, 2020 08:14

arielshaqed requested a review from ozkatz July 27, 2020 08:16

ozkatz requested changes Jul 28, 2020

View reviewed changes

arielshaqed mentioned this pull request Jul 28, 2020

Monitor repo retention #325

Closed

arielshaqed commented Jul 29, 2020

View reviewed changes

arielshaqed requested a review from ozkatz July 29, 2020 11:48

ozkatz previously approved these changes Jul 29, 2020

View reviewed changes

arielshaqed added 8 commits July 30, 2020 09:26

[refactor] Move retention policy data objects to catalog

f1a7019

Prevent dependency loops. They are *parsed* to API objects in retention, but catalog uses them and is at a lower level.

Write expiry to a write-once read-many file-backed stream

325a874

Create per-bucket expiry tagging CSVs

10a103b

Test deduped expiry

262197c

Post-rebase fixes: CreateEntryParams

790596b

Validate CreateJobInput object

dcc11ee

Find some trivial errors in the object, doesn't cost much to perform. Does *not* actually discover that no other fields are allowed in `Report` when setting `Enabled: false`.

arielshaqed added 2 commits July 30, 2020 09:41

[CR] skip repos with no retention, clean APIs and unused module

162c498

- Repos with no retention should be skipped with no error - Use `logger.FromContext` - Remove unused go-nanoid

[CR] Fail "lakefs expire" if errors occur during expiration

9d6b701

Tested by using IAM to fail CreateJob and seeing an appropriately FATAL report.

arielshaqed dismissed ozkatz’s stale review via 9d6b701 July 30, 2020 06:41

arielshaqed force-pushed the feature/retention-policy-expire-from-s3 branch from 0677198 to 9d6b701 Compare July 30, 2020 06:41

ozkatz approved these changes Jul 30, 2020

View reviewed changes

arielshaqed merged commit 9133796 into master Jul 30, 2020

arielshaqed deleted the feature/retention-policy-expire-from-s3 branch July 30, 2020 07:17

nopcoder pushed a commit that referenced this pull request Aug 27, 2020

Merge pull request #309 from treeverse/feature/retention-policy-expir…

72e6ef7

…e-from-s3 Expire objects from S3 according to retention policy Former-commit-id: 3cc43a763b7ac551977418b922a5281109251eab

Expire objects from S3 according to retention policy #309

Expire objects from S3 according to retention policy #309

Conversation

arielshaqed commented Jul 27, 2020

arielshaqed commented Jul 27, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ozkatz Jul 28, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arielshaqed left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arielshaqed commented Jul 29, 2020

ozkatz left a comment

Choose a reason for hiding this comment

arielshaqed commented Jul 30, 2020

arielshaqed commented Jul 30, 2020

ozkatz Jul 28, 2020 •

edited

Loading