Merge pull request #434 from pa250194/gcp-bucket-provider

Author: hiddeco

api/v1beta1/bucket_types.go

@@ -30,7 +30,7 @@ const (
 // BucketSpec defines the desired state of an S3 compatible bucket
 type BucketSpec struct {
 	// The S3 compatible storage provider name, default ('generic').
-	// +kubebuilder:validation:Enum=generic;aws
+	// +kubebuilder:validation:Enum=generic;aws;gcp
 	// +kubebuilder:default:=generic
 	// +optional
 	Provider string `json:"provider,omitempty"`
@@ -79,6 +79,7 @@ type BucketSpec struct {
 const (
 	GenericBucketProvider string = "generic"
 	AmazonBucketProvider  string = "aws"
+	GoogleBucketProvider  string = "gcp"
 )
 
 // BucketStatus defines the observed state of a bucket

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

@@ -66,6 +66,7 @@ spec:
                 enum:
                 - generic
                 - aws
+                - gcp
                 type: string
               region:
                 description: The bucket region.

controllers/bucket_controller.go

@@ -29,6 +29,7 @@ import (
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/minio-go/v7/pkg/s3utils"
+	"google.golang.org/api/option"
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -46,6 +47,7 @@ import (
 	"github.com/fluxcd/pkg/runtime/events"
 	"github.com/fluxcd/pkg/runtime/metrics"
 	"github.com/fluxcd/pkg/runtime/predicates"
+	"github.com/fluxcd/source-controller/pkg/gcp"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
 	"github.com/fluxcd/source-controller/pkg/sourceignore"
@@ -176,77 +178,25 @@ func (r *BucketReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 }
 
 func (r *BucketReconciler) reconcile(ctx context.Context, bucket sourcev1.Bucket) (sourcev1.Bucket, error) {
-	s3Client, err := r.auth(ctx, bucket)
-	if err != nil {
-		err = fmt.Errorf("auth error: %w", err)
-		return sourcev1.BucketNotReady(bucket, sourcev1.AuthenticationFailedReason, err.Error()), err
-	}
-
-	// create tmp dir
+	var err error
+	var sourceBucket sourcev1.Bucket
 	tempDir, err := os.MkdirTemp("", bucket.Name)
 	if err != nil {
 		err = fmt.Errorf("tmp dir error: %w", err)
 		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
 	}
 	defer os.RemoveAll(tempDir)
-
-	ctxTimeout, cancel := context.WithTimeout(ctx, bucket.Spec.Timeout.Duration)
-	defer cancel()
-
-	exists, err := s3Client.BucketExists(ctxTimeout, bucket.Spec.BucketName)
-	if err != nil {
-		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
-	}
-	if !exists {
-		err = fmt.Errorf("bucket '%s' not found", bucket.Spec.BucketName)
-		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
-	}
-
-	// Look for file with ignore rules first
-	// NB: S3 has flat filepath keys making it impossible to look
-	// for files in "subdirectories" without building up a tree first.
-	path := filepath.Join(tempDir, sourceignore.IgnoreFile)
-	if err := s3Client.FGetObject(ctxTimeout, bucket.Spec.BucketName, sourceignore.IgnoreFile, path, minio.GetObjectOptions{}); err != nil {
-		if resp, ok := err.(minio.ErrorResponse); ok && resp.Code != "NoSuchKey" {
-			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
-		}
-	}
-	ps, err := sourceignore.ReadIgnoreFile(path, nil)
-	if err != nil {
-		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
-	}
-	// In-spec patterns take precedence
-	if bucket.Spec.Ignore != nil {
-		ps = append(ps, sourceignore.ReadPatterns(strings.NewReader(*bucket.Spec.Ignore), nil)...)
-	}
-	matcher := sourceignore.NewMatcher(ps)
-
-	// download bucket content
-	for object := range s3Client.ListObjects(ctxTimeout, bucket.Spec.BucketName, minio.ListObjectsOptions{
-		Recursive: true,
-		UseV1:     s3utils.IsGoogleEndpoint(*s3Client.EndpointURL()),
-	}) {
-		if object.Err != nil {
-			err = fmt.Errorf("listing objects from bucket '%s' failed: %w", bucket.Spec.BucketName, object.Err)
-			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
-		}
-
-		if strings.HasSuffix(object.Key, "/") || object.Key == sourceignore.IgnoreFile {
-			continue
-		}
-
-		if matcher.Match(strings.Split(object.Key, "/"), false) {
-			continue
+	if bucket.Spec.Provider == sourcev1.GoogleBucketProvider {
+		sourceBucket, err = r.reconcileWithGCP(ctx, bucket, tempDir)
+		if err != nil {
+			return sourceBucket, err
 		}
-
-		localPath := filepath.Join(tempDir, object.Key)
-		err := s3Client.FGetObject(ctxTimeout, bucket.Spec.BucketName, object.Key, localPath, minio.GetObjectOptions{})
+	} else {
+		sourceBucket, err = r.reconcileWithMinio(ctx, bucket, tempDir)
 		if err != nil {
-			err = fmt.Errorf("downloading object from bucket '%s' failed: %w", bucket.Spec.BucketName, err)
-			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+			return sourceBucket, err
 		}
 	}
-
 	revision, err := r.checksum(tempDir)
 	if err != nil {
 		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
@@ -315,7 +265,177 @@ func (r *BucketReconciler) reconcileDelete(ctx context.Context, bucket sourcev1.
 	return ctrl.Result{}, nil
 }
 
-func (r *BucketReconciler) auth(ctx context.Context, bucket sourcev1.Bucket) (*minio.Client, error) {
+// reconcileWithGCP handles getting objects from a Google Cloud Platform bucket
+// using a gcp client
+func (r *BucketReconciler) reconcileWithGCP(ctx context.Context, bucket sourcev1.Bucket, tempDir string) (sourcev1.Bucket, error) {
+	log := logr.FromContext(ctx)
+	gcpClient, err := r.authGCP(ctx, bucket)
+	if err != nil {
+		err = fmt.Errorf("auth error: %w", err)
+		return sourcev1.BucketNotReady(bucket, sourcev1.AuthenticationFailedReason, err.Error()), err
+	}
+	defer gcpClient.Close(log)
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, bucket.Spec.Timeout.Duration)
+	defer cancel()
+
+	exists, err := gcpClient.BucketExists(ctxTimeout, bucket.Spec.BucketName)
+	if err != nil {
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+	if !exists {
+		err = fmt.Errorf("bucket '%s' not found", bucket.Spec.BucketName)
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+
+	// Look for file with ignore rules first.
+	path := filepath.Join(tempDir, sourceignore.IgnoreFile)
+	if err := gcpClient.FGetObject(ctxTimeout, bucket.Spec.BucketName, sourceignore.IgnoreFile, path); err != nil {
+		if err == gcp.ErrorObjectDoesNotExist && sourceignore.IgnoreFile != ".sourceignore" {
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+	}
+	ps, err := sourceignore.ReadIgnoreFile(path, nil)
+	if err != nil {
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+	// In-spec patterns take precedence
+	if bucket.Spec.Ignore != nil {
+		ps = append(ps, sourceignore.ReadPatterns(strings.NewReader(*bucket.Spec.Ignore), nil)...)
+	}
+	matcher := sourceignore.NewMatcher(ps)
+	objects := gcpClient.ListObjects(ctxTimeout, bucket.Spec.BucketName, nil)
+	// download bucket content
+	for {
+		object, err := objects.Next()
+		if err == gcp.IteratorDone {
+			break
+		}
+		if err != nil {
+			err = fmt.Errorf("listing objects from bucket '%s' failed: %w", bucket.Spec.BucketName, err)
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+
+		if strings.HasSuffix(object.Name, "/") || object.Name == sourceignore.IgnoreFile {
+			continue
+		}
+
+		if matcher.Match(strings.Split(object.Name, "/"), false) {
+			continue
+		}
+
+		localPath := filepath.Join(tempDir, object.Name)
+		if err = gcpClient.FGetObject(ctxTimeout, bucket.Spec.BucketName, object.Name, localPath); err != nil {
+			err = fmt.Errorf("downloading object from bucket '%s' failed: %w", bucket.Spec.BucketName, err)
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+	}
+	return sourcev1.Bucket{}, nil
+}
+
+// reconcileWithMinio handles getting objects from an S3 compatible bucket
+// using a minio client
+func (r *BucketReconciler) reconcileWithMinio(ctx context.Context, bucket sourcev1.Bucket, tempDir string) (sourcev1.Bucket, error) {
+	s3Client, err := r.authMinio(ctx, bucket)
+	if err != nil {
+		err = fmt.Errorf("auth error: %w", err)
+		return sourcev1.BucketNotReady(bucket, sourcev1.AuthenticationFailedReason, err.Error()), err
+	}
+
+	ctxTimeout, cancel := context.WithTimeout(ctx, bucket.Spec.Timeout.Duration)
+	defer cancel()
+
+	exists, err := s3Client.BucketExists(ctxTimeout, bucket.Spec.BucketName)
+	if err != nil {
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+	if !exists {
+		err = fmt.Errorf("bucket '%s' not found", bucket.Spec.BucketName)
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+
+	// Look for file with ignore rules first
+	// NB: S3 has flat filepath keys making it impossible to look
+	// for files in "subdirectories" without building up a tree first.
+	path := filepath.Join(tempDir, sourceignore.IgnoreFile)
+	if err := s3Client.FGetObject(ctxTimeout, bucket.Spec.BucketName, sourceignore.IgnoreFile, path, minio.GetObjectOptions{}); err != nil {
+		if resp, ok := err.(minio.ErrorResponse); ok && resp.Code != "NoSuchKey" {
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+	}
+	ps, err := sourceignore.ReadIgnoreFile(path, nil)
+	if err != nil {
+		return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+	}
+	// In-spec patterns take precedence
+	if bucket.Spec.Ignore != nil {
+		ps = append(ps, sourceignore.ReadPatterns(strings.NewReader(*bucket.Spec.Ignore), nil)...)
+	}
+	matcher := sourceignore.NewMatcher(ps)
+
+	// download bucket content
+	for object := range s3Client.ListObjects(ctxTimeout, bucket.Spec.BucketName, minio.ListObjectsOptions{
+		Recursive: true,
+		UseV1:     s3utils.IsGoogleEndpoint(*s3Client.EndpointURL()),
+	}) {
+		if object.Err != nil {
+			err = fmt.Errorf("listing objects from bucket '%s' failed: %w", bucket.Spec.BucketName, object.Err)
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+
+		if strings.HasSuffix(object.Key, "/") || object.Key == sourceignore.IgnoreFile {
+			continue
+		}
+
+		if matcher.Match(strings.Split(object.Key, "/"), false) {
+			continue
+		}
+
+		localPath := filepath.Join(tempDir, object.Key)
+		err := s3Client.FGetObject(ctxTimeout, bucket.Spec.BucketName, object.Key, localPath, minio.GetObjectOptions{})
+		if err != nil {
+			err = fmt.Errorf("downloading object from bucket '%s' failed: %w", bucket.Spec.BucketName, err)
+			return sourcev1.BucketNotReady(bucket, sourcev1.BucketOperationFailedReason, err.Error()), err
+		}
+	}
+	return sourcev1.Bucket{}, nil
+}
+
+// authGCP creates a new Google Cloud Platform storage client
+// to interact with the storage service.
+func (r *BucketReconciler) authGCP(ctx context.Context, bucket sourcev1.Bucket) (*gcp.GCPClient, error) {
+	var client *gcp.GCPClient
+	var err error
+	if bucket.Spec.SecretRef != nil {
+		secretName := types.NamespacedName{
+			Namespace: bucket.GetNamespace(),
+			Name:      bucket.Spec.SecretRef.Name,
+		}
+
+		var secret corev1.Secret
+		if err := r.Get(ctx, secretName, &secret); err != nil {
+			return nil, fmt.Errorf("credentials secret error: %w", err)
+		}
+		if err := gcp.ValidateSecret(secret.Data, secret.Name); err != nil {
+			return nil, err
+		}
+		client, err = gcp.NewClient(ctx, option.WithCredentialsJSON(secret.Data["serviceaccount"]))
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		client, err = gcp.NewClient(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return client, nil
+
+}
+
+// authMinio creates a new Minio client to interact with S3
+// compatible storage services.
+func (r *BucketReconciler) authMinio(ctx context.Context, bucket sourcev1.Bucket) (*minio.Client, error) {
 	opt := minio.Options{
 		Region: bucket.Spec.Region,
 		Secure: !bucket.Spec.Insecure,

docs/api/source.md

@@ -2032,4 +2032,4 @@ string
 <p>Source interface must be supported by all API types.</p>
 <div class="admonition note">
 <p class="last">This page was automatically generated with <code>gen-crd-api-reference-docs</code></p>
-</div>
+</div>

docs/spec/v1alpha1/buckets.md

@@ -231,4 +231,4 @@ Wait for ready condition:
 
 ```bash
 kubectl -n gitios-system wait bucket/podinfo --for=condition=ready --timeout=1m
-```
+```