Skip to content

Commit

Permalink
switch MCP watching to full EventHandler implementation
Browse files Browse the repository at this point in the history 
The main motivation behind this is to give ourselves much more transparency
and control over when precisely we reconcile on an MCP change.

The EnqueueRequestsFromMapFunc() based mechanism used so far invokes the
registered function (mapKataConfigToRequests()) on any and all MCP changes.
Combined with the mapKataConfigToRequests() implementation that filed a
reconcile.Request any time it was invoked without discrimination we were
triggering unnecessarily many reconciliations.

With this change, we only reconcile on relevant MCP changes ("worker" and
"kata-oc") unlike so far when we reconciled on any MCP change, however
unrelated to this controller the changed MCP might have been.  In addition,
we only reconcile on MCP update since reconciling on MCP creation or
deletion doesn't seem useful to this controller.  We also guard against
spurious MCP changes that are irrelevant to us (e.g. timestamps) by making
sure that we only reconcile when values actually used by the controller
change (machine counts and Updating/Updated conditions).

Signed-off-by: Pavel Mores <pmores@redhat.com>
  • Loading branch information
Pavel Mores authored and Pavel Mores committed Mar 2, 2023
1 parent ef92c21 commit ab11dce
Showing 1 changed file with 164 additions and 15 deletions.
179 changes: 164 additions & 15 deletions controllers/openshift_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"encoding/json"
"fmt"
"os"
"reflect"
"time"

appsv1 "k8s.io/api/apps/v1"
Expand All @@ -39,10 +40,11 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/util/workqueue"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
)
Expand Down Expand Up @@ -1041,32 +1043,179 @@ func (r *KataConfigOpenShiftReconciler) createExtensionMc(machinePool string) (c
return ctrl.Result{}, nil, false
}

