-
Notifications
You must be signed in to change notification settings - Fork 138
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor reconcilers and controller manager construction logic (#674)
These changes tidy up the tinker controller implementation in several ways: 1. Reduce the manager to a single `NewManager` function with the workflow controller pre-registerred. 2. Replace the full `manager.Manager` dependency in the Tink Server with a `cluster.Cluster`. [Cluster](https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/cluster#Cluster) objects are only really concerned with providing an API to retrieve a client and facilitate caching unlike the [manager object](https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/manager#Manager) that's concerned with all components related to controllers, webhooks, leader election and more. This makes the cluster object more appropriate for the Tink Server. 3. Colocate used indexes in the packages that consume them and remove unused indexes. Historically the unused indexes have been consumed by other Tinkerbell project; the projects can copy the indexes if still needed. 4. Rename `workflow.Controller` to `workflow.Reconciler` to be representative of the implementation. [Controllers](https://pkg.go.dev/sigs.k8s.io/controller-runtime#hdr-Controllers) contain more than just reconciliation components. 5. Patch through metrics and prob endpoint configuration to the controller manager. The CLI flags are the common flags exposed from controllers: `--metrics-bind-address`, `--health-probe-bind-address`.
- Loading branch information
Showing
11 changed files
with
275 additions
and
521 deletions.
There are no files selected for viewing
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 |
|---|---|---|
| @@ -1,225 +1,53 @@ | ||
| package controller | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/go-logr/zapr" | ||
| "github.com/tinkerbell/tink/api/v1alpha1" | ||
| "github.com/tinkerbell/tink/internal/workflow" | ||
| "k8s.io/apimachinery/pkg/runtime" | ||
| clientgoscheme "k8s.io/client-go/kubernetes/scheme" | ||
| "k8s.io/client-go/rest" | ||
| "knative.dev/pkg/logging" | ||
| controllerruntime "sigs.k8s.io/controller-runtime" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| ctrl "sigs.k8s.io/controller-runtime" | ||
| "sigs.k8s.io/controller-runtime/pkg/healthz" | ||
| "sigs.k8s.io/controller-runtime/pkg/manager" | ||
| ) | ||
|
|
||
| const ( | ||
| WorkflowWorkerAddrIndex = ".status.tasks.workerAddr" | ||
| WorkflowWorkerNonTerminalStateIndex = ".status.state.nonTerminalWorker" | ||
| WorkflowStateIndex = ".status.state" | ||
| HardwareMACAddrIndex = ".spec.interfaces.dhcp.mac" | ||
| HardwareIPAddrIndex = ".spec.interfaces.dhcp.ip" | ||
| var schemeBuilder = runtime.NewSchemeBuilder( | ||
| clientgoscheme.AddToScheme, | ||
| v1alpha1.AddToScheme, | ||
| ) | ||
|
|
||
| var ( | ||
| runtimescheme = runtime.NewScheme() | ||
| options = Options{} | ||
| ) | ||
|
|
||
| func init() { | ||
| _ = clientgoscheme.AddToScheme(runtimescheme) | ||
| _ = v1alpha1.AddToScheme(runtimescheme) | ||
| } | ||
|
|
||
| // Options for running this binary. | ||
| type Options struct { | ||
| MetricsPort int | ||
| HealthProbePort int | ||
| } | ||
|
|
||
| // GetControllerOptions returns a set of options used by the Tink controller. | ||
| // These options include leader election enabled. | ||
| func GetControllerOptions() controllerruntime.Options { | ||
| return controllerruntime.Options{ | ||
| Logger: zapr.NewLogger(logging.FromContext(context.Background()).Desugar()), | ||
| LeaderElection: true, | ||
| LeaderElectionID: "tink-leader-election", | ||
| Scheme: runtimescheme, | ||
| MetricsBindAddress: fmt.Sprintf(":%d", options.MetricsPort), | ||
| HealthProbeBindAddress: fmt.Sprintf(":%d", options.HealthProbePort), | ||
| } | ||
| // DefaultScheme returns a scheme with all the types necessary for the tink controller. | ||
| func DefaultScheme() *runtime.Scheme { | ||
| s := runtime.NewScheme() | ||
| _ = schemeBuilder.AddToScheme(s) | ||
| return s | ||
| } | ||
|
|
||
| // GetNamespacedControllerOptions returns a set of options used by the Tink controller. | ||
| // These options include leader election enabled. | ||
| func GetNamespacedControllerOptions(namespace string) controllerruntime.Options { | ||
| opts := GetControllerOptions() | ||
| opts.Namespace = namespace | ||
| return opts | ||
| } | ||
|
|
||
| // GetServerOptions returns a set of options used by the Tink API. | ||
| // These options include leader election disabled. | ||
| func GetServerOptions() controllerruntime.Options { | ||
| return controllerruntime.Options{ | ||
| Logger: zapr.NewLogger(logging.FromContext(context.Background()).Desugar()), | ||
| LeaderElection: false, | ||
| Scheme: runtimescheme, | ||
| MetricsBindAddress: fmt.Sprintf(":%d", options.MetricsPort), | ||
| HealthProbeBindAddress: fmt.Sprintf(":%d", options.HealthProbePort), | ||
| // NewManager creates a new controller manager with tink controller controllers pre-registered. | ||
| // If opts.Scheme is nil, DefaultScheme() is used. | ||
| func NewManager(cfg *rest.Config, opts ctrl.Options) (ctrl.Manager, error) { | ||
| if opts.Scheme == nil { | ||
| opts.Scheme = DefaultScheme() | ||
| } | ||
| } | ||
|
|
||
| // NewManagerOrDie instantiates a controller manager. | ||
| func NewManagerOrDie(config *rest.Config, options controllerruntime.Options) Manager { | ||
| m, err := NewManager(config, options) | ||
| mgr, err := ctrl.NewManager(cfg, opts) | ||
| if err != nil { | ||
| panic(err) | ||
| return nil, fmt.Errorf("controller manager: %w", err) | ||
| } | ||
| return m | ||
| } | ||
|
|
||
| // NewManager instantiates a controller manager. | ||
| func NewManager(config *rest.Config, options controllerruntime.Options) (Manager, error) { | ||
| m, err := controllerruntime.NewManager(config, options) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| indexers := []struct { | ||
| obj client.Object | ||
| field string | ||
| extractValue client.IndexerFunc | ||
| }{ | ||
| { | ||
| &v1alpha1.Workflow{}, | ||
| WorkflowWorkerAddrIndex, | ||
| WorkflowWorkerAddrIndexFunc, | ||
| }, | ||
| { | ||
| &v1alpha1.Workflow{}, | ||
| WorkflowWorkerNonTerminalStateIndex, | ||
| WorkflowWorkerNonTerminalStateIndexFunc, | ||
| }, | ||
| { | ||
| &v1alpha1.Workflow{}, | ||
| WorkflowStateIndex, | ||
| WorkflowStateIndexFunc, | ||
| }, | ||
| { | ||
| &v1alpha1.Hardware{}, | ||
| HardwareIPAddrIndex, | ||
| HardwareIPIndexFunc, | ||
| }, | ||
| { | ||
| &v1alpha1.Hardware{}, | ||
| HardwareMACAddrIndex, | ||
| HardwareMacIndexFunc, | ||
| }, | ||
| if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { | ||
| return nil, fmt.Errorf("set up health check: %w", err) | ||
| } | ||
| for _, indexer := range indexers { | ||
| if err := m.GetFieldIndexer().IndexField( | ||
| context.Background(), | ||
| indexer.obj, | ||
| indexer.field, | ||
| indexer.extractValue, | ||
| ); err != nil { | ||
| return nil, fmt.Errorf("failed to setup %s indexer, %w", indexer.field, err) | ||
| } | ||
| } | ||
| return &GenericControllerManager{Manager: m}, nil | ||
| } | ||
|
|
||
| // GenericControllerManager is a manager.Manager that allows for registering of controllers. | ||
| type GenericControllerManager struct { | ||
| manager.Manager | ||
| } | ||
|
|
||
| // RegisterControllers registers a set of controllers to the controller manager. | ||
| func (m *GenericControllerManager) RegisterControllers(ctx context.Context, controllers ...Controller) Manager { | ||
| for _, c := range controllers { | ||
| if err := c.Register(ctx, m); err != nil { | ||
| panic(err) | ||
| } | ||
| if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil { | ||
| return nil, fmt.Errorf("set up ready check: %w", err) | ||
| } | ||
| if err := m.AddHealthzCheck("healthz", healthz.Ping); err != nil { | ||
| panic(fmt.Sprintf("Failed to add readiness probe, %s", err.Error())) | ||
| } | ||
| return m | ||
| } | ||
|
|
||
| // WorkflowWorkerAddrIndexFunc func returns a list of worker addresses from a workflow. | ||
| func WorkflowWorkerAddrIndexFunc(obj client.Object) []string { | ||
| wf, ok := obj.(*v1alpha1.Workflow) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| resp := []string{} | ||
| for _, task := range wf.Status.Tasks { | ||
| if task.WorkerAddr != "" { | ||
| resp = append(resp, task.WorkerAddr) | ||
| } | ||
| } | ||
| return resp | ||
| } | ||
|
|
||
| // WorkflowWorkerNonTerminalStateIndexFunc func returns a list of worker addresses for workflows | ||
| // in a running or pending state. | ||
| func WorkflowWorkerNonTerminalStateIndexFunc(obj client.Object) []string { | ||
| wf, ok := obj.(*v1alpha1.Workflow) | ||
| if !ok { | ||
| return nil | ||
| } | ||
|
|
||
| resp := []string{} | ||
| if !(wf.Status.State == v1alpha1.WorkflowStateRunning || wf.Status.State == v1alpha1.WorkflowStatePending) { | ||
| return resp | ||
| } | ||
| for _, task := range wf.Status.Tasks { | ||
| if task.WorkerAddr != "" { | ||
| resp = append(resp, task.WorkerAddr) | ||
| } | ||
| } | ||
| return resp | ||
| } | ||
|
|
||
| // WorkflowStateIndexFunc func returns the workflow state. | ||
| func WorkflowStateIndexFunc(obj client.Object) []string { | ||
| wf, ok := obj.(*v1alpha1.Workflow) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| return []string{string(wf.Status.State)} | ||
| } | ||
|
|
||
| // HardwareMacIndexFunc returns a list of mac addresses from a hardware. | ||
| func HardwareMacIndexFunc(obj client.Object) []string { | ||
| hw, ok := obj.(*v1alpha1.Hardware) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| resp := []string{} | ||
| for _, iface := range hw.Spec.Interfaces { | ||
| if iface.DHCP != nil && iface.DHCP.MAC != "" { | ||
| resp = append(resp, iface.DHCP.MAC) | ||
| } | ||
| err = workflow.NewReconciler(mgr.GetClient()).SetupWithManager(mgr) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("setup workflow reconciler: %w", err) | ||
| } | ||
| return resp | ||
| } | ||
|
|
||
| // HardwareIPIndexFunc returns a list of IP addresses from a hardware. | ||
| func HardwareIPIndexFunc(obj client.Object) []string { | ||
| hw, ok := obj.(*v1alpha1.Hardware) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| resp := []string{} | ||
| for _, iface := range hw.Spec.Interfaces { | ||
| if iface.DHCP != nil && iface.DHCP.IP != nil && iface.DHCP.IP.Address != "" { | ||
| resp = append(resp, iface.DHCP.IP.Address) | ||
| } | ||
| } | ||
| return resp | ||
| return mgr, nil | ||
| } |
Oops, something went wrong.