Merge branch 'main' into feat/resources

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.

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