-
Notifications
You must be signed in to change notification settings - Fork 137
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
Signed-off-by: Chris Doherty <[email protected]>
- Loading branch information
1 parent
5126bd0
commit 9cba4cf
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.