fluxcd
/
source-controller
mirror of https://github.com/fluxcd/source-controller.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #413 from fluxcd/helmrepository-reconciler 

Rewrite `HelmRepositoryReconciler` to new standards
This commit is contained in:
Hidde Beydals 2021-08-12 14:12:31 +02:00 committed by GitHub
parent 04c125f1da 49639638b9
commit 7c95db88f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 977 additions and 693 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
api/v1beta1/helmrepository_types.go
View File

@ -21,8 +21,6 @@ import (
"github.com/fluxcd/pkg/apis/acl"
"github.com/fluxcd/pkg/apis/meta"
"github.com/fluxcd/pkg/runtime/conditions"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@ -107,47 +105,6 @@ const (
IndexationSucceededReason string = "IndexationSucceed"
)
// HelmRepositoryProgressing resets the conditions of the HelmRepository to
// metav1.Condition of type meta.ReadyCondition with status 'Unknown' and
// meta.ProgressingReason reason and message. It returns the modified
// HelmRepository.
func HelmRepositoryProgressing(repository HelmRepository) HelmRepository {
repository.Status.ObservedGeneration = repository.Generation
repository.Status.URL = ""
repository.Status.Conditions = []metav1.Condition{}
conditions.MarkUnknown(&repository, meta.ReadyCondition, meta.ProgressingReason, "reconciliation in progress")
return repository
}
// HelmRepositoryReady sets the given Artifact and URL on the HelmRepository and
// sets the meta.ReadyCondition to 'True', with the given reason and message. It
// returns the modified HelmRepository.
func HelmRepositoryReady(repository HelmRepository, artifact Artifact, url, reason, message string) HelmRepository {
repository.Status.Artifact = &artifact
repository.Status.URL = url
conditions.MarkTrue(&repository, meta.ReadyCondition, reason, message)
return repository
}
// HelmRepositoryNotReady sets the meta.ReadyCondition on the given
// HelmRepository to 'False', with the given reason and message. It returns the
// modified HelmRepository.
func HelmRepositoryNotReady(repository HelmRepository, reason, message string) HelmRepository {
conditions.MarkFalse(&repository, meta.ReadyCondition, reason, message)
return repository
}
// HelmRepositoryReadyMessage returns the message of the metav1.Condition of type
// meta.ReadyCondition with status 'True' if present, or an empty string.
func HelmRepositoryReadyMessage(repository HelmRepository) string {
if c := apimeta.FindStatusCondition(repository.Status.Conditions, meta.ReadyCondition); c != nil {
if c.Status == metav1.ConditionTrue {
return c.Message
}
}
return ""
}
// GetConditions returns the status conditions of the object.
func (in HelmRepository) GetConditions() []metav1.Condition {
return in.Status.Conditions

18
controllers/helmchart_controller.go
View File

@ -307,12 +307,17 @@ func (r *HelmChartReconciler) reconcileFromHelmRepository(ctx context.Context,
if secret, err := r.getHelmRepositorySecret(ctx, &repository); err != nil {
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
} else if secret != nil {
opts, cleanup, err := helm.ClientOptionsFromSecret(*secret)
tmpDir, err := os.MkdirTemp("", fmt.Sprintf("%s-%s-source-", chart.Name, chart.Namespace))
if err != nil {
err = fmt.Errorf("failed to create temporary directory for auth: %w", err)
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
}
defer os.RemoveAll(tmpDir)
opts, err := helm.ClientOptionsFromSecret(*secret, tmpDir)
if err != nil {
err = fmt.Errorf("auth options error: %w", err)
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
}
defer cleanup()
clientOpts = append(clientOpts, opts...)
}
@ -634,12 +639,17 @@ func (r *HelmChartReconciler) reconcileFromTarballArtifact(ctx context.Context,
if secret, err := r.getHelmRepositorySecret(ctx, repository); err != nil {
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
} else if secret != nil {
opts, cleanup, err := helm.ClientOptionsFromSecret(*secret)
tmpDir, err := os.MkdirTemp("", fmt.Sprintf("%s-%s-source-", chart.Name, chart.Namespace))
if err != nil {
err = fmt.Errorf("failed to create temporary directory for auth: %w", err)
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
}
defer os.RemoveAll(tmpDir)
opts, err := helm.ClientOptionsFromSecret(*secret, tmpDir)
if err != nil {
err = fmt.Errorf("auth options error: %w", err)
return sourcev1.HelmChartNotReady(chart, sourcev1.AuthenticationFailedReason, err.Error()), err
}
defer cleanup()
clientOpts = append(clientOpts, opts...)
}

579
controllers/helmrepository_controller.go
View File

@ -21,17 +21,15 @@ import (
"context"
"fmt"
"net/url"
"os"
"time"
"github.com/go-logr/logr"
"helm.sh/helm/v3/pkg/getter"
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"
kerrors "k8s.io/apimachinery/pkg/util/errors"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
@ -40,8 +38,10 @@ import (
"sigs.k8s.io/yaml"
"github.com/fluxcd/pkg/apis/meta"
"github.com/fluxcd/pkg/runtime/conditions"
helper "github.com/fluxcd/pkg/runtime/controller"
"github.com/fluxcd/pkg/runtime/events"
"github.com/fluxcd/pkg/runtime/metrics"
"github.com/fluxcd/pkg/runtime/patch"
"github.com/fluxcd/pkg/runtime/predicates"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
@ -56,12 +56,11 @@ import (
// HelmRepositoryReconciler reconciles a HelmRepository object
type HelmRepositoryReconciler struct {
client.Client
Scheme *runtime.Scheme
Storage *Storage
Getters getter.Providers
EventRecorder kuberecorder.EventRecorder
ExternalEventRecorder *events.Recorder
MetricsRecorder *metrics.Recorder
helper.Events
helper.Metrics
Getters getter.Providers
Storage *Storage
}
type HelmRepositoryReconcilerOptions struct {
@ -80,306 +79,392 @@ func (r *HelmRepositoryReconciler) SetupWithManagerAndOptions(mgr ctrl.Manager,
Complete(r)
}
func (r *HelmRepositoryReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
func (r *HelmRepositoryReconciler) Reconcile(ctx context.Context, req ctrl.Request) (result ctrl.Result, retErr error) {
start := time.Now()
log := logr.FromContext(ctx)
var repository sourcev1.HelmRepository
if err := r.Get(ctx, req.NamespacedName, &repository); err != nil {
// Fetch the HelmRepository
obj := &sourcev1.HelmRepository{}
if err := r.Get(ctx, req.NamespacedName, obj); err != nil {
return ctrl.Result{}, client.IgnoreNotFound(err)
}
// Add our finalizer if it does not exist
if !controllerutil.ContainsFinalizer(&repository, sourcev1.SourceFinalizer) {
controllerutil.AddFinalizer(&repository, sourcev1.SourceFinalizer)
if err := r.Update(ctx, &repository); err != nil {
log.Error(err, "unable to register finalizer")
return ctrl.Result{}, err
}
}
// Record suspended status metric
r.RecordSuspend(ctx, obj, obj.Spec.Suspend)
// Examine if the object is under deletion
if !repository.ObjectMeta.DeletionTimestamp.IsZero() {
return r.reconcileDelete(ctx, repository)
}
// Return early if the object is suspended.
if repository.Spec.Suspend {
// Return early if the object is suspended
if obj.Spec.Suspend {
log.Info("Reconciliation is suspended for this object")
return ctrl.Result{}, nil
}
// record reconciliation duration
if r.MetricsRecorder != nil {
objRef, err := reference.GetReference(r.Scheme, &repository)
if err != nil {
return ctrl.Result{}, err
// Initialize the patch helper
patchHelper, err := patch.NewHelper(obj, r.Client)
if err != nil {
return ctrl.Result{}, err
}
// Always attempt to patch the object and status after each reconciliation
defer func() {
// Record the value of the reconciliation request, if any
if v, ok := meta.ReconcileAnnotationValue(obj.GetAnnotations()); ok {
obj.Status.SetLastHandledReconcileRequest(v)
}
defer r.MetricsRecorder.RecordDuration(*objRef, start)
}
// set initial status
if resetRepository, ok := r.resetStatus(repository); ok {
repository = resetRepository
if err := r.updateStatus(ctx, req, repository.Status); err != nil {
log.Error(err, "unable to update status")
return ctrl.Result{Requeue: true}, err
// Summarize Ready condition
conditions.SetSummary(obj,
meta.ReadyCondition,
conditions.WithConditions(
sourcev1.FetchFailedCondition,
sourcev1.ArtifactOutdatedCondition,
sourcev1.ArtifactUnavailableCondition,
),
conditions.WithNegativePolarityConditions(
sourcev1.FetchFailedCondition,
sourcev1.ArtifactOutdatedCondition,
sourcev1.ArtifactUnavailableCondition,
),
)
// Patch the object, ignoring conflicts on the conditions owned by this controller
patchOpts := []patch.Option{
patch.WithOwnedConditions{
Conditions: []string{
sourcev1.FetchFailedCondition,
sourcev1.ArtifactOutdatedCondition,
sourcev1.ArtifactUnavailableCondition,
meta.ReadyCondition,
meta.ReconcilingCondition,
meta.StalledCondition,
},
},
}
r.recordReadiness(ctx, repository)
// Determine if the resource is still being reconciled, or if it has stalled, and record this observation
if retErr == nil && (result.IsZero() || !result.Requeue) {
// We are no longer reconciling
conditions.Delete(obj, meta.ReconcilingCondition)
// We have now observed this generation
patchOpts = append(patchOpts, patch.WithStatusObservedGeneration{})
readyCondition := conditions.Get(obj, meta.ReadyCondition)
switch readyCondition.Status {
case metav1.ConditionFalse:
// As we are no longer reconciling and the end-state
// is not ready, the reconciliation has stalled
conditions.MarkStalled(obj, readyCondition.Reason, readyCondition.Message)
case metav1.ConditionTrue:
// As we are no longer reconciling and the end-state
// is ready, the reconciliation is no longer stalled
conditions.Delete(obj, meta.StalledCondition)
}
}
// Finally, patch the resource
if err := patchHelper.Patch(ctx, obj, patchOpts...); err != nil {
retErr = kerrors.NewAggregate([]error{retErr, err})
}
// Always record readiness and duration metrics
r.Metrics.RecordReadiness(ctx, obj)
r.Metrics.RecordDuration(ctx, obj, start)
}()
// Add finalizer first if not exist to avoid the race condition
// between init and delete
if !controllerutil.ContainsFinalizer(obj, sourcev1.SourceFinalizer) {
controllerutil.AddFinalizer(obj, sourcev1.SourceFinalizer)
return ctrl.Result{Requeue: true}, nil
}
// record the value of the reconciliation request, if any
// TODO(hidde): would be better to defer this in combination with
// always patching the status sub-resource after a reconciliation.
if v, ok := meta.ReconcileAnnotationValue(repository.GetAnnotations()); ok {
repository.Status.SetLastHandledReconcileRequest(v)
// Examine if the object is under deletion
if !obj.ObjectMeta.DeletionTimestamp.IsZero() {
return r.reconcileDelete(ctx, obj)
}
// purge old artifacts from storage
if err := r.gc(repository); err != nil {
log.Error(err, "unable to purge old artifacts")
}
// reconcile repository by downloading the index.yaml file
reconciledRepository, reconcileErr := r.reconcile(ctx, *repository.DeepCopy())
// update status with the reconciliation result
if err := r.updateStatus(ctx, req, reconciledRepository.Status); err != nil {
log.Error(err, "unable to update status")
return ctrl.Result{Requeue: true}, err
}
// if reconciliation failed, record the failure and requeue immediately
if reconcileErr != nil {
r.event(ctx, reconciledRepository, events.EventSeverityError, reconcileErr.Error())
r.recordReadiness(ctx, reconciledRepository)
return ctrl.Result{Requeue: true}, reconcileErr
}
// emit revision change event
if repository.Status.Artifact == nil || reconciledRepository.Status.Artifact.Revision != repository.Status.Artifact.Revision {
r.event(ctx, reconciledRepository, events.EventSeverityInfo, sourcev1.HelmRepositoryReadyMessage(reconciledRepository))
}
r.recordReadiness(ctx, reconciledRepository)
log.Info(fmt.Sprintf("Reconciliation finished in %s, next run in %s",
time.Now().Sub(start).String(),
repository.GetRequeueAfter().String(),
))
return ctrl.Result{RequeueAfter: repository.GetRequeueAfter()}, nil
// Reconcile actual object
return r.reconcile(ctx, obj)
}
func (r *HelmRepositoryReconciler) reconcile(ctx context.Context, repository sourcev1.HelmRepository) (sourcev1.HelmRepository, error) {
clientOpts := []getter.Option{
getter.WithURL(repository.Spec.URL),
getter.WithTimeout(repository.Spec.Timeout.Duration),
getter.WithPassCredentialsAll(repository.Spec.PassCredentials),
// reconcile steps through the actual reconciliation tasks for the object, it returns early on the first step that
// produces an error.
func (r *HelmRepositoryReconciler) reconcile(ctx context.Context, obj *sourcev1.HelmRepository) (ctrl.Result, error) {
// Mark the resource as under reconciliation
conditions.MarkReconciling(obj, meta.ProgressingReason, "")
// Reconcile the storage data
if result, err := r.reconcileStorage(ctx, obj); err != nil {
return result, err
}
if repository.Spec.SecretRef != nil {
// Reconcile the source from upstream
var index helm.ChartRepository
var artifact sourcev1.Artifact
if result, err := r.reconcileSource(ctx, obj, &artifact, &index); err != nil || result.IsZero() {
return result, err
}
// Reconcile the artifact to storage
if result, err := r.reconcileArtifact(ctx, obj, artifact, index); err != nil || result.IsZero() {
return result, err
}
return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
}
// reconcileStorage ensures the current state of the storage matches the desired and previously observed state.
//
// All artifacts for the resource except for the current one are garbage collected from the storage.
// If the artifact in the Status object of the resource disappeared from storage, it is removed from the object.
// If the object does not have an artifact in its Status object, a v1beta1.ArtifactUnavailableCondition is set.
// If the hostname of any of the URLs on the object do not match the current storage server hostname, they are updated.
//
// The caller should assume a failure if an error is returned, or the Result is zero.
func (r *HelmRepositoryReconciler) reconcileStorage(ctx context.Context, obj *sourcev1.HelmRepository) (ctrl.Result, error) {
// Garbage collect previous advertised artifact(s) from storage
_ = r.garbageCollect(ctx, obj)
// Determine if the advertised artifact is still in storage
if artifact := obj.GetArtifact(); artifact != nil && !r.Storage.ArtifactExist(*artifact) {
obj.Status.Artifact = nil
obj.Status.URL = ""
}
// Record that we do not have an artifact
if obj.GetArtifact() == nil {
conditions.MarkTrue(obj, sourcev1.ArtifactUnavailableCondition, "NoArtifact", "No artifact for resource in storage")
return ctrl.Result{Requeue: true}, nil
}
conditions.Delete(obj, sourcev1.ArtifactUnavailableCondition)
// Always update URLs to ensure hostname is up-to-date
// TODO(hidde): we may want to send out an event only if we notice the URL has changed
r.Storage.SetArtifactURL(obj.GetArtifact())
obj.Status.URL = r.Storage.SetHostname(obj.Status.URL)
return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
}
// reconcileSource ensures the upstream Helm repository can be reached and downloaded out using the declared
// configuration, and observes its state.
//
// The Helm repository index is downloaded using the defined configuration, and in case of an error during this process
// (including transient errors), it records v1beta1.FetchFailedCondition=True and returns early.
// If the download is successful, the given artifact pointer is set to a new artifact with the available metadata, and
// the index pointer is set to the newly downloaded index.
//
// The caller should assume a failure if an error is returned, or the Result is zero.
func (r *HelmRepositoryReconciler) reconcileSource(ctx context.Context, obj *sourcev1.HelmRepository, artifact *sourcev1.Artifact, index *helm.ChartRepository) (ctrl.Result, error) {
// Configure Helm client to access repository
clientOpts := []getter.Option{
getter.WithTimeout(obj.Spec.Timeout.Duration),
getter.WithURL(obj.Spec.URL),
getter.WithPassCredentialsAll(obj.Spec.PassCredentials),
}
// Configure any authentication related options
if obj.Spec.SecretRef != nil {
// Attempt to retrieve secret
name := types.NamespacedName{
Namespace: repository.GetNamespace(),
Name: repository.Spec.SecretRef.Name,
Namespace: obj.GetNamespace(),
Name: obj.Spec.SecretRef.Name,
}
var secret corev1.Secret
err := r.Client.Get(ctx, name, &secret)
if err != nil {
err = fmt.Errorf("auth secret error: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.AuthenticationFailedReason, err.Error()), err
if err := r.Client.Get(ctx, name, &secret); err != nil {
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, sourcev1.AuthenticationFailedReason,
"Failed to get secret '%s': %s", name.String(), err.Error())
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.AuthenticationFailedReason,
"Failed to get secret '%s': %s", name.String(), err.Error())
// Return error as the world as observed may change
return ctrl.Result{}, err
}
opts, cleanup, err := helm.ClientOptionsFromSecret(secret)
// Get client options from secret
tmpDir, err := os.MkdirTemp("", fmt.Sprintf("%s-%s-auth-", obj.Name, obj.Namespace))
if err != nil {
err = fmt.Errorf("auth options error: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.AuthenticationFailedReason, err.Error()), err
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, sourcev1.AuthenticationFailedReason,
"Failed to create temporary directory for credentials: %s", err.Error())
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.AuthenticationFailedReason,
"Failed to create temporary directory for credentials: %s", err.Error())
return ctrl.Result{}, err
}
defer os.RemoveAll(tmpDir)
// Construct actual options
opts, err := helm.ClientOptionsFromSecret(secret, tmpDir)
if err != nil {
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, sourcev1.AuthenticationFailedReason,
"Failed to configure Helm client with secret data: %s", err)
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.AuthenticationFailedReason,
"Failed to configure Helm client with secret data: %s", err)
// Return err as the content of the secret may change
return ctrl.Result{}, err
}
defer cleanup()
clientOpts = append(clientOpts, opts...)
}
chartRepo, err := helm.NewChartRepository(repository.Spec.URL, r.Getters, clientOpts)
// Construct Helm chart repository with options and download index
newIndex, err := helm.NewChartRepository(obj.Spec.URL, r.Getters, clientOpts)
if err != nil {
switch err.(type) {
case *url.Error:
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.URLInvalidReason, err.Error()), err
ctrl.LoggerFrom(ctx).Error(err, "invalid Helm repository URL")
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, sourcev1.URLInvalidReason,
"Invalid Helm repository URL: %s", err.Error())
return ctrl.Result{}, nil
default:
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
ctrl.LoggerFrom(ctx).Error(err, "failed to construct Helm client")
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, meta.FailedReason,
"Failed to construct Helm client: %s", err.Error())
return ctrl.Result{}, nil
}
}
if err := chartRepo.DownloadIndex(); err != nil {
err = fmt.Errorf("failed to download repository index: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.IndexationFailedReason, err.Error()), err
if err := newIndex.DownloadIndex(); err != nil {
conditions.MarkTrue(obj, sourcev1.FetchFailedCondition, meta.FailedReason,
"Failed to download Helm repository index: %s", err.Error())
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.FetchFailedCondition,
"Failed to download Helm repository index: %s", err.Error())
// Coin flip on transient or persistent error, return error and hope for the best
return ctrl.Result{}, err
}
*index = *newIndex
conditions.Delete(obj, sourcev1.FetchFailedCondition)
// Mark observations about the revision on the object
if !obj.GetArtifact().HasRevision(index.Checksum) {
conditions.MarkTrue(obj, sourcev1.ArtifactOutdatedCondition, "NewRevision",
"New index revision '%s'", index.Checksum)
}
indexBytes, err := yaml.Marshal(&chartRepo.Index)
if err != nil {
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err
}
hash := r.Storage.Checksum(bytes.NewReader(indexBytes))
artifact := r.Storage.NewArtifactFor(repository.Kind,
repository.ObjectMeta.GetObjectMeta(),
hash,
fmt.Sprintf("index-%s.yaml", hash))
// return early on unchanged index
if apimeta.IsStatusConditionTrue(repository.Status.Conditions, meta.ReadyCondition) && repository.GetArtifact().HasRevision(artifact.Revision) {
if artifact.URL != repository.GetArtifact().URL {
r.Storage.SetArtifactURL(repository.GetArtifact())
repository.Status.URL = r.Storage.SetHostname(repository.Status.URL)
// Create potential new artifact
*artifact = r.Storage.NewArtifactFor(obj.Kind,
obj.ObjectMeta.GetObjectMeta(),
index.Checksum,
fmt.Sprintf("index-%s.yaml", index.Checksum))
return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
}
// reconcileArtifact stores a new artifact in the storage, if the current observation on the object does not match the
// given data.
//
// The inspection of the given data to the object is differed, ensuring any stale observations as
// v1beta1.ArtifactUnavailableCondition and v1beta1.ArtifactOutdatedCondition are always deleted.
// If the given artifact does not differ from the object's current, it returns early.
// On a successful write, the artifact in the status of the given object is set, and the symlink in the storage is
// updated to its path.
//
// The caller should assume a failure if an error is returned, or the Result is zero.
func (r *HelmRepositoryReconciler) reconcileArtifact(ctx context.Context, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index helm.ChartRepository) (ctrl.Result, error) {
// Always restore the Ready condition in case it got removed due to a transient error
defer func() {
if obj.GetArtifact() != nil {
conditions.Delete(obj, sourcev1.ArtifactUnavailableCondition)
}
return repository, nil
if obj.GetArtifact().HasRevision(artifact.Revision) {
conditions.Delete(obj, sourcev1.ArtifactOutdatedCondition)
conditions.MarkTrue(obj, meta.ReadyCondition, meta.SucceededReason,
"Stored artifact for revision '%s'", artifact.Revision)
}
}()
// The artifact is up-to-date
if obj.GetArtifact().HasRevision(artifact.Revision) {
ctrl.LoggerFrom(ctx).Info(fmt.Sprintf("Already up to date, current revision '%s'", artifact.Revision))
return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
}
// create artifact dir
err = r.Storage.MkdirAll(artifact)
if err != nil {
err = fmt.Errorf("unable to create repository index directory: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err
// Confirm the wrapper contains a loaded index
if index.Index == nil {
err := fmt.Errorf("failed to reconcile artifact: no repository index provided")
return ctrl.Result{}, err
}
// acquire lock
// Ensure artifact directory exists and acquire lock
if err := r.Storage.MkdirAll(artifact); err != nil {
err = fmt.Errorf("failed to create artifact directory: %w", err)
return ctrl.Result{}, err
}
unlock, err := r.Storage.Lock(artifact)
if err != nil {
err = fmt.Errorf("unable to acquire lock: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err
err = fmt.Errorf("failed to acquire lock for artifact: %w", err)
return ctrl.Result{}, err
}
defer unlock()
// save artifact to storage
if err := r.Storage.AtomicWriteFile(&artifact, bytes.NewReader(indexBytes), 0644); err != nil {
err = fmt.Errorf("unable to write repository index file: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err
}
// update index symlink
indexURL, err := r.Storage.Symlink(artifact, "index.yaml")
// Save artifact to storage
b, err := yaml.Marshal(index.Index)
if err != nil {
err = fmt.Errorf("storage error: %w", err)
return sourcev1.HelmRepositoryNotReady(repository, sourcev1.StorageOperationFailedReason, err.Error()), err
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.StorageOperationFailedReason,
"Failed to marshal downloaded index: %s", err)
return ctrl.Result{}, err
}
if err := r.Storage.AtomicWriteFile(&artifact, bytes.NewReader(b), 0644); err != nil {
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.StorageOperationFailedReason,
"Failed to write Helm repository index to storage: %s", err.Error())
return ctrl.Result{}, err
}
r.EventWithMetaf(ctx, obj, map[string]string{
"revision": artifact.Revision,
"checksum": artifact.Checksum,
}, events.EventSeverityInfo, "NewArtifact", "Stored artifact for revision '%s'", artifact.Revision)
message := fmt.Sprintf("Fetched revision: %s", artifact.Revision)
return sourcev1.HelmRepositoryReady(repository, artifact, indexURL, sourcev1.IndexationSucceededReason, message), nil
// Record it on the object
obj.Status.Artifact = artifact.DeepCopy()
// Update symlink on a "best effort" basis
url, err := r.Storage.Symlink(artifact, "latest.tar.gz")
if err != nil {
r.Eventf(ctx, obj, events.EventSeverityError, sourcev1.StorageOperationFailedReason,
"Failed to update status URL symlink: %s", err)
}
if url != "" {
obj.Status.URL = url
}
return ctrl.Result{RequeueAfter: obj.GetRequeueAfter()}, nil
}
func (r *HelmRepositoryReconciler) reconcileDelete(ctx context.Context, repository sourcev1.HelmRepository) (ctrl.Result, error) {
// Our finalizer is still present, so lets handle garbage collection
if err := r.gc(repository); err != nil {
r.event(ctx, repository, events.EventSeverityError,
fmt.Sprintf("garbage collection for deleted resource failed: %s", err.Error()))
// reconcileDelete handles the delete of an object. It first garbage collects all artifacts for the object from the
// artifact storage, if successful, the finalizer is removed from the object.
func (r *HelmRepositoryReconciler) reconcileDelete(ctx context.Context, obj *sourcev1.HelmRepository) (ctrl.Result, error) {
// Garbage collect the resource's artifacts
if err := r.garbageCollect(ctx, obj); err != nil {
// Return the error so we retry the failed garbage collection
return ctrl.Result{}, err
}
// Record deleted status
r.recordReadiness(ctx, repository)
// Remove our finalizer from the list and update it
controllerutil.RemoveFinalizer(&repository, sourcev1.SourceFinalizer)
if err := r.Update(ctx, &repository); err != nil {
return ctrl.Result{}, err
}
// Remove our finalizer from the list
controllerutil.RemoveFinalizer(obj, sourcev1.SourceFinalizer)
// Stop reconciliation as the object is being deleted
return ctrl.Result{}, nil
}
// resetStatus returns a modified v1beta1.HelmRepository and a boolean indicating
// if the status field has been reset.
func (r *HelmRepositoryReconciler) resetStatus(repository sourcev1.HelmRepository) (sourcev1.HelmRepository, bool) {
// We do not have an artifact, or it does no longer exist
if repository.GetArtifact() == nil || !r.Storage.ArtifactExist(*repository.GetArtifact()) {
repository = sourcev1.HelmRepositoryProgressing(repository)
repository.Status.Artifact = nil
return repository, true
// garbageCollect performs a garbage collection for the given v1beta1.HelmRepository. It removes all but the current
// artifact except for when the deletion timestamp is set, which will result in the removal of all artifacts for the
// resource.
func (r *HelmRepositoryReconciler) garbageCollect(ctx context.Context, obj *sourcev1.HelmRepository) error {
if !obj.DeletionTimestamp.IsZero() {
if err := r.Storage.RemoveAll(r.Storage.NewArtifactFor(obj.Kind, obj.GetObjectMeta(), "", "*")); err != nil {
r.Eventf(ctx, obj, events.EventSeverityError, "GarbageCollectionFailed",
"Garbage collection for deleted resource failed: %s", err)
return err
}
obj.Status.Artifact = nil
// TODO(hidde): we should only push this event if we actually garbage collected something
r.Eventf(ctx, obj, events.EventSeverityInfo, "GarbageCollectionSucceeded",
"Garbage collected artifacts for deleted resource")
return nil
}
if repository.Generation != repository.Status.ObservedGeneration {
return sourcev1.HelmRepositoryProgressing(repository), true
}
return repository, false
}
// gc performs a garbage collection for the given v1beta1.HelmRepository.
// It removes all but the current artifact except for when the
// deletion timestamp is set, which will result in the removal of
// all artifacts for the resource.
func (r *HelmRepositoryReconciler) gc(repository sourcev1.HelmRepository) error {
if !repository.DeletionTimestamp.IsZero() {
return r.Storage.RemoveAll(r.Storage.NewArtifactFor(repository.Kind, repository.GetObjectMeta(), "", "*"))
}
if repository.GetArtifact() != nil {
return r.Storage.RemoveAllButCurrent(*repository.GetArtifact())
if obj.GetArtifact() != nil {
if err := r.Storage.RemoveAllButCurrent(*obj.GetArtifact()); err != nil {
r.Eventf(ctx, obj, events.EventSeverityError, "GarbageCollectionFailed",
"Garbage collection of old artifacts failed: %s", err)
return err
}
// TODO(hidde): we should only push this event if we actually garbage collected something
r.Eventf(ctx, obj, events.EventSeverityInfo, "GarbageCollectionSucceeded",
"Garbage collected old artifacts")
}
return nil
}
// event emits a Kubernetes event and forwards the event to notification controller if configured
func (r *HelmRepositoryReconciler) event(ctx context.Context, repository sourcev1.HelmRepository, severity, msg string) {
log := logr.FromContext(ctx)
if r.EventRecorder != nil {
r.EventRecorder.Eventf(&repository, "Normal", severity, msg)
}
if r.ExternalEventRecorder != nil {
objRef, err := reference.GetReference(r.Scheme, &repository)
if err != nil {
log.Error(err, "unable to send event")
return
}
if err := r.ExternalEventRecorder.Eventf(*objRef, nil, severity, severity, msg); err != nil {
log.Error(err, "unable to send event")
return
}
}
}
func (r *HelmRepositoryReconciler) recordReadiness(ctx context.Context, repository sourcev1.HelmRepository) {
log := logr.FromContext(ctx)
if r.MetricsRecorder == nil {
return
}
objRef, err := reference.GetReference(r.Scheme, &repository)
if err != nil {
log.Error(err, "unable to record readiness metric")
return
}
if rc := apimeta.FindStatusCondition(repository.Status.Conditions, meta.ReadyCondition); rc != nil {
r.MetricsRecorder.RecordCondition(*objRef, *rc, !repository.DeletionTimestamp.IsZero())
} else {
r.MetricsRecorder.RecordCondition(*objRef, metav1.Condition{
Type: meta.ReadyCondition,
Status: metav1.ConditionUnknown,
}, !repository.DeletionTimestamp.IsZero())
}
}
func (r *HelmRepositoryReconciler) updateStatus(ctx context.Context, req ctrl.Request, newStatus sourcev1.HelmRepositoryStatus) error {
var repository sourcev1.HelmRepository
if err := r.Get(ctx, req.NamespacedName, &repository); err != nil {
return err
}
patch := client.MergeFrom(repository.DeepCopy())
repository.Status = newStatus
return r.Status().Patch(ctx, &repository, patch)
}
func (r *HelmRepositoryReconciler) recordSuspension(ctx context.Context, hr sourcev1.HelmRepository) {
if r.MetricsRecorder == nil {
return
}
log := logr.FromContext(ctx)
objRef, err := reference.GetReference(r.Scheme, &hr)
if err != nil {
log.Error(err, "unable to record suspended metric")
return
}
if !hr.DeletionTimestamp.IsZero() {
r.MetricsRecorder.RecordSuspend(*objRef, false)
} else {
r.MetricsRecorder.RecordSuspend(*objRef, hr.Spec.Suspend)
}
}

879
controllers/helmrepository_controller_test.go
View File

@ -18,394 +18,593 @@ package controllers
import (
"context"
"fmt"
"net/http"
"os"
"path"
"path/filepath"
"strings"
"time"
"testing"
. "github.com/onsi/ginkgo"
"github.com/go-logr/logr"
. "github.com/onsi/gomega"
"helm.sh/helm/v3/pkg/getter"
"helm.sh/helm/v3/pkg/repo"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
fakeclient "sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/yaml"
"github.com/fluxcd/pkg/apis/meta"
"github.com/fluxcd/pkg/helmtestserver"
"github.com/fluxcd/pkg/runtime/conditions"
sourcev1 "github.com/fluxcd/source-controller/api/v1beta1"
"github.com/fluxcd/source-controller/internal/helm"
)
var _ = Describe("HelmRepositoryReconciler", func() {
var (
testGetters = getter.Providers{
getter.Provider{
Schemes: []string{"http", "https"},
New: getter.NewHTTPGetter,
},
}
)
const (
timeout = time.Second * 30
interval = time.Second * 1
indexInterval = time.Second * 2
repositoryTimeout = time.Second * 5
)
func TestHelmRepositoryReconciler_Reconcile(t *testing.T) {
g := NewWithT(t)
Context("HelmRepository", func() {
var (
namespace *corev1.Namespace
helmServer *helmtestserver.HelmServer
err error
)
testServer, err := helmtestserver.NewTempHelmServer()
g.Expect(err).NotTo(HaveOccurred())
defer os.RemoveAll(testServer.Root())
BeforeEach(func() {
namespace = &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{Name: "helm-repository-" + randStringRunes(5)},
g.Expect(testServer.PackageChart("testdata/charts/helmchart")).To(Succeed())
g.Expect(testServer.GenerateIndex()).To(Succeed())
testServer.Start()
defer testServer.Stop()
obj := &sourcev1.HelmRepository{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "helmrepository-reconcile-",
Namespace: "default",
},
Spec: sourcev1.HelmRepositorySpec{
Interval: metav1.Duration{Duration: interval},
URL: testServer.URL(),
},
}
g.Expect(testEnv.Create(ctx, obj)).To(Succeed())
key := client.ObjectKey{Name: obj.Name, Namespace: obj.Namespace}
// Wait for finalizer to be set
g.Eventually(func() bool {
if err := testEnv.Get(ctx, key, obj); err != nil {
return false
}
return len(obj.Finalizers) > 0
}, timeout).Should(BeTrue())
// Wait for HelmRepository to be Ready
g.Eventually(func() bool {
if err := testEnv.Get(ctx, key, obj); err != nil {
return false
}
if !conditions.IsReady(obj) && obj.Status.Artifact == nil {
return false
}
readyCondition := conditions.Get(obj, meta.ReadyCondition)
return readyCondition.Status == metav1.ConditionTrue &&
obj.Generation == readyCondition.ObservedGeneration
}, timeout).Should(BeTrue())
g.Expect(testEnv.Delete(ctx, obj)).To(Succeed())
// Wait for HelmRepository to be deleted
g.Eventually(func() bool {
if err := testEnv.Get(ctx, key, obj); err != nil {
return apierrors.IsNotFound(err)
}
return false
}, timeout).Should(BeTrue())
}
func TestHelmRepositoryReconciler_reconcileStorage(t *testing.T) {
tests := []struct {
name string
beforeFunc func(obj *sourcev1.HelmRepository, storage *Storage) error
want ctrl.Result
wantErr bool
assertArtifact *sourcev1.Artifact
assertConditions []metav1.Condition
assertPaths []string
}{
{
name: "garbage collects",
beforeFunc: func(obj *sourcev1.HelmRepository, storage *Storage) error {
revisions := []string{"a", "b", "c"}
for n := range revisions {
v := revisions[n]
obj.Status.Artifact = &sourcev1.Artifact{
Path: fmt.Sprintf("/reconcile-storage/%s.txt", v),
Revision: v,
}
if err := testStorage.MkdirAll(*obj.Status.Artifact); err != nil {
return err
}
if err := testStorage.AtomicWriteFile(obj.Status.Artifact, strings.NewReader(v), 0644); err != nil {
return err
}
}
testStorage.SetArtifactURL(obj.Status.Artifact)
return nil
},
assertArtifact: &sourcev1.Artifact{
Path: "/reconcile-storage/c.txt",
Revision: "c",
Checksum: "2e7d2c03a9507ae265ecf5b5356885a53393a2029d241394997265a1a25aefc6",
URL: testStorage.Hostname + "/reconcile-storage/c.txt",
},
assertPaths: []string{
"/reconcile-storage/c.txt",
"!/reconcile-storage/b.txt",
"!/reconcile-storage/a.txt",
},
},
{
name: "notices missing artifact in storage",
beforeFunc: func(obj *sourcev1.HelmRepository, storage *Storage) error {
obj.Status.Artifact = &sourcev1.Artifact{
Path: fmt.Sprintf("/reconcile-storage/invalid.txt"),
Revision: "d",
}
testStorage.SetArtifactURL(obj.Status.Artifact)
return nil
},
want: ctrl.Result{Requeue: true},
assertPaths: []string{
"!/reconcile-storage/invalid.txt",
},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.ArtifactUnavailableCondition, "NoArtifact", "No artifact for resource in storage"),
},
},
{
name: "updates hostname on diff from current",
beforeFunc: func(obj *sourcev1.HelmRepository, storage *Storage) error {
obj.Status.Artifact = &sourcev1.Artifact{
Path: fmt.Sprintf("/reconcile-storage/hostname.txt"),
Revision: "f",
Checksum: "3b9c358f36f0a31b6ad3e14f309c7cf198ac9246e8316f9ce543d5b19ac02b80",
URL: "http://outdated.com/reconcile-storage/hostname.txt",
}
if err := testStorage.MkdirAll(*obj.Status.Artifact); err != nil {
return err
}
if err := testStorage.AtomicWriteFile(obj.Status.Artifact, strings.NewReader("file"), 0644); err != nil {
return err
}
return nil
},
assertPaths: []string{
"/reconcile-storage/hostname.txt",
},
assertArtifact: &sourcev1.Artifact{
Path: "/reconcile-storage/hostname.txt",
Revision: "f",
Checksum: "3b9c358f36f0a31b6ad3e14f309c7cf198ac9246e8316f9ce543d5b19ac02b80",
URL: testStorage.Hostname + "/reconcile-storage/hostname.txt",
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
g := NewWithT(t)
r := &HelmRepositoryReconciler{
Storage: testStorage,
}
err = k8sClient.Create(context.Background(), namespace)
Expect(err).NotTo(HaveOccurred(), "failed to create test namespace")
helmServer, err = helmtestserver.NewTempHelmServer()
Expect(err).To(Succeed())
})
AfterEach(func() {
os.RemoveAll(helmServer.Root())
helmServer.Stop()
Eventually(func() error {
return k8sClient.Delete(context.Background(), namespace)
}, timeout, interval).Should(Succeed(), "failed to delete test namespace")
})
It("Creates artifacts for", func() {
helmServer.Start()
Expect(helmServer.PackageChart(path.Join("testdata/charts/helmchart"))).Should(Succeed())
Expect(helmServer.GenerateIndex()).Should(Succeed())
key := types.NamespacedName{
Name: "helmrepository-sample-" + randStringRunes(5),
Namespace: namespace.Name,
}
created := &sourcev1.HelmRepository{
obj := &sourcev1.HelmRepository{
ObjectMeta: metav1.ObjectMeta{
Name: key.Name,
Namespace: key.Namespace,
},
Spec: sourcev1.HelmRepositorySpec{
URL: helmServer.URL(),
Interval: metav1.Duration{Duration: indexInterval},
Timeout: &metav1.Duration{Duration: repositoryTimeout},
GenerateName: "test-",
},
}
Expect(k8sClient.Create(context.Background(), created)).Should(Succeed())
if tt.beforeFunc != nil {
g.Expect(tt.beforeFunc(obj, testStorage)).To(Succeed())
}
By("Expecting artifact")
got := &sourcev1.HelmRepository{}
Eventually(func() bool {
_ = k8sClient.Get(context.Background(), key, got)
return got.Status.Artifact != nil && ginkgoTestStorage.ArtifactExist(*got.Status.Artifact)
}, timeout, interval).Should(BeTrue())
got, err := r.reconcileStorage(context.TODO(), obj)
g.Expect(err != nil).To(Equal(tt.wantErr))
g.Expect(got).To(Equal(tt.want))
By("Updating the chart index")
// Regenerating the index is sufficient to make the revision change
Expect(helmServer.GenerateIndex()).Should(Succeed())
g.Expect(obj.Status.Artifact).To(MatchArtifact(tt.assertArtifact))
if tt.assertArtifact != nil && tt.assertArtifact.URL != "" {
g.Expect(obj.Status.Artifact.URL).To(Equal(tt.assertArtifact.URL))
}
g.Expect(obj.Status.Conditions).To(conditions.MatchConditions(tt.assertConditions))
By("Expecting revision change and GC")
Eventually(func() bool {
now := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, now)
// Test revision change and garbage collection
return now.Status.Artifact.Revision != got.Status.Artifact.Revision &&
!ginkgoTestStorage.ArtifactExist(*got.Status.Artifact)
}, timeout, interval).Should(BeTrue())
updated := &sourcev1.HelmRepository{}
Expect(k8sClient.Get(context.Background(), key, updated)).Should(Succeed())
updated.Spec.URL = "invalid#url?"
Expect(k8sClient.Update(context.Background(), updated)).Should(Succeed())
Eventually(func() bool {
_ = k8sClient.Get(context.Background(), key, updated)
for _, c := range updated.Status.Conditions {
if c.Reason == sourcev1.IndexationFailedReason {
return true
}
for _, p := range tt.assertPaths {
absoluteP := filepath.Join(testStorage.BasePath, p)
if !strings.HasPrefix(p, "!") {
g.Expect(absoluteP).To(BeAnExistingFile())
continue
}
return false
}, timeout, interval).Should(BeTrue())
Expect(updated.Status.Artifact).ToNot(BeNil())
By("Expecting to delete successfully")
got = &sourcev1.HelmRepository{}
Eventually(func() error {
_ = k8sClient.Get(context.Background(), key, got)
return k8sClient.Delete(context.Background(), got)
}, timeout, interval).Should(Succeed())
By("Expecting delete to finish")
Eventually(func() error {
r := &sourcev1.HelmRepository{}
return k8sClient.Get(context.Background(), key, r)
}, timeout, interval).ShouldNot(Succeed())
exists := func(path string) bool {
// wait for tmp sync on macOS
time.Sleep(time.Second)
_, err := os.Stat(path)
return err == nil
g.Expect(absoluteP).NotTo(BeAnExistingFile())
}
By("Expecting GC after delete")
Eventually(exists(got.Status.Artifact.Path), timeout, interval).ShouldNot(BeTrue())
})
}
}
It("Handles timeout", func() {
helmServer.Start()
func TestHelmRepository_reconcileSource(t *testing.T) {
type options struct {
username string
password string
publicKey []byte
privateKey []byte
ca []byte
}
Expect(helmServer.PackageChart(path.Join("testdata/charts/helmchart"))).Should(Succeed())
Expect(helmServer.GenerateIndex()).Should(Succeed())
key := types.NamespacedName{
Name: "helmrepository-sample-" + randStringRunes(5),
Namespace: namespace.Name,
}
created := &sourcev1.HelmRepository{
tests := []struct {
name string
protocol string
server options
secret *corev1.Secret
beforeFunc func(obj *sourcev1.HelmRepository)
want ctrl.Result
wantErr bool
assertConditions []metav1.Condition
}{
{
name: "HTTP without secretRef makes ArtifactOutdated=True",
protocol: "http",
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.ArtifactOutdatedCondition, "NewRevision", "New index revision"),
},
},
{
name: "HTTP with Basic Auth secret makes ArtifactOutdated=True",
protocol: "http",
server: options{
username: "git",
password: "1234",
},
secret: &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: key.Name,
Namespace: key.Namespace,
Name: "basic-auth",
},
Spec: sourcev1.HelmRepositorySpec{
URL: helmServer.URL(),
Interval: metav1.Duration{Duration: indexInterval},
Data: map[string][]byte{
"username": []byte("git"),
"password": []byte("1234"),
},
}
Expect(k8sClient.Create(context.Background(), created)).Should(Succeed())
defer k8sClient.Delete(context.Background(), created)
},
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.SecretRef = &meta.LocalObjectReference{Name: "basic-auth"}
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.ArtifactOutdatedCondition, "NewRevision", "New index revision"),
},
},
{
name: "HTTPS with CAFile secret makes ArtifactOutdated=True",
protocol: "https",
server: options{
publicKey: tlsPublicKey,
privateKey: tlsPrivateKey,
ca: tlsCA,
},
secret: &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "ca-file",
},
Data: map[string][]byte{
"caFile": tlsCA,
},
},
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.SecretRef = &meta.LocalObjectReference{Name: "ca-file"}
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.ArtifactOutdatedCondition, "NewRevision", "New index revision"),
},
},
{
name: "HTTPS with invalid CAFile secret makes FetchFailed=True and returns error",
protocol: "https",
server: options{
publicKey: tlsPublicKey,
privateKey: tlsPrivateKey,
ca: tlsCA,
},
secret: &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "invalid-ca",
},
Data: map[string][]byte{
"caFile": []byte("invalid"),
},
},
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.SecretRef = &meta.LocalObjectReference{Name: "invalid-ca"}
},
wantErr: true,
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.FetchFailedCondition, meta.FailedReason, "can't create TLS config for client: failed to append certificates from file"),
},
},
{
name: "Invalid URL makes FetchFailed=True and returns zero Result",
protocol: "http",
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.URL = strings.ReplaceAll(obj.Spec.URL, "http://", "")
},
want: ctrl.Result{},
wantErr: false,
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.FetchFailedCondition, sourcev1.URLInvalidReason, "first path segment in URL cannot contain colon"),
},
},
{
name: "Unsupported scheme makes FetchFailed=True and returns zero Result",
protocol: "http",
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.URL = strings.ReplaceAll(obj.Spec.URL, "http://", "ftp://")
},
want: ctrl.Result{},
wantErr: false,
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.FetchFailedCondition, meta.FailedReason, "scheme \"ftp\" not supported"),
},
},
{
name: "Missing secret returns FetchFailed=True and returns error",
protocol: "http",
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.SecretRef = &meta.LocalObjectReference{Name: "non-existing"}
},
wantErr: true,
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.FetchFailedCondition, sourcev1.AuthenticationFailedReason, "secrets \"non-existing\" not found"),
},
},
{
name: "Malformed secret returns FetchFailed=True and returns error",
protocol: "http",
secret: &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "malformed-basic-auth",
},
Data: map[string][]byte{
"username": []byte("git"),
},
},
beforeFunc: func(obj *sourcev1.HelmRepository) {
obj.Spec.SecretRef = &meta.LocalObjectReference{Name: "malformed-basic-auth"}
},
wantErr: true,
assertConditions: []metav1.Condition{
*conditions.TrueCondition(sourcev1.FetchFailedCondition, sourcev1.AuthenticationFailedReason, "required fields 'username' and 'password"),
},
},
}
By("Expecting index download to succeed")
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, condition := range got.Status.Conditions {
if condition.Reason == sourcev1.IndexationSucceededReason {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
for _, tt := range tests {
obj := &sourcev1.HelmRepository{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "auth-strategy-",
},
Spec: sourcev1.HelmRepositorySpec{
Interval: metav1.Duration{Duration: interval},
Timeout: &metav1.Duration{Duration: interval},
},
}
By("Expecting index download to timeout")
updated := &sourcev1.HelmRepository{}
Expect(k8sClient.Get(context.Background(), key, updated)).Should(Succeed())
updated.Spec.Timeout = &metav1.Duration{Duration: time.Microsecond}
Expect(k8sClient.Update(context.Background(), updated)).Should(Succeed())
Eventually(func() string {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, condition := range got.Status.Conditions {
if condition.Reason == sourcev1.IndexationFailedReason {
return condition.Message
}
}
return ""
}, timeout, interval).Should(MatchRegexp("(?i)timeout"))
})
t.Run(tt.name, func(t *testing.T) {
g := NewWithT(t)
It("Authenticates when basic auth credentials are provided", func() {
helmServer, err = helmtestserver.NewTempHelmServer()
Expect(err).NotTo(HaveOccurred())
server, err := helmtestserver.NewTempHelmServer()
g.Expect(err).NotTo(HaveOccurred())
defer os.RemoveAll(server.Root())
var username, password = "john", "doe"
helmServer.WithMiddleware(func(handler http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
u, p, ok := r.BasicAuth()
if !ok || username != u || password != p {
w.WriteHeader(401)
return
}
handler.ServeHTTP(w, r)
g.Expect(server.PackageChart("testdata/charts/helmchart")).To(Succeed())
g.Expect(server.GenerateIndex()).To(Succeed())
if len(tt.server.username+tt.server.password) > 0 {
server.WithMiddleware(func(handler http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
u, p, ok := r.BasicAuth()
if !ok || u != tt.server.username || p != tt.server.password {
w.WriteHeader(401)
return
}
handler.ServeHTTP(w, r)
})
})
})
defer os.RemoveAll(helmServer.Root())
defer helmServer.Stop()
helmServer.Start()
Expect(helmServer.PackageChart(path.Join("testdata/charts/helmchart"))).Should(Succeed())
Expect(helmServer.GenerateIndex()).Should(Succeed())
secretKey := types.NamespacedName{
Name: "helmrepository-auth-" + randStringRunes(5),
Namespace: namespace.Name,
}
secret := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: secretKey.Name,
Namespace: secretKey.Namespace,
secret := tt.secret.DeepCopy()
switch tt.protocol {
case "http":
server.Start()
defer server.Stop()
obj.Spec.URL = server.URL()
case "https":
g.Expect(server.StartTLS(tt.server.publicKey, tt.server.privateKey, tt.server.ca, "example.com")).To(Succeed())
defer server.Stop()
obj.Spec.URL = server.URL()
default:
t.Fatalf("unsupported protocol %q", tt.protocol)
}
if tt.beforeFunc != nil {
tt.beforeFunc(obj)
}
builder := fakeclient.NewClientBuilder().WithScheme(testEnv.GetScheme())
if secret != nil {
builder.WithObjects(secret.DeepCopy())
}
r := &HelmRepositoryReconciler{
Client: builder.Build(),
Storage: testStorage,
Getters: testGetters,
}
var artifact sourcev1.Artifact
var index helm.ChartRepository
got, err := r.reconcileSource(logr.NewContext(ctx, log.NullLogger{}), obj, &artifact, &index)
g.Expect(obj.Status.Conditions).To(conditions.MatchConditions(tt.assertConditions))
g.Expect(err != nil).To(Equal(tt.wantErr))
g.Expect(got).To(Equal(tt.want))
g.Expect(artifact).ToNot(BeNil())
})
}
}
func TestHelmRepositoryReconciler_reconcileArtifact(t *testing.T) {
tests := []struct {
name string
beforeFunc func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository)
afterFunc func(t *WithT, obj *sourcev1.HelmRepository)
want ctrl.Result
wantErr bool
assertConditions []metav1.Condition
}{
{
name: "Archiving artifact to storage makes Ready=True",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
obj.Spec.Interval = metav1.Duration{Duration: interval}
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(meta.ReadyCondition, meta.SucceededReason, "Stored artifact for revision 'existing'"),
},
},
{
name: "Up-to-date artifact should not update status",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
obj.Spec.Interval = metav1.Duration{Duration: interval}
obj.Status.Artifact = artifact.DeepCopy()
},
afterFunc: func(t *WithT, obj *sourcev1.HelmRepository) {
t.Expect(obj.Status.URL).To(BeEmpty())
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(meta.ReadyCondition, meta.SucceededReason, "Stored artifact for revision 'existing'"),
},
},
{
name: "Removes ArtifactUnavailableCondition after creating artifact",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
obj.Spec.Interval = metav1.Duration{Duration: interval}
conditions.MarkTrue(obj, sourcev1.ArtifactUnavailableCondition, "Foo", "")
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(meta.ReadyCondition, meta.SucceededReason, "Stored artifact for revision 'existing'"),
},
},
{
name: "Removes ArtifactOutdatedCondition after creating a new artifact",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
obj.Spec.Interval = metav1.Duration{Duration: interval}
conditions.MarkTrue(obj, sourcev1.ArtifactOutdatedCondition, "Foo", "")
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(meta.ReadyCondition, meta.SucceededReason, "Stored artifact for revision 'existing'"),
},
},
{
name: "Creates latest symlink to the created artifact",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
obj.Spec.Interval = metav1.Duration{Duration: interval}
},
afterFunc: func(t *WithT, obj *sourcev1.HelmRepository) {
localPath := testStorage.LocalPath(*obj.GetArtifact())
symlinkPath := filepath.Join(filepath.Dir(localPath), "latest.tar.gz")
targetFile, err := os.Readlink(symlinkPath)
t.Expect(err).NotTo(HaveOccurred())
t.Expect(localPath).To(Equal(targetFile))
},
want: ctrl.Result{RequeueAfter: interval},
assertConditions: []metav1.Condition{
*conditions.TrueCondition(meta.ReadyCondition, meta.SucceededReason, "Stored artifact for revision 'existing'"),
},
},
{
name: "Index is empty",
beforeFunc: func(t *WithT, obj *sourcev1.HelmRepository, artifact sourcev1.Artifact, index *helm.ChartRepository) {
*index = helm.ChartRepository{}
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
g := NewWithT(t)
r := &HelmRepositoryReconciler{
Storage: testStorage,
}
obj := &sourcev1.HelmRepository{
TypeMeta: metav1.TypeMeta{
Kind: sourcev1.HelmRepositoryKind,
},
}
Expect(k8sClient.Create(context.Background(), secret)).Should(Succeed())
key := types.NamespacedName{
Name: "helmrepository-sample-" + randStringRunes(5),
Namespace: namespace.Name,
}
created := &sourcev1.HelmRepository{
ObjectMeta: metav1.ObjectMeta{
Name: key.Name,
Namespace: key.Namespace,
GenerateName: "test-bucket-",
Generation: 1,
Namespace: "default",
},
Spec: sourcev1.HelmRepositorySpec{
URL: helmServer.URL(),
SecretRef: &meta.LocalObjectReference{
Name: secretKey.Name,
},
Interval: metav1.Duration{Duration: indexInterval},
Timeout: &metav1.Duration{Duration: timeout},
},
}
Expect(k8sClient.Create(context.Background(), created)).Should(Succeed())
defer k8sClient.Delete(context.Background(), created)
By("Expecting 401")
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.IndexationFailedReason &&
strings.Contains(c.Message, "401 Unauthorized") {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
dirI, err := repo.IndexDirectory("testdata/charts", "https://example.com")
g.Expect(err).NotTo(HaveOccurred())
g.Expect(dirI).ToNot(BeNil())
By("Expecting missing field error")
secret.Data = map[string][]byte{
"username": []byte(username),
dirIB, err := yaml.Marshal(dirI)
g.Expect(err).NotTo(HaveOccurred())
i, err := helm.NewChartRepository("https://example.com", testGetters, nil)
g.Expect(err).NotTo(HaveOccurred())
g.Expect(i).ToNot(BeNil())
g.Expect(i.LoadIndex(dirIB)).To(Succeed())
artifact := testStorage.NewArtifactFor(obj.Kind, obj, "existing", "foo.tar.gz")
artifact.Checksum = i.Checksum
if tt.beforeFunc != nil {
tt.beforeFunc(g, obj, artifact, i)
}
Expect(k8sClient.Update(context.Background(), secret)).Should(Succeed())
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.AuthenticationFailedReason {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
By("Expecting artifact")
secret.Data["password"] = []byte(password)
Expect(k8sClient.Update(context.Background(), secret)).Should(Succeed())
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
return got.Status.Artifact != nil &&
ginkgoTestStorage.ArtifactExist(*got.Status.Artifact)
}, timeout, interval).Should(BeTrue())
got, err := r.reconcileArtifact(logr.NewContext(ctx, log.NullLogger{}), obj, artifact, *i)
g.Expect(err != nil).To(Equal(tt.wantErr))
g.Expect(got).To(Equal(tt.want))
By("Expecting missing secret error")
Expect(k8sClient.Delete(context.Background(), secret)).Should(Succeed())
got := &sourcev1.HelmRepository{}
Eventually(func() bool {
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.AuthenticationFailedReason {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
Expect(got.Status.Artifact).ShouldNot(BeNil())
// On error, artifact is empty. Check artifacts only on successful
// reconcile.
if !tt.wantErr {
g.Expect(obj.Status.Artifact).To(MatchArtifact(artifact.DeepCopy()))
}
g.Expect(obj.Status.Conditions).To(conditions.MatchConditions(tt.assertConditions))
if tt.afterFunc != nil {
tt.afterFunc(g, obj)
}
})
It("Authenticates when TLS credentials are provided", func() {
err = helmServer.StartTLS(examplePublicKey, examplePrivateKey, exampleCA, "example.com")
Expect(err).NotTo(HaveOccurred())
Expect(helmServer.PackageChart(path.Join("testdata/charts/helmchart"))).Should(Succeed())
Expect(helmServer.GenerateIndex()).Should(Succeed())
secretKey := types.NamespacedName{
Name: "helmrepository-auth-" + randStringRunes(5),
Namespace: namespace.Name,
}
secret := &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: secretKey.Name,
Namespace: secretKey.Namespace,
},
}
Expect(k8sClient.Create(context.Background(), secret)).Should(Succeed())
key := types.NamespacedName{
Name: "helmrepository-sample-" + randStringRunes(5),
Namespace: namespace.Name,
}
created := &sourcev1.HelmRepository{
ObjectMeta: metav1.ObjectMeta{
Name: key.Name,
Namespace: key.Namespace,
},
Spec: sourcev1.HelmRepositorySpec{
URL: helmServer.URL(),
SecretRef: &meta.LocalObjectReference{
Name: secretKey.Name,
},
Interval: metav1.Duration{Duration: indexInterval},
},
}
Expect(k8sClient.Create(context.Background(), created)).Should(Succeed())
defer k8sClient.Delete(context.Background(), created)
By("Expecting unknown authority error")
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.IndexationFailedReason &&
strings.Contains(c.Message, "certificate signed by unknown authority") {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
By("Expecting missing field error")
secret.Data = map[string][]byte{
"certFile": examplePublicKey,
}
Expect(k8sClient.Update(context.Background(), secret)).Should(Succeed())
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.AuthenticationFailedReason {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
By("Expecting artifact")
secret.Data["keyFile"] = examplePrivateKey
secret.Data["caFile"] = exampleCA
Expect(k8sClient.Update(context.Background(), secret)).Should(Succeed())
Eventually(func() bool {
got := &sourcev1.HelmRepository{}
_ = k8sClient.Get(context.Background(), key, got)
return got.Status.Artifact != nil &&
ginkgoTestStorage.ArtifactExist(*got.Status.Artifact)
}, timeout, interval).Should(BeTrue())
By("Expecting missing secret error")
Expect(k8sClient.Delete(context.Background(), secret)).Should(Succeed())
got := &sourcev1.HelmRepository{}
Eventually(func() bool {
_ = k8sClient.Get(context.Background(), key, got)
for _, c := range got.Status.Conditions {
if c.Reason == sourcev1.AuthenticationFailedReason {
return true
}
}
return false
}, timeout, interval).Should(BeTrue())
Expect(got.Status.Artifact).ShouldNot(BeNil())
})
})
})
}
}

