Merge pull request #1228 from fluxcd/bucket-prefix

bucket: Add prefix filtering capability

Author: stefanprodan

api/v1beta2/bucket_types.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/fluxcd/pkg/apis/acl"
 	"github.com/fluxcd/pkg/apis/meta"
+
 	apiv1 "github.com/fluxcd/source-controller/api/v1"
 )
 
@@ -73,6 +74,10 @@ type BucketSpec struct {
 	// +optional
 	Region string `json:"region,omitempty"`
 
+	// Prefix to use for server-side filtering of files in the Bucket.
+	// +optional
+	Prefix string `json:"prefix,omitempty"`
+
 	// SecretRef specifies the Secret containing authentication credentials
 	// for the Bucket.
 	// +optional

config/crd/bases/source.toolkit.fluxcd.io_buckets.yaml

@@ -331,6 +331,10 @@ spec:
                   to ensure efficient use of resources.
                 pattern: ^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$
                 type: string
+              prefix:
+                description: Prefix to use for server-side filtering of files in the
+                  Bucket.
+                type: string
               provider:
                 default: generic
                 description: Provider of the object storage bucket. Defaults to 'generic',

docs/api/v1beta2/source.md

@@ -138,6 +138,18 @@ string
 </tr>
 <tr>
 <td>
+<code>prefix</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Prefix to use for server-side filtering of files in the Bucket.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>secretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">
@@ -1422,6 +1434,18 @@ string
 </tr>
 <tr>
 <td>
+<code>prefix</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>Prefix to use for server-side filtering of files in the Bucket.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>secretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">

docs/spec/v1beta2/buckets.md

@@ -785,6 +785,15 @@ credentials for the object storage. For some `.spec.provider` implementations
 the presence of the field is required, see [Provider](#provider) for more
 details and examples.
 
+### Prefix
+
+`.spec.prefix` is an optional field to enable server-side filtering
+of files in the Bucket.
+
+**Note:** The server-side filtering works only with the `generic`, `aws`
+and `gcp` [provider](#provider) and is preferred over [`.spec.ignore`](#ignore)
+as a more efficient way of excluding files. 
+
 ### Ignore
 
 `.spec.ignore` is an optional field to specify rules in [the `.gitignore`

internal/controller/bucket_controller.go

@@ -145,7 +145,7 @@ type BucketProvider interface {
 	// bucket, calling visit for every item.
 	// If the underlying client or the visit callback returns an error,
 	// it returns early.
-	VisitObjects(ctx context.Context, bucketName string, visit func(key, etag string) error) error
+	VisitObjects(ctx context.Context, bucketName string, prefix string, visit func(key, etag string) error) error
 	// ObjectIsNotFound returns true if the given error indicates an object
 	// could not be found.
 	ObjectIsNotFound(error) bool
@@ -742,7 +742,7 @@ func fetchEtagIndex(ctx context.Context, provider BucketProvider, obj *bucketv1.
 	matcher := sourceignore.NewMatcher(ps)
 
 	// Build up index
-	err = provider.VisitObjects(ctxTimeout, obj.Spec.BucketName, func(key, etag string) error {
+	err = provider.VisitObjects(ctxTimeout, obj.Spec.BucketName, obj.Spec.Prefix, func(key, etag string) error {
 		if strings.HasSuffix(key, "/") || key == sourceignore.IgnoreFile {
 			return nil
 		}

internal/controller/bucket_controller_fetch_test.go

@@ -69,7 +69,7 @@ func (m mockBucketClient) ObjectIsNotFound(e error) bool {
 	return e == errMockNotFound
 }
 
-func (m mockBucketClient) VisitObjects(_ context.Context, _ string, f func(key, etag string) error) error {
+func (m mockBucketClient) VisitObjects(_ context.Context, _ string, _ string, f func(key, etag string) error) error {
 	for key, obj := range m.objects {
 		if err := f(key, obj.etag); err != nil {
 			return err

pkg/azure/blob.go

@@ -265,7 +265,7 @@ func (c *BlobClient) FGetObject(ctx context.Context, bucketName, objectName, loc
 // bucket, calling visit for every item.
 // If the underlying client or the visit callback returns an error,
 // it returns early.
-func (c *BlobClient) VisitObjects(ctx context.Context, bucketName string, visit func(path, etag string) error) error {
+func (c *BlobClient) VisitObjects(ctx context.Context, bucketName string, prefix string, visit func(path, etag string) error) error {
 	items := c.NewListBlobsFlatPager(bucketName, nil)
 	for items.More() {
 		resp, err := items.NextPage(ctx)

pkg/gcp/gcp.go

@@ -165,8 +165,10 @@ func (c *GCSClient) FGetObject(ctx context.Context, bucketName, objectName, loca
 // bucket, calling visit for every item.
 // If the underlying client or the visit callback returns an error,
 // it returns early.
-func (c *GCSClient) VisitObjects(ctx context.Context, bucketName string, visit func(path, etag string) error) error {
-	items := c.Client.Bucket(bucketName).Objects(ctx, nil)
+func (c *GCSClient) VisitObjects(ctx context.Context, bucketName string, prefix string, visit func(path, etag string) error) error {
+	items := c.Client.Bucket(bucketName).Objects(ctx, &gcpstorage.Query{
+		Prefix: prefix,
+	})
 	for {
 		object, err := items.Next()
 		if err == IteratorDone {

pkg/gcp/gcp_test.go

@@ -170,7 +170,7 @@ func TestVisitObjects(t *testing.T) {
 	}
 	keys := []string{}
 	etags := []string{}
-	err := gcpClient.VisitObjects(context.Background(), bucketName, func(key, etag string) error {
+	err := gcpClient.VisitObjects(context.Background(), bucketName, "", func(key, etag string) error {
 		keys = append(keys, key)
 		etags = append(etags, etag)
 		return nil
@@ -185,7 +185,7 @@ func TestVisitObjectsErr(t *testing.T) {
 		Client: client,
 	}
 	badBucketName := "bad-bucket"
-	err := gcpClient.VisitObjects(context.Background(), badBucketName, func(key, etag string) error {
+	err := gcpClient.VisitObjects(context.Background(), badBucketName, "", func(key, etag string) error {
 		return nil
 	})
 	assert.Error(t, err, fmt.Sprintf("listing objects from bucket '%s' failed: storage: bucket doesn't exist", badBucketName))
@@ -196,7 +196,7 @@ func TestVisitObjectsCallbackErr(t *testing.T) {
 		Client: client,
 	}
 	mockErr := fmt.Errorf("mock")
-	err := gcpClient.VisitObjects(context.Background(), bucketName, func(key, etag string) error {
+	err := gcpClient.VisitObjects(context.Background(), bucketName, "", func(key, etag string) error {
 		return mockErr
 	})
 	assert.Error(t, err, mockErr.Error())

pkg/minio/minio.go

@@ -105,9 +105,10 @@ func (c *MinioClient) FGetObject(ctx context.Context, bucketName, objectName, lo
 // bucket, calling visit for every item.
 // If the underlying client or the visit callback returns an error,
 // it returns early.
-func (c *MinioClient) VisitObjects(ctx context.Context, bucketName string, visit func(key, etag string) error) error {
+func (c *MinioClient) VisitObjects(ctx context.Context, bucketName string, prefix string, visit func(key, etag string) error) error {
 	for object := range c.Client.ListObjects(ctx, bucketName, minio.ListObjectsOptions{
 		Recursive: true,
+		Prefix:    prefix,
 		UseV1:     s3utils.IsGoogleEndpoint(*c.Client.EndpointURL()),
 	}) {
 		if object.Err != nil {