Merge pull request #1349 from stgraber/ovn

Add infrastructure for OVN events

internal/server/network/ovn/events.go

@@ -0,0 +1,14 @@
+package ovn
+
+import (
+	ovsdbModel "github.com/ovn-org/libovsdb/model"
+)
+
+// EventHandler represents an OVN database event handler.
+type EventHandler struct {
+	// Tables contains the list of OVN database tables to watch for events.
+	Tables []string
+
+	// Hook is the function being called on a matching event.
+	Hook func(action string, table string, oldObject ovsdbModel.Model, newObject ovsdbModel.Model)
+}

internal/server/network/ovn/ovn_sb.go

@@ -7,12 +7,15 @@ import (
 	"encoding/pem"
 	"fmt"
 	"runtime"
+	"slices"
 	"strings"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/go-logr/logr"
+	ovsdbCache "github.com/ovn-org/libovsdb/cache"
 	ovsdbClient "github.com/ovn-org/libovsdb/client"
+	ovsdbModel "github.com/ovn-org/libovsdb/model"
 
 	ovnSB "github.com/lxc/incus/v6/internal/server/network/ovn/schema/ovn-sb"
 )
@@ -128,11 +131,64 @@ func NewSB(dbAddr string, sslCACert string, sslClientCert string, sslClientKey s
 		return nil, err
 	}
 
-	monitorCookie, err := ovn.MonitorAll(context.TODO())
+	// Set up monitor for the tables we use.
+	monitorCookie, err := ovn.Monitor(context.TODO(), ovn.NewMonitor(
+		ovsdbClient.WithTable(&ovnSB.Chassis{}),
+		ovsdbClient.WithTable(&ovnSB.PortBinding{}),
+		ovsdbClient.WithTable(&ovnSB.ServiceMonitor{})))
 	if err != nil {
 		return nil, err
 	}
 
+	// Set up event handlers.
+	eventHandler := &ovsdbCache.EventHandlerFuncs{}
+	eventHandler.AddFunc = func(table string, newModel ovsdbModel.Model) {
+		sbEventHandlersMu.Lock()
+		defer sbEventHandlersMu.Unlock()
+
+		if sbEventHandlers == nil {
+			return
+		}
+
+		for _, handler := range sbEventHandlers {
+			if handler.Hook != nil && slices.Contains(handler.Tables, table) {
+				go handler.Hook("add", table, nil, newModel)
+			}
+		}
+	}
+
+	eventHandler.UpdateFunc = func(table string, oldModel ovsdbModel.Model, newModel ovsdbModel.Model) {
+		sbEventHandlersMu.Lock()
+		defer sbEventHandlersMu.Unlock()
+
+		if sbEventHandlers == nil {
+			return
+		}
+
+		for _, handler := range sbEventHandlers {
+			if handler.Hook != nil && slices.Contains(handler.Tables, table) {
+				go handler.Hook("update", table, oldModel, newModel)
+			}
+		}
+	}
+
+	eventHandler.DeleteFunc = func(table string, oldModel ovsdbModel.Model) {
+		sbEventHandlersMu.Lock()
+		defer sbEventHandlersMu.Unlock()
+
+		if sbEventHandlers == nil {
+			return
+		}
+
+		for _, handler := range sbEventHandlers {
+			if handler.Hook != nil && slices.Contains(handler.Tables, table) {
+				go handler.Hook("remove", table, oldModel, nil)
+			}
+		}
+	}
+
+	ovn.Cache().AddEventHandler(eventHandler)
+
 	// Create the SB struct.
 	client := &SB{
 		client: ovn,

internal/server/network/ovn/ovn_sb_events.go

@@ -0,0 +1,36 @@
+package ovn
+
+import (
+	"sync"
+)
+
+var sbEventHandlers map[string]EventHandler
+var sbEventHandlersMu sync.Mutex
+
+// AddOVNSBHandler registers a new event handler with the OVN Southbound database.
+func AddOVNSBHandler(name string, handler EventHandler) error {
+	sbEventHandlersMu.Lock()
+	defer sbEventHandlersMu.Unlock()
+
+	if sbEventHandlers == nil {
+		sbEventHandlers = map[string]EventHandler{}
+	}
+
+	sbEventHandlers[name] = handler
+
+	return nil
+}
+
+// RemoveOVNSBHandler removes a currently registered event handler.
+func RemoveOVNSBHandler(name string) error {
+	sbEventHandlersMu.Lock()
+	defer sbEventHandlersMu.Unlock()
+
+	if sbEventHandlers == nil {
+		return nil
+	}
+
+	delete(sbEventHandlers, name)
+
+	return nil
+}