Populate resources and apply from cache

Signed-off-by: Danil-Grigorev <[email protected]>

Author: Danil-Grigorev

internal/controller/genericprovider_controller.go

@@ -23,7 +23,6 @@ import (
 	"errors"
 	"fmt"
 	"hash"
-	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -33,6 +32,8 @@ import (
 	operatorv1 "sigs.k8s.io/cluster-api-operator/api/v1alpha2"
 	"sigs.k8s.io/cluster-api-operator/internal/controller/genericprovider"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
+	configclient "sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/patch"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -360,7 +361,7 @@ func calculateHash(ctx context.Context, k8sClient client.Client, provider generi
 func applyFromCache(ctx context.Context, cl client.Client, provider genericprovider.GenericProvider) (bool, error) {
 	log := log.FromContext(ctx)
 
-	configMap, err := providerConfigMap(ctx, cl, provider)
+	configMap, err := providerCacheConfigMap(ctx, cl, provider)
 	if err != nil {
 		log.Error(err, "failed to get provider config map")
 
@@ -393,37 +394,81 @@ func applyFromCache(ctx context.Context, cl client.Client, provider genericprovi
 		return false, nil
 	}
 
-	components, err := getComponentsData(*configMap)
-	if err != nil {
-		log.Error(err, "failed to get provider components")
+	log.Info("Applying provider configuration from cache")
+	errs := []error{}
+
+	mr := configclient.NewMemoryReader()
 
+	if err := mr.Init(ctx, ""); err != nil {
 		return false, err
 	}
 
-	additionalManifests, err := fetchAdditionalManifests(ctx, cl, provider)
-	if err != nil {
-		log.Error(err, "failed to get additional manifests")
+	// Fetch configuration variables from the secret. See API field docs for more info.
+	initReaderVariables(ctx, cl, mr, provider)
 
-		return false, err
-	}
+	processor := yamlprocessor.NewSimpleProcessor()
+	for _, manifest := range configMap.Data {
+		manifest, err := processor.Process([]byte(manifest), mr.Get)
+		if err != nil {
+			log.Error(err, "failed to process manifest")
 
-	if additionalManifests != "" {
-		components = components + "\n---\n" + additionalManifests
-	}
+			return false, err
+		}
 
-	for _, manifests := range strings.Split(components, "---") {
-		manifests, err := utilyaml.ToUnstructured([]byte(manifests))
+		manifests, err := utilyaml.ToUnstructured(manifest)
 		if err != nil {
 			log.Error(err, "failed to convert yaml to unstructured")
 
 			return false, err
 		}
 
+		if len(manifest) > 1 {
+			return false, fmt.Errorf("multiple manifests found: %d", len(manifests))
+		} else if len(manifests) == 0 {
+			continue
+		}
+
 		if err := cl.Patch(ctx, &manifests[0], client.Apply, client.ForceOwnership, client.FieldOwner(cacheOwner)); err != nil {
-			log.Error(err, "failed to apply object from cache")
+			errs = append(errs, err)
+		}
+	}
 
-			return false, nil
+	for _, binaryManifest := range configMap.BinaryData {
+		manifest, err := decompressYaml(binaryManifest)
+		if err != nil {
+			log.Error(err, "failed to decompress yaml")
+
+			return false, err
 		}
+
+		manifest, err = processor.Process([]byte(manifest), mr.Get)
+		if err != nil {
+			log.Error(err, "failed to process manifest")
+
+			return false, err
+		}
+		manifests, err := utilyaml.ToUnstructured([]byte(manifest))
+		if err != nil {
+			log.Error(err, "failed to convert yaml to unstructured")
+
+			return false, err
+		}
+
+		if len(manifest) > 1 {
+			return false, fmt.Errorf("multiple manifests found: %d", len(manifests))
+		} else if len(manifests) == 0 {
+			continue
+		}
+
+		if err := cl.Patch(ctx, &manifests[0], client.Apply, client.ForceOwnership, client.FieldOwner(cacheOwner)); err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	if err := kerrors.NewAggregate(errs); err != nil {
+		log.Error(err, "failed to apply objects from cache")
+
+		return false, err
 	}
 
 	log.Info("Applied all objects from cache")
@@ -433,7 +478,7 @@ func applyFromCache(ctx context.Context, cl client.Client, provider genericprovi
 
 // setCacheHash calculates current provider and configMap hash, and updates it on the configMap.
 func setCacheHash(ctx context.Context, cl client.Client, provider genericprovider.GenericProvider) error {
-	configMap, err := providerConfigMap(ctx, cl, provider)
+	configMap, err := providerCacheConfigMap(ctx, cl, provider)
 	if err != nil {
 		return err
 	}

internal/controller/manifests_downloader.go

@@ -21,8 +21,10 @@ import (
 	"compress/gzip"
 	"context"
 	"fmt"
+	"io"
 
 	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/labels"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -40,6 +42,11 @@ const (
 	configMapSourceLabel      = "provider.cluster.x-k8s.io/source"
 	configMapSourceAnnotation = "provider.cluster.x-k8s.io/source"
 	operatorManagedLabel      = "managed-by.operator.cluster.x-k8s.io"
+	operatorCacheLabel        = "cached-by.operator.cluster.x-k8s.io"
+
+	cacheVersionLabelName = "provider-cache.cluster.x-k8s.io/version"
+	cacheTypeLabel        = "provider-cache.cluster.x-k8s.io/type"
+	cacheNameLabel        = "provider-cache.cluster.x-k8s.io/name"
 
 	maxConfigMapSize = 1 * 1024 * 1024
 	ociSource        = "oci"
@@ -188,17 +195,10 @@ func TemplateManifestsConfigMap(provider operatorv1.GenericProvider, labels map[
 		configMap.Data[operatorv1.ComponentsConfigMapKey] = string(components)
 	} else {
 		var componentsBuf bytes.Buffer
-		zw := gzip.NewWriter(&componentsBuf)
-
-		_, err := zw.Write(components)
-		if err != nil {
+		if err := compressYaml(&componentsBuf, components); err != nil {
 			return nil, fmt.Errorf("cannot compress data for provider %s/%s: %w", provider.GetNamespace(), provider.GetName(), err)
 		}
 
-		if err := zw.Close(); err != nil {
-			return nil, err
-		}
-
 		configMap.BinaryData = map[string][]byte{
 			operatorv1.ComponentsConfigMapKey: componentsBuf.Bytes(),
 		}
@@ -221,6 +221,41 @@ func TemplateManifestsConfigMap(provider operatorv1.GenericProvider, labels map[
 	return configMap, nil
 }
 
+// compressYaml takes a bytes.Buffer and data, and compresses data into it.
+func compressYaml(componentsBuf *bytes.Buffer, data []byte) (err error) {
+	zw := gzip.NewWriter(componentsBuf)
+
+	_, err = zw.Write(data)
+	defer func() {
+		err = zw.Close()
+	}()
+
+	if err != nil {
+		return fmt.Errorf("cannot compress data: %w", err)
+	}
+
+	return
+}
+
+// decompressYaml takes a compressed data, and decompresses it.
+func decompressYaml(compressedData []byte) (data []byte, err error) {
+	zr, err := gzip.NewReader(bytes.NewReader(compressedData))
+	if err != nil {
+		return nil, fmt.Errorf("cannot open gzip reader from data: %w", err)
+	}
+
+	defer func() {
+		err = zr.Close()
+	}()
+
+	decompressedData, err := io.ReadAll(zr)
+	if err != nil {
+		return nil, fmt.Errorf("cannot decompress data: %w", err)
+	}
+
+	return decompressedData, nil
+}
+
 // OCIConfigMap templates config from the OCI source.
 func OCIConfigMap(ctx context.Context, provider operatorv1.GenericProvider, auth *auth.Credential) (*corev1.ConfigMap, error) {
 	store, err := FetchOCI(ctx, provider, auth)
@@ -295,25 +330,16 @@ func providerLabelSelector(provider operatorv1.GenericProvider) *metav1.LabelSel
 	}
 }
 
-// providerConfigMap finds a ConfigMap the given provider label selector.
-func providerConfigMap(ctx context.Context, cl client.Client, provider operatorv1.GenericProvider) (*corev1.ConfigMap, error) {
-	labelSelector := providerLabelSelector(provider)
-	labelSet := labels.Set(labelSelector.MatchLabels)
-	listOpts := []client.ListOption{
-		client.MatchingLabelsSelector{Selector: labels.SelectorFromSet(labelSet)},
-		client.InNamespace(provider.GetNamespace()),
-	}
-
-	configMapList := &corev1.ConfigMapList{}
-	if err := cl.List(ctx, configMapList, listOpts...); err != nil {
+// providerCacheConfigMap finds a cached ConfigMap the given provider label selector.
+func providerCacheConfigMap(ctx context.Context, cl client.Client, provider operatorv1.GenericProvider) (*corev1.ConfigMap, error) {
+	configMap := &corev1.ConfigMap{}
+	if err := cl.Get(ctx, client.ObjectKey{Name: ProviderCacheName(provider), Namespace: provider.GetNamespace()}, configMap); apierrors.IsNotFound(err) {
+		return nil, nil
+	} else if err != nil {
 		return nil, fmt.Errorf("failed to list ConfigMaps: %w", err)
 	}
 
-	if len(configMapList.Items) > 1 {
-		return nil, fmt.Errorf("multiple ConfigMaps found for provider %q", provider.GetName())
-	}
-
-	return &configMapList.Items[0], nil
+	return configMap, nil
 }
 
 // ProviderLabels returns default set of labels that identify a config map with downloaded manifests.
@@ -332,6 +358,12 @@ func ProviderLabels(provider operatorv1.GenericProvider) map[string]string {
 	return labels
 }
 
+// ProviderCacheName generates a cache name for a given provider.
+
+func ProviderCacheName(provider operatorv1.GenericProvider) string {
+	return fmt.Sprintf("%s-%s-%s-cache", provider.GetType(), provider.GetName(), provider.GetSpec().Version)
+}
+
 // needToCompress checks whether the input data exceeds the maximum configmap
 // size limit and returns whether it should be compressed.
 func needToCompress(bs ...[]byte) bool {