feat: add crd management library

docs/libs/crds.md

@@ -1,3 +1,62 @@
 # Custom Resource Definitions
 
-The `pkg/init/crds` package allows user to deploy CRDs from yaml files to a target cluster. It uses `embed.FS` to provide the files for deployment.
+The `pkg/crds` package allows user to deploy CRDs from yaml files to a target cluster.
+A typical use case is to use `embed.FS` to embed the CRDs in the controller binary and deploy them to the target clusters.
+
+The decision on which cluster a CRD should be deployed to is made by the `CRDManager` based on the labels of the CRDs and the labels of the clusters.
+The label key is passed to the `CRDManager` when it is created.
+Each cluster is then registered with a label value at the `CRDManager`.
+
+## Example
+
+```go
+package main
+
+import (
+	"context"
+	"embed"
+
+	"github.com/openmcp-project/controller-utils/pkg/clusters"
+	"github.com/openmcp-project/controller-utils/pkg/crds"
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+)
+
+//go:embed crds
+var crdsFS embed.FS
+var crdsPath = "crds"
+
+func main() {
+	ctx := context.Background()
+
+	onboardingCluster := clusters.NewTestClusterFromClient("onboarding", getOnboardingClient())
+	workloadCluster := clusters.NewTestClusterFromClient("workload", getWorkloadClient())
+
+	// use "openmcp.cloud/cluster" as the CRD label key
+	crdManager := crds.NewCRDManager("openmcp.cloud/cluster", func() ([]*apiextv1.CustomResourceDefinition, error) {
+		return crds.CRDsFromFileSystem(crdsFS, crdsPath)
+	})
+
+	// register the onboarding cluster with label value "onboarding"
+	crdManager.AddCRDLabelToClusterMapping("onboarding", onboardingCluster)
+	// register the workload cluster with label value "workload"
+	crdManager.AddCRDLabelToClusterMapping("workload", workloadCluster)
+
+	// create/update the CRDs in all clusters
+	err := crdManager.CreateOrUpdateCRDs(ctx, nil)
+	if err != nil {
+        panic(err)
+    }
+}
+```
+
+The CRDs need to be annotated with the label key and the label value of the cluster they should be deployed to.
+
+```yaml
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: testresources.example.com
+  labels:
+    openmcp.cloud/cluster: "onboarding"
+...
+```

pkg/crds/crds.go

@@ -0,0 +1,113 @@
+package crds
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"path/filepath"
+
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"sigs.k8s.io/yaml"
+
+	"github.com/openmcp-project/controller-utils/pkg/clusters"
+	"github.com/openmcp-project/controller-utils/pkg/controller"
+	"github.com/openmcp-project/controller-utils/pkg/errors"
+	"github.com/openmcp-project/controller-utils/pkg/logging"
+	"github.com/openmcp-project/controller-utils/pkg/resources"
+)
+
+type (
+	CRDLabelToClusterMappings map[string]*clusters.Cluster
+	CRDList                   func() ([]*apiextv1.CustomResourceDefinition, error)
+)
+
+type CRDManager struct {
+	mappingLabelName           string
+	crdLabelsToClusterMappings CRDLabelToClusterMappings
+	crdList                    CRDList
+}
+
+func NewCRDManager(mappingLabelName string, crdList CRDList) *CRDManager {
+	return &CRDManager{
+		mappingLabelName:           mappingLabelName,
+		crdLabelsToClusterMappings: make(CRDLabelToClusterMappings),
+		crdList:                    crdList,
+	}
+}
+
+func (m *CRDManager) AddCRDLabelToClusterMapping(labelValue string, cluster *clusters.Cluster) {
+	m.crdLabelsToClusterMappings[labelValue] = cluster
+}
+
+func (m *CRDManager) CreateOrUpdateCRDs(ctx context.Context, log *logging.Logger) error {
+	crds, err := m.crdList()
+	if err != nil {
+		return fmt.Errorf("error getting CRDs: %w", err)
+	}
+
+	var errs error
+
+	for _, crd := range crds {
+		c, err := m.getClusterForCRD(crd)
+		if err != nil {
+			errs = errors.Join(errs, err)
+			continue
+		}
+
+		if log != nil {
+			log.Info("creating/updating CRD", "name", crd.Name, "cluster", c.ID())
+		}
+		err = resources.CreateOrUpdateResource(ctx, c.Client(), resources.NewCRDMutator(crd, crd.Labels, crd.Annotations))
+		errs = errors.Join(errs, err)
+	}
+
+	if errs != nil {
+		return fmt.Errorf("error creating/updating CRDs: %w", errs)
+	}
+	return nil
+}
+
+func (m *CRDManager) getClusterForCRD(crd *apiextv1.CustomResourceDefinition) (*clusters.Cluster, error) {
+	labelValue, ok := controller.GetLabel(crd, m.mappingLabelName)
+	if !ok {
+		return nil, fmt.Errorf("missing label '%s' for CRD '%s'", m.mappingLabelName, crd.Name)
+	}
+
+	cluster, ok := m.crdLabelsToClusterMappings[labelValue]
+	if !ok {
+		return nil, fmt.Errorf("no cluster mapping found for label value '%s' in CRD '%s'", labelValue, crd.Name)
+	}
+
+	return cluster, nil
+}
+
+// CRDsFromFileSystem reads CRDs from the specified filesystem path.
+func CRDsFromFileSystem(fsys fs.FS, path string) ([]*apiextv1.CustomResourceDefinition, error) {
+	var crds []*apiextv1.CustomResourceDefinition
+
+	entries, err := fs.ReadDir(fsys, path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read directory %s: %w", path, err)
+	}
+
+	for _, entry := range entries {
+		if entry.IsDir() {
+			continue
+		}
+
+		filePath := filepath.Join(path, entry.Name())
+		data, err := fs.ReadFile(fsys, filePath)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read file %s: %w", filePath, err)
+		}
+
+		var crd apiextv1.CustomResourceDefinition
+		if err := yaml.Unmarshal(data, &crd); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal CRD from file %s: %w", filePath, err)
+		}
+
+		crds = append(crds, &crd)
+	}
+
+	return crds, nil
+}

