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
}
authGCP creates a new Google Cloud Platform storage client
to interact with the storage service.
package controllers
import (
	"context"
	"crypto/sha1"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"
	"github.com/go-logr/logr"
	"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"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	kuberecorder "k8s.io/client-go/tools/record"
	"k8s.io/client-go/tools/reference"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"github.com/fluxcd/pkg/apis/meta"
	"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"
)
type BucketReconciler struct {
	client.Client
	Scheme                *runtime.Scheme
	Storage               *Storage
	EventRecorder         kuberecorder.EventRecorder
	ExternalEventRecorder *events.Recorder
	MetricsRecorder       *metrics.Recorder
}
type BucketReconcilerOptions struct {
	MaxConcurrentReconciles int
}
func (r *BucketReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return r.SetupWithManagerAndOptions(mgr, BucketReconcilerOptions{})
}
func (r *BucketReconciler) SetupWithManagerAndOptions(mgr ctrl.Manager, opts BucketReconcilerOptions) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&sourcev1.Bucket{}).
		WithEventFilter(predicate.Or(predicate.GenerationChangedPredicate{}, predicates.ReconcileRequestedPredicate{})).
		WithOptions(controller.Options{MaxConcurrentReconciles: opts.MaxConcurrentReconciles}).
		Complete(r)
}
func (r *BucketReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	start := time.Now()
	log := ctrl.LoggerFrom(ctx)
	var bucket sourcev1.Bucket
	if err := r.Get(ctx, req.NamespacedName, &bucket); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	defer r.recordSuspension(ctx, bucket)
	if !controllerutil.ContainsFinalizer(&bucket, sourcev1.SourceFinalizer) {
		controllerutil.AddFinalizer(&bucket, sourcev1.SourceFinalizer)
		if err := r.Update(ctx, &bucket); err != nil {
			log.Error(err, "unable to register finalizer")
			return ctrl.Result{}, err
		}
	}
	if !bucket.ObjectMeta.DeletionTimestamp.IsZero() {
		return r.reconcileDelete(ctx, bucket)
	}
	if bucket.Spec.Suspend {
		log.Info("Reconciliation is suspended for this object")
		return ctrl.Result{}, nil
	}
	if r.MetricsRecorder != nil {
		objRef, err := reference.GetReference(r.Scheme, &bucket)
		if err != nil {
			return ctrl.Result{}, err
		}
		defer r.MetricsRecorder.RecordDuration(*objRef, start)
	}
	if resetBucket, ok := r.resetStatus(bucket); ok {
		bucket = resetBucket
		if err := r.updateStatus(ctx, req, bucket.Status); err != nil {
			log.Error(err, "unable to update status")
			return ctrl.Result{Requeue: true}, err
		}
		r.recordReadiness(ctx, bucket)
	}
	if v, ok := meta.ReconcileAnnotationValue(bucket.GetAnnotations()); ok {
		bucket.Status.SetLastHandledReconcileRequest(v)
	}
	if err := r.gc(bucket); err != nil {
		log.Error(err, "unable to purge old artifacts")
	}
	reconciledBucket, reconcileErr := r.reconcile(ctx, *bucket.DeepCopy())
	if err := r.updateStatus(ctx, req, reconciledBucket.Status); err != nil {
		log.Error(err, "unable to update status")
		return ctrl.Result{Requeue: true}, err
	}
	if reconcileErr != nil {
		r.event(ctx, reconciledBucket, events.EventSeverityError, reconcileErr.Error())
		r.recordReadiness(ctx, reconciledBucket)
		return ctrl.Result{Requeue: true}, reconcileErr
	}
	if bucket.Status.Artifact == nil || reconciledBucket.Status.Artifact.Revision != bucket.Status.Artifact.Revision {
		r.event(ctx, reconciledBucket, events.EventSeverityInfo, sourcev1.BucketReadyMessage(reconciledBucket))
	}
	r.recordReadiness(ctx, reconciledBucket)
	log.Info(fmt.Sprintf("Reconciliation finished in %s, next run in %s",
		time.Now().Sub(start).String(),
		bucket.GetInterval().Duration.String(),
	))
	return ctrl.Result{RequeueAfter: bucket.GetInterval().Duration}, nil
}
func (r *BucketReconciler) reconcile(ctx context.Context, bucket sourcev1.Bucket) (sourcev1.Bucket, error) {
	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)
	if bucket.Spec.Provider == sourcev1.GoogleBucketProvider {
		sourceBucket, err = r.reconcileWithGCP(ctx, bucket, tempDir)
		if err != nil {
			return sourceBucket, err
		}
	} else {
		sourceBucket, err = r.reconcileWithMinio(ctx, bucket, tempDir)
		if err != nil {
			return sourceBucket, err
		}
	}
	revision, err := r.checksum(tempDir)
	if err != nil {
		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
	}
	artifact := r.Storage.NewArtifactFor(bucket.Kind, bucket.GetObjectMeta(), revision, fmt.Sprintf("%s.tar.gz", revision))
	if apimeta.IsStatusConditionTrue(bucket.Status.Conditions, meta.ReadyCondition) && bucket.GetArtifact().HasRevision(artifact.Revision) {
		if artifact.URL != bucket.GetArtifact().URL {
			r.Storage.SetArtifactURL(bucket.GetArtifact())
			bucket.Status.URL = r.Storage.SetHostname(bucket.Status.URL)
		}
		return bucket, nil
	}
	err = r.Storage.MkdirAll(artifact)
	if err != nil {
		err = fmt.Errorf("mkdir dir error: %w", err)
		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
	}
	unlock, err := r.Storage.Lock(artifact)
	if err != nil {
		err = fmt.Errorf("unable to acquire lock: %w", err)
		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
	}
	defer unlock()
	if err := r.Storage.Archive(&artifact, tempDir, nil); err != nil {
		err = fmt.Errorf("storage archive error: %w", err)
		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
	}
	url, err := r.Storage.Symlink(artifact, "latest.tar.gz")
	if err != nil {
		err = fmt.Errorf("storage symlink error: %w", err)
		return sourcev1.BucketNotReady(bucket, sourcev1.StorageOperationFailedReason, err.Error()), err
	}
	message := fmt.Sprintf("Fetched revision: %s", artifact.Revision)
	return sourcev1.BucketReady(bucket, artifact, url, sourcev1.BucketOperationSucceedReason, message), nil
}
func (r *BucketReconciler) reconcileDelete(ctx context.Context, bucket sourcev1.Bucket) (ctrl.Result, error) {
	if err := r.gc(bucket); err != nil {
		r.event(ctx, bucket, events.EventSeverityError,
			fmt.Sprintf("garbage collection for deleted resource failed: %s", err.Error()))
		return ctrl.Result{}, err
	}
	r.recordReadiness(ctx, bucket)
	controllerutil.RemoveFinalizer(&bucket, sourcev1.SourceFinalizer)
	if err := r.Update(ctx, &bucket); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}
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
	}
	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
	}
	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)
	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
}
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
	}
	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
	}
	if bucket.Spec.Ignore != nil {
		ps = append(ps, sourceignore.ReadPatterns(strings.NewReader(*bucket.Spec.Ignore), nil)...)
	}
	matcher := sourceignore.NewMatcher(ps)
	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
}