Merge branch 'main' into feat/readinesscheck

Author: reshnm

.github/workflows/ci.yaml

@@ -14,12 +14,12 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4
         with:
           submodules: recursive
 
       - name: Set up Go
-        uses: actions/setup-go@v5
+        uses: actions/setup-go@0aaccfd150d50ccaeb58ebd88d36e91967a5f35b # v5
         with:
           go-version-file: go.mod
 

.github/workflows/release.yaml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-24.04
     steps:
       - name: Checkout code
-        uses: actions/checkout@v4
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4
         with:
           ssh-key: ${{ secrets.PUSH_KEY }}
           fetch-tags: true
@@ -72,7 +72,7 @@ jobs:
 
       - name: Build Changelog
         id: github_release
-        uses: mikepenz/release-changelog-builder-action@v5
+        uses: mikepenz/release-changelog-builder-action@e92187bd633e680ebfdd15961a7c30b2d097e7ad # v5
         with:
           mode: "PR"
           configurationJson: |
@@ -106,7 +106,7 @@ jobs:
 
       - name: Create GitHub release
         if: ${{ env.SKIP != 'true' }}
-        uses: softprops/action-gh-release@v2
+        uses: softprops/action-gh-release@da05d552573ad5aba039eaac05058a918a7bf631 # v2
         with:
           tag_name: ${{ env.version }}
           name: Release ${{ env.version }}

.github/workflows/reuse.yaml

@@ -6,6 +6,6 @@ jobs:
   test:
     runs-on: ubuntu-latest
     steps: 
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4
     - name: REUSE Compliance Check
-      uses: fsfe/reuse-action@v5
+      uses: fsfe/reuse-action@bb774aa972c2a89ff34781233d275075cbddf542 # v5

README.md

@@ -270,6 +270,47 @@ The library provides a wrapper around `logr.Logger`, exposing additional helper
 - There are several `FromContext...` functions for retrieving a logger from a `context.Context` object.
 - `InitFlags(...)` can be used to add the configuration flags for this logger to a cobra `FlagSet`.
 
+### threads
+
+The `threads` package provides a simple thread managing library. It can be used to run go routines in a non-blocking manner and provides the possibility to react if the routine has exited.
+
+The most relevant use-case for this library in the context of k8s controllers is to handle dynamic watches on multiple clusters. To start a watch, that cluster's cache's `Start` method has to be used. Because this method is blocking, it has to be executed in a different go routine, and because it can return an error, a simple `go cache.Start(...)` is not enough, because it would hide the error.
+
+#### Noteworthy Functions
+
+- `NewThreadManager` creates a new thread manager.
+	- The first argument is a `context.Context` used by the manager itself. Cancelling this context will stop the manager, and if the context contains a `logging.Logger`, the manager will use it for logging.
+	- The second argument is an optional function that is executed after any go routine executed with this manager has finished. It is also possible to provide such a function for a specific go routine, instead for all of them, see below.
+- Use the `Run` method to start a new go routine.
+	- Starting a go routine cancels the context of any running go routine with the same id.
+	- This method also takes an optional function to be executed after the actual workload is done.
+		- A on-finish function specified here is executed before the on-finish function of the manager is executed.
+	- Note that go routines will wait for the thread manager to be started, if that has not yet happened. If the manager has been started, they will be executed immediately.
+	- The thread manager will cancel the context that is passed into the workload function when the manager is being stopped. If any long-running commands are being run as part of the workload, it is strongly recommended to listen to the context's `Done` channel.
+- Use `Start()` to start the thread manager.
+	- If any go routines have been added before this is called, they will be started now. New go routines added afterwards will be started immediately.
+	- Calling this multiple times doesn't have any effect, unless the manager has already been stopped, in which case `Start()` will panic.
+- There are three ways to stop the thread manager again:
+	- Use its `Stop()` method.
+		- This is a blocking method that waits for all remaining go routines to finish. Their context is cancelled to notify them of the manager being stopped.
+	- Cancel the context that was passed into `NewThreadManager` as the first argument.
+	- Send a `SIGTERM` or `SIGINT` signal to the process.
+- The `TaskManager`'s `Restart`, `RestartOnError`, and `RestartOnSuccess` methods are pre-defined on-finish functions. They are not meant to be used directly, but instead be used as an argument to `Run`. See the example below.
+
+#### Examples
+
+```golang
+mgr := threads.NewThreadManager(ctx, nil)
+mgr.Start()
+// do other stuff
+// start a go routine that is restarted automatically if it finishes with an error
+mgr.Run(myCtx, "myTask", func(ctx context.Context) error {
+	// my task coding
+}, mgr.RestartOnError)
+// do more other stuff
+mgr.Stop()
+```
+
 ### testing
 
 This package contains useful functionality to aid with writing tests.