func (r *KataConfigOpenShiftReconciler) mapKataConfigToRequests(kataConfigObj client.Object) []reconcile.Request {
func (r *KataConfigOpenShiftReconciler) makeReconcileRequest() reconcile.Request {
return reconcile.Request{
NamespacedName: types.NamespacedName{
Name: r.kataConfig.Name,
},
}
}

func isMcpRelevant(mcp client.Object) bool {
mcpName := mcp.GetName()
if mcpName == "kata-oc" || mcpName == "worker" {
return true
}
return false
}

kataConfigList := &kataconfigurationv1.KataConfigList{}
const missingMcpStatusConditionStr = "<missing>"

err := r.Client.List(context.TODO(), kataConfigList)
if err != nil {
return []reconcile.Request{}
func logMcpChange(log logr.Logger, statusOld, statusNew mcfgv1.MachineConfigPoolStatus) {

log.Info("MCP updated")

if statusOld.MachineCount != statusNew.MachineCount {
log.Info("MachineCount changed", "old", statusOld.MachineCount, "new", statusNew.MachineCount)
} else {
log.Info("MachineCount", "#", statusNew.MachineCount)
}
if statusOld.ReadyMachineCount != statusNew.ReadyMachineCount {
log.Info("ReadyMachineCount changed", "old", statusOld.ReadyMachineCount, "new", statusNew.ReadyMachineCount)
} else {
log.Info("ReadyMachineCount", "#", statusNew.ReadyMachineCount)
}
if statusOld.UpdatedMachineCount != statusNew.UpdatedMachineCount {
log.Info("UpdatedMachineCount changed", "old", statusOld.UpdatedMachineCount, "new", statusNew.UpdatedMachineCount)
} else {
log.Info("UpdatedMachineCount", "#", statusNew.UpdatedMachineCount)
}
if statusOld.DegradedMachineCount != statusNew.DegradedMachineCount {
log.Info("DegradedMachineCount changed", "old", statusOld.DegradedMachineCount, "new", statusNew.DegradedMachineCount)
} else {
log.Info("DegradedMachineCount", "#", statusNew.DegradedMachineCount)
}
if statusOld.ObservedGeneration != statusNew.ObservedGeneration {
log.Info("ObservedGeneration changed", "old", statusOld.ObservedGeneration, "new", statusNew.ObservedGeneration)
}

reconcileRequests := make([]reconcile.Request, len(kataConfigList.Items))
for _, kataconfig := range kataConfigList.Items {
reconcileRequests = append(reconcileRequests, reconcile.Request{
NamespacedName: types.NamespacedName{
Name: kataconfig.Name,
},
})
if !reflect.DeepEqual(statusOld.Conditions, statusNew.Conditions) {

for _, condType := range []mcfgv1.MachineConfigPoolConditionType{"Updating", "Updated"} {
condOld := mcfgv1.GetMachineConfigPoolCondition(statusOld, condType)
condNew := mcfgv1.GetMachineConfigPoolCondition(statusNew, condType)
condStatusOld := missingMcpStatusConditionStr
if condOld != nil {
condStatusOld = string(condOld.Status)
}
condStatusNew := missingMcpStatusConditionStr
if condNew != nil {
condStatusNew = string(condNew.Status)
}

if condStatusOld != condStatusNew {
log.Info("mcp.status.conditions[] changed", "type", condType, "old", condStatusOld, "new", condStatusNew)
}
}
}
return reconcileRequests
}

type McpEventHandler struct {
reconciler *KataConfigOpenShiftReconciler
}

func (eh *McpEventHandler) Create(event event.CreateEvent, queue workqueue.RateLimitingInterface) {
mcp := event.Object

if !isMcpRelevant(mcp) {
return
}

// Don't reconcile on MCP creation since we're unlikely to witness "worker"
// creation and "kata-oc" should be only created by this controller.
// Log the event anyway.
log := eh.reconciler.Log.WithName("McpCreate").WithValues("MCP name", mcp.GetName())
log.Info("MCP created")
}

func (eh *McpEventHandler) Update(event event.UpdateEvent, queue workqueue.RateLimitingInterface) {
mcpOld := event.ObjectOld
mcpNew := event.ObjectNew

if !isMcpRelevant(mcpNew) {
return
}

statusOld := mcpOld.(*mcfgv1.MachineConfigPool).Status
statusNew := mcpNew.(*mcfgv1.MachineConfigPool).Status
if reflect.DeepEqual(statusOld, statusNew) {
return
}

foundRelevantChange := false

if statusOld.MachineCount != statusNew.MachineCount {
foundRelevantChange = true
} else if statusOld.ReadyMachineCount != statusNew.ReadyMachineCount {
foundRelevantChange = true
} else if statusOld.UpdatedMachineCount != statusNew.UpdatedMachineCount {
foundRelevantChange = true
} else if statusOld.DegradedMachineCount != statusNew.DegradedMachineCount {
foundRelevantChange = true
}

if !reflect.DeepEqual(statusOld.Conditions, statusNew.Conditions) {

for _, condType := range []mcfgv1.MachineConfigPoolConditionType{"Updating", "Updated"} {
condOld := mcfgv1.GetMachineConfigPoolCondition(statusOld, condType)
condNew := mcfgv1.GetMachineConfigPoolCondition(statusNew, condType)
condStatusOld := missingMcpStatusConditionStr
if condOld != nil {
condStatusOld = string(condOld.Status)
}
condStatusNew := missingMcpStatusConditionStr
if condNew != nil {
condStatusNew = string(condNew.Status)
}

if condStatusOld != condStatusNew {
foundRelevantChange = true
}
}
}

if eh.reconciler.kataConfig != nil && foundRelevantChange {

log := eh.reconciler.Log.WithName("McpUpdate").WithValues("MCP name", mcpOld.GetName())
logMcpChange(log, statusOld, statusNew)

queue.Add(eh.reconciler.makeReconcileRequest())
}
}

func (eh *McpEventHandler) Delete(event event.DeleteEvent, queue workqueue.RateLimitingInterface) {
mcp := event.Object

if !isMcpRelevant(mcp) {
return
}

// Don't reconcile on MCP deletion since "worker" should never be deleted and
// "kata-oc" should be only deleted by this controller. Log the event anyway.
log := eh.reconciler.Log.WithName("McpDelete").WithValues("MCP name", mcp.GetName())
log.Info("MCP deleted")
}

func (eh *McpEventHandler) Generic(event event.GenericEvent, queue workqueue.RateLimitingInterface) {
mcp := event.Object

if !isMcpRelevant(mcp) {
return
}

// Don't reconcile on MCP generic event since it's not quite clear ATM
// what it even means (we might revisit this later). Log the event anyway.
log := eh.reconciler.Log.WithName("McpGenericEvt").WithValues("MCP name", mcp.GetName())
log.Info("MCP generic event")
}


func (r *KataConfigOpenShiftReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&kataconfigurationv1.KataConfig{}).
Watches(
&source.Kind{Type: &mcfgv1.MachineConfigPool{}},
handler.EnqueueRequestsFromMapFunc(r.mapKataConfigToRequests)).
&McpEventHandler{r}).
Complete(r)
}

Expand Down

0 comments on commit ab11dce

Please sign in to comment.