feat(resource): implement ResourceReconciler and associated subroutines for managing OCI and Helm repositories

Signed-off-by: Bastian Echterhölter <[email protected]>
On-behalf-of: @SAP <[email protected]>

Author: nexus49

cmd/operator.go

@@ -113,6 +113,12 @@ func RunController(_ *cobra.Command, _ []string) { // coverage-ignore
 		os.Exit(1)
 	}
 
+	resourceReconciler := controller.NewResourceReconciler(log, mgr, &operatorCfg)
+	if err := resourceReconciler.SetupWithManager(mgr, defaultCfg, log); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "PlatformMesh")
+		os.Exit(1)
+	}
+
 	if operatorCfg.PatchOIDCControllerEnabled {
 		realmReconciler := controller.NewRealmReconciler(mgr, log, &operatorCfg)
 		if err := realmReconciler.SetupWithManager(mgr, defaultCfg, log); err != nil {

go.mod

@@ -3,7 +3,7 @@ module github.com/platform-mesh/platform-mesh-operator
 go 1.25.1
 
 replace sigs.k8s.io/controller-runtime => github.com/kcp-dev/controller-runtime v0.19.0-kcp.1
-
+replace ocm.software/open-component-model/kubernetes/controller => github.com/open-component-model/open-component-model/kubernetes/controller v0.0.0-20251104083903-c6d40af9889f
 replace (
 	k8s.io/api => k8s.io/api v0.34.1
 	k8s.io/apimachinery => k8s.io/apimachinery v0.34.1

internal/controller/resource_controller.go

@@ -0,0 +1,84 @@
+/*
+Copyright 2025.
+
+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 controller
+
+import (
+	"context"
+
+	pmconfig "github.com/platform-mesh/golang-commons/config"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/controllerruntime"
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/subroutine"
+	"github.com/platform-mesh/golang-commons/logger"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
+	"github.com/platform-mesh/platform-mesh-operator/internal/config"
+	"github.com/platform-mesh/platform-mesh-operator/pkg/subroutines/resource"
+)
+
+var (
+	resourceReconcilerName = "ResourceReconciler"
+)
+
+// ResourceReconciler reconciles a PlatformMesh object
+type ResourceReconciler struct {
+	lifecycle *controllerruntime.LifecycleManager
+}
+
+var gvk = schema.GroupVersionKind{
+	Group:   "delivery.ocm.software",
+	Version: "v1alpha1",
+	Kind:    "Resource",
+}
+
+func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(gvk)
+	return r.lifecycle.Reconcile(ctx, req, obj)
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *ResourceReconciler) SetupWithManager(mgr ctrl.Manager, cfg *pmconfig.CommonServiceConfig,
+	log *logger.Logger, eventPredicates ...predicate.Predicate) error {
+
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(gvk)
+
+	mgr.GetScheme().AddKnownTypeWithName(gvk, &unstructured.Unstructured{})
+	mgr.GetScheme().AddKnownTypeWithName(gvk.GroupVersion().WithKind(gvk.Kind+"List"), &unstructured.UnstructuredList{})
+
+	builder, err := r.lifecycle.SetupWithManagerBuilder(mgr, cfg.MaxConcurrentReconciles, resourceReconcilerName, obj,
+		cfg.DebugLabelValue, log, eventPredicates...)
+	if err != nil {
+		return err
+	}
+	return builder.Complete(r)
+}
+
+func NewResourceReconciler(log *logger.Logger, mgr ctrl.Manager, cfg *config.OperatorConfig) *ResourceReconciler {
+	var subs []subroutine.Subroutine
+
+	subs = append(subs, resource.NewResourceSubroutine(mgr))
+
+	return &ResourceReconciler{
+		lifecycle: controllerruntime.NewLifecycleManager(subs, operatorName,
+			resourceReconcilerName, mgr.GetClient(), log).WithReadOnly(),
+	}
+}

pkg/subroutines/resource/subroutine.go

@@ -0,0 +1,234 @@
+package resource
+
+import (
+	"context"
+	"strings"
+
+	"github.com/platform-mesh/golang-commons/controller/lifecycle/runtimeobject"
+	"github.com/platform-mesh/golang-commons/errors"
+	"github.com/platform-mesh/golang-commons/logger"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+var ociRepoGvk = schema.GroupVersionKind{
+	Group:   "source.toolkit.fluxcd.io",
+	Version: "v1",
+	Kind:    "OCIRepository",
+}
+
+var helmRepoGvk = schema.GroupVersionKind{
+	Group:   "source.toolkit.fluxcd.io",
+	Version: "v1",
+	Kind:    "HelmRepository",
+}
+
+var helmReleaseGvk = schema.GroupVersionKind{
+	Group:   "helm.toolkit.fluxcd.io",
+	Version: "v2",
+	Kind:    "HelmRelease",
+}
+
+type ResourceSubroutine struct {
+	mgr manager.Manager
+}
+
+func NewResourceSubroutine(mgr manager.Manager) *ResourceSubroutine {
+	return &ResourceSubroutine{mgr: mgr}
+}
+
+func (r *ResourceSubroutine) GetName() string {
+	return "ResourceSubroutine"
+}
+
+func (r *ResourceSubroutine) Finalize(_ context.Context, _ runtimeobject.RuntimeObject) (ctrl.Result, errors.OperatorError) {
+	return ctrl.Result{}, nil
+}
+
+func (r *ResourceSubroutine) Finalizers() []string { // coverage-ignore
+	return []string{}
+}
+
+func (r *ResourceSubroutine) Process(ctx context.Context, runtimeObj runtimeobject.RuntimeObject) (ctrl.Result, errors.OperatorError) {
+	inst := runtimeObj.(*unstructured.Unstructured)
+	log := logger.LoadLoggerFromContext(ctx).ChildLogger("name", r.GetName())
+
+	repo := inst.GetLabels()["repo"]
+	artifact := inst.GetLabels()["artifact"]
+
+	if repo == "oci" && artifact == "chart" {
+		log.Debug().Msg("Create/Update OCI Repo")
+		result, err := r.updateOciRepo(ctx, inst, log)
+		if err != nil {
+			return result, err
+		}
+	}
+	if repo == "helm" && artifact == "chart" {
+		log.Debug().Msg("Create/Update Flux Helm Repository Repo")
+		result, err := r.updateHelmRepository(ctx, inst, log)
+		if err != nil {
+			return result, err
+		}
+		log.Debug().Msg("Update Flux Helm Release Repo")
+		result, err = r.updateHelmRelease(ctx, inst, log)
+		if err != nil {
+			return result, err
+		}
+	}
+	if repo == "helm" && artifact == "image" {
+		log.Debug().Msg("Update Helm Release with Image Tag")
+		result, err := r.updateHelmReleaseWithImageTag(ctx, inst, log)
+		if err != nil {
+			return result, err
+		}
+	}
+	return ctrl.Result{}, nil
+}
+
+func (r *ResourceSubroutine) updateHelmReleaseWithImageTag(ctx context.Context, inst *unstructured.Unstructured, log *logger.Logger) (ctrl.Result, errors.OperatorError) {
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(helmReleaseGvk)
+	obj.SetName(inst.GetName())
+	obj.SetNamespace(inst.GetNamespace())
+
+	version, found, err := unstructured.NestedString(inst.Object, "status", "resource", "version")
+	if err != nil || !found {
+		log.Info().Err(err).Msg("Failed to get version from Resource status")
+	}
+
+	err = r.mgr.GetClient().Get(ctx, client.ObjectKeyFromObject(inst), obj)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to get HelmRelease")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+
+	err = unstructured.SetNestedField(obj.Object, version, "spec", "values", "image", "tag")
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to set version in HelmRelease spec")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, false)
+	}
+
+	err = r.mgr.GetClient().Update(ctx, obj)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to update HelmRelease")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+	return ctrl.Result{}, nil
+}
+
+func (r *ResourceSubroutine) updateHelmRelease(ctx context.Context, inst *unstructured.Unstructured, log *logger.Logger) (ctrl.Result, errors.OperatorError) {
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(helmReleaseGvk)
+	obj.SetName(inst.GetName())
+	obj.SetNamespace(inst.GetNamespace())
+
+	version, found, err := unstructured.NestedString(inst.Object, "status", "resource", "version")
+	if err != nil || !found {
+		log.Info().Err(err).Msg("Failed to get version from Resource status")
+	}
+
+	err = r.mgr.GetClient().Get(ctx, client.ObjectKeyFromObject(inst), obj)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to get HelmRelease")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+
+	err = unstructured.SetNestedField(obj.Object, version, "spec", "chart", "spec", "version")
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to set version in HelmRelease spec")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, false)
+	}
+
+	err = r.mgr.GetClient().Update(ctx, obj)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to update HelmRelease")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+	return ctrl.Result{}, nil
+}
+
+func (r *ResourceSubroutine) updateHelmRepository(ctx context.Context, inst *unstructured.Unstructured, log *logger.Logger) (ctrl.Result, errors.OperatorError) {
+	url, found, err := unstructured.NestedString(inst.Object, "status", "resource", "access", "helmRepository")
+	if err != nil || !found {
+		log.Info().Err(err).Msg("Failed to get imageReference from Resource status")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, false)
+	}
+
+	log.Info().Msg("Processing OCI Chart Resource")
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(helmRepoGvk)
+	obj.SetName(inst.GetName())
+	obj.SetNamespace(inst.GetNamespace())
+	_, err = controllerutil.CreateOrUpdate(ctx, r.mgr.GetClient(), obj, func() error {
+		err := unstructured.SetNestedField(obj.Object, url, "spec", "url")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedField(obj.Object, "generic", "spec", "provider")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedField(obj.Object, "5m", "spec", "interval")
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to create or update OCIRepository")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+	return ctrl.Result{}, nil
+}
+
+func (r *ResourceSubroutine) updateOciRepo(ctx context.Context, inst *unstructured.Unstructured, log *logger.Logger) (ctrl.Result, errors.OperatorError) {
+	version, found, err := unstructured.NestedString(inst.Object, "status", "resource", "version")
+	if err != nil || !found {
+		log.Info().Err(err).Msg("Failed to get version from Resource status")
+	}
+	url, found, err := unstructured.NestedString(inst.Object, "status", "resource", "access", "imageReference")
+	if err != nil || !found {
+		log.Info().Err(err).Msg("Failed to get imageReference from Resource status")
+	}
+	url = "oci://" + url
+	url = strings.TrimSuffix(url, ":"+version)
+
+	// Update or create oci repo
+	log.Info().Msg("Processing OCI Chart Resource")
+	obj := &unstructured.Unstructured{}
+	obj.SetGroupVersionKind(ociRepoGvk)
+	obj.SetName(inst.GetName())
+	obj.SetNamespace(inst.GetNamespace())
+	_, err = controllerutil.CreateOrUpdate(ctx, r.mgr.GetClient(), obj, func() error {
+		err := unstructured.SetNestedField(obj.Object, version, "spec", "ref", "tag")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedField(obj.Object, url, "spec", "url")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedField(obj.Object, "generic", "spec", "provider")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedField(obj.Object, "1m0s", "spec", "interval")
+		if err != nil {
+			return err
+		}
+		err = unstructured.SetNestedMap(inst.Object, map[string]interface{}{
+			"mediaType": "application/vnd.cncf.helm.chart.content.v1.tar+gzip",
+			"operation": "copy",
+		}, "spec", "layerSelector")
+		return nil
+	})
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to create or update OCIRepository")
+		return ctrl.Result{}, errors.NewOperatorError(err, true, true)
+	}
+	return ctrl.Result{}, nil
+}