HA SDK terminators test

common/pb/cmd_pb/cmd.proto

@@ -123,4 +123,5 @@ message Terminator {
   string hostId = 12;
   bool isSystem = 13;
   uint32 savedPrecedence = 14;
+  string sourceCtrl = 15;
 }

common/pb/ctrl_pb/impl.go

@@ -16,7 +16,10 @@
 
 package ctrl_pb
 
-import "github.com/openziti/ziti/controller/xt"
+import (
+	"github.com/openziti/channel/v3/protobufs"
+	"github.com/openziti/ziti/controller/xt"
+)
 
 func (request *CircuitConfirmation) GetContentType() int32 {
 	return int32(ContentType_CircuitConfirmationType)
@@ -107,3 +110,29 @@ func (request *UpdateCtrlAddresses) GetContentType() int32 {
 func (request *PeerStateChanges) GetContentType() int32 {
 	return int32(ContentType_PeerStateChangeRequestType)
 }
+
+type FilterableValidateTerminatorsRequest interface {
+	protobufs.TypedMessage
+	FilterTerminators(f func(terminator *Terminator) bool)
+	GetTerminators() []*Terminator
+}
+
+func (request *ValidateTerminatorsRequest) FilterTerminators(f func(terminator *Terminator) bool) {
+	var terminators []*Terminator
+	for _, terminator := range request.Terminators {
+		if f(terminator) {
+			terminators = append(terminators, terminator)
+		}
+	}
+	request.Terminators = terminators
+}
+
+func (request *ValidateTerminatorsV2Request) FilterTerminators(f func(terminator *Terminator) bool) {
+	var terminators []*Terminator
+	for _, terminator := range request.Terminators {
+		if f(terminator) {
+			terminators = append(terminators, terminator)
+		}
+	}
+	request.Terminators = terminators
+}

common/router_data_model.go

@@ -18,9 +18,12 @@ package common
 
 import (
 	"compress/gzip"
+	"crypto"
+	"crypto/x509"
 	"encoding/json"
 	"fmt"
 	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/foundation/v2/concurrenz"
 	"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/sirupsen/logrus"
@@ -93,14 +96,15 @@ type RouterDataModel struct {
 	EventCache
 	listeners map[chan *edge_ctrl_pb.DataState_ChangeSet]struct{}
 
-	ConfigTypes     cmap.ConcurrentMap[string, *ConfigType]                        `json:"configTypes"`
-	Configs         cmap.ConcurrentMap[string, *Config]                            `json:"configs"`
-	Identities      cmap.ConcurrentMap[string, *Identity]                          `json:"identities"`
-	Services        cmap.ConcurrentMap[string, *Service]                           `json:"services"`
-	ServicePolicies cmap.ConcurrentMap[string, *ServicePolicy]                     `json:"servicePolicies"`
-	PostureChecks   cmap.ConcurrentMap[string, *PostureCheck]                      `json:"postureChecks"`
-	PublicKeys      cmap.ConcurrentMap[string, *edge_ctrl_pb.DataState_PublicKey]  `json:"publicKeys"`
-	Revocations     cmap.ConcurrentMap[string, *edge_ctrl_pb.DataState_Revocation] `json:"revocations"`
+	ConfigTypes      cmap.ConcurrentMap[string, *ConfigType]                        `json:"configTypes"`
+	Configs          cmap.ConcurrentMap[string, *Config]                            `json:"configs"`
+	Identities       cmap.ConcurrentMap[string, *Identity]                          `json:"identities"`
+	Services         cmap.ConcurrentMap[string, *Service]                           `json:"services"`
+	ServicePolicies  cmap.ConcurrentMap[string, *ServicePolicy]                     `json:"servicePolicies"`
+	PostureChecks    cmap.ConcurrentMap[string, *PostureCheck]                      `json:"postureChecks"`
+	PublicKeys       cmap.ConcurrentMap[string, *edge_ctrl_pb.DataState_PublicKey]  `json:"publicKeys"`
+	Revocations      cmap.ConcurrentMap[string, *edge_ctrl_pb.DataState_Revocation] `json:"revocations"`
+	CachedPublicKeys concurrenz.AtomicValue[map[string]crypto.PublicKey]
 
 	listenerBufferSize uint
 	lastSaveIndex      *uint64
@@ -433,6 +437,7 @@ func (rdm *RouterDataModel) HandlePublicKeyEvent(event *edge_ctrl_pb.DataState_E
 	} else {
 		rdm.PublicKeys.Set(model.PublicKey.Kid, model.PublicKey)
 	}
+	rdm.recalculateCachedPublicKeys()
 }
 
 // HandleRevocationEvent will apply the delta event to the router data model. It is not restricted by index calculations.
@@ -507,8 +512,33 @@ func (rdm *RouterDataModel) HandleServicePolicyChange(index uint64, model *edge_
 	})
 }
 
-func (rdm *RouterDataModel) GetPublicKeys() map[string]*edge_ctrl_pb.DataState_PublicKey {
-	return rdm.PublicKeys.Items()
+func (rdm *RouterDataModel) GetPublicKeys() map[string]crypto.PublicKey {
+	return rdm.CachedPublicKeys.Load()
+}
+
+func (rdm *RouterDataModel) recalculateCachedPublicKeys() {
+	publicKeys := map[string]crypto.PublicKey{}
+	rdm.PublicKeys.IterCb(func(kid string, pubKey *edge_ctrl_pb.DataState_PublicKey) {
+		log := pfxlog.Logger().WithField("format", pubKey.Format).WithField("kid", kid)
+
+		switch pubKey.Format {
+		case edge_ctrl_pb.DataState_PublicKey_X509CertDer:
+			if cert, err := x509.ParseCertificate(pubKey.GetData()); err != nil {
+				log.WithError(err).Error("error parsing x509 certificate DER")
+			} else {
+				publicKeys[kid] = cert.PublicKey
+			}
+		case edge_ctrl_pb.DataState_PublicKey_PKIXPublicKey:
+			if pub, err := x509.ParsePKIXPublicKey(pubKey.GetData()); err != nil {
+				log.WithError(err).Error("error parsing PKIX public key DER")
+			} else {
+				publicKeys[kid] = pub
+			}
+		default:
+			log.Error("unknown public key format")
+		}
+	})
+	rdm.CachedPublicKeys.Store(publicKeys)
 }
 
 func (rdm *RouterDataModel) GetDataState() *edge_ctrl_pb.DataState {

controller/command/command.go

@@ -52,6 +52,7 @@ type Dispatcher interface {
 	Dispatch(command Command) error
 	IsLeaderOrLeaderless() bool
 	IsLeaderless() bool
+	IsLeader() bool
 	GetPeers() map[string]channel.Channel
 	GetRateLimiter() rate.RateLimiter
 	Bootstrap() error
@@ -67,6 +68,10 @@ func (self *LocalDispatcher) Bootstrap() error {
 	return nil
 }
 
+func (self *LocalDispatcher) IsLeader() bool {
+	return true
+}
+
 func (self *LocalDispatcher) IsLeaderOrLeaderless() bool {
 	return true
 }

controller/config/config.go

@@ -51,7 +51,7 @@ const (
 	DefaultHealthChecksBoltCheckTimeout      = 20 * time.Second
 	DefaultHealthChecksBoltCheckInitialDelay = 30 * time.Second
 
-	DefaultRaftCommandHandlerMaxQueueSize = 1000
+	DefaultRaftCommandHandlerMaxQueueSize = 250
 
 	// DefaultTlsHandshakeRateLimiterEnabled is whether the tls handshake rate limiter is enabled by default
 	DefaultTlsHandshakeRateLimiterEnabled = false
@@ -204,6 +204,10 @@ func LoadConfig(path string) (*Config, error) {
 	if value, found := cfgmap["raft"]; found {
 		if submap, ok := value.(map[interface{}]interface{}); ok {
 			controllerConfig.Raft = &RaftConfig{}
+
+			controllerConfig.Raft.ElectionTimeout = 5 * time.Second
+			controllerConfig.Raft.HeartbeatTimeout = 3 * time.Second
+			controllerConfig.Raft.LeaderLeaseTimeout = 3 * time.Second
 			controllerConfig.Raft.CommandHandlerOptions.MaxQueueSize = DefaultRaftCommandHandlerMaxQueueSize
 
 			if value, found := submap["dataDir"]; found {
@@ -243,23 +247,23 @@ func LoadConfig(path string) (*Config, error) {
 
 			if value, found := submap["electionTimeout"]; found {
 				if val, err := time.ParseDuration(fmt.Sprintf("%v", value)); err == nil {
-					controllerConfig.Raft.ElectionTimeout = &val
+					controllerConfig.Raft.ElectionTimeout = val
 				} else {
 					return nil, errors.Wrapf(err, "failed to parse raft.electionTimeout value '%v", value)
 				}
 			}
 
 			if value, found := submap["heartbeatTimeout"]; found {
 				if val, err := time.ParseDuration(fmt.Sprintf("%v", value)); err == nil {
-					controllerConfig.Raft.HeartbeatTimeout = &val
+					controllerConfig.Raft.HeartbeatTimeout = val
 				} else {
 					return nil, errors.Wrapf(err, "failed to parse raft.heartbeatTimeout value '%v", value)
 				}
 			}
 
 			if value, found := submap["leaderLeaseTimeout"]; found {
 				if val, err := time.ParseDuration(fmt.Sprintf("%v", value)); err == nil {
-					controllerConfig.Raft.LeaderLeaseTimeout = &val
+					controllerConfig.Raft.LeaderLeaseTimeout = val
 				} else {
 					return nil, errors.Wrapf(err, "failed to parse raft.leaderLeaseTimeout value '%v", value)
 				}

controller/config/config_raft.go

@@ -20,10 +20,10 @@ type RaftConfig struct {
 	TrailingLogs      *uint32
 	MaxAppendEntries  *uint32
 
-	ElectionTimeout    *time.Duration
+	ElectionTimeout    time.Duration
 	CommitTimeout      *time.Duration
-	HeartbeatTimeout   *time.Duration
-	LeaderLeaseTimeout *time.Duration
+	HeartbeatTimeout   time.Duration
+	LeaderLeaseTimeout time.Duration
 
 	LogLevel *string
 	Logger   hclog.Logger

controller/controller.go

@@ -540,7 +540,7 @@ func (c *Controller) GetEventDispatcher() event.Dispatcher {
 }
 
 func (c *Controller) routerDispatchCallback(evt *event.ClusterEvent) {
-	if evt.EventType == event.ClusterMembersChanged {
+	if evt.EventType == event.ClusterMembersChanged || evt.EventType == event.ClusterLeadershipGained {
 		var endpoints []string
 		for _, peer := range evt.Peers {
 			endpoints = append(endpoints, peer.Addr)

controller/db/controller_store.go

@@ -36,12 +36,12 @@ const (
 
 type Controller struct {
 	boltz.BaseExtEntity
-	Name         string     `json:"name"`
-	CtrlAddress  string     `json:"address"`
-	CertPem      string     `json:"certPem"`
-	Fingerprint  string     `json:"fingerprint"`
-	IsOnline     bool       `json:"isOnline"`
-	LastJoinedAt *time.Time `json:"lastJoinedAt"`
+	Name         string    `json:"name"`
+	CtrlAddress  string    `json:"address"`
+	CertPem      string    `json:"certPem"`
+	Fingerprint  string    `json:"fingerprint"`
+	IsOnline     bool      `json:"isOnline"`
+	LastJoinedAt time.Time `json:"lastJoinedAt"`
 	ApiAddresses map[string][]ApiAddress
 }
 
@@ -105,7 +105,7 @@ func (store *controllerStoreImpl) FillEntity(entity *Controller, bucket *boltz.T
 	entity.CertPem = bucket.GetStringOrError(FieldControllerCertPem)
 	entity.Fingerprint = bucket.GetStringOrError(FieldControllerFingerprint)
 	entity.IsOnline = bucket.GetBoolWithDefault(FieldControllerIsOnline, false)
-	entity.LastJoinedAt = bucket.GetTime(FieldControllerLastJoinedAt)
+	entity.LastJoinedAt = bucket.GetTimeOrError(FieldControllerLastJoinedAt)
 	entity.ApiAddresses = map[string][]ApiAddress{}
 
 	apiListBucket := bucket.GetBucket(FieldControllerApiAddresses)
@@ -142,7 +142,7 @@ func (store *controllerStoreImpl) PersistEntity(entity *Controller, ctx *boltz.P
 	ctx.SetString(FieldControllerCertPem, entity.CertPem)
 	ctx.SetString(FieldControllerFingerprint, entity.Fingerprint)
 	ctx.SetBool(FieldControllerIsOnline, entity.IsOnline)
-	ctx.SetTimeP(FieldControllerLastJoinedAt, entity.LastJoinedAt)
+	ctx.SetTimeP(FieldControllerLastJoinedAt, &entity.LastJoinedAt)
 
 	apiListBucket := ctx.Bucket.GetOrCreateBucket(FieldControllerApiAddresses)
 

controller/db/terminator_store.go

@@ -19,10 +19,10 @@ package db
 import (
 	"encoding/binary"
 	"github.com/michaelquigley/pfxlog"
-	"github.com/openziti/ziti/controller/xt"
 	"github.com/openziti/foundation/v2/sequence"
 	"github.com/openziti/storage/ast"
 	"github.com/openziti/storage/boltz"
+	"github.com/openziti/ziti/controller/xt"
 	"go.etcd.io/bbolt"
 )
 
@@ -39,6 +39,7 @@ const (
 	FieldServerPeerData            = "peerData"
 	FieldTerminatorHostId          = "hostId"
 	FieldTerminatorSavedPrecedence = "savedPrecedence"
+	FieldTerminatorsSourceCtrl     = "sourceCtrl"
 )
 
 type Terminator struct {
@@ -54,6 +55,7 @@ type Terminator struct {
 	PeerData        xt.PeerData `json:"peerData"`
 	HostId          string      `json:"hostId"`
 	SavedPrecedence *string     `json:"savedPrecedence"`
+	SourceCtrl      string      `json:"sourceCtrl"`
 }
 
 func (entity *Terminator) GetCost() uint16 {
@@ -100,6 +102,10 @@ func (entity *Terminator) GetEntityType() string {
 	return EntityTypeTerminators
 }
 
+func (entity *Terminator) GetSourceCtrl() string {
+	return entity.SourceCtrl
+}
+
 type TerminatorStore interface {
 	boltz.EntityStore[*Terminator]
 	GetTerminatorsInIdentityGroup(tx *bbolt.Tx, terminatorId string) ([]*Terminator, error)
@@ -133,6 +139,7 @@ func (store *terminatorStoreImpl) initializeLocal() {
 	store.AddSymbol(FieldTerminatorAddress, ast.NodeTypeString)
 	store.AddSymbol(FieldTerminatorInstanceId, ast.NodeTypeString)
 	store.AddSymbol(FieldTerminatorHostId, ast.NodeTypeString)
+	store.AddSymbol(FieldTerminatorsSourceCtrl, ast.NodeTypeString)
 
 	store.serviceSymbol = store.AddFkSymbol(FieldTerminatorService, store.stores.service)
 	store.routerSymbol = store.AddFkSymbol(FieldTerminatorRouter, store.stores.router)
@@ -164,6 +171,7 @@ func (store *terminatorStoreImpl) FillEntity(entity *Terminator, bucket *boltz.T
 	entity.Precedence = bucket.GetStringWithDefault(FieldTerminatorPrecedence, xt.Precedences.Default.String())
 	entity.HostId = bucket.GetStringWithDefault(FieldTerminatorHostId, "")
 	entity.SavedPrecedence = bucket.GetString(FieldTerminatorSavedPrecedence)
+	entity.SourceCtrl = bucket.GetStringWithDefault(FieldTerminatorsSourceCtrl, "")
 
 	data := bucket.GetBucket(FieldServerPeerData)
 	if data != nil {
@@ -201,6 +209,7 @@ func (store *terminatorStoreImpl) PersistEntity(entity *Terminator, ctx *boltz.P
 	ctx.SetRequiredString(FieldTerminatorPrecedence, entity.Precedence)
 	ctx.SetString(FieldTerminatorHostId, entity.HostId)
 	ctx.SetStringP(FieldTerminatorSavedPrecedence, entity.SavedPrecedence)
+	ctx.SetString(FieldTerminatorsSourceCtrl, entity.SourceCtrl)
 
 	if ctx.ProceedWithSet(FieldServerPeerData) {
 		_ = ctx.Bucket.DeleteBucket([]byte(FieldServerPeerData))

controller/env/appenv.go

@@ -413,6 +413,10 @@ func (ae *AppEnv) AddRouterPresenceHandler(h model.RouterPresenceHandler) {
 	ae.HostController.GetNetwork().AddRouterPresenceHandler(h)
 }
 
+func (ae *AppEnv) GetId() string {
+	return ae.HostController.GetNetwork().GetAppId()
+}
+
 type HostController interface {
 	GetConfig() *config.Config
 	GetEnv() *AppEnv