@@ -325,6 +366,81 @@ if readiness.IsReady() {
 }
 ```
 
+### Kubernetes resource management
+
+The `pkg/resource` package contains some useful functions for working with Kubernetes resources. The `Mutator` interface can be used to modify resources in a generic way. It is used by the `Mutate` function, which takes a resource and a mutator and applies the mutator to the resource.
+The package also contains convenience types for the most common resource types, e.g. `ConfigMap`, `Secret`, `ClusterRole`, `ClusterRoleBinding`, etc. These types implement the `Mutator` interface and can be used to modify the corresponding resources.
+
+#### Examples
+
+Create or update a `ConfigMap`, a `ServiceAccount` and a `Deployment` using the `Mutator` interface:
+
+```go
+type myDeploymentMutator struct {
+}
+
+var _ resource.Mutator[*appsv1.Deployment] = &myDeploymentMutator{}
+
+func newDeploymentMutator() resources.Mutator[*appsv1.Deployment] {
+	return &MyDeploymentMutator{}
+}
+
+func (m *MyDeploymentMutator) String() string {
+	return "deployment default/test"
+}
+
+func (m *MyDeploymentMutator) Empty() *appsv1.Deployment {
+	return &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{	
+			Name:      "test",
+			Namespace: "default",
+		},
+	}
+}
+
+func (m *MyDeploymentMutator) Mutate(deployment *appsv1.Deployment) error {
+	// create one container with an image
+	deployment.Spec.Template.Spec.Containers = []corev1.Container{
+		{
+			Name:  "test",
+			Image: "test-image:latest",
+		},
+	}
+	return nil
+}
+
+
+func ReconcileResources(ctx context.Context, client client.Client) error {
+	configMapResource := resource.NewConfigMap("my-configmap", "my-namespace", map[string]string{)
+		"label1": "value1",
+		"label2": "value2",
+	}, nil)
+
+	serviceAccountResource := resource.NewServiceAccount("my-serviceaccount", "my-namespace", nil, nil)
+	
+	myDeploymentMutator := newDeploymentMutator()
+	
+	var err error
+	
+	err = resources.CreateOrUpdateResource(ctx, client, configMapResource)
+	if err != nil {
+		return err
+	}
+	
+	resources.CreateOrUpdateResource(ctx, client, serviceAccountResource)
+	if err != nil {
+		return err
+	}
+	
+	err = resources.CreateOrUpdateResource(ctx, client, myDeploymentMutator)
+	if err != nil {
+		return err
+	}
+	
+	return nil
+}
+```
+
 ## Support, Feedback, Contributing
 
 This project is open to feature requests/suggestions, bug reports etc. via [GitHub issues](https://github.com/openmcp-project/controller-utils/issues). Contribution and feedback are encouraged and always welcome. For more information about how to contribute, the project structure, as well as additional contribution information, see our [Contribution Guidelines](CONTRIBUTING.md).

pkg/clusters/cluster.go

@@ -23,6 +23,9 @@ type Cluster struct {
 	client client.Client
 	// cluster
 	cluster cluster.Cluster
+
+	clientOpts  *client.Options
+	clusterOpts []cluster.Option
 }
 
 // Initializes a new cluster.
@@ -56,6 +59,36 @@ func (c *Cluster) RegisterConfigPathFlag(flags *flag.FlagSet) {
 	flags.StringVar(&c.cfgPath, fmt.Sprintf("%s-cluster", c.id), "", fmt.Sprintf("Path to the %s cluster kubeconfig file or directory containing either a kubeconfig or host, token, and ca file. Leave empty to use in-cluster config.", c.id))
 }
 
+// WithClientOptions allows to overwrite the default client options.
+// It must be called before InitializeClient().
+// Note that using this method disables the the scheme injection during client initialization.
+// This means that the required scheme should already be set in the options that are passed into this method.
+// Returns the cluster for chaining.
+func (c *Cluster) WithClientOptions(opts client.Options) *Cluster {
+	c.clientOpts = &opts
+	return c
+}
+
+// WithClusterOptions allows to overwrite the default cluster options.
+// It must be called before InitializeClient().
+// Note that using this method disables the the scheme injection during client initialization.
+// This means that the required scheme should be set by the cluster options that are passed into this method.
+// The DefaultClusterOptions function can be passed in as a cluster option to set the scheme.
+// Returns the cluster for chaining.
+func (c *Cluster) WithClusterOptions(opts ...cluster.Option) *Cluster {
+	c.clusterOpts = opts
+	return c
+}
+
+// DefaultClusterOptions returns the default cluster options.
+// This is useful when one wants to add custom cluster options without overwriting the default ones via WithClusterOptions().
+func DefaultClusterOptions(scheme *runtime.Scheme) cluster.Option {
+	return func(o *cluster.Options) {
+		o.Scheme = scheme
+		o.Cache.Scheme = scheme
+	}
+}
+
 ///////////////////
 // STATUS CHECKS //
 ///////////////////
@@ -92,15 +125,11 @@ func (c *Cluster) InitializeID(id string) {
 }
 
 // InitializeRESTConfig loads the cluster's REST config.
-// If the config has already been loaded, this is a no-op.
 // Panics if the cluster's id is not set (InitializeID must be called first).
 func (c *Cluster) InitializeRESTConfig() error {
 	if !c.HasID() {
 		panic("cluster id must be set before loading the config")
 	}
-	if c.HasRESTConfig() {
-		return nil
-	}
 	cfg, err := controller.LoadKubeconfig(c.cfgPath)
 	if err != nil {
 		return fmt.Errorf("failed to load '%s' cluster kubeconfig: %w", c.ID(), err)
@@ -111,20 +140,29 @@ func (c *Cluster) InitializeRESTConfig() error {
 
 // InitializeClient creates a new client for the cluster.
 // This also initializes the cluster's controller-runtime 'Cluster' representation.
-// If the client has already been initialized, this is a no-op.
 // Panics if the cluster's REST config has not been loaded (InitializeRESTConfig must be called first).
 func (c *Cluster) InitializeClient(scheme *runtime.Scheme) error {
 	if !c.HasRESTConfig() {
 		panic("cluster REST config must be set before creating the client")
 	}
-	if c.HasClient() {
-		return nil
+	if c.clientOpts == nil {
+		c.clientOpts = &client.Options{
+			Scheme: scheme,
+		}
 	}
-	cli, err := client.New(c.restCfg, client.Options{Scheme: scheme})
+	cli, err := client.New(c.restCfg, *c.clientOpts)
 	if err != nil {
 		return fmt.Errorf("failed to create '%s' cluster client: %w", c.ID(), err)
 	}
-	clu, err := cluster.New(c.restCfg, func(o *cluster.Options) { o.Scheme = scheme; o.Cache.Scheme = scheme })
+	if c.clusterOpts == nil {
+		c.clusterOpts = []cluster.Option{
+			func(o *cluster.Options) {
+				o.Scheme = scheme
+				o.Cache.Scheme = scheme
+			},
+		}
+	}
+	clu, err := cluster.New(c.restCfg, c.clusterOpts...)
 	if err != nil {
 		return fmt.Errorf("failed to create '%s' cluster Cluster representation: %w", c.ID(), err)
 	}

pkg/resources/clusterrole.go

@@ -0,0 +1,47 @@
+package resources
+
+import (
+	"fmt"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	v1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type ClusterRoleMutator struct {
+	Name  string
+	Rules []v1.PolicyRule
+	meta  Mutator[client.Object]
+}
+
+var _ Mutator[*v1.ClusterRole] = &ClusterRoleMutator{}
+
+func NewClusterRoleMutator(name string, rules []v1.PolicyRule, labels map[string]string, annotations map[string]string) Mutator[*v1.ClusterRole] {
+	return &ClusterRoleMutator{
+		Name:  name,
+		Rules: rules,
+		meta:  NewMetadataMutator(labels, annotations),
+	}
+}
+
+func (m *ClusterRoleMutator) String() string {
+	return fmt.Sprintf("clusterrole %s", m.Name)
+}
+
+func (m *ClusterRoleMutator) Empty() *v1.ClusterRole {
+	return &v1.ClusterRole{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "rbac.authorization.k8s.io/v1",
+			Kind:       "ClusterRole",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: m.Name,
+		},
+	}
+}
+
+func (m *ClusterRoleMutator) Mutate(r *v1.ClusterRole) error {
+	r.Rules = m.Rules
+	return m.meta.Mutate(r)
+}