pkg/crds/crds_test.go

@@ -0,0 +1,86 @@
+package crds_test
+
+import (
+	"context"
+	"embed"
+	"testing"
+
+	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/openmcp-project/controller-utils/pkg/clusters"
+	utilstest "github.com/openmcp-project/controller-utils/pkg/testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/openmcp-project/controller-utils/pkg/crds"
+)
+
+//go:embed testdata/*
+var testFS embed.FS
+
+func TestCRDs(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "CRDs Test Suite")
+}
+
+var _ = Describe("CRDsFromFileSystem", func() {
+	It("should correctly read and parse CRDs from the filesystem", func() {
+		crdPath := "testdata"
+		crdsList, err := crds.CRDsFromFileSystem(testFS, crdPath)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(crdsList).To(HaveLen(2))
+
+		// Validate the first CRD
+		Expect(crdsList[0].Name).To(Equal("testresources.example.com"))
+		Expect(crdsList[0].Spec.Names.Kind).To(Equal("TestResource"))
+
+		// Validate the second CRD
+		Expect(crdsList[1].Name).To(Equal("sampleresources.example.com"))
+		Expect(crdsList[1].Spec.Names.Kind).To(Equal("SampleResource"))
+	})
+})
+
+var _ = Describe("CRDManager", func() {
+	It("should correctly manage CRD mappings and create/update CRDs", func() {
+		scheme := runtime.NewScheme()
+		err := apiextv1.AddToScheme(scheme)
+		Expect(err).NotTo(HaveOccurred())
+
+		// Create fake clients
+		clientA, err := utilstest.GetFakeClient(scheme)
+		Expect(err).NotTo(HaveOccurred())
+
+		clientB, err := utilstest.GetFakeClient(scheme)
+		Expect(err).NotTo(HaveOccurred())
+
+		// Create fake clusters
+		clusterA := clusters.NewTestClusterFromClient("cluster_a", clientA)
+		clusterB := clusters.NewTestClusterFromClient("cluster_b", clientB)
+
+		crdManager := crds.NewCRDManager("openmcp.cloud/cluster", func() ([]*apiextv1.CustomResourceDefinition, error) {
+			return crds.CRDsFromFileSystem(testFS, "testdata")
+		})
+
+		crdManager.AddCRDLabelToClusterMapping("cluster_a", clusterA)
+		crdManager.AddCRDLabelToClusterMapping("cluster_b", clusterB)
+
+		ctx := context.Background()
+
+		err = crdManager.CreateOrUpdateCRDs(ctx, nil)
+		Expect(err).NotTo(HaveOccurred())
+
+		// Verify that the CRDs were created in the respective clusters
+		crdA := &apiextv1.CustomResourceDefinition{}
+		err = clientA.Get(ctx, types.NamespacedName{Name: "testresources.example.com"}, crdA)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(crdA.Labels["openmcp.cloud/cluster"]).To(Equal("cluster_a"))
+
+		crdB := &apiextv1.CustomResourceDefinition{}
+		err = clientB.Get(ctx, types.NamespacedName{Name: "sampleresources.example.com"}, crdB)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(crdB.Labels["openmcp.cloud/cluster"]).To(Equal("cluster_b"))
+	})
+})

pkg/crds/testdata/crd_a.yaml

@@ -0,0 +1,30 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: testresources.example.com
+  labels:
+    openmcp.cloud/cluster: "cluster_a"
+spec:
+  group: example.com
+  names:
+    kind: TestResource
+    listKind: TestResourceList
+    plural: testresources
+    singular: testresource
+  scope: Namespaced
+  versions:
+    - name: v1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                name:
+                  type: string
+                replicas:
+                  type: integer
+                  minimum: 1

pkg/crds/testdata/crd_b.yaml

@@ -0,0 +1,29 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: sampleresources.example.com
+  labels:
+    openmcp.cloud/cluster: "cluster_b"
+spec:
+  group: example.com
+  names:
+    kind: SampleResource
+    listKind: SampleResourceList
+    plural: sampleresources
+    singular: sampleresource
+  scope: Namespaced
+  versions:
+    - name: v1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                type:
+                  type: string
+                enabled:
+                  type: boolean