k8s: Add OnAddNode handlers for CiliumNodeUpdater and EndpointManager

The k8s Node watcher receives events related to the local node from the
Resource[T] framework Events() method.

The Resource[T] framework starts a single informer for each type,
alongside a workqueue for each new subscriber that calls the Events()
method. The workqueue is used to accumulate the keys related to the
objects that are added and/or updated.

When Store() is called before Events(), the informer is started, but no
queue exists to receive the events yet. Only when the Events() is
called, the framework iterates over all the keys in the store to send
the related updates to the new subscriber. Doing that, events like an
ADD followed by one or more UPDATE might be coalesced in a single ADD
event carrying the last available version of the object.

What this means for the consumers of the NodeChain subscription
mechanism, is that discarding the ADD events with an empty handler might
lead to missing update events too.  In order to avoid this, all the
NodeChain subscribers like the CiliumNodeUpdater and the EndpointManager
should have a ADD handler that mimics what their UPDATE handler is
already doing.

Fixes: #26082

Signed-off-by: Fabio Falzoi <fabio.falzoi@isovalent.com>

pkg/endpointmanager/host.go

@@ -4,9 +4,12 @@
 package endpointmanager
 
 import (
+	"context"
+
 	"github.com/cilium/cilium/pkg/endpoint"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	"github.com/cilium/cilium/pkg/labels"
+	"github.com/cilium/cilium/pkg/labelsfilter"
 	"github.com/cilium/cilium/pkg/lock"
 	"github.com/cilium/cilium/pkg/node"
 )
@@ -29,11 +32,23 @@ func (mgr *endpointManager) HostEndpointExists() bool {
 }
 
 // OnAddNode implements the endpointManager's logic for reacting to new nodes
-// from K8s. It is currently not implemented as the endpointManager has not
-// need for it. This adheres to the subscriber.NodeHandler interface.
-func (mgr *endpointManager) OnAddNode(node *slim_corev1.Node,
+// from K8s.
+// This adheres to the subscriber.NodeHandler interface.
+func (mgr *endpointManager) OnAddNode(newNode *slim_corev1.Node,
 	swg *lock.StoppableWaitGroup) error {
 
+	node.SetLabels(newNode.GetLabels())
+
+	nodeEP := mgr.GetHostEndpoint()
+	if nodeEP == nil {
+		// if host endpoint does not exist yet, labels will be set when it'll be created.
+		return nil
+	}
+
+	newLabels := labels.Map2Labels(newNode.GetLabels(), labels.LabelSourceK8s)
+	newIdtyLabels, _ := labelsfilter.Filter(newLabels)
+	nodeEP.UpdateLabels(context.TODO(), newIdtyLabels, nil, false)
+
 	return nil
 }
 

pkg/k8s/watchers/node.go

@@ -127,11 +127,8 @@ func NewCiliumNodeUpdater(kvStoreNodeUpdater NodeUpdate) *ciliumNodeUpdater {
 }
 
 func (u *ciliumNodeUpdater) OnAddNode(newNode *slim_corev1.Node, swg *lock.StoppableWaitGroup) error {
-	// We don't need to run OnAddNode because Cilium will fetch the state from
-	// k8s upon initialization and will populate the KVStore [1] node with this
-	// information or create a Cilium Node CR [2].
-	// [1] https://github.com/cilium/cilium/blob/2bea69a54a00f10bec093347900cc66395269154/daemon/cmd/daemon.go#L1102
-	// [2] https://github.com/cilium/cilium/blob/2bea69a54a00f10bec093347900cc66395269154/daemon/cmd/daemon.go#L864-L868
+	u.updateCiliumNode(newNode)
+
 	return nil
 }