S3 object sync keeps per-bucket totals, not every object

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## Unreleased
+
+### Changed
+
+- S3 object sync keeps a count and a total size per bucket instead of every object,
+  so its memory no longer grows with the number of objects in the account. It used
+  to hold them all until the end of the sync, plus a copy of the bucket per object,
+  and an account with a few busy buckets took a 512 MiB container to its limit in
+  seconds. `inspect -i bucket_sizer` reads the totals, and the sync line counts the
+  objects from them. (OpenSystemsLab/awless-ro#4)
+
+### Added
+
+- `aws.storage.s3object.keep` (off by default) keeps every object in the local graph
+  as before, for `list s3objects --local`. Without it, `list s3objects --local` says
+  where the sizes are and how to list the objects.
+
 ## v0.6.1
 
 ### Fixed

README.md

@@ -511,6 +511,13 @@ and skipped; the rest still land.
 Two resource types are off by default because they cost an API call per parent — one
 per bucket, one per hosted zone. The sync says so rather than reporting zero of them.
 
+With S3 object sync on, the sync keeps the number and total size of the objects in
+each bucket, which is what `inspect -i bucket_sizer` reads, and not the objects
+themselves, so memory stays flat however much the buckets hold. To keep every object
+too, for `list s3objects --local` on small buckets, set
+`aws.storage.s3object.keep true`. `list s3objects` without `--local` asks AWS
+directly either way.
+
 You rarely need to run it by hand. `show` and `inspect` refresh the copy on their own
 for the services they touch, and changing region with `switch` syncs the new one
 once. The first run does not sync. `--no-sync` skips that for one command, and

aws/fetch/config.go

@@ -66,3 +66,10 @@ func (c *Config) getBoolDefaultTrue(key string) bool {
 
 	return true
 }
+
+// getBool is false unless the key is set to true: for the settings that cost
+// something to turn on.
+func (c *Config) getBool(key string) bool {
+	b, _ := c.Extra[key].(bool)
+	return b
+}

aws/fetch/manual_fetchers.go

@@ -676,8 +676,13 @@ func addManualStorageFetchFuncs(conf *Config, funcs map[string]fetch.Func) {
 			}
 		}()
 
+		// A sync keeps per-bucket totals only, unless asked for the objects too.
+		// `list s3objects` fetches by type and prints the objects themselves.
+		fetchingType, _ := fetch.IsFetchingByType(ctx)
+		keepObjects := fetchingType == "s3object" || conf.getBool("aws.storage.s3object.keep")
+
 		err := forEachBucketParallel(ctx, cache, conf.APIs.S3, func(b s3types.Bucket) error {
-			return fetchObjectsForBucket(ctx, conf.APIs.S3, b, resourcesC)
+			return fetchObjectsForBucket(ctx, conf.APIs.S3, b, keepObjects, resourcesC)
 		})
 
 		close(resourcesC)

aws/fetch/s3_helpers.go

@@ -10,6 +10,7 @@ import (
 	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 
 	"github.com/theazz/awless-ro/aws/conv"
+	"github.com/theazz/awless-ro/cloud/properties"
 	"github.com/theazz/awless-ro/cloud/rdf"
 	"github.com/theazz/awless-ro/fetch"
 	"github.com/theazz/awless-ro/graph"
@@ -33,29 +34,50 @@ func forEachBucketParallel(ctx context.Context, cache fetch.Cache, api S3API, f
 	})
 }
 
