gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-15 08:31:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #530 from openziti/modular.api.session.sync

Browse Source 
modular edge router sync
This commit is contained in:
Andrew
2021-02-15 20:18:07 -05:00
committed by GitHub
parent 7eccd2f34b fdf9221f89
commit 69326cbbed
33 changed files with 2453 additions and 2935 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/env/appenv.go
Vendored
+1 -1
View File
@@ -118,7 +118,7 @@ func (ae *AppEnv) GetEnrollRegistry() model.EnrollmentRegistry {
}
func (ae *AppEnv) IsEdgeRouterOnline(id string) bool {
return ae.Broker.GetOnlineEdgeRouter(id) != nil
return ae.Broker.IsEdgeRouterOnline(id)
}
func (ae *AppEnv) GetApiClientCsrSigner() cert.Signer {
controller/env/broker.go
Vendored
+74 -704
View File
@@ -17,22 +17,14 @@
package env
import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/kataras/go-events"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/build"
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/storage/boltz"
"github.com/openziti/foundation/util/concurrenz"
"go.etcd.io/bbolt"
"strings"
"sync"
"time"
)
const (
@@ -44,382 +36,103 @@ const (
ApiSessionRemovedType = int32(edge_ctrl_pb.ContentType_ApiSessionRemovedType)
ApiSessionAddedType = int32(edge_ctrl_pb.ContentType_ApiSessionAddedType)
ApiSessionUpdatedType = int32(edge_ctrl_pb.ContentType_ApiSessionUpdatedType)
RequestClientReSyncType = int32(edge_ctrl_pb.ContentType_RequestClientReSyncType)
ServerHelloType = int32(edge_ctrl_pb.ContentType_ServerHelloType)
ClientHelloType = int32(edge_ctrl_pb.ContentType_ClientHelloType)
)
type edgeRouterMap struct {
internalMap *sync.Map //edgeRouterId -> edgeRouterEntry
}
type edgeRouterEntry struct {
EdgeRouter *model.EdgeRouter
Channel channel2.Channel
send chan *channel2.Message
stop chan interface{}
running concurrenz.AtomicBoolean
stopping concurrenz.AtomicBoolean
}
func newEdgeRouterEntry(router *model.EdgeRouter, ch channel2.Channel, sendBufferSize int) *edgeRouterEntry {
return &edgeRouterEntry{
EdgeRouter: router,
Channel: ch,
send: make(chan *channel2.Message, sendBufferSize),
stop: make(chan interface{}, 0),
running: concurrenz.AtomicBoolean(0),
stopping: concurrenz.AtomicBoolean(0),
}
}
func (entry *edgeRouterEntry) Start() {
running := entry.running.Get()
if !running {
entry.running.Set(true)
go entry.run()
}
}
func (entry *edgeRouterEntry) Stop() {
stopping := entry.stopping.Get()
if !stopping {
entry.stopping.Set(true)
go func() {
entry.stop <- struct{}{}
}()
}
}
func (entry *edgeRouterEntry) run() {
for {
select {
case <-entry.stop:
entry.running.Set(false)
entry.stopping.Set(false)
return
case msg := <-entry.send:
if !entry.Channel.IsClosed() {
_ = entry.Channel.Send(msg)
}
}
}
}
func (entry *edgeRouterEntry) Send(msg *channel2.Message) {
entry.send <- msg
}
func (m *edgeRouterMap) AddEntry(edgeRouterEntry *edgeRouterEntry) {
m.internalMap.Store(edgeRouterEntry.EdgeRouter.Id, edgeRouterEntry)
edgeRouterEntry.Start()
}
func (m *edgeRouterMap) GetEntry(edgeRouterId string) *edgeRouterEntry {
val, found := m.internalMap.Load(edgeRouterId)
if !found {
return nil
}
return val.(*edgeRouterEntry)
}
func (m *edgeRouterMap) GetOnlineEntries() []*edgeRouterEntry {
var entries []*edgeRouterEntry
m.internalMap.Range(func(_, vi interface{}) bool {
if entry, ok := vi.(*edgeRouterEntry); ok && !entry.Channel.IsClosed() {
entries = append(entries, entry)
} else {
panic("edge router map contains a non *edgeRouterEntry value")
}
return true
})
return entries
}
func (m *edgeRouterMap) RemoveEntry(edgeRouterId string) {
entry := m.GetEntry(edgeRouterId)
if entry != nil {
entry.Stop()
m.internalMap.Delete(edgeRouterId)
}
}
func (m *edgeRouterMap) RangeEdgeRouterEntries(f func(entries *edgeRouterEntry) bool) {
m.internalMap.Range(func(edgeRouterId, value interface{}) bool {
if edgeRouterEntry, ok := value.(*edgeRouterEntry); ok {
return f(edgeRouterEntry)
}
pfxlog.Logger().Panic("could not convert edge router entry")
return false
})
}
// The Broker delegates Ziti Edge events to a RouterSyncStrategy. Handling the details of which events to watch
// and dealing with casting arguments to their proper concrete types.
type Broker struct {
ae *AppEnv
events map[events.EventEmmiter]map[events.EventName][]events.Listener
edgeRouterMap *edgeRouterMap
sessionChunkSize int
apiSessionChunkSize int
routerMsgBufferSize int
routerSyncStrategy RouterSyncStrategy
}
func NewBroker(ae *AppEnv) *Broker {
b := &Broker{
ae: ae,
edgeRouterMap: &edgeRouterMap{
internalMap: &sync.Map{},
},
func NewBroker(ae *AppEnv, synchronizer RouterSyncStrategy) *Broker {
broker := &Broker{
ae: ae,
routerSyncStrategy: synchronizer,
sessionChunkSize: 100,
apiSessionChunkSize: 100,
routerMsgBufferSize: 100,
}
b.events = map[events.EventEmmiter]map[events.EventName][]events.Listener{
b.ae.GetStores().Session: {
boltz.EventDelete: []events.Listener{
b.sessionDeleteEventHandler,
},
boltz.EventCreate: []events.Listener{
b.sessionCreateEventHandler,
},
},
b.ae.GetStores().ApiSession: {
boltz.EventCreate: []events.Listener{
b.apiSessionCreateEventHandler,
},
boltz.EventDelete: []events.Listener{
b.apiSessionDeleteEventHandler,
},
},
b.ae.GetStores().ApiSessionCertificate: {
boltz.EventCreate: []events.Listener{
b.apiSessionCertificateEventHandler,
},
boltz.EventDelete: []events.Listener{
b.apiSessionCertificateEventHandler,
},
},
}
broker.ae.GetStores().Session.AddListener(boltz.EventDelete, broker.sessionDeleted)
broker.ae.GetStores().Session.AddListener(boltz.EventCreate, broker.sessionCreated)
broker.ae.GetStores().ApiSession.AddListener(boltz.EventCreate, broker.apiSessionCreated)
broker.ae.GetStores().ApiSession.AddListener(boltz.EventDelete, broker.apiSessionDeleted)
broker.ae.GetStores().ApiSessionCertificate.AddListener(boltz.EventCreate, broker.apiSessionCertificateHandler)
broker.ae.GetStores().ApiSessionCertificate.AddListener(boltz.EventDelete, broker.apiSessionCertificateHandler)
b.registerEventHandlers()
ae.HostController.GetNetwork().AddRouterPresenceHandler(broker)
ae.HostController.GetNetwork().AddRouterPresenceHandler(b)
return b
return broker
}
func (b *Broker) AddEdgeRouter(ch channel2.Channel, edgeRouter *model.EdgeRouter) {
logger := pfxlog.Logger()
edgeRouterEntry := newEdgeRouterEntry(edgeRouter, ch, b.routerMsgBufferSize)
edgeRouterEntry.Start()
b.edgeRouterMap.AddEntry(edgeRouterEntry)
//stream so we don't hold hundreds of thousands in memory at once
beforeFullStreamApiSessionTime := time.Now()
err := b.streamApiSessions(b.apiSessionChunkSize, func(addedMsg *edge_ctrl_pb.ApiSessionAdded) {
logger.Infof("sending edge router session updates [%d]", len(addedMsg.ApiSessions))
if buf, err := proto.Marshal(addedMsg); err == nil {
beforeApiSessionSend := time.Now()
msg := channel2.NewMessage(ApiSessionAddedType, buf)
edgeRouterEntry.Send(msg)
apiSessionSendDelta := time.Now().Sub(beforeApiSessionSend)
logger.Debugf("broker api session send timing (micro-sec): %v\n", apiSessionSendDelta.Microseconds())
} else {
logger.WithError(err).Error("error sending session added, could not marshal message content")
func (broker *Broker) RouterConnected(router *network.Router) {
go func() {
log := pfxlog.Logger().WithField("routerId", router.Id).WithField("routerName", router.Name).WithField("routerFingerprint", router.Fingerprint)
if router.Fingerprint != nil {
if edgeRouter, _ := broker.ae.Handlers.EdgeRouter.ReadOneByFingerprint(*router.Fingerprint); edgeRouter != nil {
pfxlog.Logger().WithField("routerId", router.Id).
WithField("routerName", router.Name).
WithField("routerFingerprint", router.Fingerprint).
Infof("broker detected edge router with id %s connecting", router.Id)
broker.routerSyncStrategy.RouterConnected(edgeRouter, router)
} else {
log.Debugf("broker detected non-edge router with id %s connecting", router.Id)
}
}
})
fullStreamApiSessionDelta := time.Now().Sub(beforeFullStreamApiSessionTime)
logger.Debugf("broker api session FULL stream timing (micro-sec): %v\n", fullStreamApiSessionDelta.Microseconds())
if err != nil {
logger.WithError(err).Error("could not get stream sessions")
return
}
beforeFullStreamSessionTime := time.Now()
err = b.streamSessions(edgeRouter.Id, b.sessionChunkSize, func(addedMsg *edge_ctrl_pb.SessionAdded) {
logger.Infof("sending edge router network session updates [%d]", len(addedMsg.Sessions))
if buf, err := proto.Marshal(addedMsg); err == nil {
beforeSessionSend := time.Now()
msg := channel2.NewMessage(SessionAddedType, buf)
edgeRouterEntry.Send(msg)
sessionSendDelta := time.Now().Sub(beforeSessionSend)
logger.Debugf(fmt.Sprintf("broker session send timing (micro-sec): %v\n", sessionSendDelta.Microseconds()))
} else {
logger.WithError(err).Error("error sending network session added, could not marshal message content")
}
})
fullStreamSessionTimeDelta := time.Now().Sub(beforeFullStreamSessionTime)
logger.Debugf(fmt.Sprintf("broker session FULL stream timing (micro-sec): %v\n", fullStreamSessionTimeDelta.Microseconds()))
if err != nil {
logger.WithError(err).Errorf("could not stream current network sessions for edge router id [%s], connected but potentially out of sync", edgeRouter.Id)
return
}
logger.Infof("edge router connection finalized and synchronized [%s] [%s]", edgeRouter.Id, edgeRouter.Name)
}()
}
func (b *Broker) apiSessionCreateEventHandler(args ...interface{}) {
func (broker *Broker) RouterDisconnected(r *network.Router) {
// if disconnected but, by id it is still connected then it may have been a dupe
// router connecting and being disconnected
if !broker.ae.HostController.GetNetwork().ConnectedRouter(r.Id) {
go func() {
pfxlog.Logger().WithField("routerId", r.Id).
WithField("routerName", r.Name).
WithField("routerFingerprint", r.Fingerprint).
Debugf("broker detected router with id %s disconnecting", r.Id)
broker.routerSyncStrategy.RouterDisconnected(r)
}()
}
}
func (broker *Broker) apiSessionCreated(args ...interface{}) {
var apiSession *persistence.ApiSession
if len(args) == 1 {
apiSession, _ = args[0].(*persistence.ApiSession)
}
if apiSession == nil {
log := pfxlog.Logger()
log.Error("could not cast event args to event details")
pfxlog.Logger().Error("during broker apiSessionCreated could not cast arg[0] to *persistence.ApiSession")
return
}
b.sendApiSessionCreates(apiSession)
broker.routerSyncStrategy.ApiSessionAdded(apiSession)
}
func (b *Broker) sendApiSessionCreates(apiSession *persistence.ApiSession) {
apiSessionMsg := &edge_ctrl_pb.ApiSessionAdded{}
apiSessionMsg.IsFullState = false
fingerprints, err := b.getFingerprints(apiSession.IdentityId, apiSession.Id)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not get session fingerprints")
return
}
if len(fingerprints) == 0 {
return
}
apiSessionMsg.ApiSessions = append(apiSessionMsg.ApiSessions, &edge_ctrl_pb.ApiSession{
Token: apiSession.Token,
CertFingerprints: fingerprints,
Id: apiSession.Id,
})
byteMsg, err := proto.Marshal(apiSessionMsg)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not marshal api session added message")
return
}
channelMsg := channel2.NewMessage(ApiSessionAddedType, byteMsg)
b.sendToAllEdgeRouters(channelMsg)
}
// todo: use this once cert rolling happens
func (b *Broker) apiSessionUpdateEventHandler(args ...interface{}) {
func (broker *Broker) apiSessionDeleted(args ...interface{}) {
var apiSession *persistence.ApiSession
if len(args) == 1 {
apiSession, _ = args[0].(*persistence.ApiSession)
}
if apiSession == nil {
log := pfxlog.Logger()
log.Error("could not cast event args to event details")
pfxlog.Logger().Error("during broker apiSessionDeleted could not cast arg[0] to *persistence.ApiSession")
return
}
b.sendApiSessionUpdates(apiSession)
broker.routerSyncStrategy.ApiSessionDeleted(apiSession)
}
func (b *Broker) sendApiSessionUpdates(apiSession *persistence.ApiSession) {
apiSessionMsg := &edge_ctrl_pb.ApiSessionUpdated{}
fingerprints, err := b.getFingerprints(apiSession.IdentityId, apiSession.Id)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not get api session fingerprints")
return
}
if len(fingerprints) == 0 {
pfxlog.Logger().WithError(err).Debug("api session has no fingerprints, not sending to edge routers")
return
}
apiSessionMsg.ApiSessions = append(apiSessionMsg.ApiSessions, &edge_ctrl_pb.ApiSession{
Token: apiSession.Token,
CertFingerprints: fingerprints,
Id: apiSession.Id,
})
byteMsg, err := proto.Marshal(apiSessionMsg)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not marshal api session updated message")
return
}
channelMsg := channel2.NewMessage(ApiSessionUpdatedType, byteMsg)
b.sendToAllEdgeRouters(channelMsg)
}
func (b *Broker) sendToAllEdgeRouters(msg *channel2.Message) {
b.edgeRouterMap.RangeEdgeRouterEntries(func(edgeRouterEntry *edgeRouterEntry) bool {
edgeRouterEntry.Send(msg)
return true
})
}
func (b *Broker) sendToAllEdgeRoutersForSession(sessionId string, msg *channel2.Message) {
edgeRouterList, err := b.ae.Handlers.EdgeRouter.ListForSession(sessionId)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not get edge routers for session [%s]", sessionId)
return
}
for _, edgeRouter := range edgeRouterList.EdgeRouters {
edgeRouterEntry := b.edgeRouterMap.GetEntry(edgeRouter.Id)
if edgeRouterEntry != nil && !edgeRouterEntry.Channel.IsClosed() {
edgeRouterEntry.Send(msg)
}
}
}
func (b *Broker) apiSessionDeleteEventHandler(args ...interface{}) {
var apiSession *persistence.ApiSession
if len(args) == 1 {
apiSession, _ = args[0].(*persistence.ApiSession)
}
if apiSession == nil {
log := pfxlog.Logger()
log.Error("could not cast event args to event details")
return
}
b.sendApiSessionDeletes(apiSession)
}
func (b *Broker) sendApiSessionDeletes(apiSession *persistence.ApiSession) {
apiSessionMsg := &edge_ctrl_pb.ApiSessionRemoved{}
apiSessionMsg.Tokens = append(apiSessionMsg.Tokens, apiSession.Token)
byteMsg, err := proto.Marshal(apiSessionMsg)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not marshal session removed message")
return
}
channelMsg := channel2.NewMessage(ApiSessionRemovedType, byteMsg)
b.sendToAllEdgeRouters(channelMsg)
}
func (b *Broker) sessionDeleteEventHandler(args ...interface{}) {
func (broker *Broker) sessionCreated(args ...interface{}) {
var session *persistence.Session
if len(args) == 1 {
session, _ = args[0].(*persistence.Session)
@@ -431,400 +144,57 @@ func (b *Broker) sessionDeleteEventHandler(args ...interface{}) {
return
}
b.sendSessionDeletes(session)
broker.routerSyncStrategy.SessionAdded(session)
}
func (b *Broker) sendSessionDeletes(session *persistence.Session) {
sessionsRemoved := &edge_ctrl_pb.SessionRemoved{}
sessionsRemoved.Tokens = append(sessionsRemoved.Tokens, session.Token)
if buf, err := proto.Marshal(sessionsRemoved); err == nil {
msg := channel2.NewMessage(SessionRemovedType, buf)
// can't use sendToAllEdgeRoutersForSession b/c the session is gone
b.sendToAllEdgeRouters(msg)
} else {
pfxlog.Logger().WithError(err).Error("error sending session removed, could not marshal message content")
}
}
func (b *Broker) sessionCreateEventHandler(args ...interface{}) {
func (broker *Broker) sessionDeleted(args ...interface{}) {
var session *persistence.Session
if len(args) == 1 {
session, _ = args[0].(*persistence.Session)
}
if session == nil {
log := pfxlog.Logger()
log.Error("could not cast event args to event details")
pfxlog.Logger().Error("during broker sessionDeleted could not cast arg[0] to *persistence.Session")
return
}
b.sendSessionCreates(session)
broker.routerSyncStrategy.SessionDeleted(session)
}
func (b *Broker) sendSessionCreates(session *persistence.Session) {
sessionAdded := &edge_ctrl_pb.SessionAdded{}
service, err := b.ae.Handlers.EdgeService.Read(session.ServiceId)
if err != nil {
pfxlog.Logger().WithError(err).Error("could not send network session added, could not find service")
return
}
fps, err := b.getFingerprintsByApiSessionId(session.ApiSessionId)
if err != nil {
pfxlog.Logger().Errorf("could not obtain a fingerprint for the api session [%s] and session [%s]", session.ApiSessionId, session.Id)
return
}
svc, err := b.modelServiceToProto(service)
if err != nil {
pfxlog.Logger().Errorf("could not convert service [%s] to proto: %s", service.Id, err)
return
}
sessionType := edge_ctrl_pb.SessionType_Dial
if session.Type == persistence.SessionTypeBind {
sessionType = edge_ctrl_pb.SessionType_Bind
}
sessionAdded.Sessions = append(sessionAdded.Sessions, &edge_ctrl_pb.Session{
Id: session.Id,
Token: session.Token,
Service: svc,
CertFingerprints: fps,
Type: sessionType,
ApiSessionId: session.ApiSessionId,
})
if buf, err := proto.Marshal(sessionAdded); err == nil {
msg := channel2.NewMessage(SessionAddedType, buf)
b.sendToAllEdgeRoutersForSession(session.Id, msg)
} else {
pfxlog.Logger().WithError(err).Error("error sending network session added, could not marshal message content")
}
}
func (b *Broker) modelServiceToProto(service *model.Service) (*edge_ctrl_pb.Service, error) {
return &edge_ctrl_pb.Service{
Name: service.Name,
Id: service.Id,
EncryptionRequired: service.EncryptionRequired,
}, nil
}
func (b *Broker) registerEventHandlers() {
for emitter, eventNameMap := range b.events {
for en, ls := range eventNameMap {
for _, l := range ls {
emitter.AddListener(en, l)
}
}
}
}
func (b *Broker) streamApiSessions(chunkSize int, callback func(addedMsg *edge_ctrl_pb.ApiSessionAdded)) error {
ret := &edge_ctrl_pb.ApiSessionAdded{
IsFullState: true,
}
logger := pfxlog.Logger()
return b.ae.GetHandlers().ApiSession.Stream("true", func(apiSession *model.ApiSession, err error) error {
if err != nil {
logger.Errorf("error reading api sessions for router: %v", err)
return nil
}
if apiSession == nil {
//done
callback(ret)
return nil
}
apiSessionProto, err := b.modelApiSessionToProto(apiSession.Token, apiSession.IdentityId, apiSession.Id)
if err != nil {
logger.Errorf("error converting api session to proto: %v", err)
return nil
}
ret.ApiSessions = append(ret.ApiSessions, apiSessionProto)
if len(ret.ApiSessions) > chunkSize {
callback(ret)
ret = &edge_ctrl_pb.ApiSessionAdded{}
}
return nil
})
}
func (b *Broker) streamSessions(edgeRouterId string, chunkSize int, callback func(*edge_ctrl_pb.SessionAdded)) error {
logger := pfxlog.Logger()
ret := &edge_ctrl_pb.SessionAdded{
IsFullState: true,
}
return b.ae.Handlers.Session.StreamAll(func(session *model.Session, err error) error {
if err != nil {
logger.Errorf("error reading sessions for router [%s]: %v", edgeRouterId, err)
return nil
}
if session == nil {
//done
callback(ret)
return nil
}
sessionProto, err := b.modelSessionToProto(session)
if err != nil {
logger.Errorf("error converting session to proto: %v", err)
return nil
}
ret.Sessions = append(ret.Sessions, sessionProto)
if len(ret.Sessions) > chunkSize {
callback(ret)
ret = &edge_ctrl_pb.SessionAdded{}
}
return nil
})
}
func (b *Broker) getFingerprints(identityId, apiSessionId string) ([]string, error) {
identityPrints, err := b.getIdentityAuthenticatorFingerprints(identityId)
if err != nil {
return nil, err
}
apiSessionPrints, err := b.getApiSessionCertificateFingerprints(apiSessionId)
if err != nil {
return nil, err
}
for _, apiSessionPrint := range apiSessionPrints {
identityPrints = append(identityPrints, apiSessionPrint)
}
return identityPrints, nil
}
func (b *Broker) getFingerprintsByApiSessionId(apiSessionId string) ([]string, error) {
apiSession, err := b.ae.GetHandlers().ApiSession.Read(apiSessionId)
if err != nil {
return nil, fmt.Errorf("could not query fingerprints by api session id [%s]: %s", apiSessionId, err)
}
return b.getFingerprints(apiSession.IdentityId, apiSessionId)
}
func (b *Broker) getIdentityAuthenticatorFingerprints(identityId string) ([]string, error) {
fingerprintsMap := map[string]struct{}{}
err := b.ae.Handlers.Identity.CollectAuthenticators(identityId, func(authenticator *model.Authenticator) error {
for _, authPrint := range authenticator.Fingerprints() {
fingerprintsMap[authPrint] = struct{}{}
}
return nil
})
if err != nil {
return nil, err
}
var fingerprints []string
for fingerprint := range fingerprintsMap {
fingerprints = append(fingerprints, fingerprint)
}
return fingerprints, nil
}
func (b *Broker) getApiSessionCertificateFingerprints(apiSessionId string) ([]string, error) {
apiSessionCerts, err := b.ae.GetHandlers().ApiSessionCertificate.ReadByApiSessionId(apiSessionId)
if err != nil {
return nil, err
}
var validPrints []string
now := time.Now()
for _, apiSessionCert := range apiSessionCerts {
if apiSessionCert.ValidAfter != nil && now.After(*apiSessionCert.ValidAfter) &&
apiSessionCert.ValidBefore != nil && now.Before(*apiSessionCert.ValidBefore) {
validPrints = append(validPrints, apiSessionCert.Fingerprint)
}
}
return validPrints, nil
}
func (b *Broker) modelApiSessionToProto(token, identityId, apiSessionId string) (*edge_ctrl_pb.ApiSession, error) {
fingerprints, err := b.getFingerprints(identityId, apiSessionId)
if err != nil {
return nil, err
}
return &edge_ctrl_pb.ApiSession{
Token: token,
CertFingerprints: fingerprints,
Id: apiSessionId,
}, nil
}
func (b *Broker) modelSessionToProto(ns *model.Session) (*edge_ctrl_pb.Session, error) {
service, err := b.ae.Handlers.EdgeService.Read(ns.ServiceId)
if err != nil {
return nil, fmt.Errorf("could not convert to session proto, could not find service: %s", err)
}
fps, err := b.getFingerprintsByApiSessionId(ns.ApiSessionId)
if err != nil {
return nil, fmt.Errorf("could not get fingerprints for network session: %s", err)
}
svc, err := b.modelServiceToProto(service)
if err != nil {
return nil, fmt.Errorf("could not convert service [%s] to proto: %s", service.Id, err)
}
sessionType := edge_ctrl_pb.SessionType_Dial
if ns.Type == persistence.SessionTypeBind {
sessionType = edge_ctrl_pb.SessionType_Bind
}
return &edge_ctrl_pb.Session{
Id: ns.Id,
Token: ns.Token,
Service: svc,
CertFingerprints: fps,
Type: sessionType,
ApiSessionId: ns.ApiSessionId,
}, nil
}
func (b *Broker) GetOnlineEdgeRouter(id string) *model.EdgeRouter {
result, found := b.edgeRouterMap.internalMap.Load(id)
if !found {
return nil
}
entry, ok := result.(*edgeRouterEntry)
if !ok || entry.Channel.IsClosed() {
return nil
}
return entry.EdgeRouter
}
func (b *Broker) RouterConnected(r *network.Router) {
go func() {
if r.Fingerprint != nil {
if edgeRouter, _ := b.ae.Handlers.EdgeRouter.ReadOneByFingerprint(*r.Fingerprint); edgeRouter != nil {
b.sendHello(r)
}
}
}()
}
func (b *Broker) sendHello(r *network.Router) {
serverVersion := build.GetBuildInfo().Version()
serverHello := &edge_ctrl_pb.ServerHello{
Version: serverVersion,
}
buf, err := proto.Marshal(serverHello)
if err != nil {
pfxlog.Logger().WithError(err).Error("could not marshal serverHello")
return
}
if err = r.Control.SendWithTimeout(channel2.NewMessage(ServerHelloType, buf), 10*time.Second); err != nil {
pfxlog.Logger().WithError(err).Error("timed out sending serverHello message for edge router")
return
}
}
func (b *Broker) ReceiveHello(r *network.Router, edgeRouter *model.EdgeRouter, respHello *edge_ctrl_pb.ClientHello) {
serverVersion := build.GetBuildInfo().Version()
entry := pfxlog.Logger().
WithField("hostname", respHello.Hostname).
WithField("protocols", respHello.Protocols).
WithField("protocolPorts", respHello.ProtocolPorts).
WithField("data", respHello.Data)
if r.VersionInfo != nil {
entry.WithField("version", r.VersionInfo.Version).
WithField("revision", r.VersionInfo.Revision).
WithField("buildDate", r.VersionInfo.BuildDate).
WithField("os", r.VersionInfo.OS).
WithField("arch", r.VersionInfo.Arch)
}
entry.Info("edge router responded with client hello")
protocols := map[string]string{}
for _, p := range respHello.ProtocolPorts {
parts := strings.Split(p, ":")
ingressUrl := fmt.Sprintf("%s://%s:%s", parts[0], respHello.Hostname, parts[1])
protocols[parts[0]] = ingressUrl
}
edgeRouter.Hostname = &respHello.Hostname
edgeRouter.EdgeRouterProtocols = protocols
edgeRouter.VersionInfo = r.VersionInfo
//todo: restrict version?
pfxlog.Logger().Infof("edge router connecting with version [%s] to controller with version [%s]", respHello.Version, serverVersion)
b.AddEdgeRouter(r.Control, edgeRouter)
}
func (b *Broker) RouterDisconnected(r *network.Router) {
go func() {
//ensure this wasn't the same router being started multiple times and the 2nd connection being refused w/ a close
if r.Fingerprint != nil && !b.ae.HostController.GetNetwork().ConnectedRouter(r.Id) {
if edgeRouter, _ := b.ae.Handlers.EdgeRouter.ReadOneByFingerprint(*r.Fingerprint); edgeRouter != nil {
b.edgeRouterMap.RemoveEntry(edgeRouter.Id)
}
}
}()
}
func (b *Broker) apiSessionCertificateEventHandler(args ...interface{}) {
func (broker *Broker) apiSessionCertificateHandler(args ...interface{}) {
var apiSessionCert *persistence.ApiSessionCertificate
if len(args) == 1 {
apiSessionCert, _ = args[0].(*persistence.ApiSessionCertificate)
}
if apiSessionCert == nil {
log := pfxlog.Logger()
log.Error("could not cast event args to event details: create/delete api session cert")
pfxlog.Logger().Error("during broker apiSessionCertificateEvent could not cast arg[0] to *persistence.ApiSessionCertificate")
return
}
var apiSession *persistence.ApiSession
var err error
err = b.ae.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error {
apiSession, err = b.ae.GetStores().ApiSession.LoadOneById(tx, apiSessionCert.ApiSessionId)
err = broker.ae.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error {
apiSession, err = broker.ae.GetStores().ApiSession.LoadOneById(tx, apiSessionCert.ApiSessionId)
return err
})
if err != nil {
pfxlog.Logger().WithError(err).Error("could not process API Session certificate creation. Failed to query for parent API Session")
pfxlog.Logger().WithError(err).Error("could not process API Session certificate event, failed to query for parent API Session")
return
}
b.sendApiSessionUpdates(apiSession)
broker.routerSyncStrategy.ApiSessionUpdated(apiSession, apiSessionCert)
}
func (broker *Broker) IsEdgeRouterOnline(id string) bool {
er, _ := broker.GetOnlineEdgeRouter(id)
return er != nil
}
func (broker *Broker) GetOnlineEdgeRouter(id string) (*model.EdgeRouter, RouterSyncStatus) {
return broker.routerSyncStrategy.GetOnlineEdgeRouter(id)
}
func (broker *Broker) Stop() {
broker.routerSyncStrategy.Stop()
}
controller/env/sync.go
Vendored
+78
View File
@@ -0,0 +1,78 @@
/*
Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package env
import (
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/fabric/controller/network"
)
// Aliased type for router strategies
type RouterSyncStrategyType string
// Aliased type for router sync status
type RouterSyncStatus string
const (
RouterSyncNew RouterSyncStatus = "SYNC_NEW" //connection accepted but no strategy actions have been taken
RouterSyncQueued RouterSyncStatus = "SYNC_QUEUED" //connection handed to strategy, but not processed
RouterSyncHello RouterSyncStatus = "SYNC_HELLO" //connection is beginning hello cycle
RouterSyncHelloWait RouterSyncStatus = "SYNC_HELLO_WAIT" //hello received from router, but there are too many synchronizing routers
RouterSyncResyncWait RouterSyncStatus = "SYNC_RESYNC_WAIT" //router requested a resync, in queue
RouterSynInProgress RouterSyncStatus = "SYNC_IN_PROGRESS" //hello finished, starting to send state
RouterSyncDone RouterSyncStatus = "SYNC_DONE" //initial state sent
//Error states
RouterSyncUnknown RouterSyncStatus = "SYNC_UNKNOWN" //the router is currently unknown
RouterSyncDisconnected RouterSyncStatus = "SYNC_DISCONNECTED" //strategy was disconnected before finishing
RouterSyncHelloTimeout RouterSyncStatus = "SYNC_HELLO_TIMEOUT" //sync failed due to a hello timeout.
//msg headers
SyncStrategyTypeHeader = 1013
SyncStrategyStateHeader = 1014
)
// RouterSyncStrategy handles the life cycle of an Edge Router connecting to the controller, synchronizing
// any upfront state and then maintaining state after that.
type RouterSyncStrategy interface {
Type() RouterSyncStrategyType
Status(id string) RouterSyncStatus
GetOnlineEdgeRouter(id string) (*model.EdgeRouter, RouterSyncStatus)
Stop()
RouterConnectionHandler
RouterSynchronizerEventHandler
}
// RouterConnectionHandler is responsible for handling router connect/disconnect for synchronizing state.
// This is intended for API Session but additional state is possible. Implementations may bind additional
// handlers to the channel.
type RouterConnectionHandler interface {
RouterConnected(edgeRouter *model.EdgeRouter, router *network.Router)
RouterDisconnected(router *network.Router)
}
// RouterSynchronizerEventHandler is responsible for keeping Edge Routers up to date on API Sessions
type RouterSynchronizerEventHandler interface {
ApiSessionAdded(apiSession *persistence.ApiSession)
ApiSessionUpdated(apiSession *persistence.ApiSession, apiSessionCert *persistence.ApiSessionCertificate)
ApiSessionDeleted(apiSession *persistence.ApiSession)
//todo: To be removed inline session creation
SessionAdded(session *persistence.Session)
SessionDeleted(session *persistence.Session)
}
controller/handler_edge_ctrl/hello.go
+7 -16
View File
@@ -21,16 +21,19 @@ import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
)
type helloHandler struct {
appEnv *env.AppEnv
appEnv *env.AppEnv
callback func(r *network.Router, respHello *edge_ctrl_pb.ClientHello)
}
func NewHelloHandler(appEnv *env.AppEnv) *helloHandler {
func NewHelloHandler(appEnv *env.AppEnv, callback func(r *network.Router, respHello *edge_ctrl_pb.ClientHello)) *helloHandler {
return &helloHandler{
appEnv: appEnv,
appEnv: appEnv,
callback: callback,
}
}
@@ -52,17 +55,5 @@ func (h *helloHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel)
return
}
if r.Fingerprint == nil {
pfxlog.Logger().Errorf("router %v is not enrolled, closing channel", ch.Id().Token)
_ = ch.Close()
return
}
edgeRouter, _ := h.appEnv.Handlers.EdgeRouter.ReadOneByFingerprint(*r.Fingerprint)
if edgeRouter == nil {
pfxlog.Logger().Errorf("edge hello received for non-edge router %v", ch.Id().Token)
return
}
h.appEnv.Broker.ReceiveHello(r, edgeRouter, respHello)
h.callback(r, respHello)
}
controller/handler_edge_ctrl/resync.go
+59
View File
@@ -0,0 +1,59 @@
/*
Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package handler_edge_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
)
type resyncHandler struct {
appEnv *env.AppEnv
callback func(r *network.Router, respHello *edge_ctrl_pb.RequestClientReSync)
}
func NewResyncHandler(appEnv *env.AppEnv, callback func(r *network.Router, respHello *edge_ctrl_pb.RequestClientReSync)) *resyncHandler {
return &resyncHandler{
appEnv: appEnv,
callback: callback,
}
}
func (h *resyncHandler) ContentType() int32 {
return env.RequestClientReSyncType
}
func (h *resyncHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
resyncReq := &edge_ctrl_pb.RequestClientReSync{}
if err := proto.Unmarshal(msg.Body, resyncReq); err != nil {
pfxlog.Logger().WithError(err).Error("could not unmarshal RequestClientReSync")
return
}
r, err := h.appEnv.GetHostController().GetNetwork().GetRouter(ch.Id().Token)
if err != nil {
pfxlog.Logger().WithError(err).Errorf("could not find router %v, closing channel", ch.Id().Token)
_ = ch.Close()
return
}
h.callback(r, resyncReq)
}
controller/internal/routes/current_identity_api_model.go
+1 -1
View File
@@ -163,7 +163,7 @@ func MapCurrentIdentityEdgeRouterToRestEntity(ae *env.AppEnv, _ *response.Reques
func MapCurrentIdentityEdgeRouterToRestModel(ae *env.AppEnv, router *model.EdgeRouter) (*rest_model.CurrentIdentityEdgeRouterDetail, error) {
hostname := ""
protocols := map[string]string{}
onlineEdgeRouter := ae.Broker.GetOnlineEdgeRouter(router.Id)
onlineEdgeRouter, _ := ae.Broker.GetOnlineEdgeRouter(router.Id)
isOnline := onlineEdgeRouter != nil
controller/internal/routes/edge_router_api_model.go
+4 -1
View File
@@ -119,7 +119,8 @@ func MapEdgeRouterToRestModel(ae *env.AppEnv, router *model.EdgeRouter) (*rest_m
revision := ""
version := ""
onlineEdgeRouter := ae.Broker.GetOnlineEdgeRouter(router.Id)
onlineEdgeRouter, syncStatus := ae.Broker.GetOnlineEdgeRouter(router.Id)
syncStatusStr := string(syncStatus)
isOnline := onlineEdgeRouter != nil
@@ -134,6 +135,7 @@ func MapEdgeRouterToRestModel(ae *env.AppEnv, router *model.EdgeRouter) (*rest_m
revision = onlineEdgeRouter.VersionInfo.Revision
version = onlineEdgeRouter.VersionInfo.Version
}
}
ret := &rest_model.EdgeRouterDetail{
@@ -149,6 +151,7 @@ func MapEdgeRouterToRestModel(ae *env.AppEnv, router *model.EdgeRouter) (*rest_m
Fingerprint: stringz.OrEmpty(router.Fingerprint),
Hostname: &hostname,
SupportedProtocols: protocols,
SyncStatus: &syncStatusStr,
VersionInfo: &rest_model.VersionInfo{
Os: &os,
Arch: &arch,
controller/internal/routes/session_api_model.go
+1 -1
View File
@@ -122,7 +122,7 @@ func getSessionEdgeRouters(ae *env.AppEnv, ns *model.Session) ([]*rest_model.Ses
}
for _, edgeRouter := range edgeRoutersForSession.EdgeRouters {
onlineEdgeRouter := ae.Broker.GetOnlineEdgeRouter(edgeRouter.Id)
onlineEdgeRouter, _ := ae.Broker.GetOnlineEdgeRouter(edgeRouter.Id)
if onlineEdgeRouter != nil {
restModel := &rest_model.SessionEdgeRouter{
controller/model/api_session_handlers.go
+2
View File
@@ -18,6 +18,7 @@ package model
import (
"fmt"
"github.com/lucsky/cuid"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/fabric/controller/models"
"github.com/openziti/foundation/storage/ast"
@@ -43,6 +44,7 @@ func (handler *ApiSessionHandler) newModelEntity() boltEntitySink {
}
func (handler *ApiSessionHandler) Create(entity *ApiSession) (string, error) {
entity.Id = cuid.New() //use cuids which are longer than shortids but are monotonic
return handler.createEntity(entity)
}
controller/model/session_handlers.go
+2
View File
@@ -18,6 +18,7 @@ package model
import (
"fmt"
"github.com/lucsky/cuid"
"github.com/openziti/fabric/controller/models"
"github.com/openziti/foundation/storage/ast"
"github.com/openziti/foundation/storage/boltz"
@@ -41,6 +42,7 @@ func (handler *SessionHandler) newModelEntity() boltEntitySink {
}
func (handler *SessionHandler) Create(entity *Session) (string, error) {
entity.Id = cuid.New() //use cuids which are longer than shortids but are monotonic
return handler.createEntity(entity)
}
controller/server/controller.go
+12 -2
View File
@@ -23,6 +23,7 @@ import (
"github.com/openziti/edge/controller"
"github.com/openziti/edge/controller/apierror"
"github.com/openziti/edge/controller/response"
sync2 "github.com/openziti/edge/controller/sync_strats"
"github.com/openziti/edge/controller/timeout"
"github.com/openziti/edge/rest_server"
"github.com/openziti/fabric/controller/xtv"
@@ -129,7 +130,6 @@ func (c *Controller) SetHostController(h env.HostController) {
func (c *Controller) GetCtrlHandlers() []channel2.ReceiveHandler {
return []channel2.ReceiveHandler{
handler_edge_ctrl.NewSessionHeartbeatHandler(c.AppEnv),
handler_edge_ctrl.NewHelloHandler(c.AppEnv),
}
}
@@ -188,7 +188,15 @@ func (c *Controller) Initialize() {
}
//after InitPersistence
c.AppEnv.Broker = env.NewBroker(c.AppEnv)
c.AppEnv.Broker = env.NewBroker(c.AppEnv, sync2.NewInstantStrategy(c.AppEnv, sync2.InstantStrategyOptions{
MaxQueuedRouterConnects: 100,
MaxQueuedClientHellos: 100,
RouterConnectWorkerCount: 10,
SyncWorkerCount: 10,
RouterTxBufferSize: 100,
HelloSendTimeout: 10 * time.Second,
SessionChunkSize: 100,
}))
servicePolicyEnforcer := policy.NewServicePolicyEnforcer(c.AppEnv, policyAppWanFreq)
if err := c.policyEngine.AddOperation(servicePolicyEnforcer); err != nil {
@@ -357,6 +365,8 @@ func (c *Controller) Shutdown() {
c.apiServer.Shutdown(ctx)
_ = c.policyEngine.Stop()
c.AppEnv.Broker.Stop()
pfxlog.Logger().Info("edge controller shutting down")
}
controller/sync_strats/marshal.go
+144
View File
@@ -0,0 +1,144 @@
/*
Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package sync_strats
import (
"fmt"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"time"
)
func apiSessionToProto(ae *env.AppEnv, token, identityId, apiSessionId string) (*edge_ctrl_pb.ApiSession, error) {
fingerprints, err := getFingerprints(ae, identityId, apiSessionId)
if err != nil {
return nil, err
}
return &edge_ctrl_pb.ApiSession{
Token: token,
CertFingerprints: fingerprints,
Id: apiSessionId,
}, nil
}
func sessionToProto(ae *env.AppEnv, session *persistence.Session) (*edge_ctrl_pb.Session, error) {
service, err := ae.Handlers.EdgeService.Read(session.ServiceId)
if err != nil {
return nil, fmt.Errorf("could not convert to session proto, could not find service: %s", err)
}
fps, err := getFingerprintsByApiSessionId(ae, session.ApiSessionId)
if err != nil {
return nil, fmt.Errorf("could not get fingerprints for network session: %s", err)
}
serviceProto := &edge_ctrl_pb.Service{
Name: service.Name,
Id: service.Id,
EncryptionRequired: service.EncryptionRequired,
}
sessionType := edge_ctrl_pb.SessionType_Dial
if session.Type == persistence.SessionTypeBind {
sessionType = edge_ctrl_pb.SessionType_Bind
}
return &edge_ctrl_pb.Session{
Id: session.Id,
Token: session.Token,
Service: serviceProto,
CertFingerprints: fps,
Type: sessionType,
ApiSessionId: session.ApiSessionId,
}, nil
}
func getFingerprintsByApiSessionId(ae *env.AppEnv, apiSessionId string) ([]string, error) {
apiSession, err := ae.GetHandlers().ApiSession.Read(apiSessionId)
if err != nil {
return nil, fmt.Errorf("could not query fingerprints by api session id [%s]: %s", apiSessionId, err)
}
return getFingerprints(ae, apiSession.IdentityId, apiSessionId)
}
func getFingerprints(ae *env.AppEnv, identityId, apiSessionId string) ([]string, error) {
identityPrints, err := getIdentityAuthenticatorFingerprints(ae, identityId)
if err != nil {
return nil, err
}
apiSessionPrints, err := getApiSessionCertificateFingerprints(ae, apiSessionId)
if err != nil {
return nil, err
}
for _, apiSessionPrint := range apiSessionPrints {
identityPrints = append(identityPrints, apiSessionPrint)
}
return identityPrints, nil
}
func getIdentityAuthenticatorFingerprints(ae *env.AppEnv, identityId string) ([]string, error) {
fingerprintsMap := map[string]struct{}{}
err := ae.Handlers.Identity.CollectAuthenticators(identityId, func(authenticator *model.Authenticator) error {
for _, authPrint := range authenticator.Fingerprints() {
fingerprintsMap[authPrint] = struct{}{}
}
return nil
})
if err != nil {
return nil, err
}
var fingerprints []string
for fingerprint := range fingerprintsMap {
fingerprints = append(fingerprints, fingerprint)
}
return fingerprints, nil
}
func getApiSessionCertificateFingerprints(ae *env.AppEnv, apiSessionId string) ([]string, error) {
apiSessionCerts, err := ae.GetHandlers().ApiSessionCertificate.ReadByApiSessionId(apiSessionId)
if err != nil {
return nil, err
}
var validPrints []string
now := time.Now()
for _, apiSessionCert := range apiSessionCerts {
if apiSessionCert.ValidAfter != nil && now.After(*apiSessionCert.ValidAfter) &&
apiSessionCert.ValidBefore != nil && now.Before(*apiSessionCert.ValidBefore) {
validPrints = append(validPrints, apiSessionCert.Fingerprint)
}
}
return validPrints, nil
}
controller/sync_strats/rtx.go
+156
View File
@@ -0,0 +1,156 @@
/*
Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package sync_strats
import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/eid"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/util/concurrenz"
"github.com/sirupsen/logrus"
"sync"
)
// RouterSender represents a connection from an Edge Router to the controller. Used
// to asynchronously buffer and send messages to an Edge Router via Start() then Send()
type RouterSender struct {
Id string
EdgeRouter *model.EdgeRouter
Router *network.Router
Status env.RouterSyncStatus
send chan *channel2.Message
stop chan interface{}
running concurrenz.AtomicBoolean
stopping concurrenz.AtomicBoolean
}
func newRouterSender(edgeRouter *model.EdgeRouter, router *network.Router, sendBufferSize int) *RouterSender {
return &RouterSender{
Id: eid.New(),
EdgeRouter: edgeRouter,
Router: router,
Status: env.RouterSyncNew,
send: make(chan *channel2.Message, sendBufferSize),
stop: make(chan interface{}, 0),
running: concurrenz.AtomicBoolean(0),
stopping: concurrenz.AtomicBoolean(0),
}
}
func (rtx *RouterSender) Start() {
if rtx.running.CompareAndSwap(false, true) {
go rtx.run()
}
}
func (rtx *RouterSender) Stop() {
if rtx.stopping.CompareAndSwap(false, true) {
go func() {
rtx.stop <- struct{}{}
}()
}
}
func (rtx *RouterSender) run() {
for {
select {
case <-rtx.stop:
rtx.running.Set(false)
rtx.stopping.Set(false)
return
case msg := <-rtx.send:
if !rtx.Router.Control.IsClosed() {
_ = rtx.Router.Control.Send(msg)
}
}
}
}
func (rtx *RouterSender) logger() *logrus.Entry {
return pfxlog.Logger().
WithField("routerTxId", rtx.Id).
WithField("routerId", rtx.Router.Id).
WithField("routerName", rtx.Router.Name).
WithField("routerFingerprint", rtx.Router.Fingerprint).
WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed())
}
func (rtx *RouterSender) Send(msg *channel2.Message) {
rtx.send <- msg
}
// Map used make working with internal RouterSender easier as sync.Map accepts and returns interface{}
type routerTxMap struct {
internalMap *sync.Map //id -> RouterSender
}
func (m *routerTxMap) Add(id string, routerMessageTxer *RouterSender) {
m.internalMap.Store(id, routerMessageTxer)
routerMessageTxer.Start()
}
func (m *routerTxMap) Get(id string) *RouterSender {
val, found := m.internalMap.Load(id)
if !found {
return nil
}
return val.(*RouterSender)
}
func (m *routerTxMap) Remove(id string) {
entry := m.Get(id)
if entry != nil {
entry.Stop()
m.internalMap.Delete(id)
}
}
func (m *routerTxMap) Range(f func(entries *RouterSender) bool) {
m.internalMap.Range(func(edgeRouterId, value interface{}) bool {
if rtx, ok := value.(*RouterSender); ok {
return f(rtx)
}
pfxlog.Logger().Panic("could not convert edge router entry")
return false
})
}
// Helper to generate channel2.ReceiveHandler instances from a contentType and function
func newReceiveHandlerFunc(contentType int32, handler func(m *channel2.Message, ch channel2.Channel)) channel2.ReceiveHandler {
return receiveHandlerFunc{
contentType: contentType,
handler: handler,
}
}
var _ channel2.ReceiveHandler = &receiveHandlerFunc{}
type receiveHandlerFunc struct {
contentType int32
handler func(m *channel2.Message, ch channel2.Channel)
}
func (r receiveHandlerFunc) ContentType() int32 {
return r.contentType
}
func (r receiveHandlerFunc) HandleReceive(m *channel2.Message, ch channel2.Channel) {
r.handler(m, ch)
}
controller/sync_strats/sync_instant.go
+562
View File
@@ -0,0 +1,562 @@
/*
Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package sync_strats
import (
"encoding/json"
"fmt"
"github.com/golang/protobuf/proto"
"github.com/lucsky/cuid"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/build"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/handler_edge_ctrl"
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/storage/ast"
"go.etcd.io/bbolt"
"strings"
"sync"
"time"
)
const (
RouterSyncStrategyInstant env.RouterSyncStrategyType = "instant"
)
var _ env.RouterSyncStrategy = &InstantStrategy{}
// Options for the instant strategy.
// - MaxQueuedRouterConnects - max number of router connected events to buffer
// - MaxQueuedClientHellos - max number of client hello messages to buffer
// - RouterConnectWorkerCount - max number of workers used to process router connections
// - SyncWorkerCount - max number of workers used to send api sessions/session data
// - RouterTxBufferSize - max number of messages buffered to be send to a router
// - HelloSendTimeout - the max amount of time per worker to wait to send hellos
// - SessionChunkSize - the number of sessions to send in each message
type InstantStrategyOptions struct {
MaxQueuedRouterConnects int32
MaxQueuedClientHellos int32
RouterConnectWorkerCount int32
SyncWorkerCount int32
RouterTxBufferSize int
HelloSendTimeout time.Duration
SessionChunkSize int
}
// Original API Session synchronization implementation. Assumes that on connect, the router requires and instant
// and full set of API Sessions. Send individual create, update, delete events for sessions after synchronization.
//
// This strategy uses a series of queus and workers to managed sychronization state. The order of events is as follows:
// 1. An edge router connects to the controller, triggering RouterConnected()
// 2. A RouterSender is created encapsulating the Edge Router, Router, and Sync State
// 3. The RouterSender is queued on the routerConnectedQueue channel which buffers up to options.MaxQueuedRouterConnects
// 4. The routerConnectedQueue is read and the edge server hello is sent
// 5. The controller waits for a client hello to be recieved via ReceiveClientHello message
// 6. The client hello is used to identity the RouterSender associated with the client and is queued on
// the receivedClientHelloQueue channel which buffers up to options.MaxQueuedClientHellos
// 7. A startSynchronizeWorker will pick up the RouterSender from the receivedClientHelloQueue and being to
// send data to the edge router via the RouterSender
type InstantStrategy struct {
InstantStrategyOptions
rtxMap *routerTxMap
helloHandler channel2.ReceiveHandler
resyncHandler channel2.ReceiveHandler
ae *env.AppEnv
routerConnectedQueue chan *RouterSender
receivedClientHelloQueue chan *RouterSender
stop chan struct{}
}
func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *InstantStrategy {
if options.MaxQueuedRouterConnects <= 0 {
pfxlog.Logger().Panicf("MaxQueuedRouterConnects for InstantStrategy cannot be less than 1, got %d", options.MaxQueuedRouterConnects)
}
if options.MaxQueuedClientHellos <= 0 {
pfxlog.Logger().Panicf("MaxQueuedClientHellos for InstantStrategy cannot be less than 1, got %d", options.MaxQueuedClientHellos)
}
if options.RouterConnectWorkerCount <= 0 {
pfxlog.Logger().Panicf("RouterConnectWorkerCount for InstantStrategy cannot be less than 1, got %d", options.RouterConnectWorkerCount)
}
if options.SyncWorkerCount <= 0 {
pfxlog.Logger().Panicf("SyncWorkerCount for InstantStrategy cannot be less than 1, got %d", options.SyncWorkerCount)
}
if options.RouterTxBufferSize < 0 {
pfxlog.Logger().Panicf("RouterTxBufferSize for InstantStrategy cannot be less than 0, got %d", options.MaxQueuedRouterConnects)
}
if options.SessionChunkSize <= 0 {
pfxlog.Logger().Panicf("SessionChunkSize for InstantStrategy cannot be less than 1, got %d", options.SessionChunkSize)
}
strategy := &InstantStrategy{
InstantStrategyOptions: options,
rtxMap: &routerTxMap{
internalMap: &sync.Map{},
},
ae: ae,
routerConnectedQueue: make(chan *RouterSender, options.MaxQueuedRouterConnects),
receivedClientHelloQueue: make(chan *RouterSender, options.MaxQueuedClientHellos),
stop: make(chan struct{}, 0),
}
strategy.helloHandler = handler_edge_ctrl.NewHelloHandler(ae, strategy.ReceiveClientHello)
strategy.resyncHandler = handler_edge_ctrl.NewResyncHandler(ae, strategy.ReceiveResync)
for i := int32(0); i < options.RouterConnectWorkerCount; i++ {
go strategy.startHandleRouterConnectWorker()
}
for i := int32(0); i < options.SyncWorkerCount; i++ {
go strategy.startSynchronizeWorker()
}
return strategy
}
func (strategy *InstantStrategy) GetOnlineEdgeRouter(id string) (*model.EdgeRouter, env.RouterSyncStatus) {
rtx := strategy.rtxMap.Get(id)
if rtx != nil {
return rtx.EdgeRouter, rtx.Status
}
return nil, env.RouterSyncUnknown
}
func (strategy *InstantStrategy) Status(id string) env.RouterSyncStatus {
rtx := strategy.rtxMap.Get(id)
if rtx == nil {
return env.RouterSyncUnknown
}
return rtx.Status
}
func (strategy *InstantStrategy) Type() env.RouterSyncStrategyType {
return RouterSyncStrategyInstant
}
func (strategy *InstantStrategy) Stop() {
if strategy.stop != nil {
close(strategy.stop)
strategy.stop = nil
}
}
func (strategy *InstantStrategy) RouterConnected(edgeRouter *model.EdgeRouter, router *network.Router) {
rtx := newRouterSender(edgeRouter, router, strategy.RouterTxBufferSize)
rtx.Status = env.RouterSyncQueued
log := pfxlog.Logger().WithField("sync_strategy", strategy.Type()).
WithField("syncStatus", rtx.Status).
WithField("routerId", rtx.Router.Id).
WithField("routerName", rtx.Router.Name).
WithField("routerFingerprint", rtx.Router.Fingerprint)
log.Info("edge router connected, adding to sync routerConnectedQueue")
strategy.rtxMap.Add(router.Id, rtx)
strategy.routerConnectedQueue <- rtx
}
func (strategy *InstantStrategy) RouterDisconnected(router *network.Router) {
strategy.rtxMap.Remove(router.Id)
}
func (strategy *InstantStrategy) ApiSessionAdded(apiSession *persistence.ApiSession) {
logger := pfxlog.Logger().WithField("strategy", strategy.Type())
apiSessionProto, err := apiSessionToProto(strategy.ae, apiSession.Token, apiSession.IdentityId, apiSession.Id)
if err != nil {
logger.WithField("apiSessionId", apiSession.Id).
Errorf("error for individual api session added, could not convert to proto: %v", err)
return
}
state := &InstantSyncState{
Id: cuid.New(),
IsLast: true,
Sequence: 0,
}
strategy.rtxMap.Range(func(rtx *RouterSender) bool {
strategy.sendApiSessionAdded(rtx, false, state, []*edge_ctrl_pb.ApiSession{apiSessionProto})
return true
})
}
func (strategy *InstantStrategy) ApiSessionUpdated(apiSession *persistence.ApiSession, _ *persistence.ApiSessionCertificate) {
logger := pfxlog.Logger().WithField("strategy", strategy.Type())
apiSessionProto, err := apiSessionToProto(strategy.ae, apiSession.Token, apiSession.IdentityId, apiSession.Id)
if err != nil {
logger.WithField("apiSessionId", apiSession.Id).
Errorf("error for individual api session added, could not convert to proto: %v", err)
return
}
apiSessionAdded := &edge_ctrl_pb.ApiSessionAdded{
IsFullState: false,
ApiSessions: []*edge_ctrl_pb.ApiSession{apiSessionProto},
}
strategy.rtxMap.Range(func(rtx *RouterSender) bool {
content, _ := proto.Marshal(apiSessionAdded)
msg := channel2.NewMessage(env.ApiSessionUpdatedType, content)
msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
msg.Headers[env.SyncStrategyStateHeader] = nil
rtx.Send(msg)
return true
})
}
func (strategy *InstantStrategy) ApiSessionDeleted(apiSession *persistence.ApiSession) {
sessionRemoved := &edge_ctrl_pb.ApiSessionRemoved{
Tokens: []string{apiSession.Token},
}
strategy.rtxMap.Range(func(rtx *RouterSender) bool {
content, _ := proto.Marshal(sessionRemoved)
msg := channel2.NewMessage(env.ApiSessionRemovedType, content)
rtx.Send(msg)
return true
})
}
func (strategy *InstantStrategy) SessionAdded(session *persistence.Session) {
logger := pfxlog.Logger().WithField("strategy", strategy.Type())
sessionProto, err := sessionToProto(strategy.ae, session)
if err != nil {
logger.WithField("sessionId", session.Id).
Errorf("error for individual session added, could not convert to proto: %v", err)
return
}
state := &InstantSyncState{
Id: cuid.New(),
IsLast: true,
Sequence: 0,
}
strategy.rtxMap.Range(func(rtx *RouterSender) bool {
strategy.sendSessionAdded(rtx, false, state, []*edge_ctrl_pb.Session{sessionProto})
return true
})
}
func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) {
sessionRemoved := &edge_ctrl_pb.SessionRemoved{
Tokens: []string{session.Token},
}
strategy.rtxMap.Range(func(rtx *RouterSender) bool {
content, _ := proto.Marshal(sessionRemoved)
msg := channel2.NewMessage(env.SessionRemovedType, content)
rtx.Send(msg)
return true
})
}
func (strategy *InstantStrategy) startHandleRouterConnectWorker() {
select {
case <-strategy.stop:
return
case rtx := <-strategy.routerConnectedQueue:
strategy.hello(rtx)
}
}
func (strategy *InstantStrategy) startSynchronizeWorker() {
select {
case <-strategy.stop:
return
case rtx := <-strategy.receivedClientHelloQueue:
strategy.synchronize(rtx)
}
}
func (strategy *InstantStrategy) hello(rtx *RouterSender) {
logger := rtx.logger().WithField("strategy", strategy.Type())
logger.Info("edge router sync starting")
if rtx.Router.Control.IsClosed() {
rtx.Status = env.RouterSyncDisconnected
logger.WithField("syncStatus", rtx.Status).Info("edge router sync aborting, edge router disconnected before sync started")
return
}
rtx.Router.Control.AddReceiveHandler(strategy.helloHandler)
rtx.Status = env.RouterSyncHello
logger.WithField("syncStatus", rtx.Status).Info("sending edge router hello")
strategy.sendHello(rtx)
}
func (strategy *InstantStrategy) sendHello(rtx *RouterSender) {
logger := rtx.logger().WithField("strategy", strategy.Type())
serverVersion := build.GetBuildInfo().Version()
serverHello := &edge_ctrl_pb.ServerHello{
Version: serverVersion,
}
buf, err := proto.Marshal(serverHello)
if err != nil {
logger.WithError(err).Error("could not marshal serverHello")
return
}
if err = rtx.Router.Control.SendWithTimeout(channel2.NewMessage(env.ServerHelloType, buf), strategy.HelloSendTimeout); err != nil {
if rtx.Router.Control.IsClosed() {
rtx.Status = env.RouterSyncDisconnected
rtx.logger().WithError(err).Error("timed out sending serverHello message for edge router, connection closed, giving up")
} else {
rtx.Status = env.RouterSyncHelloTimeout
rtx.logger().WithError(err).Error("timed out sending serverHello message for edge router, queuing again")
go func() {
strategy.routerConnectedQueue <- rtx
}()
}
}
}
func (strategy *InstantStrategy) ReceiveResync(r *network.Router, hello *edge_ctrl_pb.RequestClientReSync) {
rtx := strategy.rtxMap.Get(r.Id)
if rtx == nil {
pfxlog.Logger().
WithField("strategy", strategy.Type()).
WithField("routerId", r.Id).
WithField("routerName", r.Name).
Error("received resync from router that is currently not tracked by the strategy, dropping resync")
return
}
rtx.Status = env.RouterSyncResyncWait
rtx.logger().WithField("strategy", strategy.Type()).Info("received resync from router, queuing")
strategy.receivedClientHelloQueue <- rtx
}
func (strategy *InstantStrategy) ReceiveClientHello(r *network.Router, respHello *edge_ctrl_pb.ClientHello) {
rtx := strategy.rtxMap.Get(r.Id)
if rtx == nil {
pfxlog.Logger().
WithField("strategy", strategy.Type()).
WithField("routerId", r.Id).
WithField("routerName", r.Name).
Error("received hello from router that is currently not tracked by the strategy, dropping hello")
return
}
rtx.Status = env.RouterSyncHelloWait
logger := rtx.logger().WithField("strategy", strategy.Type()).
WithField("hostname", respHello.Hostname).
WithField("protocols", respHello.Protocols).
WithField("protocolPorts", respHello.ProtocolPorts).
WithField("data", respHello.Data)
serverVersion := build.GetBuildInfo().Version()
if r.VersionInfo != nil {
logger = logger.WithField("version", r.VersionInfo.Version).
WithField("revision", r.VersionInfo.Revision).
WithField("buildDate", r.VersionInfo.BuildDate).
WithField("os", r.VersionInfo.OS).
WithField("arch", r.VersionInfo.Arch)
}
protocols := map[string]string{}
for _, p := range respHello.ProtocolPorts {
parts := strings.Split(p, ":")
ingressUrl := fmt.Sprintf("%s://%s:%s", parts[0], respHello.Hostname, parts[1])
protocols[parts[0]] = ingressUrl
}
rtx.EdgeRouter.Hostname = &respHello.Hostname
rtx.EdgeRouter.EdgeRouterProtocols = protocols
rtx.EdgeRouter.VersionInfo = r.VersionInfo
logger.Infof("edge router sent hello with version [%s] to controller with version [%s]", respHello.Version, serverVersion)
strategy.receivedClientHelloQueue <- rtx
}
func (strategy *InstantStrategy) synchronize(rtx *RouterSender) {
defer func() {
rtx.logger().WithField("strategy", strategy.Type()).Infof("exiting synchronization, final status: %s", rtx.Status)
}()
if rtx.Router.Control.IsClosed() {
rtx.Status = env.RouterSyncDisconnected
rtx.logger().WithField("strategy", strategy.Type()).Error("attempting to start synchronization with edge router, but it has disconnected")
}
rtx.Status = env.RouterSynInProgress
logger := rtx.logger().WithField("strategy", strategy.Type())
logger.Info("started synchronizing edge router")
chunkSize := 100
strategy.ae.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error {
var apiSessions []*edge_ctrl_pb.ApiSession
state := &InstantSyncState{
Id: cuid.New(),
IsLast: true,
Sequence: 0,
}
for cursor := strategy.ae.GetStores().ApiSession.IterateIds(tx, ast.BoolNodeTrue); cursor.IsValid(); cursor.Next() {
current := cursor.Current()
apiSession, err := strategy.ae.GetStores().ApiSession.LoadOneById(tx, string(current))
if err != nil {
logger.WithError(err).WithField("apiSessionId", string(current)).Errorf("error querying api session [%s]: %v", string(current), err)
continue
}
apiSessionProto, err := apiSessionToProto(strategy.ae, apiSession.Token, apiSession.IdentityId, apiSession.Id)
if err != nil {
logger.WithError(err).WithField("apiSessionId", string(current)).Errorf("error turning apiSession [%s] into proto: %v", string(current), err)
continue
}
apiSessions = append(apiSessions, apiSessionProto)
if len(apiSessions) >= chunkSize {
state.IsLast = !cursor.IsValid()
strategy.sendApiSessionAdded(rtx, true, state, apiSessions)
state.Sequence = state.Sequence + 1
apiSessions = []*edge_ctrl_pb.ApiSession{}
}
}
if len(apiSessions) > 0 {
state.IsLast = true
strategy.sendApiSessionAdded(rtx, true, state, apiSessions)
}
return nil
})
strategy.ae.GetDbProvider().GetDb().View(func(tx *bbolt.Tx) error {
var sessions []*edge_ctrl_pb.Session
state := &InstantSyncState{
Id: cuid.New(),
IsLast: true,
Sequence: 0,
}
for cursor := strategy.ae.GetStores().Session.IterateIds(tx, ast.BoolNodeTrue); cursor.IsValid(); cursor.Next() {
current := cursor.Current()
session, err := strategy.ae.GetStores().Session.LoadOneById(tx, string(current))
if err != nil {
logger.WithError(err).WithField("sessionId", string(current)).Errorf("error querying session [%s]: %v", string(current), err)
continue
}
sessionProto, err := sessionToProto(strategy.ae, session)
if err != nil {
logger.WithError(err).WithField("sessionId", string(current)).Errorf("error turning session [%s] into proto: %v", string(current), err)
continue
}
sessions = append(sessions, sessionProto)
if len(sessions) >= chunkSize {
state.IsLast = !cursor.IsValid()
strategy.sendSessionAdded(rtx, true, state, sessions)
state.Sequence = state.Sequence + 1
sessions = []*edge_ctrl_pb.Session{}
}
}
if len(sessions) > 0 {
state.IsLast = true
strategy.sendSessionAdded(rtx, true, state, sessions)
}
return nil
})
rtx.Status = env.RouterSyncDone
}
func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, apiSessions []*edge_ctrl_pb.ApiSession) {
stateBytes, _ := json.Marshal(state)
msgContent := &edge_ctrl_pb.ApiSessionAdded{
IsFullState: isFullState,
ApiSessions: apiSessions,
}
msgContentBytes, _ := proto.Marshal(msgContent)
msg := channel2.NewMessage(env.ApiSessionAddedType, msgContentBytes)
msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
msg.Headers[env.SyncStrategyStateHeader] = stateBytes
rtx.Send(msg)
}
func (strategy *InstantStrategy) sendSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, sessions []*edge_ctrl_pb.Session) {
stateBytes, _ := json.Marshal(state)
msgContent := &edge_ctrl_pb.SessionAdded{
IsFullState: isFullState,
Sessions: sessions,
}
msgContentBytes, _ := proto.Marshal(msgContent)
msg := channel2.NewMessage(env.SessionAddedType, msgContentBytes)
msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
msg.Headers[env.SyncStrategyStateHeader] = stateBytes
rtx.Send(msg)
}
type InstantSyncState struct {
Id string `json:"id"` //unique id for the sync attempt
IsLast bool `json:"isLast"` //
Sequence int `json:"sequence"` //inreasing id from 0 per id for the
}
go.mod
+2 -1
View File
@@ -32,6 +32,7 @@ require (
github.com/gorilla/handlers v1.5.1
github.com/jessevdk/go-flags v1.4.0
github.com/kataras/go-events v0.0.3-0.20201007151548-c411dc70c0a6
github.com/lucsky/cuid v1.0.2
github.com/mdlayher/netlink v1.2.1
github.com/michaelquigley/pfxlog v0.3.6
github.com/miekg/dns v1.1.38
@@ -54,7 +55,7 @@ require (
golang.org/x/net v0.0.0-20210119194325-5f4716e94777
golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
golang.org/x/sys v0.0.0-20210110051926-789bb1bd4061
google.golang.org/protobuf v1.25.0
google.golang.org/protobuf v1.25.0 // indirect
gopkg.in/resty.v1 v1.12.0
gopkg.in/yaml.v2 v2.4.0
gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
go.sum
+2
View File
@@ -404,6 +404,8 @@ github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL
github.com/lucas-clemente/quic-go v0.17.1/go.mod h1:I0+fcNTdb9eS1ZcjQZbDVPGchJ86chcIxPALn9lEJqE=
github.com/lucas-clemente/quic-go v0.18.1 h1:DMR7guC0NtVS8zNZR3IO7NARZvZygkSC56GGtC6cyys=
github.com/lucas-clemente/quic-go v0.18.1/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg=
github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ=
github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE=
github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
pb/edge_ctrl_pb/edge_ctrl.pb.go
+626 -2060
View File
File diff suppressed because it is too large Load Diff
pb/edge_ctrl_pb/edge_ctrl.proto
+5 -85
View File
@@ -14,6 +14,7 @@ enum ContentType {
ApiSessionUpdatedType = 20201;
ApiSessionRemovedType = 20202;
ApiSessionHeartbeatType = 20203;
RequestClientReSyncType = 20204;
EnrollType = 20300;
EnrollCertsType = 20301;
}
@@ -75,6 +76,7 @@ message ApiSessionUpdated {
message ApiSessionRemoved {
repeated string tokens = 1;
repeated string ids = 2;
}
message ApiSessionHeartbeat {
@@ -88,6 +90,7 @@ message SessionAdded {
message SessionRemoved {
repeated string tokens = 1;
repeated string ids = 2;
}
message SessionUpdated {
@@ -96,89 +99,6 @@ message SessionUpdated {
Service service = 3;
}
message PostureQuery {
string id = 1;
string postureCheckId = 20;
string displayName = 30;
PostureQuery.Type type = 40;
string version = 50;
oneof parameters {
PostureQuery.None na = 60;
PostureQuery.Process process = 70;
}
enum Type {
OS = 0;
MAC = 10;
WINDOWS_DOMAIN = 20;
PROCESS = 30;
}
enum HashAlgorithm {
SHA512 = 0;
}
message None {
}
message Process {
repeated string paths = 1;
HashAlgorithm hashAlgorithm = 10;
}
message RequestClientReSync {
string reason = 1;
}
message PostureResponse {
string id = 1;
oneof payload {
PostureResponse.Error error = 20;
PostureResponse.Proc process = 30;
PostureResponse.MacAddresses mac = 40;
PostureResponse.WindowsDomain domain = 50;
PostureResponse.OperatingSystem os = 60;
}
message Error {
enum Code {
UNKNOWN = 0;
UNSUPPORTED_QUERY_TYPE = 10;
UNSUPPORTED_HASH_ALGORITHM = 20;
}
PostureResponse.Error.Code code = 1;
string message = 10;
map<string, string> data = 20;
}
message Proc {
string binaryHash = 1;
string signerFingerprint = 10;
}
message MacAddresses {
repeated string addresses = 1;
}
message WindowsDomain {
string name = 1;
}
message OperatingSystem {
PostureResponse.OperatingSystem.Type type = 1;
string version = 10;
string build = 20;
enum Type {
Android = 0;
iOS = 10;
Linux = 20;
macOS = 30;
Windows = 40;
WindowsServer = 50;
}
}
}
rest_model/edge_router_detail.go
+25
View File
@@ -83,6 +83,10 @@ type EdgeRouterDetail struct {
// Required: true
SupportedProtocols map[string]string `json:"supportedProtocols"`
// sync status
// Required: true
SyncStatus *string `json:"syncStatus"`
// version info
VersionInfo *VersionInfo `json:"versionInfo,omitempty"`
}
@@ -120,6 +124,8 @@ func (m *EdgeRouterDetail) UnmarshalJSON(raw []byte) error {
SupportedProtocols map[string]string `json:"supportedProtocols"`
SyncStatus *string `json:"syncStatus"`
VersionInfo *VersionInfo `json:"versionInfo,omitempty"`
}
if err := swag.ReadJSON(raw, &dataAO1); err != nil {
@@ -148,6 +154,8 @@ func (m *EdgeRouterDetail) UnmarshalJSON(raw []byte) error {
m.SupportedProtocols = dataAO1.SupportedProtocols
m.SyncStatus = dataAO1.SyncStatus
m.VersionInfo = dataAO1.VersionInfo
return nil
@@ -185,6 +193,8 @@ func (m EdgeRouterDetail) MarshalJSON() ([]byte, error) {
SupportedProtocols map[string]string `json:"supportedProtocols"`
SyncStatus *string `json:"syncStatus"`
VersionInfo *VersionInfo `json:"versionInfo,omitempty"`
}
@@ -210,6 +220,8 @@ func (m EdgeRouterDetail) MarshalJSON() ([]byte, error) {
dataAO1.SupportedProtocols = m.SupportedProtocols
dataAO1.SyncStatus = m.SyncStatus
dataAO1.VersionInfo = m.VersionInfo
jsonDataAO1, errAO1 := swag.WriteJSON(dataAO1)
@@ -261,6 +273,10 @@ func (m *EdgeRouterDetail) Validate(formats strfmt.Registry) error {
res = append(res, err)
}
if err := m.validateSyncStatus(formats); err != nil {
res = append(res, err)
}
if err := m.validateVersionInfo(formats); err != nil {
res = append(res, err)
}
@@ -354,6 +370,15 @@ func (m *EdgeRouterDetail) validateSupportedProtocols(formats strfmt.Registry) e
return nil
}
func (m *EdgeRouterDetail) validateSyncStatus(formats strfmt.Registry) error {
if err := validate.Required("syncStatus", "body", m.SyncStatus); err != nil {
return err
}
return nil
}
func (m *EdgeRouterDetail) validateVersionInfo(formats strfmt.Registry) error {
if swag.IsZero(m.VersionInfo) { // not required
rest_server/embedded_spec.go
+10 -2
View File
@@ -7645,7 +7645,8 @@ func init() {
"version",
"arch",
"buildDate",
"revision"
"revision",
"syncStatus"
],
"properties": {
"enrollmentCreatedAt": {
@@ -7690,6 +7691,9 @@ func init() {
"type": "string"
}
},
"syncStatus": {
"type": "string"
},
"versionInfo": {
"$ref": "#/definitions/versionInfo"
}
@@ -27181,7 +27185,8 @@ func init() {
"version",
"arch",
"buildDate",
"revision"
"revision",
"syncStatus"
],
"properties": {
"enrollmentCreatedAt": {
@@ -27226,6 +27231,9 @@ func init() {
"type": "string"
}
},
"syncStatus": {
"type": "string"
},
"versionInfo": {
"$ref": "#/definitions/versionInfo"
}
router/handler_edge_ctrl/apiSessionAdded.go
+285 -12
View File
@@ -17,21 +17,57 @@
package handler_edge_ctrl
import (
"encoding/json"
"errors"
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/edge/controller/sync_strats"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/foundation/channel2"
"sync"
"time"
)
type apiSessionAddedHandler struct {
sm fabric.StateManager
control channel2.Channel
sm fabric.StateManager
syncTracker *apiSessionSyncTracker
reqChan chan *apiSessionAddedWithState
syncReady chan []*edge_ctrl_pb.ApiSession
syncFail chan error
stop chan struct{}
}
func NewApiSessionAddedHandler(sm fabric.StateManager) *apiSessionAddedHandler {
return &apiSessionAddedHandler{
sm: sm,
func NewApiSessionAddedHandler(sm fabric.StateManager, control channel2.Channel) *apiSessionAddedHandler {
handler := &apiSessionAddedHandler{
control: control,
sm: sm,
reqChan: make(chan *apiSessionAddedWithState, 100),
syncReady: make(chan []*edge_ctrl_pb.ApiSession, 0),
syncFail: make(chan error, 0),
stop: make(chan struct{}, 0),
}
go handler.startReceiveSync()
go handler.startSyncApplier()
go handler.startSyncFail()
control.AddCloseHandler(handler)
return handler
}
func (h *apiSessionAddedHandler) HandleClose(_ channel2.Channel) {
if h.stop != nil {
close(h.stop)
h.stop = nil
}
}
@@ -39,19 +75,256 @@ func (h *apiSessionAddedHandler) ContentType() int32 {
return env.ApiSessionAddedType
}
func (h *apiSessionAddedHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *apiSessionAddedHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
go func() {
req := &edge_ctrl_pb.ApiSessionAdded{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
for _, session := range req.ApiSessions {
h.sm.AddApiSession(session)
}
if req.IsFullState {
h.sm.RemoveMissingApiSessions(req.ApiSessions)
reqWithState := &apiSessionAddedWithState{
ApiSessionAdded: req,
}
if syncStrategyType, syncState, err := parseInstantSyncHeaders(msg); err == nil {
reqWithState.SyncStrategyType = syncStrategyType
reqWithState.InstantSyncState = syncState
} else {
pfxlog.Logger().WithField("msgContentType", msg.ContentType).WithError(err).Errorf("sync headers not present (old controller) or only partial present(error), treating as legacy: %v", err)
}
h.reqChan <- reqWithState
} else {
for _, session := range req.ApiSessions {
h.sm.AddApiSession(session)
}
}
} else {
pfxlog.Logger().Panic("could not convert message as network session added")
pfxlog.Logger().Panic("could not convert message as api session added")
}
}()
}
func (h *apiSessionAddedHandler) startSyncApplier() {
for {
select {
case <-h.stop:
return
case apiSessions := <-h.syncReady:
for _, apiSession := range apiSessions {
h.sm.AddApiSession(apiSession)
}
h.sm.RemoveMissingApiSessions(apiSessions)
pfxlog.Logger().Infof("finished sychronizing api sessions [count: %d]", len(apiSessions))
}
}
}
func (h *apiSessionAddedHandler) startSyncFail() {
for {
select {
case <-h.stop:
return
case err := <-h.syncFail:
pfxlog.Logger().Errorf("failed to synchronize api sessions: %v", err)
h.syncTracker.Stop()
h.syncTracker = nil
resync := &edge_ctrl_pb.RequestClientReSync{
Reason: fmt.Sprintf("error during api session sync: %v", err),
}
resyncProto, _ := proto.Marshal(resync)
resyncMsg := channel2.NewMessage(env.RequestClientReSyncType, resyncProto)
_ = h.control.Send(resyncMsg)
}
}
}
func (h *apiSessionAddedHandler) legacySync(reqWithState *apiSessionAddedWithState) {
pfxlog.Logger().Warn("using legacy sync logic some connections may be dropped")
for _, apiSession := range reqWithState.ApiSessions {
h.sm.AddApiSession(apiSession)
}
h.sm.RemoveMissingApiSessions(reqWithState.ApiSessions)
}
func (h *apiSessionAddedHandler) startReceiveSync() {
for {
select {
case <-h.stop:
return
case reqWithState := <-h.reqChan:
switch reqWithState.SyncStrategyType {
case string(sync_strats.RouterSyncStrategyInstant):
h.instantSync(reqWithState)
case "":
pfxlog.Logger().Warn("syncStrategy is not specified, old controller?")
h.legacySync(reqWithState)
default:
pfxlog.Logger().Warnf("syncStrategy [%s] is not supported", reqWithState.SyncStrategyType)
h.legacySync(reqWithState)
}
}
}
}
func (h *apiSessionAddedHandler) instantSync(reqWithState *apiSessionAddedWithState) {
logger := pfxlog.Logger().WithField("strategy", reqWithState.SyncStrategyType)
if reqWithState.InstantSyncState == nil {
logger.Panic("syncState is empty, cannot continue")
}
if reqWithState.InstantSyncState.Id == "" {
logger.Panic("syncState id is empty, cannot continue")
}
//if no id or the sync id is newer, reset
if h.syncTracker == nil || h.syncTracker.syncId == "" || h.syncTracker.isDone || h.syncTracker.syncId < reqWithState.Id {
if h.syncTracker == nil || h.syncTracker.syncId == "" {
logger.Infof("first api session syncId [%s], starting", reqWithState.Id)
} else if h.syncTracker.isDone{
logger.Infof("api session syncId [%s], starting", reqWithState.Id)
} else {
logger.Infof("api session with newer syncId [old: %s, new: %s], aborting old, starting new", h.syncTracker.syncId, reqWithState.Id)
}
if h.syncTracker != nil {
h.syncTracker.Stop()
}
h.syncTracker = newApiSessionSyncTracker(reqWithState.Id)
h.syncTracker.StartDeadline(h.syncReady, h.syncFail, 20*time.Second)
}
//ignore older syncs
if h.syncTracker.syncId > reqWithState.Id {
logger.Warnf("older syncId [%s], ignoring", reqWithState.Id)
return
}
h.syncTracker.Add(reqWithState)
}
type apiSessionSyncTracker struct {
syncId string
reqsWithState map[int]*apiSessionAddedWithState
hasLast bool
lastSeq int
stop chan struct{}
deadline sync.Once
isDone bool
}
func newApiSessionSyncTracker(id string) *apiSessionSyncTracker {
return &apiSessionSyncTracker{
syncId: id,
reqsWithState: map[int]*apiSessionAddedWithState{},
stop: make(chan struct{}, 0),
}
}
func (tracker *apiSessionSyncTracker) Add(reqWithState *apiSessionAddedWithState) {
tracker.reqsWithState[reqWithState.Sequence] = reqWithState
if reqWithState.IsLast {
tracker.hasLast = true
tracker.lastSeq = reqWithState.Sequence
}
}
func (tracker *apiSessionSyncTracker) Stop() {
if tracker != nil && tracker.stop != nil {
close(tracker.stop)
tracker.stop = nil
}
}
func (tracker *apiSessionSyncTracker) IsDone() bool {
return tracker.isDone
}
func (tracker *apiSessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_pb.ApiSession, syncFail chan error, timeout time.Duration) {
tracker.deadline.Do(func() {
go func() {
ticker := time.NewTicker(1 * time.Second)
select {
case <-tracker.stop:
tracker.reqsWithState = nil
return
case <-ticker.C:
if tracker.HasAll() {
syncReady <- tracker.all()
tracker.isDone = true
return
}
case <-time.After(timeout):
tracker.reqsWithState = nil
syncFail <- errors.New("timeout, did not receive all updates in time")
return
}
}()
})
}
func (tracker *apiSessionSyncTracker) HasAll() bool {
if !tracker.hasLast {
return false
}
hasAll := true
for i := 0; i <= tracker.lastSeq; i++ {
if req, ok := tracker.reqsWithState[i]; !ok && req == nil {
hasAll = false
break
}
}
return hasAll
}
func (tracker *apiSessionSyncTracker) all() []*edge_ctrl_pb.ApiSession {
var result []*edge_ctrl_pb.ApiSession
for i := 0; i <= tracker.lastSeq; i++ {
if req, ok := tracker.reqsWithState[i]; ok {
for _, apiSession := range req.ApiSessions {
result = append(result, apiSession)
}
} else {
pfxlog.Logger().WithField("strategy", sync_strats.RouterSyncStrategyInstant).Error("all failed to have all update sequences")
}
}
return result
}
type apiSessionAddedWithState struct {
SyncStrategyType string
*sync_strats.InstantSyncState
*edge_ctrl_pb.ApiSessionAdded
}
func parseInstantSyncHeaders(msg *channel2.Message) (string, *sync_strats.InstantSyncState, error) {
if syncStrategyType, ok := msg.Headers[env.SyncStrategyTypeHeader]; ok {
if syncStrategyState, ok := msg.Headers[env.SyncStrategyStateHeader]; ok {
state := &sync_strats.InstantSyncState{}
if err := json.Unmarshal(syncStrategyState, state); err == nil {
return string(syncStrategyType), state, nil
} else {
pfxlog.Logger().WithField("strategy", syncStrategyType).WithField("msgContentType", msg.ContentType).Panicf("could not parse sync state [%s], error: %v", syncStrategyState, err)
}
} else {
return "", nil, errors.New("received sync message with a strategy type header, but no state")
}
}
return "", nil, errors.New("received sync message with no strategy type header")
}
router/handler_edge_ctrl/apiSessionRemoved.go
+17 -3
View File
@@ -20,8 +20,8 @@ import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/foundation/channel2"
)
@@ -43,8 +43,22 @@ func (h *apiSessionRemovedHandler) HandleReceive(msg *channel2.Message, ch chann
go func() {
req := &edge_ctrl_pb.ApiSessionRemoved{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
for _, t := range req.Tokens {
h.sm.RemoveApiSession(t)
//older controllers will not provide req.Ids
hasIds := len(req.Ids) == len(req.Tokens)
for i, token := range req.Tokens {
id := "unknown"
if hasIds {
id = req.Ids[i]
}
pfxlog.Logger().
WithField("apiSessionToken", token).
WithField("apiSessionId", id).
Debugf("removing api session [token: %s] [id: %s]", token, id)
h.sm.RemoveApiSession(token)
}
} else {
pfxlog.Logger().Panic("could not convert message as session removed")
router/handler_edge_ctrl/sessionAdded.go
+249 -12
View File
@@ -17,42 +17,279 @@
package handler_edge_ctrl
import (
"errors"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/edge/controller/sync_strats"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/foundation/channel2"
"sync"
"time"
)
type sessionAddedHandler struct {
sm fabric.StateManager
sm fabric.StateManager
syncTracker *sessionSyncTracker
reqChan chan *sessionAddedWithState
syncReady chan []*edge_ctrl_pb.Session
syncFail chan error
stop chan struct{}
}
func NewSessionAddedHandler(sm fabric.StateManager) *sessionAddedHandler {
return &sessionAddedHandler{
sm: sm,
func NewSessionAddedHandler(sm fabric.StateManager, control channel2.Channel) *sessionAddedHandler {
handler := &sessionAddedHandler{
sm: sm,
reqChan: make(chan *sessionAddedWithState, 100),
syncReady: make(chan []*edge_ctrl_pb.Session, 0),
syncFail: make(chan error, 0),
stop: make(chan struct{}),
}
go handler.startReceiveSync()
go handler.startSyncApplier()
go handler.startSyncFail()
control.AddCloseHandler(handler)
return handler
}
func (h *sessionAddedHandler) ContentType() int32 {
return env.SessionAddedType
}
func (h *sessionAddedHandler) HandleClose(_ channel2.Channel) {
if h.stop != nil {
close(h.stop)
h.stop = nil
}
}
func (h *sessionAddedHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
go func() {
req := &edge_ctrl_pb.SessionAdded{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
for _, session := range req.Sessions {
pfxlog.Logger().Debugf("received new session %+v", session)
h.sm.AddSession(session)
}
if req.IsFullState {
h.sm.RemoveMissingSessions(req.Sessions)
reqWithState := &sessionAddedWithState{
SessionAdded: req,
}
if syncStrategyType, syncState, err := parseInstantSyncHeaders(msg); err == nil {
reqWithState.SyncStrategyType = syncStrategyType
reqWithState.InstantSyncState = syncState
} else {
pfxlog.Logger().WithField("strategy", syncStrategyType).WithField("msgContentType", msg.ContentType).WithError(err).Errorf("sync headers not present (old controller) or only partial present(error), treading as legacy: %v", err)
}
h.reqChan <- reqWithState
} else {
for _, session := range req.Sessions {
h.sm.AddSession(session)
}
}
} else {
pfxlog.Logger().Panic("could not convert message as network session added")
pfxlog.Logger().Panic("could not convert message as session added")
}
}()
}
func (h *sessionAddedHandler) startSyncApplier() {
for {
select {
case <-h.stop:
return
case sessions := <-h.syncReady:
for _, session := range sessions {
h.sm.AddSession(session)
}
h.sm.RemoveMissingSessions(sessions)
pfxlog.Logger().Infof("finished sychronizing sessions [count: %d]", len(sessions))
}
}
}
func (h *sessionAddedHandler) startSyncFail() {
for {
select {
case <-h.stop:
return
case err := <-h.syncFail:
pfxlog.Logger().Errorf("failed to synchronize sessions, aborting: %v", err)
h.syncTracker.Stop()
}
}
}
func (h *sessionAddedHandler) legacySync(reqWithState *sessionAddedWithState) {
pfxlog.Logger().Warn("using legacy sync logic some connections may be dropped")
for _, session := range reqWithState.Sessions {
h.sm.AddSession(session)
}
h.sm.RemoveMissingSessions(reqWithState.Sessions)
}
func (h *sessionAddedHandler) startReceiveSync() {
for {
select {
case <-h.stop:
return
case reqWithState := <-h.reqChan:
switch reqWithState.SyncStrategyType {
case string(sync_strats.RouterSyncStrategyInstant):
h.instantSync(reqWithState)
case "":
pfxlog.Logger().Warn("syncStrategy is not specified, old controller?")
h.legacySync(reqWithState)
default:
pfxlog.Logger().Warnf("syncStrategy [%s] is not supported", reqWithState.SyncStrategyType)
h.legacySync(reqWithState)
}
}
}
}
func (h *sessionAddedHandler) instantSync(reqWithState *sessionAddedWithState) {
logger := pfxlog.Logger().WithField("strategy", reqWithState.SyncStrategyType)
if reqWithState.InstantSyncState == nil {
logger.Panic("syncState is empty, cannot continue")
}
if reqWithState.InstantSyncState.Id == "" {
logger.Panic("syncState id is empty, cannot continue")
}
//if no id or the sync id is newer, reset
if h.syncTracker == nil || h.syncTracker.syncId == "" || h.syncTracker.IsDone() || h.syncTracker.syncId < reqWithState.Id {
if h.syncTracker == nil || h.syncTracker.syncId == "" {
logger.Infof("first session syncId [%s], starting", reqWithState.Id)
} else if h.syncTracker.isDone{
logger.Infof("session syncId [%s], starting", reqWithState.Id)
} else {
logger.Infof("session with newer syncId [old: %s, new: %s], aborting old, starting new", h.syncTracker.syncId, reqWithState.Id)
}
if h.syncTracker != nil {
h.syncTracker.Stop()
}
h.syncTracker = newSessionSyncTracker(reqWithState.Id)
h.syncTracker.StartDeadline(h.syncReady, h.syncFail, 20*time.Second)
}
//ignore older syncs
if h.syncTracker.syncId > reqWithState.Id {
logger.Warnf("older syncId [%s], ignoring", reqWithState.Id)
return
}
h.syncTracker.Add(reqWithState)
}
type sessionSyncTracker struct {
syncId string
reqsWithState map[int]*sessionAddedWithState
hasLast bool
lastSeq int
stop chan struct{}
deadline sync.Once
isDone bool
}
func newSessionSyncTracker(id string) *sessionSyncTracker {
return &sessionSyncTracker{
syncId: id,
reqsWithState: map[int]*sessionAddedWithState{},
stop: make(chan struct{}, 0),
}
}
func (tracker *sessionSyncTracker) Add(reqWithState *sessionAddedWithState) {
tracker.reqsWithState[reqWithState.Sequence] = reqWithState
if reqWithState.IsLast {
tracker.hasLast = true
tracker.lastSeq = reqWithState.Sequence
}
}
func (tracker *sessionSyncTracker) Stop() {
if tracker != nil && tracker.stop != nil {
close(tracker.stop)
tracker.stop = nil
}
}
func (tracker *sessionSyncTracker) IsDone() bool {
return tracker.isDone
}
func (tracker *sessionSyncTracker) StartDeadline(syncReady chan []*edge_ctrl_pb.Session, syncFail chan error, timeout time.Duration) {
tracker.deadline.Do(func() {
go func() {
ticker := time.NewTicker(1 * time.Second)
select {
case <-tracker.stop:
tracker.reqsWithState = nil
return
case <-ticker.C:
if tracker.HasAll() {
syncReady <- tracker.all()
tracker.isDone = true
return
}
case <-time.After(timeout):
tracker.reqsWithState = nil
syncFail <- errors.New("timeout, did not receive all updates in time")
return
}
}()
})
}
func (tracker *sessionSyncTracker) HasAll() bool {
if !tracker.hasLast {
return false
}
hasAll := true
for i := 0; i <= tracker.lastSeq; i++ {
if req, ok := tracker.reqsWithState[i]; !ok && req == nil {
hasAll = false
break
}
}
return hasAll
}
func (tracker *sessionSyncTracker) all() []*edge_ctrl_pb.Session {
var result []*edge_ctrl_pb.Session
for i := 0; i <= tracker.lastSeq; i++ {
if req, ok := tracker.reqsWithState[i]; ok {
for _, session := range req.Sessions {
result = append(result, session)
}
} else {
pfxlog.Logger().WithField("strategy", sync_strats.RouterSyncStrategyInstant).Error("all failed to have all update sequences")
}
}
return result
}
type sessionAddedWithState struct {
SyncStrategyType string
*sync_strats.InstantSyncState
*edge_ctrl_pb.SessionAdded
}
router/handler_edge_ctrl/sessionRemoved.go
+13 -3
View File
@@ -20,8 +20,8 @@ import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/internal/fabric"
"github.com/openziti/foundation/channel2"
)
@@ -43,8 +43,18 @@ func (h *sessionRemovedHandler) HandleReceive(msg *channel2.Message, ch channel2
go func() {
req := &edge_ctrl_pb.SessionRemoved{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
for _, t := range req.Tokens {
h.sm.RemoveSession(t)
hasIds := len(req.Ids) == len(req.Tokens)
for i, token := range req.Tokens {
id := "unknown"
if hasIds {
id = req.Ids[i]
}
pfxlog.Logger().
WithField("sessionToken", token).
WithField("sessionId", id).
Debugf("removing api session [token: %s] [id: %s]", token, id)
h.sm.RemoveSession(token)
}
} else {
pfxlog.Logger().Panic("could not convert message as network session removed")
router/internal/fabric/manager.go
+51 -29
View File
@@ -55,8 +55,8 @@ type StateManager interface {
//ApiSessions
GetApiSession(token string) *edge_ctrl_pb.ApiSession
GetApiSessionWithTimeout(token string, timeout time.Duration) *edge_ctrl_pb.ApiSession
AddApiSession(ns *edge_ctrl_pb.ApiSession)
UpdateApiSession(ns *edge_ctrl_pb.ApiSession)
AddApiSession(apiSession *edge_ctrl_pb.ApiSession)
UpdateApiSession(apiSession *edge_ctrl_pb.ApiSession)
RemoveApiSession(token string)
RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession)
AddConnectedApiSession(token string, removeCB func(), ch channel2.Channel)
@@ -97,20 +97,28 @@ func GetStateManager() StateManager {
return singleStateManager
}
func (sm *StateManagerImpl) AddApiSession(session *edge_ctrl_pb.ApiSession) {
pfxlog.Logger().Debugf("adding session [%s] fingerprints [%s]", session.Token, session.CertFingerprints)
sm.apiSessionsByToken.Store(session.Token, session)
sm.Emit(EventAddedSession, session)
func (sm *StateManagerImpl) AddApiSession(apiSession *edge_ctrl_pb.ApiSession) {
pfxlog.Logger().
WithField("apiSessionId", apiSession.Id).
WithField("apiSessionToken", apiSession.Token).
WithField("apiSessionCertFingerprints", apiSession.CertFingerprints).
Debugf("adding apiSession [id: %s] [token: %s] fingerprints [%s]", apiSession.Id, apiSession.Token, apiSession.CertFingerprints)
sm.apiSessionsByToken.Store(apiSession.Token, apiSession)
sm.Emit(EventAddedSession, apiSession)
}
func (sm *StateManagerImpl) UpdateApiSession(session *edge_ctrl_pb.ApiSession) {
pfxlog.Logger().Debugf("updating session [%s] fingerprints [%s]", session.Token, session.CertFingerprints)
sm.apiSessionsByToken.Store(session.Token, session)
func (sm *StateManagerImpl) UpdateApiSession(apiSession *edge_ctrl_pb.ApiSession) {
pfxlog.Logger().
WithField("apiSessionId", apiSession.Id).
WithField("apiSessionToken", apiSession.Token).
WithField("apiSessionCertFingerprints", apiSession.CertFingerprints).
Debugf("updating apiSession [id: %s] [token: %s] fingerprints [%s]", apiSession.Id, apiSession.Token, apiSession.CertFingerprints)
sm.apiSessionsByToken.Store(apiSession.Token, apiSession)
sm.sessionsByToken.Range(func(key, value interface{}) bool {
if ns, ok := value.(*edge_ctrl_pb.Session); ok {
if ns.ApiSessionId == session.Id { //only update the specific api session's sessions
ns.CertFingerprints = session.CertFingerprints //session.CertFingerprints is all currently valid
if ns.ApiSessionId == apiSession.Id { //only update the specific api apiSession's sessions
ns.CertFingerprints = apiSession.CertFingerprints //apiSession.CertFingerprints is all currently valid
}
} else {
pfxlog.Logger().Warn("could not convert value from concurrent map sessionsByToken to Session, this should not happen")
@@ -118,19 +126,19 @@ func (sm *StateManagerImpl) UpdateApiSession(session *edge_ctrl_pb.ApiSession) {
return true
})
sm.Emit(EventUpdatedSession, session)
sm.Emit(EventUpdatedSession, apiSession)
}
func (sm *StateManagerImpl) RemoveApiSession(token string) {
if ns, ok := sm.apiSessionsByToken.Load(token); ok {
pfxlog.Logger().Debugf("removing session [%s]", token)
pfxlog.Logger().WithField("apiSessionToken", token).Debugf("removing api session [token: %s]", token)
sm.apiSessionsByToken.Delete(token)
eventName := sm.getSessionRemovedEventName(token)
sm.Emit(eventName)
sm.RemoveAllListeners(eventName)
sm.Emit(EventRemovedSession, ns)
} else {
pfxlog.Logger().Debugf("could not remove session [%s]; not found", token)
pfxlog.Logger().WithField("apiSessionToken", token).Debugf("could not remove api session [token: %s]; not found", token)
}
}
func (sm *StateManagerImpl) RemoveMissingApiSessions(knownSessions []*edge_ctrl_pb.ApiSession) {
@@ -153,28 +161,42 @@ func (sm *StateManagerImpl) RemoveMissingApiSessions(knownSessions []*edge_ctrl_
}
}
func (sm *StateManagerImpl) AddSession(ns *edge_ctrl_pb.Session) {
func (sm *StateManagerImpl) AddSession(session *edge_ctrl_pb.Session) {
// BACKWARDS_COMPATIBILITY introduced 0.15.2
// support 0.14 (and older) style controller generated fingerprints, as fingerprint format changed from 0.14 to 0.15
for i := 0; i < len(ns.CertFingerprints); i++ {
ns.CertFingerprints[i] = strings.Replace(strings.ToLower(ns.CertFingerprints[i]), ":", "", -1)
for i := 0; i < len(session.CertFingerprints); i++ {
session.CertFingerprints[i] = strings.Replace(strings.ToLower(session.CertFingerprints[i]), ":", "", -1)
}
pfxlog.Logger().Debugf("adding network session [%s] fingerprints [%s] TypeId [%v]", ns.Token, ns.CertFingerprints, ns.Type.String())
sm.sessionsByToken.Store(ns.Token, ns)
sm.Emit(EventAddedNetworkSession, ns)
pfxlog.Logger().
WithField("apiSessionId", session.ApiSessionId).
WithField("sessionId", session.Id).
WithField("sessionToken", session.Token).
WithField("serviceId", session.Service.Id).
WithField("serviceName", session.Service.Name).
WithField("serviceEncryptionRequired", session.Service.EncryptionRequired).
Debugf("adding session [token: %s] [id: %s] fingerprints [%s] TypeId [%v]", session.Token, session.Id, session.CertFingerprints, session.Type.String())
sm.sessionsByToken.Store(session.Token, session)
sm.Emit(EventAddedNetworkSession, session)
}
func (sm *StateManagerImpl) UpdateSession(ns *edge_ctrl_pb.Session) {
func (sm *StateManagerImpl) UpdateSession(session *edge_ctrl_pb.Session) {
// BACKWARDS_COMPATIBILITY introduced 0.15.2
// support 0.14 (and older) style controller generated fingerprints, as fingerprint format changed from 0.14 to 0.15
for i := 0; i < len(ns.CertFingerprints); i++ {
ns.CertFingerprints[i] = strings.Replace(strings.ToLower(ns.CertFingerprints[i]), ":", "", -1)
for i := 0; i < len(session.CertFingerprints); i++ {
session.CertFingerprints[i] = strings.Replace(strings.ToLower(session.CertFingerprints[i]), ":", "", -1)
}
pfxlog.Logger().Debugf("updating network session [%s] fingerprints [%s]", ns.Token, ns.CertFingerprints)
sm.sessionsByToken.Store(ns.Token, ns)
sm.Emit(EventUpdatedNetworkSession, ns)
pfxlog.Logger().
WithField("apiSessionId", session.ApiSessionId).
WithField("sessionId", session.Id).
WithField("sessionToken", session.Token).
WithField("serviceId", session.Service.Id).
WithField("serviceName", session.Service.Name).
WithField("serviceEncryptionRequired", session.Service.EncryptionRequired).
Debugf("updating session [token: %s] [id: %s] fingerprints [%s]", session.Token, session.Id, session.CertFingerprints)
sm.sessionsByToken.Store(session.Token, session)
sm.Emit(EventUpdatedNetworkSession, session)
}
func (sm *StateManagerImpl) RemoveMissingSessions(knownSessions []*edge_ctrl_pb.Session) {
@@ -199,14 +221,14 @@ func (sm *StateManagerImpl) RemoveMissingSessions(knownSessions []*edge_ctrl_pb.
func (sm *StateManagerImpl) RemoveSession(token string) {
if ns, ok := sm.sessionsByToken.Load(token); ok {
pfxlog.Logger().Debugf("removing network session [%s]", token)
pfxlog.Logger().Debugf("removing session [token: %s]", token)
sm.sessionsByToken.Delete(token)
sm.Emit(EventRemovedNetworkSession, ns)
eventName := sm.getNetworkSessionRemovedEventName(token)
sm.Emit(eventName)
sm.RemoveAllListeners(eventName)
} else {
pfxlog.Logger().Debugf("could not remove network session [%s]; not found", token)
pfxlog.Logger().Debugf("could not remove session [token: %s]; not found", token)
}
}
@@ -245,7 +267,7 @@ func (sm *StateManagerImpl) GetSession(token string) *edge_ctrl_pb.Session {
if ns, ok := obj.(*edge_ctrl_pb.Session); ok {
return ns
}
pfxlog.Logger().Panic("encountered non-network session in network session map")
pfxlog.Logger().Panic("encountered non-session in network session map")
}
return nil
router/xgress_edge/factory.go
+2 -2
View File
@@ -81,11 +81,11 @@ func (factory *Factory) BindChannel(ch channel2.Channel) error {
}
ch.AddReceiveHandler(handler_edge_ctrl.NewHelloHandler(hostname, supportedProtocols, protocolPorts))
ch.AddReceiveHandler(handler_edge_ctrl.NewSessionAddedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewSessionAddedHandler(factory.stateManager, ch))
ch.AddReceiveHandler(handler_edge_ctrl.NewSessionRemovedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewSessionUpdatedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewApiSessionAddedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewApiSessionAddedHandler(factory.stateManager, ch))
ch.AddReceiveHandler(handler_edge_ctrl.NewApiSessionRemovedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewApiSessionUpdatedHandler(factory.stateManager))
return nil
specs/swagger.yml
+3
View File
@@ -5197,6 +5197,7 @@ definitions:
- arch
- buildDate
- revision
- syncStatus
properties:
name:
type: string
@@ -5228,6 +5229,8 @@ definitions:
$ref: '#/definitions/attributes'
fingerprint:
type: string
syncStatus:
type: string
versionInfo:
$ref: '#/definitions/versionInfo'
example:
tests/accept_manual_start_test.go
+6
View File
@@ -49,6 +49,8 @@ func Test_ManualStart(t *testing.T) {
log.Info("starting listener1")
host1 := &host{}
host1.id, host1.context = ctx.AdminSession.RequireCreateSdkContext()
defer host1.context.Close()
listener, err := host1.context.ListenWithOptions(service.Name, &ziti.ListenOptions{
Precedence: ziti.PrecedenceRequired,
ManualStart: true,
@@ -63,6 +65,8 @@ func Test_ManualStart(t *testing.T) {
log.Info("starting listener2")
host2 := &host{}
host2.id, host2.context = ctx.AdminSession.RequireCreateSdkContext()
defer host2.context.Close()
listener, err = host2.context.ListenWithOptions(service.Name, &ziti.ListenOptions{
Precedence: ziti.PrecedenceDefault,
ManualStart: false,
@@ -116,6 +120,8 @@ func Test_ManualStart(t *testing.T) {
}()
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
for i := 0; i < 10; i++ {
log.Infof("dialing: %v", i)
conn := ctx.WrapConn(context.Dial(service.Name))
tests/addressable_terminators_test.go
+9
View File
@@ -52,6 +52,8 @@ func Test_AddressableTerminators(t *testing.T) {
hosts = append(hosts, host)
host.id, host.context = ctx.AdminSession.RequireCreateSdkContext()
defer host.context.Close()
host.listener, err = host.context.ListenWithOptions(service.Name, &ziti.ListenOptions{
BindUsingEdgeIdentity: true,
})
@@ -69,6 +71,7 @@ func Test_AddressableTerminators(t *testing.T) {
client := &client{}
clients = append(clients, client)
client.id, client.context = ctx.AdminSession.RequireCreateSdkContext()
defer client.context.Close()
}
waitForConn := func(listener net.Listener, timeout time.Duration) (net.Conn, error) {
@@ -127,6 +130,8 @@ func Test_AddressableTerminatorSameIdentity(t *testing.T) {
}
identity, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.ListenWithOptions(service.Name, &ziti.ListenOptions{
BindUsingEdgeIdentity: true,
ConnectTimeout: 5 * time.Second,
@@ -171,6 +176,8 @@ func Test_AddressableTerminatorDifferentIdentity(t *testing.T) {
}
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.ListenWithOptions(service.Name, &ziti.ListenOptions{
Identity: "foobar",
ConnectTimeout: 5 * time.Second,
@@ -180,6 +187,8 @@ func Test_AddressableTerminatorDifferentIdentity(t *testing.T) {
defer func() { _ = listener.Close() }()
_, context2 := ctx.AdminSession.RequireCreateSdkContext()
defer context2.Close()
listener2, err := context2.ListenWithOptions(service.Name, &ziti.ListenOptions{
Identity: "foobar",
ConnectTimeout: 5 * time.Second,
tests/data_flow_close_test.go
+28
View File
@@ -39,6 +39,8 @@ func Test_ServerConnClosePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
ctx.Req.NoError(err)
@@ -69,6 +71,8 @@ func Test_ServerConnClosePropagation(t *testing.T) {
clientContext := ziti.NewContextWithConfig(clientConfig)
conn := ctx.WrapConn(clientContext.Dial(service.Name))
defer conn.Close()
name := eid.New()
conn.WriteString(name, time.Second)
conn.ReadExpected("hello, "+name, time.Second)
@@ -97,7 +101,11 @@ func Test_ServerContextClosePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
defer listener.Close()
ctx.Req.NoError(err)
errC := make(chan error, 1)
@@ -128,6 +136,8 @@ func Test_ServerContextClosePropagation(t *testing.T) {
clientContext := ziti.NewContextWithConfig(clientConfig)
conn := ctx.WrapConn(clientContext.Dial(service.Name))
defer conn.Close()
name := eid.New()
conn.WriteString(name, time.Second)
conn.ReadExpected("hello, "+name, time.Second)
@@ -157,7 +167,11 @@ func Test_ServerCloseListenerPropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
defer listener.Close()
ctx.Req.NoError(err)
errC := make(chan error, 1)
@@ -190,6 +204,8 @@ func Test_ServerCloseListenerPropagation(t *testing.T) {
clientContext := ziti.NewContextWithConfig(clientConfig)
conn := ctx.WrapConn(clientContext.Dial(service.Name))
defer conn.Close()
name := eid.New()
conn.WriteString(name, time.Second)
conn.ReadExpected("hello, "+name, time.Second)
@@ -209,8 +225,11 @@ func Test_ClientConnClosePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
ctx.Req.NoError(err)
defer listener.Close()
clientIdentity := ctx.AdminSession.RequireNewIdentityWithOtt(false)
clientConfig := ctx.EnrollIdentity(clientIdentity.Id)
@@ -268,8 +287,11 @@ func Test_ClientContextClosePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
ctx.Req.NoError(err)
defer listener.Close()
clientIdentity := ctx.AdminSession.RequireNewIdentityWithOtt(false)
clientConfig := ctx.EnrollIdentity(clientIdentity.Id)
@@ -328,8 +350,11 @@ func Test_ServerConnCloseWritePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
ctx.Req.NoError(err)
defer listener.Close()
clientIdentity := ctx.AdminSession.RequireNewIdentityWithOtt(false)
clientConfig := ctx.EnrollIdentity(clientIdentity.Id)
@@ -393,8 +418,11 @@ func Test_ClientConnCloseWritePropagation(t *testing.T) {
service := ctx.AdminSession.RequireNewServiceAccessibleToAll("smartrouting")
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
listener, err := context.Listen(service.Name)
ctx.Req.NoError(err)
defer listener.Close()
defer func() {
ctx.Req.NoError(listener.Close())
tests/data_flow_hs_rotating_test.go
+6
View File
@@ -73,6 +73,8 @@ func testClientFirstWithStrategy(t *testing.T, strategy string) {
var serverContexts []ziti.Context
for i := 0; i < 3; i++ {
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
serverContexts = append(serverContexts, context)
serverContextC <- context
}
@@ -220,6 +222,8 @@ func testServerFirstWithStrategy(t *testing.T, strategy string) {
for i := 0; i < 3; i++ {
_, context := ctx.AdminSession.RequireCreateSdkContext()
defer context.Close()
serverContexts = append(serverContexts, context)
serverContextC <- context
}
@@ -279,6 +283,8 @@ func testServerFirstWithStrategy(t *testing.T, strategy string) {
}
conn := ctx.WrapConn(clientContext.Dial(service.Name))
defer conn.Close()
name := conn.ReadString(1024, time.Second)
conn.WriteString("hello, "+name, time.Second)
logger.Debugf("%v: done", i+1)
tests/data_flow_hs_test.go
+5
View File
@@ -38,12 +38,17 @@ func Test_HSDataflow(t *testing.T) {
ctx.CreateEnrollAndStartEdgeRouter()
_, hostContext1 := ctx.AdminSession.RequireCreateSdkContext()
defer hostContext1.Close()
listener1, err := hostContext1.Listen(service.Name)
ctx.Req.NoError(err)
_, hostContext2 := ctx.AdminSession.RequireCreateSdkContext()
defer hostContext2.Close()
listener2, err := hostContext2.Listen(service.Name)
ctx.Req.NoError(err)
defer listener2.Close()
serverHandler := func(conn *testServerConn) error {
for {
tests/data_flow_test.go
+6
View File
@@ -35,8 +35,11 @@ func Test_Dataflow(t *testing.T) {
ctx.CreateEnrollAndStartEdgeRouter()
_, hostContext := ctx.AdminSession.RequireCreateSdkContext()
defer hostContext.Close()
listener, err := hostContext.Listen(service.Name)
ctx.Req.NoError(err)
defer listener.Close()
testServer := newTestServer(listener, func(conn *testServerConn) error {
for {
@@ -57,7 +60,10 @@ func Test_Dataflow(t *testing.T) {
testServer.start()
_, clientContext := ctx.AdminSession.RequireCreateSdkContext()
defer clientContext.Close()
conn := ctx.WrapConn(clientContext.Dial(service.Name))
defer conn.Close()
name := eid.New()
conn.WriteString(name, time.Second)