Merge pull request #115 from rbtr/feat/bbolt-persistence

feat: bbolt persistent pod device configs

Author: k8s-ci-robot

README.md

@@ -103,6 +103,15 @@ helm upgrade --install dranet ./deployments/helm/dranet -n kube-system
 
 For available configuration options, see the [chart README](deployments/helm/dranet/README.md).
 
+DRANET persists prepared pod device configuration in a local bbolt database so
+NRI hooks can continue initialization after a driver restart. The default DB
+path is `/var/run/dranet/dranet.db`, so deployment manifests must mount
+`/var/run/dranet` as writable storage (hostPath with `DirectoryOrCreate` in the
+provided manifests).
+
+You can override the DB path with `--db-path`; set it to an empty
+string to disable persistence and use in-memory state.
+
 ### How to Use It
 
 Once DRANET is running, you can inspect the network interfaces and their

cmd/dranet/app.go

@@ -23,6 +23,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"reflect"
 	"runtime/debug"
 	"sync/atomic"
@@ -54,6 +55,7 @@ var (
 	kubeconfig        string
 	bindAddress       string
 	celExpression     string
+	dbPath            string
 	minPollInterval   time.Duration
 	maxPollInterval   time.Duration
 	pollBurst         int
@@ -68,6 +70,7 @@ 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.StringVar(&dbPath, "db-path", filepath.Join("/var/run/dranet", "dranet.db"), "Path to the persistent bbolt database file. Set to an empty string to disable persistence and use in-memory state.")
 	flag.DurationVar(&minPollInterval, "inventory-min-poll-interval", 2*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.")
@@ -143,6 +146,11 @@ func main() {
 	signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)
 
 	opts := []driver.Option{}
+
+	if dbPath != "" {
+		opts = append(opts, driver.WithDBPath(dbPath))
+	}
+
 	if celExpression != "" {
 		env, err := cel.NewEnv(
 			ext.NativeTypes(

examples/dranetctl-install.yaml

@@ -144,6 +144,8 @@ spec:
         - name: netns
           mountPath: /var/run/netns
           mountPropagation: HostToContainer
+        - name: dranet-run
+          mountPath: /var/run/dranet
       volumes:
       - name: device-plugin
         hostPath:
@@ -157,6 +159,10 @@ spec:
       - name: netns
         hostPath:
           path: /var/run/netns
+      - name: dranet-run
+        hostPath:
+          path: /var/run/dranet
+          type: DirectoryOrCreate
       - name: etc
         hostPath:
           path: /etc

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/spf13/pflag v1.0.10
 	github.com/vishvananda/netlink v1.3.1
 	github.com/vishvananda/netns v0.0.5
+	go.etcd.io/bbolt v1.4.3
 	golang.org/x/sys v0.42.0
 	golang.org/x/time v0.15.0
 	google.golang.org/api v0.273.1

go.sum

@@ -186,6 +186,8 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
+go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo=
+go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E=
 go.etcd.io/etcd/client/pkg/v3 v3.6.5 h1:Duz9fAzIZFhYWgRjp/FgNq2gO1jId9Yae/rLn3RrBP8=
 go.etcd.io/etcd/client/pkg/v3 v3.6.5/go.mod h1:8Wx3eGRPiy0qOFMZT/hfvdos+DjEaPxdIDiCDUv/FQk=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=

install.yaml

@@ -139,6 +139,8 @@ spec:
         - name: bpf-programs
           mountPath: /sys/fs/bpf
           mountPropagation: HostToContainer
+        - name: dranet-run
+          mountPath: /var/run/dranet
       volumes:
       - name: device-plugin
         hostPath:
@@ -158,4 +160,8 @@ spec:
       - name: bpf-programs
         hostPath:
           path: /sys/fs/bpf
+      - name: dranet-run
+        hostPath:
+          path: /var/run/dranet
+          type: DirectoryOrCreate
 ---

pkg/driver/dra_hooks.go

@@ -66,7 +66,8 @@ func (np *NetworkDriver) PublishResources(ctx context.Context) {
 
 			resources := resourceslice.DriverResources{
 				Pools: map[string]resourceslice.Pool{
-					np.nodeName: {Slices: []resourceslice.Slice{{Devices: devices}}}},
+					np.nodeName: {Slices: []resourceslice.Slice{{Devices: devices}}},
+				},
 			}
 			err := np.draPlugin.PublishResources(ctx, resources)
 			if err != nil {
@@ -247,7 +248,9 @@ func (np *NetworkDriver) prepareResourceClaim(ctx context.Context, claim *resour
 			}
 			deviceCfg.RDMADevice = buildRDMAConfig(rdmaDevName, charDevices)
 			for _, uid := range podUIDs {
-				np.podConfigStore.SetDeviceConfig(uid, result.Device, deviceCfg)
+				if err := np.podConfigStore.SetDeviceConfig(uid, result.Device, deviceCfg); err != nil {
+					errorList = append(errorList, fmt.Errorf("failed to persist device config for pod %s device %s: %v", uid, result.Device, err))
+				}
 			}
 			klog.V(4).Infof("IB-only claim resources for pods %v : %#v", podUIDs, deviceCfg)
 			continue
@@ -395,7 +398,9 @@ func (np *NetworkDriver) prepareResourceClaim(ctx context.Context, claim *resour
 		// TODO: support for multiple pods sharing the same device
 		// we'll create the subinterface here
 		for _, uid := range podUIDs {
-			np.podConfigStore.SetDeviceConfig(uid, result.Device, deviceCfg)
+			if err := np.podConfigStore.SetDeviceConfig(uid, result.Device, deviceCfg); err != nil {
+				errorList = append(errorList, fmt.Errorf("failed to persist device config for pod %s device %s: %v", uid, result.Device, err))
+			}
 		}
 		klog.V(4).Infof("Claim Resources for pods %v : %#v", podUIDs, deviceCfg)
 	}

pkg/driver/dra_hooks_test.go

@@ -193,7 +193,7 @@ func TestUnprepareResourceClaimsMetrics(t *testing.T) {
 		draPluginRequestsLatencySeconds.Reset()
 
 		np := &NetworkDriver{
-			podConfigStore: NewPodConfigStore(),
+			podConfigStore: mustNewPodConfigStore(),
 		}
 		claimName := types.NamespacedName{Name: "test-claim", Namespace: "test-ns"}
 		np.podConfigStore.SetDeviceConfig("pod-uid-1", "device-a", DeviceConfig{Claim: claimName})

pkg/driver/driver.go

@@ -41,8 +41,7 @@ import (
 )
 
 const (
-	kubeletPluginRegistryPath = "/var/lib/kubelet/plugins_registry"
-	kubeletPluginPath         = "/var/lib/kubelet/plugins"
+	kubeletPluginPath = "/var/lib/kubelet/plugins"
 )
 
 const (
@@ -84,6 +83,14 @@ func WithInventory(db inventoryDB) Option {
 	}
 }
 
+// WithDBPath sets the path for the persistent pod config database.
+// If not set, an in-memory store is used.
+func WithDBPath(path string) Option {
+	return func(o *NetworkDriver) {
+		o.dbPath = path
+	}
+}
+
 type NetworkDriver struct {
 	driverName string
 	nodeName   string
@@ -98,6 +105,7 @@ type NetworkDriver struct {
 	// Cache the rdma shared mode state
 	rdmaSharedMode bool
 	podConfigStore *PodConfigStore
+	dbPath         string // path for persistent bbolt database; empty means in-memory
 
 	clock clock.WithTicker // Injectable clock for testing
 }
@@ -120,14 +128,31 @@ func Start(ctx context.Context, driverName string, kubeClient kubernetes.Interfa
 		nodeName:       nodeName,
 		kubeClient:     kubeClient,
 		rdmaSharedMode: rdmaNetnsMode == apis.RdmaNetnsModeShared,
-		podConfigStore: NewPodConfigStore(),
 		clock:          clock.RealClock{},
 	}
 
 	for _, o := range opts {
 		o(plugin)
 	}
 
+	// Initialize the pod config store with optional bbolt checkpoint backend.
+	var checkpointer Checkpointer
+	if plugin.dbPath != "" {
+		var err error
+		checkpointer, err = newBoltCheckpointer(plugin.dbPath)
+		if err != nil {
+			return nil, fmt.Errorf("failed to open pod config database at %s: %v", plugin.dbPath, err)
+		}
+	}
+	store, err := NewPodConfigStore(checkpointer)
+	if err != nil {
+		if checkpointer != nil {
+			checkpointer.Close()
+		}
+		return nil, fmt.Errorf("failed to initialize pod config store: %v", err)
+	}
+	plugin.podConfigStore = store
+
 	driverPluginPath := filepath.Join(kubeletPluginPath, driverName)
 	err = os.MkdirAll(driverPluginPath, 0750)
 	if err != nil {
@@ -312,5 +337,11 @@ func (np *NetworkDriver) Stop(ctxCancel context.CancelFunc) {
 	ctxCancel()
 
 	np.nriPlugin.Stop()
+
+	// Close the pod config store.
+	if err := np.podConfigStore.Close(); err != nil {
+		klog.Errorf("Failed to close pod config database: %v", err)
+	}
+
 	klog.Info("Driver stopped.")
 }

pkg/driver/driver_test.go

@@ -107,7 +107,7 @@ func TestStop(t *testing.T) {
 	np := &NetworkDriver{
 		draPlugin:      fakeDra,
 		nriPlugin:      fakeNri,
-		podConfigStore: NewPodConfigStore(),
+		podConfigStore: mustNewPodConfigStore(),
 		clock:          fakeClock,
 	}