chore: Offload handler placeholders for the great eBPF merge

Author: rg0now

config_test.go

@@ -114,7 +114,8 @@ func TestStunnerConfigFileRoundTrip(t *testing.T) {
 
 	newConf := &stnrv1.StunnerConfig{}
 	err = yaml.Unmarshal(file, newConf)
-	assert.NoError(t, err, "unmarschal config from file")
+	assert.NoError(t, err, "unmarshal config from file")
+	assert.NoError(t, newConf.Validate(), "validate")
 
 	ok := newConf.DeepEqual(c)
 	assert.True(t, ok, "config file roundtrip")

handlers.go

@@ -177,10 +177,12 @@ const (
 
 // TURN offload handler.
 type OffloadHandler interface {
-	// OffloadHandler manages offload requests.
-	OffloadHandler(src, dst net.Addr, proto, username, realm string, relayAddr, peer net.Addr, chanNum uint16, event OffloadEventType)
+	// Start starts the offload handler.
+	Start() error
 	// Close closes the offload handler.
 	Close() error
+	// Offload manages offload requests.
+	Offload(OffloadEventType, net.Addr, net.Addr, string, string, string, net.Addr, net.Addr, uint16, string, string)
 }
 
 // NewOffloadHandler creates a offload handler that defaults to a stub.
@@ -193,17 +195,18 @@ var offloadHandlerConstructor = newOffloadHandlerStub
 // offloadHandlerStub is a stub offload handler that does nothing.
 type offloadHandlerStub struct{}
 
-func (o *offloadHandlerStub) OffloadHandler(_, _ net.Addr, _, _, _ string, _, _ net.Addr, _ uint16, _ OffloadEventType) {
+func (o *offloadHandlerStub) Offload(_ OffloadEventType, _, _ net.Addr, _, _, _ string, _, _ net.Addr, _ uint16, _, _ string) {
 }
 
+func (o *offloadHandlerStub) Start() error { return nil }
 func (o *offloadHandlerStub) Close() error { return nil }
 
 func newOffloadHandlerStub(_ *Stunner) OffloadHandler {
 	return &offloadHandlerStub{}
 }
 
 // NewEventHandler creates a set of callbcks for tracking the lifecycle of TURN allocations.
-func (s *Stunner) NewEventHandler() turn.EventHandlers {
+func (s *Stunner) NewEventHandler(l *object.Listener) turn.EventHandlers {
 	return turn.EventHandlers{
 		OnAuth: func(src, dst net.Addr, proto, username, realm string, method string, verdict bool) {
 			status := "REJECTED"
@@ -240,14 +243,33 @@ func (s *Stunner) NewEventHandler() turn.EventHandlers {
 				dumpClient(src, dst, proto, username, realm), relayAddr.String(),
 				peer.String(), chanNum)
 
-			s.offloadHandler.OffloadHandler(src, dst, proto, username, realm, relayAddr, peer, chanNum, ChannnelOffloadCreated)
+			// listener and cluster needed for monitoring
+			listener := l.Name
+			cluster := ""
+			peerAddr, ok := peer.(*net.UDPAddr)
+			if ok {
+				clusters := s.clusterManager.Keys()
+				for _, r := range l.Routes {
+					if util.Member(clusters, r) {
+						c := s.GetCluster(r)
+						if c.Route(peerAddr.IP) {
+							cluster = c.Name
+							break
+						}
+					}
+				}
+			}
+
+			s.offloadHandler.Offload(ChannnelOffloadCreated, src, dst, proto, username, realm,
+				relayAddr, peer, chanNum, listener, cluster)
 		},
 		OnChannelDeleted: func(src, dst net.Addr, proto, username, realm string, relayAddr, peer net.Addr, chanNum uint16) {
 			s.log.Infof("Channel deleted: client=%s, relay-addr=%s, peer=%s, channel-num=%d",
 				dumpClient(src, dst, proto, username, realm), relayAddr.String(),
 				peer.String(), chanNum)
 
-			s.offloadHandler.OffloadHandler(src, dst, proto, username, realm, relayAddr, peer, chanNum, ChannnelOffloadDeleted)
+			s.offloadHandler.Offload(ChannnelOffloadDeleted, src, dst, proto, username, realm,
+				relayAddr, peer, chanNum, "", "")
 		},
 	}
 }

internal/object/admin.go

@@ -8,7 +8,9 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"reflect"
 	"strconv"
+	"strings"
 
 	"github.com/pion/logging"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -27,6 +29,8 @@ type Admin struct {
 	metricsServer, healthCheckServer     *http.Server
 	health                               *http.ServeMux
 	quota                                int
+	offload                              stnrv1.OffloadMode
+	offloadIntfs                         []string
 	LicenseManager                       licensecfg.ConfigManager
 	licenseConfig                        *stnrv1.LicenseConfig
 	log                                  logging.LeveledLogger
@@ -43,6 +47,7 @@ func NewAdmin(conf stnrv1.Config, dryRun bool, rc ReadinessHandler, status Statu
 		DryRun:         dryRun,
 		health:         http.NewServeMux(),
 		LicenseManager: licensecfg.New(logger.NewLogger("license")),
+		offload:        stnrv1.OffloadEngineNone,
 		log:            logger.NewLogger("admin"),
 	}
 	admin.log.Tracef("NewAdmin: %s", req.String())
@@ -56,7 +61,7 @@ func NewAdmin(conf stnrv1.Config, dryRun bool, rc ReadinessHandler, status Statu
 	})
 
 	// readniness checker calls the checker from the factory
-	admin.health.HandleFunc("/ready", func(w http.ResponseWriter, req *http.Request) {
+	admin.health.HandleFunc("/ready", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "application/json; charset=utf-8")
 		if err := rc(); err != nil {
 			w.WriteHeader(http.StatusServiceUnavailable)
@@ -71,7 +76,7 @@ func NewAdmin(conf stnrv1.Config, dryRun bool, rc ReadinessHandler, status Statu
 	})
 
 	// status handler returns the status
-	admin.health.HandleFunc("/status", func(w http.ResponseWriter, req *http.Request) {
+	admin.health.HandleFunc("/status", func(w http.ResponseWriter, _ *http.Request) {
 		w.Header().Set("Content-Type", "application/json; charset=utf-8")
 		if js, err := json.Marshal(status()); err != nil {
 			w.WriteHeader(http.StatusInternalServerError)
@@ -93,7 +98,25 @@ func NewAdmin(conf stnrv1.Config, dryRun bool, rc ReadinessHandler, status Statu
 // Inspect examines whether a configuration change requires a reconciliation (returns true if it
 // does) or restart (returns ErrRestartRequired).
 func (a *Admin) Inspect(old, new, full stnrv1.Config) (bool, error) {
-	return !old.DeepEqual(new), nil
+	oldConf, ok := old.(*stnrv1.AdminConfig)
+	if !ok {
+		return false, stnrv1.ErrInvalidConf
+	}
+
+	newConf, ok := new.(*stnrv1.AdminConfig)
+	if !ok {
+		return false, stnrv1.ErrInvalidConf
+	}
+
+	changed := !oldConf.DeepEqual(newConf)
+
+	// restart the admin object if the offload engine or the interfaces change
+	var restart error
+	if oldConf.OffloadEngine != newConf.OffloadEngine ||
+		!reflect.DeepEqual(oldConf.OffloadInterfaces, newConf.OffloadInterfaces) {
+		restart = ErrRestartRequired
+	}
+	return changed, restart
 }
 
 // Reconcile updates the authenticator for a new configuration. Requires a valid reconciliation
@@ -127,6 +150,9 @@ func (a *Admin) Reconcile(conf stnrv1.Config) error {
 
 	a.quota = req.UserQuota
 
+	a.offload, _ = stnrv1.NewOffloadEngine(req.OffloadEngine)
+	a.offloadIntfs = req.OffloadInterfaces
+
 	a.LicenseManager.Reconcile(req.LicenseConfig)
 	a.licenseConfig = req.LicenseConfig
 
@@ -157,6 +183,8 @@ func (a *Admin) GetConfig() stnrv1.Config {
 		MetricsEndpoint:     a.MetricsEndpoint,
 		HealthCheckEndpoint: &h,
 		UserQuota:           a.quota,
+		OffloadEngine:       a.offload.String(),
+		OffloadInterfaces:   a.offloadIntfs,
 		LicenseConfig:       a.licenseConfig,
 	}
 }
@@ -186,12 +214,17 @@ func (a *Admin) Close() error {
 
 // Status returns the status of the object.
 func (a *Admin) Status() stnrv1.Status {
+	offloadIntfs := ""
+	if a.offload != stnrv1.OffloadEngineNone && len(a.offloadIntfs) > 0 {
+		offloadIntfs = strings.Join(a.offloadIntfs, ",")
+	}
 	s := stnrv1.AdminStatus{
 		Name:                a.Name,
 		LogLevel:            a.LogLevel,
 		MetricsEndpoint:     a.MetricsEndpoint,
 		HealthCheckEndpoint: a.HealthCheckEndpoint,
 		UserQuota:           fmt.Sprintf("%d", a.quota),
+		OffloadStatus:       fmt.Sprintf("offload=%s%s", a.offload.String(), offloadIntfs),
 		LicensingInfo:       a.LicenseManager.Status(),
 	}
 	return &s

pkg/apis/v1/admin.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net/url"
 	"reflect"
+	"sort"
 	"strings"
 )
 
@@ -28,6 +29,12 @@ type AdminConfig struct {
 	// UserQuota defines the number of permitted TURN allocatoins per username. Affects
 	// allocation created on any listener. Default is 0, meaning no quota is enforced.
 	UserQuota int `json:"user_quota,omitempty"`
+	// OffloadEngine defines the dataplane offload mode, either "None", "XDP", "TC", or
+	// "Auto". Set to "Auto" to let STUNner find the optimal offload mode. Default is "None".
+	OffloadEngine string `json:"offload_engine,omitempty"`
+	// OffloadInterfaces explicitly specifies the interfaces on which to enable the offload
+	// engine. Empty list means to enable offload on all interfaces (this is the default).
+	OffloadInterfaces []string `json:"offload_interfaces,omitempty"`
 	// LicenseConfig describes the licensing info to be used to check subscription status with
 	// the license server.
 	LicenseConfig *LicenseConfig `json:"license_config,omitempty"`
@@ -76,6 +83,21 @@ func (req *AdminConfig) Validate() error {
 		req.UserQuota = 0
 	}
 
+	// Normalize
+	if req.OffloadEngine == "" {
+		req.OffloadEngine = OffloadEngineNone.String()
+	}
+	t, err := NewOffloadEngine(req.OffloadEngine)
+	if err != nil {
+		return err
+	}
+	req.OffloadEngine = t.String()
+
+	if req.OffloadInterfaces == nil {
+		req.OffloadInterfaces = []string{}
+	}
+	sort.Strings(req.OffloadInterfaces)
+
 	return nil
 }
 
@@ -94,6 +116,8 @@ func (req *AdminConfig) DeepEqual(other Config) bool {
 func (req *AdminConfig) DeepCopyInto(dst Config) {
 	ret := dst.(*AdminConfig)
 	*ret = *req
+	ret.OffloadInterfaces = make([]string, len(req.OffloadInterfaces))
+	copy(ret.OffloadInterfaces, req.OffloadInterfaces)
 }
 
 // String stringifies the configuration.
@@ -114,6 +138,13 @@ func (req *AdminConfig) String() string {
 	if req.UserQuota > 0 {
 		status = append(status, fmt.Sprintf("quota=%d", req.UserQuota))
 	}
+	if req.OffloadEngine != "" {
+		intfs := ""
+		if req.OffloadEngine != "None" && len(req.OffloadInterfaces) > 0 {
+			intfs = fmt.Sprintf("<%s>", strings.Join(req.OffloadInterfaces, ","))
+		}
+		status = append(status, fmt.Sprintf("offload=%s%s", req.OffloadEngine, intfs))
+	}
 	status = append(status, fmt.Sprintf("license_info=%s", LicensingStatus(req.LicenseConfig)))
 
 	return fmt.Sprintf("admin:{%s}", strings.Join(status, ","))
@@ -142,6 +173,7 @@ type AdminStatus struct {
 	MetricsEndpoint     string `json:"metrics_endpoint,omitempty"`
 	HealthCheckEndpoint string `json:"healthcheck_endpoint,omitempty"`
 	UserQuota           string `json:"quota,omitempty"`
+	OffloadStatus       string `json:"offload,omitempty"`
 	LicensingInfo       string `json:"licensing_info,omitempty"`
 }
 
@@ -161,6 +193,9 @@ func (a *AdminStatus) String() string {
 	if a.LicensingInfo != "" {
 		status = append(status, fmt.Sprintf("license-info=%s", a.LicensingInfo))
 	}
+	if a.OffloadStatus != "" {
+		status = append(status, fmt.Sprintf("offload=%s", a.OffloadStatus))
+	}
 
 	return fmt.Sprintf("admin:{%s}", strings.Join(status, ","))
 }

pkg/apis/v1/util.go

@@ -148,7 +148,8 @@ func NewClusterType(raw string) (ClusterType, error) {
 	case clusterTypeStrictDNSStr:
 		return ClusterTypeStrictDNS, nil
 	default:
-		return ClusterType(ClusterTypeUnknown), fmt.Errorf("unknown cluster type: \"%s\"", raw)
+		return ClusterType(ClusterTypeUnknown),
+			fmt.Errorf("unknown cluster type: \"%s\"", raw)
 	}
 }
 
@@ -201,3 +202,53 @@ func (p ClusterProtocol) String() string {
 		return "<unknown>"
 	}
 }
+
+// OffloadEngine specifies the type of TURN offload mode.
+type OffloadMode int
+
+const (
+	OffloadEngineNone OffloadMode = iota
+	OffloadEngineXDP
+	OffloadEngineTC
+	OffloadEngineAuto
+)
+
+const (
+	offloadEngineNoneStr = "none"
+	offloadEngineXDPStr  = "XDP"
+	offloadEngineTCStr   = "TC"
+	offloadEngineAutoStr = "Auto"
+)
+
+// NewOffloadEngine parses the offload mode.
+func NewOffloadEngine(raw string) (OffloadMode, error) {
+	switch strings.ToLower(raw) {
+	case strings.ToLower(offloadEngineNoneStr):
+		return OffloadEngineNone, nil
+	case strings.ToLower(offloadEngineXDPStr):
+		return OffloadEngineXDP, nil
+	case strings.ToLower(offloadEngineTCStr):
+		return OffloadEngineTC, nil
+	case strings.ToLower(offloadEngineAutoStr):
+		return OffloadEngineAuto, nil
+	default:
+		return OffloadEngineNone,
+			fmt.Errorf("unknown offload mode: %q", raw)
+	}
+}
+
+// String returns a string representation of a cluster protocol.
+func (p OffloadMode) String() string {
+	switch p {
+	case OffloadEngineNone:
+		return offloadEngineNoneStr
+	case OffloadEngineXDP:
+		return offloadEngineXDPStr
+	case OffloadEngineTC:
+		return offloadEngineTCStr
+	case OffloadEngineAuto:
+		return offloadEngineAutoStr
+	default:
+		return "<unknown>"
+	}
+}