Add RDM config, including a ctrl side disable flag. Fixes #2596

common/capabilities/consts.go

@@ -20,4 +20,5 @@ const (
 	ControllerCreateTerminatorV2     int = 1
 	ControllerSingleRouterLinkSource int = 2
 	ControllerCreateCircuitV2        int = 3
+	RouterDataModel                  int = 4
 )

common/router_data_model.go

@@ -36,6 +36,13 @@ import (
 	"sync/atomic"
 )
 
+// RouterDataModelConfig contains the configuration values for a RouterDataModel
+type RouterDataModelConfig struct {
+	Enabled            bool
+	LogSize            uint64
+	ListenerBufferSize uint
+}
+
 // AccessPolicies represents the Identity's access to a Service through many Policies. The PostureChecks provided
 // are referenced by the granting Policies. The PostureChecks for each of the Policies may be evaluated to determine
 // a valid policy and posture access path.

controller/config/config.go

@@ -30,6 +30,7 @@ import (
 	"github.com/openziti/storage/boltz"
 	"github.com/openziti/transport/v2"
 	transporttls "github.com/openziti/transport/v2/tls"
+	"github.com/openziti/ziti/common"
 	"github.com/openziti/ziti/common/config"
 	"github.com/openziti/ziti/common/pb/ctrl_pb"
 	"github.com/openziti/ziti/common/pb/mgmt_pb"
@@ -75,6 +76,10 @@ const (
 
 	// DefaultTlsHandshakeRateLimiterMaxWindow is the default max size for the tls handshake rate limiter
 	DefaultTlsHandshakeRateLimiterMaxWindow = 1000
+
+	DefaultRouterDataModelEnabled            = true
+	DefaultRouterDataModelLogSize            = 10_1000
+	DefaultRouterDataModelListenerBufferSize = 1000
 )
 
 type Config struct {
@@ -109,6 +114,7 @@ type Config struct {
 			InitialDelay time.Duration
 		}
 	}
+	RouterDataModel         common.RouterDataModelConfig
 	CommandRateLimiter      command.RateLimiterConfig
 	TlsHandshakeRateLimiter command.AdaptiveRateLimiterConfig
 	Src                     map[interface{}]interface{}
@@ -661,6 +667,50 @@ func LoadConfig(path string) (*Config, error) {
 		}
 	}
 
+	controllerConfig.RouterDataModel.Enabled = DefaultRouterDataModelEnabled
+	controllerConfig.RouterDataModel.LogSize = DefaultRouterDataModelLogSize
+	controllerConfig.RouterDataModel.ListenerBufferSize = DefaultRouterDataModelListenerBufferSize
+
+	if value, found := cfgmap["routerDataModel"]; found {
+		if submap, ok := value.(map[interface{}]interface{}); ok {
+			if value, found := submap["enabled"]; found {
+				controllerConfig.RouterDataModel.Enabled = strings.EqualFold("true", fmt.Sprintf("%v", value))
+			}
+
+			if value, found := submap["logSize"]; found {
+				if val, ok := value.(int); ok {
+					if val < 0 {
+						return nil, errors.Wrapf(err, "failed to parse routerDataModel.logSize, must be >= 0 %v", value)
+					}
+					controllerConfig.RouterDataModel.LogSize = uint64(val)
+				} else {
+					return nil, errors.Wrapf(err, "failed to parse routerDataModel.logSize, should be int not value %T", value)
+				}
+			}
+
+			if value, found := submap["listenerBufferSize"]; found {
+				if val, ok := value.(int); ok {
+					if val < 0 {
+						return nil, errors.Wrapf(err, "failed to parse routerDataModel.listenerBufferSize, must be >= 0 %v", value)
+					}
+					controllerConfig.RouterDataModel.ListenerBufferSize = uint(val)
+				} else {
+					return nil, errors.Wrapf(err, "failed to parse routerDataModel.listenerBufferSize, should be int not value %T", value)
+				}
+			}
+		} else {
+			return nil, errors.Errorf("invalid raft configuration")
+		}
+	} else if value, found := cfgmap["db"]; found {
+		str, err := db.Open(value.(string))
+		if err != nil {
+			return nil, err
+		}
+		controllerConfig.Db = str
+	} else {
+		panic("controllerConfig must provide [db] or [raft]")
+	}
+
 	edgeConfig, err := LoadEdgeConfigFromMap(cfgmap)
 	if err != nil {
 		return nil, err

controller/controller.go

@@ -359,6 +359,11 @@ func (c *Controller) Run() error {
 	capabilityMask.SetBit(capabilityMask, capabilities.ControllerCreateTerminatorV2, 1)
 	capabilityMask.SetBit(capabilityMask, capabilities.ControllerSingleRouterLinkSource, 1)
 	capabilityMask.SetBit(capabilityMask, capabilities.ControllerCreateCircuitV2, 1)
+
+	if c.config.RouterDataModel.Enabled || c.raftController != nil {
+		capabilityMask.SetBit(capabilityMask, capabilities.RouterDataModel, 1)
+	}
+
 	headers := map[int32][]byte{
 		channel.HelloVersionHeader:                       versionHeader,
 		int32(ctrl_pb.ControlHeaders_CapabilitiesHeader): capabilityMask.Bytes(),
@@ -703,7 +708,6 @@ func (c *Controller) GetApiAddresses() (map[string][]event.ApiAddress, []byte) {
 func (c *Controller) GetHelloHeaderProviders() []mesh.HeaderProvider {
 	providerFunc := func(headers map[int32][]byte) {
 		_, apiDataBytes := c.GetApiAddresses()
-
 		headers[mesh.ApiAddressesHeader] = apiDataBytes
 	}
 

controller/sync_strats/sync_instant.go

@@ -269,7 +269,7 @@ func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *Instant
 		changeSets:               map[uint64]*edge_ctrl_pb.DataState_ChangeSet{},
 	}
 
-	err := strategy.Initialize(10000, 1000)
+	err := strategy.Initialize(ae.GetConfig().RouterDataModel.LogSize, ae.GetConfig().RouterDataModel.ListenerBufferSize)
 
 	if err != nil {
 		pfxlog.Logger().WithError(err).Fatal("could not build initial data model for router synchronization")

router/handler_ctrl/bind.go

@@ -17,6 +17,7 @@
 package handler_ctrl
 
 import (
+	"github.com/openziti/ziti/common/capabilities"
 	"runtime/debug"
 	"time"
 
@@ -117,11 +118,8 @@ func (self *bindHandler) BindChannel(binding channel.Binding) error {
 		}
 	}
 
-	if ok, _ := ctrl.GetVersion().HasMinimumVersion("1.3.0"); ok {
-		self.env.GetRouterDataModelEnabledConfig().Store(true)
-	} else {
-		self.env.GetRouterDataModelEnabledConfig().Store(false)
-	}
+	enableRouterDataModel := capabilities.IsCapable(binding.GetChannel(), capabilities.RouterDataModel)
+	self.env.GetRouterDataModelEnabledConfig().Store(enableRouterDataModel)
 
 	return nil
 }