Init helm release drift manager POC based on dynamic manifest objects…

… watch

api/v1alpha1/helmchartproxy_types.go

@@ -84,6 +84,8 @@ type HelmChartProxySpec struct {
 	// +optional
 	ReconcileStrategy string `json:"reconcileStrategy,omitempty"`
 
+	ReleaseDrift bool `json:"releaseDrift,omitempty"`
+
 	// Options represents CLI flags passed to Helm operations (i.e. install, upgrade, delete) and
 	// include options such as wait, skipCRDs, timeout, waitForJobs, etc.
 	// +optional

api/v1alpha1/helmreleaseproxy_types.go

@@ -78,6 +78,8 @@ type HelmReleaseProxySpec struct {
 	// +optional
 	ReconcileStrategy string `json:"reconcileStrategy,omitempty"`
 
+	ReleaseDrift bool `json:"releaseDrift,omitempty"`
+
 	// Options represents the helm setting options which can be used to control behaviour of helm operations(Install, Upgrade, Delete, etc)
 	// via options like wait, skipCrds, timeout, waitForJobs, etc.
 	// +optional

config/crd/bases/addons.cluster.x-k8s.io_helmchartproxies.yaml

@@ -270,6 +270,8 @@ spec:
                 - InstallOnce
                 - Continuous
                 type: string
+              releaseDrift:
+                type: boolean
               releaseName:
                 description: ReleaseName is the release name of the installed Helm
                   chart. If it is not specified, a name will be generated.

config/crd/bases/addons.cluster.x-k8s.io_helmreleaseproxies.yaml

@@ -277,6 +277,8 @@ spec:
                 - InstallOnce
                 - Continuous
                 type: string
+              releaseDrift:
+                type: boolean
               releaseName:
                 description: ReleaseName is the release name of the installed Helm
                   chart. If it is not specified, a name will be generated.

controllers/helmchartproxy/helmchartproxy_controller_phases.go

@@ -227,6 +227,7 @@ func constructHelmReleaseProxy(existing *addonsv1alpha1.HelmReleaseProxy, helmCh
 	}
 
 	helmReleaseProxy.Spec.ReconcileStrategy = helmChartProxy.Spec.ReconcileStrategy
+	helmReleaseProxy.Spec.ReleaseDrift = helmChartProxy.Spec.ReleaseDrift
 	helmReleaseProxy.Spec.Version = helmChartProxy.Spec.Version
 	helmReleaseProxy.Spec.Values = parsedValues
 	helmReleaseProxy.Spec.Options = helmChartProxy.Spec.Options

controllers/helmreleasedrift/manager.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helmreleasedrift
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/ironcore-dev/controller-utils/unstructuredutils"
+	"golang.org/x/exp/slices"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	addonsv1alpha1 "sigs.k8s.io/cluster-api-addon-provider-helm/api/v1alpha1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+	"sigs.k8s.io/kustomize/api/konfig"
+)
+
+const (
+	InstanceLabelKey = "app.kubernetes.io/instance"
+)
+
+var (
+	managers = map[string]options{}
+	mutex    sync.Mutex
+)
+
+type options struct {
+	gvks   []schema.GroupVersionKind
+	cancel context.CancelFunc
+}
+
+func Add(ctx context.Context, restConfig *rest.Config, helmReleaseProxy *addonsv1alpha1.HelmReleaseProxy, releaseManifest string, eventChannel chan event.GenericEvent) error {
+	log := ctrl.LoggerFrom(ctx)
+	gvks, err := extractGVKsFromManifest(releaseManifest)
+	if err != nil {
+		return err
+	}
+
+	manager, exist := managers[managerKey(helmReleaseProxy)]
+	if exist {
+		if slices.Equal(manager.gvks, gvks) {
+			return nil
+		}
+		Remove(helmReleaseProxy)
+	}
+
+	mutex.Lock()
+	defer mutex.Unlock()
+	k8sManager, err := ctrl.NewManager(restConfig, ctrl.Options{
+		Scheme: scheme.Scheme,
+		Metrics: metricsserver.Options{
+			BindAddress: "0",
+		},
+		HealthProbeBindAddress: "0",
+		Cache: cache.Options{
+			DefaultLabelSelector: labels.SelectorFromSet(map[string]string{
+				konfig.ManagedbyLabelKey: "Helm",
+				InstanceLabelKey:         helmReleaseProxy.Spec.ReleaseName,
+			}),
+		},
+	})
+	if err != nil {
+		return err
+	}
+	if err = (&releaseDriftReconciler{
+		Client:                k8sManager.GetClient(),
+		Scheme:                k8sManager.GetScheme(),
+		HelmReleaseProxyKey:   client.ObjectKeyFromObject(helmReleaseProxy),
+		HelmReleaseProxyEvent: eventChannel,
+	}).setupWithManager(k8sManager, gvks); err != nil {
+		return err
+	}
+	log.V(2).Info("Starting release drift controller manager")
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		if err = k8sManager.Start(ctx); err != nil {
+			log.V(2).Error(err, "failed to start release drift manager")
+			objectMeta := metav1.ObjectMeta{
+				Name:      helmReleaseProxy.Name,
+				Namespace: helmReleaseProxy.Namespace,
+			}
+			eventChannel <- event.GenericEvent{Object: &addonsv1alpha1.HelmReleaseProxy{ObjectMeta: objectMeta}}
+		}
+	}()
+
+	managers[managerKey(helmReleaseProxy)] = options{
+		gvks:   gvks,
+		cancel: cancel,
+	}
+
+	return nil
+}
+
+func Remove(helmReleaseProxy *addonsv1alpha1.HelmReleaseProxy) {
+	mutex.Lock()
+	defer mutex.Unlock()
+
+	manager, exist := managers[managerKey(helmReleaseProxy)]
+	if exist {
+		manager.cancel()
+		delete(managers, managerKey(helmReleaseProxy))
+	}
+}
+
+func managerKey(helmReleaseProxy *addonsv1alpha1.HelmReleaseProxy) string {
+	return fmt.Sprintf("%s-%s-%s", helmReleaseProxy.Spec.ClusterRef.Name, helmReleaseProxy.Namespace, helmReleaseProxy.Spec.ReleaseName)
+}
+
+func extractGVKsFromManifest(manifest string) ([]schema.GroupVersionKind, error) {
+	objects, err := unstructuredutils.Read(strings.NewReader(manifest))
+	if err != nil {
+		return nil, err
+	}
+	var gvks []schema.GroupVersionKind
+	for _, obj := range objects {
+		if !slices.Contains(gvks, obj.GroupVersionKind()) {
+			gvks = append(gvks, obj.GroupVersionKind())
+		}
+	}
+
+	return gvks, nil
+}

