-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Switch to builder focused cluster support
Signed-off-by: Dr. Stefan Schimanski <[email protected]>
- Loading branch information
Showing
13 changed files
with
452 additions
and
373 deletions.
There are no files selected for viewing
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| /* | ||
| Copyright 2024 The Kubernetes Authors. | ||
| 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" | ||
| "sync" | ||
|
|
||
| "sigs.k8s.io/controller-runtime/pkg/cluster" | ||
| ) | ||
|
|
||
| // MultiClusterController is a Controller that is aware of the Cluster it is | ||
| // running in. It engage and disengage clusters dynamically, starting the | ||
| // watches and stopping them. | ||
| type MultiClusterController interface { | ||
| cluster.AwareRunnable | ||
| Controller | ||
| } | ||
|
|
||
| // ClusterWatcher starts watches for a given Cluster. The ctx should be | ||
| // used to cancel the watch when the Cluster is disengaged. | ||
| type ClusterWatcher interface { | ||
| Watch(ctx context.Context, cl cluster.Cluster) error | ||
| } | ||
|
|
||
| // NewMultiClusterController creates a new MultiClusterController for the given | ||
| // controller with the given ClusterWatcher. | ||
| func NewMultiClusterController(c Controller, watcher ClusterWatcher) MultiClusterController { | ||
| return &multiClusterController{ | ||
| Controller: c, | ||
| watcher: watcher, | ||
| clusters: map[string]struct{}{}, | ||
| } | ||
| } | ||
|
|
||
| type multiClusterController struct { | ||
| Controller | ||
| watcher ClusterWatcher | ||
|
|
||
| lock sync.Mutex | ||
| clusters map[string]struct{} | ||
| } | ||
|
|
||
| // Engage gets called when the runnable should start operations for the given Cluster. | ||
| func (c *multiClusterController) Engage(clusterCtx context.Context, cl cluster.Cluster) error { | ||
| c.lock.Lock() | ||
| defer c.lock.Unlock() | ||
|
|
||
| if _, ok := c.clusters[cl.Name()]; ok { | ||
| return nil | ||
| } | ||
|
|
||
| // pass through in case the controller itself is cluster aware | ||
| if ctrl, ok := c.Controller.(cluster.AwareRunnable); ok { | ||
| if err := ctrl.Engage(clusterCtx, cl); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| // start watches on the cluster | ||
| if err := c.watcher.Watch(clusterCtx, cl); err != nil { | ||
| if ctrl, ok := c.Controller.(cluster.AwareRunnable); ok { | ||
| if err := ctrl.Disengage(clusterCtx, cl); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return err | ||
| } | ||
| c.clusters[cl.Name()] = struct{}{} | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Disengage gets called when the runnable should stop operations for the given Cluster. | ||
| func (c *multiClusterController) Disengage(ctx context.Context, cl cluster.Cluster) error { | ||
| c.lock.Lock() | ||
| defer c.lock.Unlock() | ||
|
|
||
| if _, ok := c.clusters[cl.Name()]; !ok { | ||
| return nil | ||
| } | ||
| delete(c.clusters, cl.Name()) | ||
|
|
||
| // pass through in case the controller itself is cluster aware | ||
| if ctrl, ok := c.Controller.(cluster.AwareRunnable); ok { | ||
| if err := ctrl.Disengage(ctx, cl); err != nil { | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| Copyright 2024 The Kubernetes Authors. | ||
| 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 handler | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "k8s.io/client-go/util/workqueue" | ||
| "sigs.k8s.io/controller-runtime/pkg/event" | ||
| "sigs.k8s.io/controller-runtime/pkg/reconcile" | ||
| ) | ||
|
|
||
| // ForCluster wraps an EventHandler and adds the cluster name to the reconcile.Requests. | ||
| func ForCluster(clusterName string, h EventHandler) EventHandler { | ||
| return &clusterAwareHandler{ | ||
| clusterName: clusterName, | ||
| handler: h, | ||
| } | ||
| } | ||
|
|
||
| type clusterAwareHandler struct { | ||
| handler EventHandler | ||
| clusterName string | ||
| } | ||
|
|
||
| var _ EventHandler = &clusterAwareHandler{} | ||
|
|
||
| func (c *clusterAwareHandler) Create(ctx context.Context, evt event.CreateEvent, q workqueue.RateLimitingInterface) { | ||
| c.handler.Create(ctx, evt, &clusterWorkqueue{RateLimitingInterface: q, clusterName: c.clusterName}) | ||
| } | ||
|
|
||
| func (c *clusterAwareHandler) Update(ctx context.Context, evt event.UpdateEvent, q workqueue.RateLimitingInterface) { | ||
| c.handler.Update(ctx, evt, &clusterWorkqueue{RateLimitingInterface: q, clusterName: c.clusterName}) | ||
| } | ||
|
|
||
| func (c *clusterAwareHandler) Delete(ctx context.Context, evt event.DeleteEvent, q workqueue.RateLimitingInterface) { | ||
| c.handler.Delete(ctx, evt, &clusterWorkqueue{RateLimitingInterface: q, clusterName: c.clusterName}) | ||
| } | ||
|
|
||
| func (c *clusterAwareHandler) Generic(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) { | ||
| c.handler.Generic(ctx, evt, &clusterWorkqueue{RateLimitingInterface: q, clusterName: c.clusterName}) | ||
| } | ||
|
|
||
| // clusterWorkqueue is a wrapper around a RateLimitingInterface that adds the | ||
| // cluster name to the reconcile.Requests | ||
| type clusterWorkqueue struct { | ||
| workqueue.RateLimitingInterface | ||
| clusterName string | ||
| } | ||
|
|
||
| func (q *clusterWorkqueue) AddAfter(item interface{}, duration time.Duration) { | ||
| req := item.(reconcile.Request) | ||
| req.ClusterName = q.clusterName | ||
| q.RateLimitingInterface.AddAfter(req, duration) | ||
| } | ||
|
|
||
| func (q *clusterWorkqueue) Add(item interface{}) { | ||
| req := item.(reconcile.Request) | ||
| req.ClusterName = q.clusterName | ||
| q.RateLimitingInterface.Add(req) | ||
| } | ||
|
|
||
| func (q *clusterWorkqueue) AddRateLimited(item interface{}) { | ||
| req := item.(reconcile.Request) | ||
| req.ClusterName = q.clusterName | ||
| q.RateLimitingInterface.AddRateLimited(req) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.