-func fetchObjectsForBucket(ctx context.Context, api S3API, bucket s3types.Bucket, resourcesC chan<- *graph.Resource) error {
+// fetchObjectsForBucket pages through a bucket's objects and sends the bucket with
+// two totals on it, ObjectCount and Size in bytes, which is all bucket_sizer reads.
+//
+// It used to send every object as a resource, plus a fresh copy of the bucket per
+// object, and a sync held all of them until it wrote the graph: memory grew with the
+// number of objects in the account, and a few busy buckets were enough to take a
+// 512 MiB container to its limit. The objects themselves are only sent when
+// keepObjects is set, for `list s3objects` and whoever turns on
+// aws.storage.s3object.keep.
+//
+// A listing that fails part way sends no totals, since a partial count would read as
+// the bucket's size.
+func fetchObjectsForBucket(ctx context.Context, api S3API, bucket s3types.Bucket, keepObjects bool, resourcesC chan<- *graph.Resource) error {
+	parent, err := awsconv.InitResource(bucket)
+	if err != nil {
+		return err
+	}
+
+	var count, size int
 	paginator := s3.NewListObjectsV2Paginator(api, &s3.ListObjectsV2Input{Bucket: bucket.Name})
 	for paginator.HasMorePages() {
 		page, err := paginator.NextPage(ctx)
 		if err != nil {
 			return err
 		}
 		for _, output := range page.Contents {
-			res, err := awsconv.NewResource(output)
-			if err != nil {
-				return err
+			count++
+			size += int(aws.ToInt64(output.Size))
+			if !keepObjects {
+				continue
 			}
-			res.SetProperty("Bucket", aws.ToString(bucket.Name))
-			resourcesC <- res
-
-			parent, err := awsconv.InitResource(bucket)
+			res, err := awsconv.NewResource(output)
 			if err != nil {
 				return err
 			}
+			res.SetProperty(properties.Bucket, aws.ToString(bucket.Name))
 			res.AddRelation(rdf.ChildrenOfRel, parent)
-			resourcesC <- parent
+			resourcesC <- res
 		}
 	}
+
+	parent.SetProperty(properties.ObjectCount, count)
+	parent.SetProperty(properties.Size, size)
+	resourcesC <- parent
 	return nil
 }
 

aws/fetch/s3_objects_test.go

@@ -0,0 +1,202 @@
+package awsfetch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"testing"
+
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+
+	"github.com/theazz/awless-ro/cloud"
+	"github.com/theazz/awless-ro/cloud/properties"
+	"github.com/theazz/awless-ro/fetch"
+	"github.com/theazz/awless-ro/graph"
+)
+
+// pagedS3 lists each bucket's objects a page at a time, as S3 does past 1000 keys,
+// and can fail on a given page of a given bucket.
+type pagedS3 struct {
+	stubS3
+	pageSize   int
+	failBucket string
+	failPage   int
+}
+
+func (p *pagedS3) ListObjectsV2(_ context.Context, in *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+	name := awssdk.ToString(in.Bucket)
+	page := tokenIndex(awssdk.ToString(in.ContinuationToken))
+	if name == p.failBucket && page == p.failPage {
+		return nil, errors.New("listing failed")
+	}
+	all := p.objects[name]
+	start, end := page*p.pageSize, (page+1)*p.pageSize
+	if end > len(all) {
+		end = len(all)
+	}
+	out := &s3.ListObjectsV2Output{Contents: all[start:end], IsTruncated: awssdk.Bool(end < len(all))}
+	if end < len(all) {
+		out.NextContinuationToken = awssdk.String(strconv.Itoa(page + 1))
+	}
+	return out, nil
+}
+
+func objectsOfSize(n int, size int64) []s3types.Object {
+	objects := make([]s3types.Object, n)
+	for i := range objects {
+		objects[i] = s3types.Object{Key: awssdk.String(fmt.Sprintf("key-%d", i)), Size: awssdk.Int64(size)}
+	}
+	return objects
+}
+
+func newPagedS3() *pagedS3 {
+	return &pagedS3{
+		pageSize: 1000,
+		stubS3: stubS3{
+			buckets: map[string][]s3types.Bucket{"eu-west-1": {{Name: awssdk.String("logs")}, {Name: awssdk.String("empty")}}},
+			objects: map[string][]s3types.Object{"logs": objectsOfSize(2500, 3)},
+			grants:  map[string][]s3types.Grant{"logs": {{Permission: "READ", Grantee: &s3types.Grantee{ID: awssdk.String("owner"), Type: "CanonicalUser"}}}},
+		},
+		failPage: -1,
+	}
+}
+
+func fetchObjects(t *testing.T, api S3API, extra map[string]interface{}, ctx context.Context) []*graph.Resource {
+	t.Helper()
+	conf := NewConfig(&AWSAPI{S3: api})
+	conf.Extra["aws.storage.s3object.sync"] = true
+	for k, v := range extra {
+		conf.Extra[k] = v
+	}
+	funcs := BuildStorageFetchFuncs(conf)
+	resources, _, err := funcs["s3object"](context.WithValue(ctx, "region", "eu-west-1"), fetch.NewFetcher(funcs))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return resources
+}
+
+func byType(resources []*graph.Resource, typ string) []*graph.Resource {
+	var out []*graph.Resource
+	for _, r := range resources {
+		if r.Type() == typ {
+			out = append(out, r)
+		}
+	}
+	return out
+}
+
+// Object sync used to keep every object of every bucket in memory until the end of
+// the sync, and an account with a few busy buckets took a 512 MiB container to its
+// limit in seconds. bucket_sizer, the only reader, needs two numbers per bucket.
+func TestObjectSyncKeepsPerBucketTotalsNotObjects(t *testing.T) {
+	resources := fetchObjects(t, newPagedS3(), nil, context.Background())
+
+	if objects := byType(resources, cloud.S3Object); len(objects) != 0 {
+		t.Errorf("got %d object resources, want none: memory would grow with the object count", len(objects))
+	}
+
+	buckets := byType(resources, cloud.Bucket)
+	totals := map[string][2]interface{}{}
+	for _, b := range buckets {
+		if _, seen := totals[b.Id()]; seen {
+			t.Errorf("bucket %s was sent more than once", b.Id())
+		}
+		totals[b.Id()] = [2]interface{}{b.Properties()[properties.ObjectCount], b.Properties()[properties.Size]}
+	}
+	want := map[string][2]interface{}{
+		"logs":  {2500, 7500},
+		"empty": {0, 0},
+	}
+	if fmt.Sprint(totals) != fmt.Sprint(want) {
+		t.Errorf("got totals %v, want %v", totals, want)
+	}
+}
+
+// The totals go into the graph on the bucket the bucket fetcher builds, not on a
+// second node.
+func TestObjectTotalsLandOnTheSyncedBucket(t *testing.T) {
+	conf := NewConfig(&AWSAPI{S3: newPagedS3()})
+	conf.Extra["aws.storage.s3object.sync"] = true
+	ctx := context.WithValue(context.Background(), "region", "eu-west-1")
+
+	g, err := fetch.NewFetcher(BuildStorageFetchFuncs(conf)).Fetch(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	logs, err := g.GetResource(cloud.Bucket, "logs")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := logs.Properties()[properties.ObjectCount]; got != 2500 {
+		t.Errorf("object count %v, want 2500", got)
+	}
+	if got := logs.Properties()[properties.Size]; got != 7500 {
+		t.Errorf("size %v, want 7500", got)
+	}
+	if _, ok := logs.Properties()[properties.Grants]; !ok {
+		t.Error("the bucket lost what the bucket fetcher put on it")
+	}
+}
+
+// Keeping the objects themselves stays possible, for `list s3objects --local` on small
+// buckets. Each one still points at its bucket, which is built once.
+func TestObjectSyncKeepsObjectsWhenAskedTo(t *testing.T) {
+	resources := fetchObjects(t, newPagedS3(), map[string]interface{}{"aws.storage.s3object.keep": true}, context.Background())
+
+	objects := byType(resources, cloud.S3Object)
+	if len(objects) != 2500 {
+		t.Fatalf("got %d objects, want 2500", len(objects))
+	}
+	if got := objects[0].Properties()[properties.Bucket]; got != "logs" {
+		t.Errorf("object names bucket %v, want logs", got)
+	}
+	if n := len(byType(resources, cloud.Bucket)); n != 2 {
+		t.Errorf("got %d bucket resources, want one per bucket", n)
+	}
+}
+
+// `list s3objects` without --local asks AWS for the objects and prints them, so they
+// have to come back whatever the sync setting says.
+func TestLiveObjectListingReturnsObjects(t *testing.T) {
+	conf := NewConfig(&AWSAPI{S3: newPagedS3()})
+	ctx := context.WithValue(context.Background(), "region", "eu-west-1")
+	ctx = context.WithValue(ctx, "force", true)
+
+	g, err := fetch.NewFetcher(BuildStorageFetchFuncs(conf)).FetchByType(ctx, cloud.S3Object)
+	if err != nil {
+		t.Fatal(err)
+	}
+	objects, err := g.GetAllResources(cloud.S3Object)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(objects) != 2500 {
+		t.Errorf("got %d objects, want 2500", len(objects))
+	}
+}
+
+// A listing that fails half way gives no totals for that bucket: a partial count
+// would read as the bucket's size.
+func TestFailedListingGivesNoPartialTotals(t *testing.T) {
+	api := newPagedS3()
+	api.failBucket, api.failPage = "logs", 1
+
+	conf := NewConfig(&AWSAPI{S3: api})
+	conf.Extra["aws.storage.s3object.sync"] = true
+	funcs := BuildStorageFetchFuncs(conf)
+	ctx := context.WithValue(context.Background(), "region", "eu-west-1")
+
+	resources, _, err := funcs["s3object"](ctx, fetch.NewFetcher(funcs))
+	if err == nil {
+		t.Fatal("the listing error was swallowed")
+	}
+	for _, b := range byType(resources, cloud.Bucket) {
+		if n, ok := b.Properties()[properties.ObjectCount]; ok && b.Id() == "logs" {
+			t.Errorf("logs got a count of %v from a listing that failed", n)
+		}
+	}
+}