controllers/helmreleasedrift/manager_test.go

@@ -0,0 +1,93 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helmreleasedrift_test
+
+import (
+	"github.com/ironcore-dev/controller-utils/metautils"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+	addonsv1alpha1 "sigs.k8s.io/cluster-api-addon-provider-helm/api/v1alpha1"
+	"sigs.k8s.io/cluster-api-addon-provider-helm/controllers/helmreleasedrift"
+	"sigs.k8s.io/cluster-api-addon-provider-helm/controllers/helmreleasedrift/test/fake"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+const (
+	releaseName                = "ahoy"
+	objectName                 = "ahoy-hello-world"
+	originalDeploymentReplicas = 1
+	patchedDeploymentReplicas  = 3
+)
+
+var _ = Describe("Testing HelmReleaseProxy drift manager with fake manifest", func() {
+	It("Adding HelmReleaseProxy drift manager and validating its lifecycle", func() {
+		objectMeta := metav1.ObjectMeta{
+			Name:      releaseName,
+			Namespace: metav1.NamespaceDefault,
+		}
+		fake.ManifestEventChannel <- event.GenericEvent{Object: &addonsv1alpha1.HelmReleaseProxy{ObjectMeta: objectMeta}}
+
+		helmReleaseProxy := &addonsv1alpha1.HelmReleaseProxy{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "ahoy-release-proxy",
+				Namespace: metav1.NamespaceDefault,
+			},
+			Spec: addonsv1alpha1.HelmReleaseProxySpec{
+				ReleaseName: releaseName,
+			},
+		}
+
+		// TODO (dvolodin) Find way how to wait manager to start for testing
+		err := helmreleasedrift.Add(ctx, restConfig, helmReleaseProxy, manifest, fake.ManifestEventChannel)
+		Expect(err).NotTo(HaveOccurred())
+
+		Eventually(func() bool {
+			for _, objectList := range []client.ObjectList{&corev1.ServiceList{}, &appsv1.DeploymentList{}, &corev1.ServiceAccountList{}} {
+				err := k8sClient.List(ctx, objectList, client.InNamespace(metav1.NamespaceDefault), client.MatchingLabels(map[string]string{helmreleasedrift.InstanceLabelKey: releaseName}))
+				if err != nil {
+					return false
+				}
+				objects, err := metautils.ExtractList(objectList)
+				if err != nil || len(objects) == 0 {
+					return false
+				}
+			}
+
+			return true
+		}, timeout, interval).Should(BeTrue())
+
+		deployment := &appsv1.Deployment{}
+		err = k8sClient.Get(ctx, client.ObjectKey{Name: objectName, Namespace: metav1.NamespaceDefault}, deployment)
+		Expect(err).NotTo(HaveOccurred())
+		patch := client.MergeFrom(deployment.DeepCopy())
+		deployment.Spec.Replicas = ptr.To(int32(patchedDeploymentReplicas))
+		err = k8sClient.Patch(ctx, deployment, patch)
+		Expect(err).NotTo(HaveOccurred())
+
+		Eventually(func() bool {
+			err = k8sClient.Get(ctx, client.ObjectKey{Name: objectName, Namespace: metav1.NamespaceDefault}, deployment)
+			return err == nil && *deployment.Spec.Replicas == originalDeploymentReplicas
+		}, timeout, interval).Should(BeTrue())
+
+		helmreleasedrift.Remove(helmReleaseProxy)
+	})
+})

