Merge branch 'main' into release-0-7-0

Author: reshnm

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.
@@ -298,6 +339,108 @@ env := testing.NewEnvironmentBuilder().
 env.ShouldReconcile(testing.RequestFromStrings("testresource"))
 ```
 
+### Readiness Checks
+
+The `pkg/readiness` package provides a simple way to check if a kubernetes resource is ready.
+The meaning of readiness depends on the resource type.
+
+#### Example
+
+```go
+deployment := &appsv1.Deployment{}
+err := r.Client.Get(ctx, types.NamespacedName{
+  Name:      "my-deployment",
+  Namespace: "my-namespace",
+}, deployment)
+
+if err != nil {
+  return err
+}
+
+readiness := readiness.CheckDeployment(deployment)
+
+if readiness.IsReady() {
+  fmt.Println("Deployment is ready")
+} else {
+  fmt.Printf("Deployment is not ready: %s\n", readiness.Message())
+}
+```
+
+### 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).

go.mod

@@ -19,6 +19,7 @@ require (
 	k8s.io/client-go v0.33.0
 	k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979
 	sigs.k8s.io/controller-runtime v0.20.4
+	sigs.k8s.io/gateway-api v1.3.0
 	sigs.k8s.io/yaml v1.4.0
 )
 
@@ -66,13 +67,13 @@ require (
 	golang.org/x/time v0.10.0 // indirect
 	golang.org/x/tools v0.33.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/protobuf v1.36.5 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff // indirect
 	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.6.0 // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.7.0 // indirect
 )

go.sum

@@ -193,8 +193,8 @@ google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQ
 google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
 google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
-google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
+google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
@@ -226,12 +226,14 @@ k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979 h1:jgJW5IePPXLGB8e/1wvd0Ich9QE97
 k8s.io/utils v0.0.0-20250502105355-0f33e8f1c979/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/controller-runtime v0.20.4 h1:X3c+Odnxz+iPTRobG4tp092+CvBU9UK0t/bRf+n0DGU=
 sigs.k8s.io/controller-runtime v0.20.4/go.mod h1:xg2XB0K5ShQzAgsoujxuKN4LNXR2LfwwHsPj7Iaw+XY=
+sigs.k8s.io/gateway-api v1.3.0 h1:q6okN+/UKDATola4JY7zXzx40WO4VISk7i9DIfOvr9M=
+sigs.k8s.io/gateway-api v1.3.0/go.mod h1:d8NV8nJbaRbEKem+5IuxkL8gJGOZ+FJ+NvOIltV8gDk=
 sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 h1:gBQPwqORJ8d8/YNZWEjoZs7npUVDpVXUUOFfW6CgAqE=
 sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
 sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
 sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
-sigs.k8s.io/structured-merge-diff/v4 v4.6.0 h1:IUA9nvMmnKWcj5jl84xn+T5MnlZKThmUW1TdblaLVAc=
-sigs.k8s.io/structured-merge-diff/v4 v4.6.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
+sigs.k8s.io/structured-merge-diff/v4 v4.7.0 h1:qPeWmscJcXP0snki5IYF79Z8xrl8ETFxgMd7wez1XkI=
+sigs.k8s.io/structured-merge-diff/v4 v4.7.0/go.mod h1:dDy58f92j70zLsuZVuUX5Wp9vtxXpaZnkPGWeqDfCps=
 sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=
 sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY=

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)
 	}