Skip to content

fix: node watcher should only reconcile on important events #1522

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions pkg/controllers/daemon/nodereconciler/cell_linux.go
Original file line number Diff line number Diff line change
@@ -34,7 +34,6 @@ type params struct {
cell.In

Config config.RetinaHubbleConfig
Logger logrus.FieldLogger
Client client.Client
IPCache *ipcache.IPCache
}
@@ -51,7 +50,7 @@ func newNodeController(params params) (*NodeReconciler, error) {
n := &NodeReconciler{
Client: params.Client,
clusterName: params.Config.ClusterName,
l: params.Logger.WithField("component", "node-controller"),
l: log.Logger().Named("node-controller"),
nodes: make(map[string]types.Node),
handlers: make(map[string]datapath.NodeHandler),
c: params.IPCache,
72 changes: 54 additions & 18 deletions pkg/controllers/daemon/nodereconciler/node_controller_linux.go
Original file line number Diff line number Diff line change
@@ -11,12 +11,14 @@ import (
"sync"

"github.com/microsoft/retina/pkg/common/apiretry"
"github.com/sirupsen/logrus"
"github.com/microsoft/retina/pkg/log"
"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
errors "k8s.io/apimachinery/pkg/api/errors"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/predicate"

cmtypes "github.com/cilium/cilium/pkg/clustermesh/types"
datapath "github.com/cilium/cilium/pkg/datapath/types"
@@ -35,28 +37,14 @@ type NodeReconciler struct {

clusterName string

l logrus.FieldLogger
l *log.ZapLogger
handlers map[string]datapath.NodeHandler
nodes map[string]types.Node
c *ipc.IPCache
localNodeIP string
m sync.RWMutex
}

// isNodeUpdated checks if the node has been updated.
// This is a simple check for labels and annotations
// being updated. Those are the only fields that are mutable.
// AKS specific for now.
func isNodeUpdated(n1, n2 types.Node) bool {
if !reflect.DeepEqual(n1.Labels, n2.Labels) {
return true
}
if !reflect.DeepEqual(n1.Annotations, n2.Annotations) {
return true
}
return false
}

func (r *NodeReconciler) addNode(node *corev1.Node) {
r.m.Lock()
defer r.m.Unlock()
@@ -89,7 +77,7 @@ func (r *NodeReconciler) addNode(node *corev1.Node) {
nd.Cluster = r.clusterName

// Check if the node already exists.
if curNode, ok := r.nodes[node.Name]; ok && !isNodeUpdated(curNode, nd) {
if _, ok := r.nodes[node.Name]; ok {
r.l.Debug("Node already exists", zap.String("Node", node.Name))
}

@@ -117,7 +105,7 @@ func (r *NodeReconciler) addNode(node *corev1.Node) {
r.l.Debug("Added IP to ipcache", zap.String("IP", address.ToString()))
}

r.l.Info("Added Node", zap.String("Node", node.Name))
r.l.Info("Added Node", zap.String("name", nd.Name))
}

func (r *NodeReconciler) deleteNode(node *corev1.Node) {
@@ -204,8 +192,45 @@ func (r *NodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
// SetupWithManager sets up the controller with the Manager.
func (r *NodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
r.l.Debug("Setting up Node controller")

// Create a predicate to filter node events
nodePredicate := predicate.Funcs{
CreateFunc: func(event.CreateEvent) bool {
// Always reconcile on node creation
return true
},
DeleteFunc: func(event.DeleteEvent) bool {
// Always reconcile on node deletion
return true
},
UpdateFunc: func(e event.UpdateEvent) bool {
oldNode, ok := e.ObjectOld.(*corev1.Node)
if !ok {
r.l.Error("Failed to convert old object to Node")
return false
}

newNode, ok := e.ObjectNew.(*corev1.Node)
if !ok {
r.l.Error("Failed to convert new object to Node")
return false
}

// Compare node IP addresses
oldIPs := extractNodeIPs(oldNode)
newIPs := extractNodeIPs(newNode)

// Only reconcile if IPs changed or labels/annotations changed
return !reflect.DeepEqual(oldIPs, newIPs) || !reflect.DeepEqual(oldNode.Labels, newNode.Labels) || !reflect.DeepEqual(oldNode.Annotations, newNode.Annotations)
},
GenericFunc: func(event.GenericEvent) bool {
return false
},
}

err := ctrl.NewControllerManagedBy(mgr).
For(&corev1.Node{}).
WithEventFilter(nodePredicate).
Complete(r)
if err != nil {
return fmt.Errorf("setting up node controller: %w", err)
@@ -244,3 +269,14 @@ func (r *NodeReconciler) StartNeighborRefresh(datapath.NodeNeighbors) {}
func (r *NodeReconciler) StartNodeNeighborLinkUpdater(datapath.NodeNeighbors) {}

func (r *NodeReconciler) SetPrefixClusterMutatorFn(func(*types.Node) []cmtypes.PrefixClusterOpts) {}

// extractNodeIPs extracts IP addresses from a node
func extractNodeIPs(node *corev1.Node) map[string]string {
ips := make(map[string]string)
for _, addr := range node.Status.Addresses {
if addr.Type == corev1.NodeInternalIP || addr.Type == corev1.NodeExternalIP {
ips[string(addr.Type)] = addr.Address
}
}
return ips
}
Loading