8
controllers/legacy_suite_test.go
View File

@ -87,12 +87,6 @@ var _ = BeforeSuite(func(done Done) {
err = sourcev1.AddToScheme(scheme.Scheme)
Expect(err).NotTo(HaveOccurred())
err = sourcev1.AddToScheme(scheme.Scheme)
Expect(err).NotTo(HaveOccurred())
err = sourcev1.AddToScheme(scheme.Scheme)
Expect(err).NotTo(HaveOccurred())
// +kubebuilder:scaffold:scheme
Expect(loadExampleKeys()).To(Succeed())
@ -124,7 +118,7 @@ var _ = BeforeSuite(func(done Done) {
err = (&HelmRepositoryReconciler{
Client: k8sManager.GetClient(),
Scheme: scheme.Scheme,
Events: testEventsH,
Storage: ginkgoTestStorage,
Getters: getter.Providers{getter.Provider{
Schemes: []string{"http", "https"},

10
controllers/suite_test.go
View File

@ -98,6 +98,16 @@ func TestMain(m *testing.M) {
panic(fmt.Sprintf("Failed to start GitRepositoryReconciler: %v", err))
}
if err := (&HelmRepositoryReconciler{
Client: testEnv,
Events: testEventsH,
Metrics: testMetricsH,
Getters: testGetters,
Storage: testStorage,
}).SetupWithManager(testEnv); err != nil {
panic(fmt.Sprintf("Failed to start HelmRepositoryReconciler: %v", err))
}
if err := (&BucketReconciler{
Client: testEnv,
Events: testEventsH,

4
go.mod
View File

@ -30,9 +30,9 @@ require (
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
gotest.tools v2.2.0+incompatible
helm.sh/helm/v3 v3.6.3
k8s.io/api v0.21.2
k8s.io/api v0.21.3
k8s.io/apimachinery v0.21.3
k8s.io/client-go v0.21.2
k8s.io/client-go v0.21.3
k8s.io/utils v0.0.0-20210527160623-6fdb442a123b
sigs.k8s.io/controller-runtime v0.9.3
sigs.k8s.io/yaml v1.2.0

6
go.sum
View File

@ -1268,8 +1268,9 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
k8s.io/api v0.21.0/go.mod h1:+YbrhBBGgsxbF6o6Kj4KJPJnBmAKuXDeS3E18bgHNVU=
k8s.io/api v0.21.2 h1:vz7DqmRsXTCSa6pNxXwQ1IYeAZgdIsua+DZU+o+SX3Y=
k8s.io/api v0.21.2/go.mod h1:Lv6UGJZ1rlMI1qusN8ruAp9PUBFyBwpEHAdG24vIsiU=
k8s.io/api v0.21.3 h1:cblWILbLO8ar+Fj6xdDGr603HRsf8Wu9E9rngJeprZQ=
k8s.io/api v0.21.3/go.mod h1:hUgeYHUbBp23Ue4qdX9tR8/ANi/g3ehylAqDn9NWVOg=
k8s.io/apiextensions-apiserver v0.21.0/go.mod h1:gsQGNtGkc/YoDG9loKI0V+oLZM4ljRPjc/sql5tmvzc=
k8s.io/apiextensions-apiserver v0.21.2 h1:+exKMRep4pDrphEafRvpEi79wTnCFMqKf8LBtlA3yrE=
k8s.io/apiextensions-apiserver v0.21.2/go.mod h1:+Axoz5/l3AYpGLlhJDfcVQzCerVYq3K3CvDMvw6X1RA=
@ -1283,8 +1284,9 @@ k8s.io/apiserver v0.21.2/go.mod h1:lN4yBoGyiNT7SC1dmNk0ue6a5Wi6O3SWOIw91TsucQw=
k8s.io/cli-runtime v0.21.0 h1:/V2Kkxtf6x5NI2z+Sd/mIrq4FQyQ8jzZAUD6N5RnN7Y=
k8s.io/cli-runtime v0.21.0/go.mod h1:XoaHP93mGPF37MkLbjGVYqg3S1MnsFdKtiA/RZzzxOo=
k8s.io/client-go v0.21.0/go.mod h1:nNBytTF9qPFDEhoqgEPaarobC8QPae13bElIVHzIglA=
k8s.io/client-go v0.21.2 h1:Q1j4L/iMN4pTw6Y4DWppBoUxgKO8LbffEMVEV00MUp0=
k8s.io/client-go v0.21.2/go.mod h1:HdJ9iknWpbl3vMGtib6T2PyI/VYxiZfq936WNVHBRrA=
k8s.io/client-go v0.21.3 h1:J9nxZTOmvkInRDCzcSNQmPJbDYN/PjlxXT9Mos3HcLg=
k8s.io/client-go v0.21.3/go.mod h1:+VPhCgTsaFmGILxR/7E1N0S+ryO010QBeNCv5JwRGYU=
k8s.io/code-generator v0.21.0/go.mod h1:hUlps5+9QaTrKx+jiM4rmq7YmH8wPOIko64uZCHDh6Q=
k8s.io/code-generator v0.21.2/go.mod h1:8mXJDCB7HcRo1xiEQstcguZkbxZaqeUOrO9SsicWs3U=
k8s.io/component-base v0.21.0/go.mod h1:qvtjz6X0USWXbgmbfXR+Agik4RZ3jv2Bgr5QnZzdPYw=

73
internal/helm/getter.go
View File

@ -17,33 +17,35 @@ limitations under the License.
package helm
import (
"bytes"
"fmt"
"io"
"os"
"path/filepath"
"helm.sh/helm/v3/pkg/getter"
corev1 "k8s.io/api/core/v1"
)
// ClientOptionsFromSecret constructs a getter.Option slice for the given secret.
// ClientOptionsFromSecret constructs a getter.Option slice for the given secret,
// storing any temporary credentials .
// It returns the slice, and a callback to remove temporary files.
func ClientOptionsFromSecret(secret corev1.Secret) ([]getter.Option, func(), error) {
func ClientOptionsFromSecret(secret corev1.Secret, dir string) ([]getter.Option, error) {
var opts []getter.Option
basicAuth, err := BasicAuthFromSecret(secret)
if err != nil {
return opts, nil, err
return opts, err
}
if basicAuth != nil {
opts = append(opts, basicAuth)
}
tlsClientConfig, cleanup, err := TLSClientConfigFromSecret(secret)
tlsClientConfig, err := TLSClientConfigFromSecret(secret, dir)
if err != nil {
return opts, nil, err
return opts, err
}
if tlsClientConfig != nil {
opts = append(opts, tlsClientConfig)
}
return opts, cleanup, nil
return opts, nil
}
// BasicAuthFromSecret attempts to construct a basic auth getter.Option for the
@ -68,45 +70,54 @@ func BasicAuthFromSecret(secret corev1.Secret) (getter.Option, error) {
//
// Secrets with no certFile, keyFile, AND caFile are ignored, if only a
// certBytes OR keyBytes is defined it returns an error.
func TLSClientConfigFromSecret(secret corev1.Secret) (getter.Option, func(), error) {
func TLSClientConfigFromSecret(secret corev1.Secret, dir string) (getter.Option, error) {
certBytes, keyBytes, caBytes := secret.Data["certFile"], secret.Data["keyFile"], secret.Data["caFile"]
switch {
case len(certBytes)+len(keyBytes)+len(caBytes) == 0:
return nil, func() {}, nil
return nil, nil
case (len(certBytes) > 0 && len(keyBytes) == 0) || (len(keyBytes) > 0 && len(certBytes) == 0):
return nil, nil, fmt.Errorf("invalid '%s' secret data: fields 'certFile' and 'keyFile' require each other's presence",
return nil, fmt.Errorf("invalid '%s' secret data: fields 'certFile' and 'keyFile' require each other's presence",
secret.Name)
}
// create tmp dir for TLS files
tmp, err := os.MkdirTemp("", "helm-tls-"+secret.Name)
if err != nil {
return nil, nil, err
}
cleanup := func() { os.RemoveAll(tmp) }
var certFile, keyFile, caFile string
if len(certBytes) > 0 && len(keyBytes) > 0 {
certFile = filepath.Join(tmp, "cert.crt")
if err := os.WriteFile(certFile, certBytes, 0644); err != nil {
cleanup()
return nil, nil, err
f, err := os.CreateTemp(dir, "cert-")
if err != nil {
}
keyFile = filepath.Join(tmp, "key.crt")
if err := os.WriteFile(keyFile, keyBytes, 0644); err != nil {
cleanup()
return nil, nil, err
if _, err = io.Copy(f, bytes.NewReader(certBytes)); err != nil {
f.Close()
return nil, err
}
f.Close()
certFile = f.Name()
f, err = os.CreateTemp(dir, "key-")
if err != nil {
f.Close()
return nil, err
}
if _, err = io.Copy(f, bytes.NewReader(keyBytes)); err != nil {
f.Close()
return nil, err
}
f.Close()
keyFile = f.Name()
}
if len(caBytes) > 0 {
caFile = filepath.Join(tmp, "ca.pem")
if err := os.WriteFile(caFile, caBytes, 0644); err != nil {
cleanup()
return nil, nil, err
f, err := os.CreateTemp(dir, "ca-")
if err != nil {
f.Close()
return nil, err
}
if _, err = io.Copy(f, bytes.NewReader(caBytes)); err != nil {
f.Close()
}
f.Close()
caFile = f.Name()
}
return getter.WithTLSClientConfig(certFile, keyFile, caFile), cleanup, nil
return getter.WithTLSClientConfig(certFile, keyFile, caFile), nil
}

17
internal/helm/getter_test.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package helm
import (
"os"
"testing"
corev1 "k8s.io/api/core/v1"
@ -56,10 +57,12 @@ func TestClientOptionsFromSecret(t *testing.T) {
secret.Data[k] = v
}
}
got, cleanup, err := ClientOptionsFromSecret(secret)
if cleanup != nil {
defer cleanup()
tmpDir, err := os.MkdirTemp("", "")
if err != nil {
t.Fatalf("Failed to create temporary directory: %v", err)
}
defer os.RemoveAll(tmpDir)
got, err := ClientOptionsFromSecret(secret, tmpDir)
if err != nil {
t.Errorf("ClientOptionsFromSecret() error = %v", err)
return
@ -123,10 +126,12 @@ func TestTLSClientConfigFromSecret(t *testing.T) {
if tt.modify != nil {
tt.modify(secret)
}
got, cleanup, err := TLSClientConfigFromSecret(*secret)
if cleanup != nil {
defer cleanup()
tmpDir, err := os.MkdirTemp("", "")
if err != nil {
t.Fatalf("Failed to create temporary directory: %v", err)
}
defer os.RemoveAll(tmpDir)
got, err := TLSClientConfigFromSecret(*secret, tmpDir)
if (err != nil) != tt.wantErr {
t.Errorf("TLSClientConfigFromSecret() error = %v, wantErr %v", err, tt.wantErr)
return

21
internal/helm/repository.go
View File

@ -18,9 +18,11 @@ package helm
import (
"bytes"
"crypto/sha256"
"fmt"
"io"
"net/url"
"os"
"path"
"sort"
"strings"
@ -36,10 +38,11 @@ import (
// ChartRepository represents a Helm chart repository, and the configuration
// required to download the chart index, and charts from the repository.
type ChartRepository struct {
URL string
Index *repo.IndexFile
Client getter.Getter
Options []getter.Option
URL string
Index *repo.IndexFile
Checksum string
Client getter.Getter
Options []getter.Option
}
// NewChartRepository constructs and returns a new ChartRepository with
@ -175,6 +178,15 @@ func (r *ChartRepository) DownloadChart(chart *repo.ChartVersion) (*bytes.Buffer
return r.Client.Get(u.String(), r.Options...)
}
// LoadIndexFile takes a file at the given path and loads it using LoadIndex.
func (r *ChartRepository) LoadIndexFile(path string) error {
b, err := os.ReadFile(path)
if err != nil {
return err
}
return r.LoadIndex(b)
}
// LoadIndex loads the given bytes into the Index while performing
// minimal validity checks. It fails if the API version is not set
// (repo.ErrNoAPIVersion), or if the unmarshal fails.
@ -191,6 +203,7 @@ func (r *ChartRepository) LoadIndex(b []byte) error {
}
i.SortEntries()
r.Index = i
r.Checksum = fmt.Sprintf("%x", sha256.Sum256(b))
return nil
}

12
main.go
View File

@ -166,13 +166,11 @@ func main() {
os.Exit(1)
}
if err = (&controllers.HelmRepositoryReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
Storage: storage,
Getters: getters,
EventRecorder: mgr.GetEventRecorderFor(controllerName),
ExternalEventRecorder: eventRecorder,
MetricsRecorder: metricsH.MetricsRecorder,
Client: mgr.GetClient(),
Events: eventsH,
Metrics: metricsH,
Storage: storage,
Getters: getters,
}).SetupWithManagerAndOptions(mgr, controllers.HelmRepositoryReconcilerOptions{
MaxConcurrentReconciles: concurrent,
}); err != nil {