feat(inventory): Add configurable discovery rate limiting

This commit introduces configurable rate limiting for the inventory discovery
process. Previously, a fixed 5-second rate limit with a burst of 1 could delay
processing of netlink updates, leading to failures during high pod churn
scenarios.

Command-line flags have been added to control the inventory discovery rate limit
and burst size. The default values have been adjusted to be more responsive to
rapid pod lifecycle events, ensuring that device state is updated promptly.

Author: gauravkghildiyal

cmd/dranet/app.go

@@ -27,12 +27,15 @@ import (
 	"runtime/debug"
 	"sync/atomic"
 	"syscall"
+	"time"
 
 	"github.com/google/cel-go/cel"
 	"github.com/google/cel-go/ext"
 	"github.com/google/dranet/pkg/driver"
+	"github.com/google/dranet/pkg/inventory"
 	"github.com/google/dranet/pkg/pcidb"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"golang.org/x/time/rate"
 
 	resourcev1 "k8s.io/api/resource/v1"
 	"k8s.io/client-go/kubernetes"
@@ -51,6 +54,9 @@ var (
 	kubeconfig       string
 	bindAddress      string
 	celExpression    string
+	minPollInterval  time.Duration
+	maxPollInterval  time.Duration
+	pollBurst        int
 
 	ready atomic.Bool
 )
@@ -60,6 +66,9 @@ func init() {
 	flag.StringVar(&bindAddress, "bind-address", ":9177", "The IP address and port for the metrics and healthz server to serve on")
 	flag.StringVar(&hostnameOverride, "hostname-override", "", "If non-empty, will be used as the name of the Node that kube-network-policies is running on. If unset, the node name is assumed to be the same as the node's hostname.")
 	flag.StringVar(&celExpression, "filter", `!("dra.net/type" in attributes) || attributes["dra.net/type"].StringValue  != "veth"`, "CEL expression to filter network interface attributes (v1.DeviceAttribute).")
+	flag.DurationVar(&minPollInterval, "inventory-min-poll-interval", 5*time.Second, "The minimum interval between two consecutive polls of the inventory.")
+	flag.DurationVar(&maxPollInterval, "inventory-max-poll-interval", 1*time.Minute, "The maximum interval between two consecutive polls of the inventory.")
+	flag.IntVar(&pollBurst, "inventory-poll-burst", 5, "The number of polls that can be run in a burst.")
 
 	flag.Usage = func() {
 		fmt.Fprint(os.Stderr, "Usage: dranet [options]\n\n")
@@ -150,6 +159,11 @@ func main() {
 		}
 		opts = append(opts, driver.WithFilter(prg))
 	}
+	db := inventory.New(
+		inventory.WithRateLimiter(rate.NewLimiter(rate.Every(minPollInterval), pollBurst)),
+		inventory.WithMaxPollInterval(maxPollInterval),
+	)
+	opts = append(opts, driver.WithInventory(db))
 	dranet, err := driver.Start(ctx, driverName, clientset, nodeName, opts...)
 	if err != nil {
 		klog.Fatalf("driver failed to start: %v", err)

pkg/driver/driver.go

@@ -73,6 +73,13 @@ func WithFilter(filter cel.Program) Option {
 	}
 }
 
+// WithInventory sets the inventory database for the driver.
+func WithInventory(db inventoryDB) Option {
+	return func(o *NetworkDriver) {
+		o.netdb = db
+	}
+}
+
 type NetworkDriver struct {
 	driverName string
 	nodeName   string
@@ -174,7 +181,9 @@ func Start(ctx context.Context, driverName string, kubeClient kubernetes.Interfa
 	}()
 
 	// register the host network interfaces
-	plugin.netdb = inventory.New()
+	if plugin.netdb == nil {
+		plugin.netdb = inventory.New()
+	}
 	go func() {
 		for i := 0; i < maxAttempts; i++ {
 			err = plugin.netdb.Run(ctx)

pkg/inventory/db.go

@@ -42,9 +42,15 @@ import (
 )
 
 const (
-	// database poll period
-	minInterval = 5 * time.Second
-	maxInterval = 1 * time.Minute
+	// defaultMinPollInterval is the default minimum interval between two
+	// consecutive polls of the inventory.
+	defaultMinPollInterval = 2 * time.Second
+	// defaultMaxPollInterval is the default maximum interval between two
+	// consecutive polls of the inventory.
+	defaultMaxPollInterval = 1 * time.Minute
+	// defaultPollBurst is the default number of polls that can be run in a
+	// burst.
+	defaultPollBurst = 5
 )
 
 var (
@@ -66,18 +72,38 @@ type DB struct {
 	// The deviceStore is periodically updated by the Run method.
 	deviceStore map[string]resourceapi.Device
 
-	rateLimiter   *rate.Limiter
-	notifications chan []resourceapi.Device
-	hasDevices    bool
+	rateLimiter     *rate.Limiter
+	maxPollInterval time.Duration
+	notifications   chan []resourceapi.Device
+	hasDevices      bool
 }
 
-func New() *DB {
-	return &DB{
-		podNetNsStore: map[string]string{},
-		deviceStore:   map[string]resourceapi.Device{},
-		rateLimiter:   rate.NewLimiter(rate.Every(minInterval), 1),
-		notifications: make(chan []resourceapi.Device),
+type Option func(*DB)
+
+func WithRateLimiter(limiter *rate.Limiter) Option {
+	return func(db *DB) {
+		db.rateLimiter = limiter
+	}
+}
+
+func WithMaxPollInterval(d time.Duration) Option {
+	return func(db *DB) {
+		db.maxPollInterval = d
+	}
+}
+
+func New(opts ...Option) *DB {
+	db := &DB{
+		podNetNsStore:   map[string]string{},
+		deviceStore:     map[string]resourceapi.Device{},
+		rateLimiter:     rate.NewLimiter(rate.Every(defaultMinPollInterval), defaultPollBurst),
+		notifications:   make(chan []resourceapi.Device),
+		maxPollInterval: defaultMaxPollInterval,
+	}
+	for _, o := range opts {
+		o(db)
 	}
+	return db
 }
 
 func (db *DB) AddPodNetNs(pod string, netNsPath string) {
@@ -114,7 +140,7 @@ func (db *DB) Run(ctx context.Context) error {
 	doneCh := make(chan struct{})
 	defer close(doneCh)
 	if err := netlink.LinkSubscribe(nlChannel, doneCh); err != nil {
-		klog.Error(err, "error subscribing to netlink interfaces, only syncing periodically", "interval", maxInterval.String())
+		klog.Error(err, "error subscribing to netlink interfaces, only syncing periodically", "interval", db.maxPollInterval.String())
 	}
 
 	// Obtain data that will not change after the startup
@@ -159,7 +185,7 @@ func (db *DB) Run(ctx context.Context) error {
 			for len(nlChannel) > 0 {
 				<-nlChannel
 			}
-		case <-time.After(maxInterval):
+		case <-time.After(db.maxPollInterval):
 		case <-ctx.Done():
 			return ctx.Err()
 		}

tests/e2e.bats

@@ -110,6 +110,9 @@ setup_tcx_filter() {
   docker exec "$CLUSTER_NAME"-worker bash -c "ip link add mlx5_6 type dummy"
   docker exec "$CLUSTER_NAME"-worker bash -c "ip link set up dev mlx5_6"
 
+  echo "DEBUG: This is debug info before creation of deviceclass and resourceclaim"
+  dump_debug_info_on_failure
+
   kubectl apply -f "$BATS_TEST_DIRNAME"/../tests/manifests/deviceclass.yaml
   kubectl apply -f "$BATS_TEST_DIRNAME"/../tests/manifests/resourceclaim.yaml
   kubectl wait --timeout=30s --for=condition=ready pods -l app=pod