controllers/helmreleasedrift/releasedrift_controller.go

@@ -0,0 +1,80 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package helmreleasedrift
+
+import (
+	"context"
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	addonsv1alpha1 "sigs.k8s.io/cluster-api-addon-provider-helm/api/v1alpha1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+// releaseDriftReconciler reconciles an event from the all helm objects managed by the HelmReleaseProxy.
+type releaseDriftReconciler struct {
+	client.Client
+	Scheme                *runtime.Scheme
+	HelmReleaseProxyKey   client.ObjectKey
+	HelmReleaseProxyEvent chan event.GenericEvent
+}
+
+var excludeCreateEventsPredicate = predicate.Funcs{
+	CreateFunc: func(e event.CreateEvent) bool {
+		return false
+	},
+}
+
+// setupWithManager sets up the controller with the Manager.
+func (r *releaseDriftReconciler) setupWithManager(mgr ctrl.Manager, gvks []schema.GroupVersionKind) error {
+	controllerBuilder := ctrl.NewControllerManagedBy(mgr).
+		Named(fmt.Sprintf("%s-%s-release-drift-controller", r.HelmReleaseProxyKey.Name, r.HelmReleaseProxyKey.Namespace))
+	for _, gvk := range gvks {
+		watch := &unstructured.Unstructured{}
+		watch.SetGroupVersionKind(gvk)
+		controllerBuilder.Watches(watch, handler.EnqueueRequestsFromMapFunc(r.WatchesToReleaseMapper), builder.OnlyMetadata)
+	}
+
+	return controllerBuilder.WithEventFilter(excludeCreateEventsPredicate).Complete(r)
+}
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+func (r *releaseDriftReconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Result, reterr error) {
+	log := ctrl.LoggerFrom(ctx)
+	log.V(2).Info("Beginning reconciliation", "requestNamespace", req.Namespace, "requestName", req.Name)
+
+	objectMeta := metav1.ObjectMeta{
+		Name:      r.HelmReleaseProxyKey.Name,
+		Namespace: r.HelmReleaseProxyKey.Namespace,
+	}
+	r.HelmReleaseProxyEvent <- event.GenericEvent{Object: &addonsv1alpha1.HelmReleaseProxy{ObjectMeta: objectMeta}}
+
+	return ctrl.Result{}, nil
+}
+
+func (r *releaseDriftReconciler) WatchesToReleaseMapper(_ context.Context, _ client.Object) []ctrl.Request {
+	return []ctrl.Request{{NamespacedName: r.HelmReleaseProxyKey}}
+}