lb; update neighbor cache on TAPA connection close

Monitor TAPA->Proxy NSM connections and update neighbor
cache upon connection close events by removing neighbor
entries associated with TAPA side IP addresses.

Currently, the connection monitor uses a wildcard filter,
matching all connections.

New stateless-lb container env variables:
- NAMESPACE
- TARGET_DISCONNECT_MONITORING (enabling the feature by default)

Author: zolug

cmd/stateless-lb/config.go

@@ -27,25 +27,27 @@ import (
 
 // Config for the proxy
 type Config struct {
-	Name                  string        `default:"load-balancer" desc:"Name of the pod"`
-	ServiceName           string        `default:"load-balancer" desc:"Name of providing service" split_words:"true"`
-	ConnectTo             url.URL       `default:"unix:///var/lib/networkservicemesh/nsm.io.sock" desc:"url to connect to NSM" split_words:"true"`
-	DialTimeout           time.Duration `default:"5s" desc:"timeout to dial NSMgr" split_words:"true"`
-	RequestTimeout        time.Duration `default:"15s" desc:"timeout to request NSE" split_words:"true"`
-	MaxTokenLifetime      time.Duration `default:"24h" desc:"maximum lifetime of tokens" split_words:"true"`
-	NSPService            string        `default:"nsp-service:7778" desc:"IP (or domain) and port of the NSP Service" split_words:"true"`
-	ConduitName           string        `default:"load-balancer" desc:"Name of the conduit" split_words:"true"`
-	TrenchName            string        `default:"default" desc:"Trench the pod is running on" split_words:"true"`
-	LogLevel              string        `default:"DEBUG" desc:"Log level" split_words:"true"`
-	Nfqueue               string        `default:"0:3" desc:"netfilter queue(s) to be used by nfqlb" split_words:"true"`
-	NfqueueFanout         bool          `default:"false" desc:"enable fanout nfqueue option" split_words:"true"`
-	IdentifierOffsetStart int           `default:"5000" desc:"Each Stream will get a unique identifier range starting from that value" split_words:"true"`
-	GRPCKeepaliveTime     time.Duration `default:"30s" desc:"gRPC keepalive timeout" envconfig:"grpc_keepalive_time"`
-	GRPCProbeRPCTimeout   time.Duration `default:"1s" desc:"RPC timeout of internal gRPC health probe" envconfig:"grpc_probe_rpc_timeout"`
-	GRPCMaxBackoff        time.Duration `default:"5s" desc:"Upper bound on gRPC connection backoff delay" envconfig:"grpc_max_backoff"`
-	MetricsEnabled        bool          `default:"false" desc:"Enable the metrics collection" split_words:"true"`
-	MetricsPort           int           `default:"2223" desc:"Specify the port used to expose the metrics" split_words:"true"`
-	Socket                url.URL       `default:"unix:///var/lib/meridio/lb.sock" desc:"Server socket to host Stream Availability Service" split_words:"true"`
+	Name                       string        `default:"load-balancer" desc:"Name of the pod"`
+	ServiceName                string        `default:"load-balancer" desc:"Name of providing service" split_words:"true"`
+	ConnectTo                  url.URL       `default:"unix:///var/lib/networkservicemesh/nsm.io.sock" desc:"url to connect to NSM" split_words:"true"`
+	DialTimeout                time.Duration `default:"5s" desc:"timeout to dial NSMgr" split_words:"true"`
+	RequestTimeout             time.Duration `default:"15s" desc:"timeout to request NSE" split_words:"true"`
+	MaxTokenLifetime           time.Duration `default:"24h" desc:"maximum lifetime of tokens" split_words:"true"`
+	NSPService                 string        `default:"nsp-service:7778" desc:"IP (or domain) and port of the NSP Service" split_words:"true"`
+	ConduitName                string        `default:"load-balancer" desc:"Name of the conduit" split_words:"true"`
+	TrenchName                 string        `default:"default" desc:"Trench the pod is running on" split_words:"true"`
+	LogLevel                   string        `default:"DEBUG" desc:"Log level" split_words:"true"`
+	Nfqueue                    string        `default:"0:3" desc:"netfilter queue(s) to be used by nfqlb" split_words:"true"`
+	NfqueueFanout              bool          `default:"false" desc:"enable fanout nfqueue option" split_words:"true"`
+	IdentifierOffsetStart      int           `default:"5000" desc:"Each Stream will get a unique identifier range starting from that value" split_words:"true"`
+	GRPCKeepaliveTime          time.Duration `default:"30s" desc:"gRPC keepalive timeout" envconfig:"grpc_keepalive_time"`
+	GRPCProbeRPCTimeout        time.Duration `default:"1s" desc:"RPC timeout of internal gRPC health probe" envconfig:"grpc_probe_rpc_timeout"`
+	GRPCMaxBackoff             time.Duration `default:"5s" desc:"Upper bound on gRPC connection backoff delay" envconfig:"grpc_max_backoff"`
+	MetricsEnabled             bool          `default:"false" desc:"Enable the metrics collection" split_words:"true"`
+	MetricsPort                int           `default:"2223" desc:"Specify the port used to expose the metrics" split_words:"true"`
+	Socket                     url.URL       `default:"unix:///var/lib/meridio/lb.sock" desc:"Server socket to host Stream Availability Service" split_words:"true"`
+	Namespace                  string        `default:"default" desc:"Namespace the pod is running on" split_words:"true"`
+	TargetDisconnectMonitoring bool          `default:"true" desc:"Enable listening to Target disconnect events to clean-up linux neighbor cache" split_words:"true"`
 }
 
 // IsValid checks if the configuration is valid

cmd/stateless-lb/internal/neighborcache/neighborcache.go

@@ -0,0 +1,90 @@
+/*
+Copyright (c) 2024 OpenInfra Foundation Europe
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package neighborcache
+
+import (
+	"context"
+	"net"
+
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"github.com/nordix/meridio/pkg/log"
+	"github.com/vishvananda/netlink"
+)
+
+// RemoveInvalid attempts to remove potentially invalid neighbor entries for
+// which NSM has reported that the connection was closed, implying that the
+// interface has either disappeared or is about to disappear along with its
+// IP and MAC addresses. Thus, even if the same IP address reappears shortly
+// due to NSM heal successfully fixing or, more accurately, re-establishing the
+// connection, communication disturbances caused by an old invalid neighbor
+// cache entry can be avoided, which would have otherwise occurred due to the
+// behavior of the neighbor state machine (DELAY state and unicast probes).
+// Note: The LB monitors TAPA -> Proxy connections, where the SrcIpAddrs refer
+// to TAPA-side IPs, including the ones used as Target IPs by the LB.
+func RemoveInvalid(ctx context.Context, connectionEvent *networkservice.ConnectionEvent) {
+	if connectionEvent.Type != networkservice.ConnectionEventType_DELETE {
+		return
+	}
+	logger := log.FromContextOrGlobal(ctx).WithValues("func", "RemoveInvalid")
+	// Fetch neighbor cache from kernel
+	neighborList, err := netlink.NeighList(0, 0)
+	if err != nil {
+		logger.Info("Could not fetch neighbor list", "err", err)
+		return
+	}
+	// Convert neighbor list to a map
+	neighborMap := make(map[string][]netlink.Neigh)
+	for _, neigh := range neighborList {
+		ipStr := neigh.IP.String()
+		neighborMap[ipStr] = append(neighborMap[ipStr], neigh)
+	}
+
+	// Remove any of the NSM SrcIpAddrs from the neighbor cache if they are present
+	eventPrinted := false
+	for _, connection := range connectionEvent.Connections {
+		if connection.GetPath() == nil || len(connection.GetPath().GetPathSegments()) < 1 {
+			continue
+		}
+		if connection.GetContext() == nil || connection.GetContext().GetIpContext() == nil {
+			continue
+		}
+		ipContext := connection.GetContext().GetIpContext()
+		for _, ipStr := range ipContext.SrcIpAddrs {
+			if ip, _, err := net.ParseCIDR(ipStr); err == nil {
+				// Check if neighbor map has an entry for this IP
+				neighs, ok := neighborMap[ip.String()]
+				if !ok {
+					continue
+				}
+				if !eventPrinted {
+					eventPrinted = true
+					logger.Info("Connection event", "event", connectionEvent)
+				}
+				for _, neigh := range neighs {
+					logger.Info("Delete from neighbor cache", "neigh", neigh, "MAC", neigh.HardwareAddr.String())
+					err := netlink.NeighDel(&netlink.Neigh{
+						LinkIndex: neigh.LinkIndex,
+						IP:        ip,
+					})
+					if err != nil {
+						logger.Info("Failed to delete from neighbor cache", "neigh", neigh, "MAC", neigh.HardwareAddr.String(), "err", err)
+					}
+				}
+			}
+		}
+	}
+}

cmd/stateless-lb/main.go

@@ -24,8 +24,10 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"net/url"
 	"os"
 	"os/signal"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -35,6 +37,7 @@ import (
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	kernelmech "github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/kernel"
 	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/noop"
+	"github.com/networkservicemesh/api/pkg/api/registry"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/mechanisms"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/mechanisms/kernel"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/mechanisms/sendfd"
@@ -43,6 +46,7 @@ import (
 	nsmlog "github.com/networkservicemesh/sdk/pkg/tools/log"
 	lbAPI "github.com/nordix/meridio/api/loadbalancer/v1"
 	nspAPI "github.com/nordix/meridio/api/nsp/v1"
+	"github.com/nordix/meridio/cmd/stateless-lb/internal/neighborcache"
 	"github.com/nordix/meridio/pkg/debug"
 	"github.com/nordix/meridio/pkg/endpoint"
 	"github.com/nordix/meridio/pkg/health"
@@ -139,13 +143,13 @@ func main() {
 
 	netUtils := &linuxKernel.KernelUtils{}
 
-	// create and start health server
+	// Create and start health server
 	ctx = health.CreateChecker(ctx)
 	if err := health.RegisterReadinessSubservices(ctx, health.LBReadinessServices...); err != nil {
 		logger.Error(err, "RegisterReadinessSubservices")
 	}
-	// note: NSM endpoint service is hosted from early on by its server, thus it can be probed
-	// irrespective of its registration status at NSM
+	// Note: The NSM endpoint service is hosted from the start by its server,
+	// so it can be probed regardless of its registration status at NSM.
 	if err := health.RegisterLivenessSubservices(ctx, health.LBLivenessServices...); err != nil {
 		logger.Error(err, "RegisterLivenessSubservices")
 	}
@@ -175,12 +179,12 @@ func main() {
 	}
 	defer conn.Close()
 
-	// monitor status of NSP connection and adjust probe status accordingly
+	// Monitor status of NSP connection and adjust probe status accordingly
 	if err := connection.Monitor(ctx, health.NSPCliSvc, conn); err != nil {
 		logger.Error(err, "NSP connection state monitor")
 	}
 
-	stream.SetInterfaceNamePrefix(config.ServiceName) // deduce the NSM interfacename prefix for the netfilter defrag rules
+	stream.SetInterfaceNamePrefix(config.ServiceName) // Determine the NSM interface name prefix for the netfilter defragmentation rules
 	targetRegistryClient := nspAPI.NewTargetRegistryClient(conn)
 	configurationManagerClient := nspAPI.NewConfigurationManagerClient(conn)
 	conduit := &nspAPI.Conduit{
@@ -225,7 +229,7 @@ func main() {
 		return
 	}
 
-	// start server to host Stream Forwarding Availability service
+	// Start server to host Stream Forwarding Availability service
 	lis, err := createStreamAvailabilityListener(config)
 	if err != nil {
 		logger.Error(err, "createStreamAvailabilityListener")
@@ -239,7 +243,7 @@ func main() {
 		}),
 	)
 	defer func() {
-		// attempt graceful shutdown to allow sending out pending msgs
+		// Attempt graceful shutdown to allow sending out pending msgs
 		stopped := make(chan struct{})
 		go func() {
 			s.GracefulStop()
@@ -248,7 +252,7 @@ func main() {
 		waitTimer := time.NewTimer(time.Second)
 		select {
 		case <-waitTimer.C:
-			s.Stop() // graceful shutdown not finished in time, force stop immediately
+			s.Stop() // Graceful shutdown not finished in time, force stop immediately
 		case <-stopped:
 			waitTimer.Stop()
 			select {
@@ -257,7 +261,8 @@ func main() {
 			}
 		}
 	}()
-	// announces forwarding availability of streams (i.e. if the LB can forward traffic towards application targets)
+	// Announces the forwarding availability of streams
+	// (i.e., whether the LB can forward traffic towards application targets)
 	streamFwdAvailabilityService := stream.NewForwardingAvailabilityService(
 		context.Background(),
 		&lbAPI.Target{
@@ -280,8 +285,8 @@ func main() {
 		configurationManagerClient,
 		conduit,
 		netUtils,
-		lbFactory, // to spawn nfqlb instance for each Stream created
-		nfa,       // netfilter kernel configuration to steer VIP traffic to nfqlb process
+		lbFactory, // To spawn nfqlb instance for each Stream created
+		nfa,       // Netfilter kernel configuration to steer VIP traffic to nfqlb process
 		config.IdentifierOffsetStart,
 		targetHitsMetrics,
 		neighborMonitor,
@@ -290,7 +295,7 @@ func main() {
 
 	interfaceMonitorEndpoint := interfacemonitor.NewServer(interfaceMonitor, sns, netUtils)
 
-	// Note: naming the interface is left to NSM (refer to getNameFromConnection())
+	// Note: Naming the interface is left to NSM (refer to getNameFromConnection())
 	// However NSM does not seem to ensure uniqueness either. Might need to revisit...
 	responderEndpoint := []networkservice.NetworkServiceServer{
 		mechanisms.NewServer(map[string]networkservice.NetworkServiceServer{
@@ -310,10 +315,10 @@ func main() {
 		MaxTokenLifetime: config.MaxTokenLifetime,
 		GRPCMaxBackoff:   config.GRPCMaxBackoff,
 	}
-	nsmAPIClient := nsm.NewAPIClient(context.Background(), apiClientConfig) // background context to allow endpoint unregistration on tear down
+	nsmAPIClient := nsm.NewAPIClient(context.Background(), apiClientConfig) // Background context to allow endpoint unregistration on tear down
 	defer nsmAPIClient.Delete()
 
-	// connect NSMgr and start NSM connection monitoring (to log events of interest)
+	// Connect local NSMgr
 	cc, err := grpc.DialContext(ctx,
 		grpcutils.URLToTarget(&nsmAPIClient.Config.ConnectTo),
 		nsmAPIClient.GRPCDialOption...,
@@ -324,8 +329,13 @@ func main() {
 		return
 	}
 	defer cc.Close()
+	// Start monitoring NSM connections the LB is part of
 	monitorClient := networkservice.NewMonitorConnectionClient(cc)
 	go nsmmonitor.ConnectionMonitor(ctx, config.Name, monitorClient)
+	// Start cluster-wide monitoring of NSM connection Delete events between TAPA and Proxy
+	if config.TargetDisconnectMonitoring {
+		go startClusterConnectionMonitor(ctx, config, cc, nsmAPIClient.GRPCDialOption)
+	}
 
 	endpointConfig := &endpoint.Config{
 		Name:             config.Name,
@@ -334,7 +344,7 @@ func main() {
 		MaxTokenLifetime: config.MaxTokenLifetime,
 	}
 	ep, err := endpoint.NewEndpoint(
-		context.Background(), // use background context to allow endpoint unregistration on tear down
+		context.Background(), // Use background context to allow endpoint unregistration on tear down
 		endpointConfig,
 		nsmAPIClient.NetworkServiceRegistryClient,
 		nsmAPIClient.NetworkServiceEndpointRegistryClient,
@@ -346,7 +356,7 @@ func main() {
 	defer func() {
 		ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second*3))
 		defer cancel()
-		ep.Delete(ctx) // let endpoint unregister with NSM to inform proxies in time
+		ep.Delete(ctx) // Let endpoint unregister with NSM to inform proxies in time
 		logger.Info("LB endpoint deleted")
 	}()
 
@@ -367,7 +377,7 @@ func main() {
 		cancel()
 	}()
 	sns.Start()
-	// monitor availibilty of frontends (advertise NSE to proxies only if there's feasible FE)
+	// Monitor availibilty of frontends (advertise NSE to proxies only if there's feasible FE)
 	fns := NewFrontendNetworkService(ctx, targetRegistryClient, ep, NewServiceControlDispatcher(sns))
 	go func() {
 		logger.Info("Start frontend monitoring service")
@@ -938,3 +948,31 @@ func (sns *SimpleNetworkService) updateVips(vips []*nspAPI.Vip) error {
 	}
 	return nil
 }
+
+// startClusterConnectionMonitor starts cluster-wide monitoring of NSM connection
+// Delete events between TAPA and Proxy, removing invalid Target entries from
+// the Linux neighbor cache to prevent connection disturbances.
+func startClusterConnectionMonitor(
+	ctx context.Context,
+	config Config,
+	cc grpc.ClientConnInterface,
+	dialOptions []grpc.DialOption,
+) {
+	nsmmonitor.ClusterConnectionMonitor(
+		ctx,
+		// Create a registry client to learn NSMgr URLs
+		registry.NewNetworkServiceEndpointRegistryClient(cc),
+		// Use provided gRPC dial options to connect NSMgrs
+		dialOptions,
+		&networkservice.MonitorScopeSelector{},
+		// Replace "local" URLs with the local NSMgr's URL to avoid connecting the local forwarder
+		func(connectTo *url.URL) *url.URL {
+			if strings.HasPrefix(connectTo.String(), "inode://") {
+				return &config.ConnectTo
+			}
+			return connectTo
+		},
+		// Callback function to remove invalid entries from the neighbor cache
+		neighborcache.RemoveInvalid,
+	)
+}

config/templates/charts/meridio/deployment/stateless-lb-frontend.yaml

@@ -115,6 +115,8 @@ spec:
               value:  # to be filled by operator
             - name: NSM_METRICS_ENABLED
               value: "true"
+            - name: NSM_NAMESPACE
+              value:  # to be filled by operator
           volumeMounts:
             - name: spire-agent-socket
               mountPath: /run/spire/sockets

deployments/helm/templates/load-balancer.yaml

@@ -57,6 +57,10 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.name
+            - name: NSM_NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
             - name: NSM_SERVICE_NAME
               value: {{ template "meridio.loadBalancer.networkServiceName" . }}
             - name: NSM_CONDUIT_NAME

docs/components/stateless-lb.md

@@ -38,6 +38,8 @@ NSM_GRPC_MAX_BACKOFF | time.Duration | Upper bound on gRPC connection backoff de
 NSM_METRICS_ENABLED | bool | Enable the metrics collection| false
 NSM_METRICS_PORT | int | Specify the port used to expose the metrics | 2223
 NSM_SOCKET | url.URL | Server socket to host Stream Availability Service | unix:///var/lib/meridio/lb.sock
+NSM_NAMESPACE | string | Namespace the pod is running on | default
+NSM_TARGET_DISCONNECT_MONITORING | bool | Enable listenting to Target disconnect events to clean-up linux neighbor cache | true
 
 ## Command Line 
 
@@ -91,6 +93,6 @@ Sysctl: net.ipv4.conf.all.rp_filter=0 | Allow packets to have a source IPv4 addr
 Sysctl: net.ipv4.conf.default.rp_filter=0 | Allow packets to have a source IPv6 address which does not correspond to any routing destination address.
 Sysctl: net.ipv4.fwmark_reflect=1 | Allow LB generated outbound ICMP Frag Needed reply to use VIP as source address.
 Sysctl: net.ipv6.fwmark_reflect=1 | Allow LB generated outbound ICMPv6 Packet Too Big reply to use VIP as source address.
-NET_ADMIN | The load balancer configures IP rules and IP routes to steer packets (processed by [nfqueue-loadbalancer program](https://github.com/Nordix/nfqueue-loadbalancer)) to targets. The user space load balancer program relies on [libnetfilter_queue](https://netfilter.org/projects/libnetfilter_queue).
+NET_ADMIN | The load balancer configures IP rules and IP routes to steer packets (processed by [nfqueue-loadbalancer program](https://github.com/Nordix/nfqueue-loadbalancer)) to targets. The user space load balancer program relies on [libnetfilter_queue](https://netfilter.org/projects/libnetfilter_queue). The load balancer can remove entries from its Linux neighbor cache that correspond to targets for which the MAC address is no longer valid.
 IPC_LOCK | The user space load balancer program uses shared memory.
 IPC_OWNER | The user space load balancer program uses shared memory.

go.mod

@@ -3,6 +3,7 @@ module github.com/nordix/meridio
 go 1.22
 
 require (
+	github.com/edwarnicke/genericsync v0.0.0-20220910010113-61a344f9bc29
 	github.com/edwarnicke/grpcfd v1.1.4
 	github.com/faisal-memon/sviddisk v0.0.0-20211007205134-77ccea0b9271
 	github.com/go-logr/logr v1.4.1
@@ -55,7 +56,6 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/edwarnicke/exechelper v1.0.2 // indirect
-	github.com/edwarnicke/genericsync v0.0.0-20220910010113-61a344f9bc29 // indirect
 	github.com/edwarnicke/serialize v1.0.7 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.8.0 // indirect