gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 23:59:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #876 from openziti/router-control-channel-update

Browse Source 
Update router side to use channel library for control channel
This commit is contained in:
Paul Lorenz
2022-01-28 17:06:33 -05:00
committed by GitHub
parent e079e38072 01fb919d68
commit 46ad354a10
19 changed files with 122 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -42,7 +42,7 @@ require (
github.com/miekg/dns v1.1.45
github.com/mitchellh/mapstructure v1.4.3
github.com/netfoundry/secretstream v0.1.2
github.com/openziti/fabric v0.17.30
github.com/openziti/fabric v0.17.31
github.com/openziti/foundation v0.16.18
github.com/openziti/sdk-golang v0.15.136
github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
@@ -56,7 +56,7 @@ require (
github.com/teris-io/shortid v0.0.0-20201117134242-e59966efd125
github.com/xeipuuv/gojsonschema v1.2.0
go.etcd.io/bbolt v1.3.5-0.20200615073812-232d8fc87f50
golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8
golang.org/x/crypto v0.0.0-20220128200615-198e4374d7ed
golang.org/x/net v0.0.0-20211209124913-491a49abca63
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
golang.org/x/sys v0.0.0-20220114195835-da31bd327af9
go.sum
+6 -3
View File
@@ -647,6 +647,8 @@ github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d/go.mod h1:01TrycV0kFyex
github.com/michaelquigley/pfxlog v0.3.1/go.mod h1:pwJmJ9nN787xk9U+0z8/5nP2GMhU6GGCTfgBWGpiXaQ=
github.com/michaelquigley/pfxlog v0.6.3 h1:/od363CtyX4z03vhzyJnfzOevXEYihGHVnLVI64a/e0=
github.com/michaelquigley/pfxlog v0.6.3/go.mod h1:D2vg1tPyPdSXWWkSnGk6Fomwh5b3clwVJDUh71tq8Sk=
github.com/michaelquigley/pfxlog v0.6.9 h1:K/weH6ARu58aEDQi0ccinItvV958CeO6Ri4jqeFwd/w=
github.com/michaelquigley/pfxlog v0.6.9/go.mod h1:D2vg1tPyPdSXWWkSnGk6Fomwh5b3clwVJDUh71tq8Sk=
github.com/microcosm-cc/bluemonday v1.0.1/go.mod h1:hsXNsILzKxV+sX77C5b8FSuKF00vh2OMYv+xgHpAMF4=
github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso=
@@ -710,8 +712,8 @@ github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTm
github.com/openziti/channel v0.18.2 h1:H0gsINZGr4AGK7ryHSECbCONQrPRQuV37sRcqbrCQZU=
github.com/openziti/channel v0.18.2/go.mod h1:wmpGSNQranDE7jsiT+Cv31UzOO3z+m69JxE2tzYwkyY=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/fabric v0.17.30 h1:7T9FoWpCynShUgZVrKbX4Yrw3CXdB8om1CRgfaWN/wY=
github.com/openziti/fabric v0.17.30/go.mod h1:snmb4suNPo+6lh/H2/4ohfPA2YWkxbcswvhdH23QczQ=
github.com/openziti/fabric v0.17.31 h1:gvPnRrejmMzMy2wyotair0lcaBYGZ2hwYWWWNF2uPU8=
github.com/openziti/fabric v0.17.31/go.mod h1:sRf0iQ9+0bpZK1HKW/1QHvRaEbijyWqK33qxBmqV5Pc=
github.com/openziti/foundation v0.16.17/go.mod h1:D8IchItaO2SvK/fc/JpX9yF0vNymmOw2gzeNqwHz4+g=
github.com/openziti/foundation v0.16.18 h1:Ietmq4VwVP+Vt+D1H/O75RAyS4XAmHWBjR8RftxlJC0=
github.com/openziti/foundation v0.16.18/go.mod h1:D8IchItaO2SvK/fc/JpX9yF0vNymmOw2gzeNqwHz4+g=
@@ -968,8 +970,9 @@ golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPh
golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8 h1:kACShD3qhmr/3rLmg1yXyt+N4HcwutKyPRB93s54TIU=
golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/crypto v0.0.0-20220128200615-198e4374d7ed h1:YoWVYYAfvQ4ddHv3OKmIvX7NCAhFGTj62VP2l2kfBbA=
golang.org/x/crypto v0.0.0-20220128200615-198e4374d7ed/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
pb/edge_ctrl_pb/messages.go
+13 -1
View File
@@ -5,6 +5,18 @@ import (
"github.com/openziti/sdk-golang/ziti"
)
func (m *ClientHello) GetContentType() int32 {
return int32(ContentType_ClientHelloType)
}
func (m *ServerHello) GetContentType() int32 {
return int32(ContentType_ServerHelloType)
}
func (m *RequestClientReSync) GetContentType() int32 {
return int32(ContentType_RequestClientReSyncType)
}
func (m *CreateCircuitRequest) GetContentType() int32 {
return int32(ContentType_CreateCircuitRequestType)
}
@@ -70,7 +82,7 @@ func (request *CreateTunnelTerminatorRequest) GetContentType() int32 {
return int32(ContentType_CreateTunnelTerminatorRequestType)
}
func (request *EnrollmentExtendRouterVerifyRequest) GetContentType() int32{
func (request *EnrollmentExtendRouterVerifyRequest) GetContentType() int32 {
return int32(ContentType_EnrollmentExtendRouterVerifyRequestType)
}
router/fabric/heartbeat.go
+6 -6
View File
@@ -19,17 +19,17 @@ package fabric
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/runner"
"github.com/openziti/foundation/channel2"
"time"
)
const maxTokensPerMessage = 10000
type heartbeatOperation struct {
ctrl channel2.Channel
ctrl channel.Channel
*runner.BaseOperation
tokenProvider TokenProvider
}
@@ -39,7 +39,7 @@ type TokenProvider interface {
flushRecentlyRemoved()
}
func newHeartbeatOperation(ctrl channel2.Channel, frequency time.Duration, tokenProvider TokenProvider) *heartbeatOperation {
func newHeartbeatOperation(ctrl channel.Channel, frequency time.Duration, tokenProvider TokenProvider) *heartbeatOperation {
return &heartbeatOperation{
ctrl: ctrl,
tokenProvider: tokenProvider,
@@ -56,7 +56,7 @@ func (operation *heartbeatOperation) Run() error {
}
func (operation *heartbeatOperation) beat(tokens []string) {
var msgs []*channel2.Message
var msgs []*channel.Message
pfxlog.Logger().Tracef("heartbeat tokens: %d", len(tokens))
@@ -72,7 +72,7 @@ func (operation *heartbeatOperation) beat(tokens []string) {
pfxlog.Logger().Panic("could not marshal SessionHeartbeat type (1)")
}
msgs = append(msgs, channel2.NewMessage(env.ApiSessionHeartbeatType, bodyBytes))
msgs = append(msgs, channel.NewMessage(env.ApiSessionHeartbeatType, bodyBytes))
tokens = nil
} else {
@@ -87,7 +87,7 @@ func (operation *heartbeatOperation) beat(tokens []string) {
}
tokens = tokens[maxTokensPerMessage:]
msgs = append(msgs, channel2.NewMessage(env.ApiSessionHeartbeatType, bodyBytes))
msgs = append(msgs, channel.NewMessage(env.ApiSessionHeartbeatType, bodyBytes))
}
}
router/fabric/manager.go
+6 -5
View File
@@ -22,6 +22,7 @@ import (
"github.com/golang/protobuf/proto"
"github.com/kataras/go-events"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/runner"
"github.com/openziti/foundation/channel2"
@@ -62,8 +63,8 @@ type StateManager interface {
RemoveConnectedApiSessionWithChannel(token string, underlay channel2.Channel)
AddApiSessionRemovedListener(token string, callBack func(token string)) RemoveListener
StartHeartbeat(channel channel2.Channel, seconds int, closeNotify <-chan struct{})
ValidateSessions(ch channel2.Channel, chunkSize uint32, minInterval, maxInterval time.Duration)
StartHeartbeat(channel channel.Channel, seconds int, closeNotify <-chan struct{})
ValidateSessions(ch channel.Channel, chunkSize uint32, minInterval, maxInterval time.Duration)
DumpApiSessions(c *bufio.ReadWriter) error
MarkSyncInProgress(trackerId string)
@@ -294,7 +295,7 @@ func (sm *StateManagerImpl) getApiSessionRemovedEventName(token string) events.E
return events.EventName(eventName)
}
func (sm *StateManagerImpl) StartHeartbeat(ctrl channel2.Channel, intervalSeconds int, closeNotify <-chan struct{}) {
func (sm *StateManagerImpl) StartHeartbeat(ctrl channel.Channel, intervalSeconds int, closeNotify <-chan struct{}) {
sm.heartbeatOperation = newHeartbeatOperation(ctrl, time.Duration(intervalSeconds)*time.Second, sm)
var err error
@@ -455,7 +456,7 @@ func (self *MapWithMutex) Put(ch channel2.Channel, f func()) {
self.m[ch] = f
}
func (sm *StateManagerImpl) ValidateSessions(ch channel2.Channel, chunkSize uint32, minInterval, maxInterval time.Duration) {
func (sm *StateManagerImpl) ValidateSessions(ch channel.Channel, chunkSize uint32, minInterval, maxInterval time.Duration) {
sessionTokens := sm.sessions.Keys()
for len(sessionTokens) > 0 {
@@ -481,7 +482,7 @@ func (sm *StateManagerImpl) ValidateSessions(ch channel2.Channel, chunkSize uint
return
}
msg := channel2.NewMessage(request.GetContentType(), body)
msg := channel.NewMessage(request.GetContentType(), body)
if err := ch.Send(msg); err != nil {
logrus.WithError(err).Error("failed to send validate sessions request")
return
router/handler_edge_ctrl/apiSessionAdded.go
+12 -13
View File
@@ -22,11 +22,12 @@ import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/sync_strats"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/util/concurrenz"
"github.com/sirupsen/logrus"
"sync"
@@ -34,7 +35,7 @@ import (
)
type apiSessionAddedHandler struct {
control channel2.Channel
control channel.Channel
sm fabric.StateManager
syncTracker *apiSessionSyncTracker
@@ -45,9 +46,9 @@ type apiSessionAddedHandler struct {
trackerLock sync.Mutex
}
func NewApiSessionAddedHandler(sm fabric.StateManager, control channel2.Channel) *apiSessionAddedHandler {
func NewApiSessionAddedHandler(sm fabric.StateManager, binding channel.Binding) *apiSessionAddedHandler {
handler := &apiSessionAddedHandler{
control: control,
control: binding.GetChannel(),
sm: sm,
reqChan: make(chan *apiSessionAddedWithState, 100),
stop: make(chan struct{}, 0),
@@ -55,12 +56,12 @@ func NewApiSessionAddedHandler(sm fabric.StateManager, control channel2.Channel)
go handler.startReceiveSync()
control.AddCloseHandler(handler)
binding.AddCloseHandler(handler)
return handler
}
func (h *apiSessionAddedHandler) HandleClose(_ channel2.Channel) {
func (h *apiSessionAddedHandler) HandleClose(_ channel.Channel) {
if h.stopped.CompareAndSwap(false, true) {
close(h.stop)
}
@@ -70,7 +71,7 @@ func (h *apiSessionAddedHandler) ContentType() int32 {
return env.ApiSessionAddedType
}
func (h *apiSessionAddedHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
func (h *apiSessionAddedHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
go func() {
req := &edge_ctrl_pb.ApiSessionAdded{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
@@ -138,11 +139,9 @@ func (h *apiSessionAddedHandler) syncFailed(err error) {
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)
if err := protobufs.MarshalTyped(resync).Send(h.control); err != nil {
logrus.WithError(err).Error("failed to send request client re-sync message")
}
}
func (h *apiSessionAddedHandler) legacySync(reqWithState *apiSessionAddedWithState) {
@@ -355,7 +354,7 @@ type apiSessionAddedWithState struct {
*edge_ctrl_pb.ApiSessionAdded
}
func parseInstantSyncHeaders(msg *channel2.Message) (string, *sync_strats.InstantSyncState, error) {
func parseInstantSyncHeaders(msg *channel.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{}
router/handler_edge_ctrl/apiSessionRemoved.go
+2 -2
View File
@@ -19,10 +19,10 @@ package handler_edge_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/foundation/channel2"
)
type apiSessionRemovedHandler struct {
@@ -39,7 +39,7 @@ func (h *apiSessionRemovedHandler) ContentType() int32 {
return env.ApiSessionRemovedType
}
func (h *apiSessionRemovedHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *apiSessionRemovedHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
go func() {
req := &edge_ctrl_pb.ApiSessionRemoved{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
router/handler_edge_ctrl/apiSessionUpdated.go
+2 -2
View File
@@ -19,10 +19,10 @@ package handler_edge_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/foundation/channel2"
)
type apiSessionUpdatedHandler struct {
@@ -39,7 +39,7 @@ func (h *apiSessionUpdatedHandler) ContentType() int32 {
return env.ApiSessionUpdatedType
}
func (h *apiSessionUpdatedHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *apiSessionUpdatedHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
go func() {
req := &edge_ctrl_pb.ApiSessionUpdated{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
router/handler_edge_ctrl/extendEnrollmentCerts.go
+5 -3
View File
@@ -21,10 +21,11 @@ import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/internal/cert"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
nfpem "github.com/openziti/foundation/util/pem"
"time"
@@ -46,7 +47,7 @@ func (h *extendEnrollmentCertsHandler) ContentType() int32 {
return env.EnrollmentCertsResponseType
}
func (h *extendEnrollmentCertsHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *extendEnrollmentCertsHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
go func() {
certs := ch.Underlay().Certificates()
@@ -89,8 +90,9 @@ func (h *extendEnrollmentCertsHandler) HandleReceive(msg *channel2.Message, ch c
verifyRequest := &edge_ctrl_pb.EnrollmentExtendRouterVerifyRequest{
ClientCertPem: enrollmentCerts.ClientCertPem,
}
replyMsg, err := protobufs.MarshalTyped(verifyRequest).WithTimeout(30 * time.Second).SendForReply(ch)
reply := &edge_ctrl_pb.Error{}
err := ch.SendForReplyAndDecode(verifyRequest, 30*time.Second, reply)
err = protobufs.TypedResponse(reply).Unmarshall(replyMsg, err)
if err != nil {
log.WithError(err).Errorf("error during enrollment extension, verification reply produced an error")
router/handler_edge_ctrl/hello.go
+6 -12
View File
@@ -19,10 +19,11 @@ package handler_edge_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/build"
"github.com/openziti/foundation/channel2"
)
type helloHandler struct {
@@ -43,7 +44,7 @@ func (h *helloHandler) ContentType() int32 {
return env.ServerHelloType
}
func (h *helloHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *helloHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
go func() {
serverHello := &edge_ctrl_pb.ServerHello{}
if err := proto.Unmarshal(msg.Body, serverHello); err == nil {
@@ -55,19 +56,12 @@ func (h *helloHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel)
Protocols: h.supportedProtocols,
ProtocolPorts: h.protocolPorts,
}
clientHelloBuff, err := proto.Marshal(clientHello)
if err != nil {
pfxlog.Logger().WithField("cause", err).Error("could not marshal client hello")
return
if err := protobufs.MarshalTyped(clientHello).ReplyTo(msg).Send(ch); err != nil {
pfxlog.Logger().WithError(err).Error("could not send client hello")
}
clientHelloMsg := channel2.NewMessage(env.ClientHelloType, clientHelloBuff)
clientHelloMsg.ReplyTo(msg)
_ = ch.Send(clientHelloMsg)
return
} else {
pfxlog.Logger().WithField("cause", err).Error("could not unmarshal server hello")
pfxlog.Logger().WithError(err).Error("could not unmarshal server hello")
}
}()
}
router/handler_edge_ctrl/serviceList.go
+2 -2
View File
@@ -3,8 +3,8 @@ package handler_edge_ctrl
import (
"encoding/json"
"github.com/golang/protobuf/proto"
"github.com/openziti/channel"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/sdk-golang/ziti/edge"
"github.com/sirupsen/logrus"
)
@@ -23,7 +23,7 @@ func (self *ServiceListHandler) ContentType() int32 {
return int32(edge_ctrl_pb.ContentType_ServiceListType)
}
func (self *ServiceListHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *ServiceListHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
serviceList := &edge_ctrl_pb.ServicesList{}
if err := proto.Unmarshal(msg.Body, serviceList); err == nil {
logrus.Debugf("received services list with %v entries", len(serviceList.Services))
router/handler_edge_ctrl/sessionRemoved.go
+2 -2
View File
@@ -19,10 +19,10 @@ package handler_edge_ctrl
import (
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/foundation/channel2"
)
type sessionRemovedHandler struct {
@@ -39,7 +39,7 @@ func (h *sessionRemovedHandler) ContentType() int32 {
return env.SessionRemovedType
}
func (h *sessionRemovedHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *sessionRemovedHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
go func() {
req := &edge_ctrl_pb.SessionRemoved{}
if err := proto.Unmarshal(msg.Body, req); err == nil {
router/xgress_common/messages.go
+4 -3
View File
@@ -2,8 +2,9 @@ package xgress_common
import (
"github.com/golang/protobuf/proto"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/pkg/errors"
)
@@ -11,7 +12,7 @@ const (
PayloadFlagsHeader uint8 = 0x10
)
func GetResultOrFailure(msg *channel2.Message, err error, result channel2.TypedMessage) error {
func GetResultOrFailure(msg *channel.Message, err error, result protobufs.TypedMessage) error {
if err != nil {
return err
}
@@ -31,7 +32,7 @@ func GetResultOrFailure(msg *channel2.Message, err error, result channel2.TypedM
return proto.Unmarshal(msg.Body, result)
}
func CheckForFailureResult(msg *channel2.Message, err error, successType edge_ctrl_pb.ContentType) error {
func CheckForFailureResult(msg *channel.Message, err error, successType edge_ctrl_pb.ContentType) error {
if err != nil {
return err
}
router/xgress_edge/certchecker.go
+4 -4
View File
@@ -21,11 +21,11 @@ import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/enroll"
"github.com/openziti/edge/router/internal/edgerouter"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/util/concurrenz"
"github.com/pkg/errors"
@@ -44,7 +44,7 @@ type CertExtender interface {
type CertExpirationChecker struct {
id *identity.TokenId
closeNotify chan struct{}
ctrl channel2.Channel
ctrl channel.Channel
edgeConfig *edgerouter.Config
certsUpdated chan struct{}
@@ -57,7 +57,7 @@ type CertExpirationChecker struct {
extender CertExtender
}
func NewCertExpirationChecker(id *identity.TokenId, edgeConfig *edgerouter.Config, ctrl channel2.Channel, closeNotify chan struct{}) *CertExpirationChecker {
func NewCertExpirationChecker(id *identity.TokenId, edgeConfig *edgerouter.Config, ctrl channel.Channel, closeNotify chan struct{}) *CertExpirationChecker {
ret := &CertExpirationChecker{
id: id,
closeNotify: closeNotify,
@@ -164,7 +164,7 @@ func (self *CertExpirationChecker) ExtendEnrollment() error {
return fmt.Errorf("could not marshal enrollment extension request: %v", err)
}
msg := channel2.NewMessage(env.EnrollmentExtendRouterRequestType, body)
msg := channel.NewMessage(env.EnrollmentExtendRouterRequestType, body)
if err = self.ctrl.Send(msg); err != nil {
return fmt.Errorf("could not send enrollment extension request, error: %v", err)
router/xgress_edge/certchecker_test.go
+11 -87
View File
@@ -7,9 +7,9 @@ import (
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"github.com/openziti/channel"
"github.com/openziti/edge/eid"
"github.com/openziti/edge/router/internal/edgerouter"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/util/concurrenz"
"github.com/openziti/foundation/util/tlz"
@@ -536,6 +536,16 @@ type simpleTestChannel struct {
isClosed bool
}
func (ch *simpleTestChannel) Send(s channel.Sendable) error {
//TODO implement me
panic("implement me")
}
func (ch *simpleTestChannel) Underlay() channel.Underlay {
//TODO implement me
panic("implement me")
}
func (ch *simpleTestChannel) StartRx() {
}
@@ -563,88 +573,6 @@ func (ch *simpleTestChannel) SetLogicalName(string) {
panic("implement SetLogicalName")
}
func (ch *simpleTestChannel) Bind(channel2.BindHandler) error {
panic("implement Bind")
}
func (ch *simpleTestChannel) AddPeekHandler(channel2.PeekHandler) {
panic("implement AddPeekHandler")
}
func (ch *simpleTestChannel) AddTransformHandler(channel2.TransformHandler) {
panic("implement AddTransformHandler")
}
func (ch *simpleTestChannel) AddReceiveHandler(channel2.ReceiveHandler) {
panic("implement AddTypedReceiveHandler")
}
func (ch *simpleTestChannel) AddErrorHandler(channel2.ErrorHandler) {
panic("implement me")
}
func (ch *simpleTestChannel) AddCloseHandler(channel2.CloseHandler) {
panic("implement AddErrorHandler")
}
func (ch *simpleTestChannel) SetUserData(interface{}) {
panic("implement SetUserData")
}
func (ch *simpleTestChannel) GetUserData() interface{} {
panic("implement GetUserData")
}
func (ch *simpleTestChannel) Send(*channel2.Message) error {
return nil
}
func (ch *simpleTestChannel) SendWithPriority(*channel2.Message, channel2.Priority) error {
return nil
}
func (ch *simpleTestChannel) SendAndSync(m *channel2.Message) (chan error, error) {
return ch.SendAndSyncWithPriority(m, channel2.Standard)
}
func (ch *simpleTestChannel) SendAndSyncWithPriority(*channel2.Message, channel2.Priority) (chan error, error) {
result := make(chan error, 1)
result <- nil
return result, nil
}
func (ch *simpleTestChannel) SendWithTimeout(*channel2.Message, time.Duration) error {
return nil
}
func (ch *simpleTestChannel) SendPrioritizedWithTimeout(*channel2.Message, channel2.Priority, time.Duration) error {
return nil
}
func (ch *simpleTestChannel) SendAndWaitWithTimeout(*channel2.Message, time.Duration) (*channel2.Message, error) {
panic("implement SendAndWaitWithTimeout")
}
func (ch *simpleTestChannel) SendPrioritizedAndWaitWithTimeout(*channel2.Message, channel2.Priority, time.Duration) (*channel2.Message, error) {
panic("implement SendPrioritizedAndWaitWithTimeout")
}
func (ch *simpleTestChannel) SendAndWait(*channel2.Message) (chan *channel2.Message, error) {
panic("implement SendAndWait")
}
func (ch *simpleTestChannel) SendAndWaitWithPriority(*channel2.Message, channel2.Priority) (chan *channel2.Message, error) {
panic("implement SendAndWaitWithPriority")
}
func (ch *simpleTestChannel) SendForReply(channel2.TypedMessage, time.Duration) (*channel2.Message, error) {
panic("implement SendForReply")
}
func (ch *simpleTestChannel) SendForReplyAndDecode(channel2.TypedMessage, time.Duration, channel2.TypedMessage) error {
return nil
}
func (ch *simpleTestChannel) Close() error {
panic("implement Close")
}
@@ -653,10 +581,6 @@ func (ch *simpleTestChannel) IsClosed() bool {
return ch.isClosed
}
func (ch *simpleTestChannel) Underlay() channel2.Underlay {
panic("implement Underlay")
}
func (ch *simpleTestChannel) GetTimeSinceLastRead() time.Duration {
return 0
}
router/xgress_edge/factory.go
+13 -12
View File
@@ -19,6 +19,7 @@ package xgress_edge
import (
"fmt"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/edge/router/handler_edge_ctrl"
"github.com/openziti/edge/router/internal/apiproxy"
@@ -37,7 +38,7 @@ import (
type Factory struct {
id *identity.TokenId
ctrl channel2.Channel
ctrl channel.Channel
enabled bool
routerConfig *router.Config
edgeRouterConfig *edgerouter.Config
@@ -48,7 +49,7 @@ type Factory struct {
metricsRegistry metrics.Registry
}
func (factory *Factory) Channel() channel2.Channel {
func (factory *Factory) Channel() channel.Channel {
return factory.ctrl
}
@@ -60,8 +61,8 @@ func (factory *Factory) Enabled() bool {
return factory.enabled
}
func (factory *Factory) BindChannel(ch channel2.Channel) error {
factory.ctrl = ch
func (factory *Factory) BindChannel(binding channel.Binding) error {
factory.ctrl = binding.GetChannel()
var parts []string
var hostname string
@@ -88,15 +89,15 @@ func (factory *Factory) BindChannel(ch channel2.Channel) error {
protocolPorts = append(protocolPorts, "ws:"+parts[1])
pfxlog.Logger().Debugf("HelloHandler will contain hostname=[%s] supportedProtocols=%v protocolPorts=%v", hostname, supportedProtocols, protocolPorts)
}
ch.AddReceiveHandler(handler_edge_ctrl.NewHelloHandler(hostname, supportedProtocols, protocolPorts))
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewHelloHandler(hostname, supportedProtocols, protocolPorts))
ch.AddReceiveHandler(handler_edge_ctrl.NewSessionRemovedHandler(factory.stateManager))
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewSessionRemovedHandler(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))
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewApiSessionAddedHandler(factory.stateManager, binding))
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewApiSessionRemovedHandler(factory.stateManager))
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewApiSessionUpdatedHandler(factory.stateManager))
ch.AddReceiveHandler(handler_edge_ctrl.NewExtendEnrollmentCertsHandler(factory.routerConfig.Id, func() {
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewExtendEnrollmentCertsHandler(factory.routerConfig.Id, func() {
factory.certChecker.CertsUpdated()
}))
@@ -107,11 +108,11 @@ func (factory *Factory) NotifyOfReconnect() {
go factory.stateManager.ValidateSessions(factory.Channel(), factory.edgeRouterConfig.SessionValidateChunkSize, factory.edgeRouterConfig.SessionValidateMinInterval, factory.edgeRouterConfig.SessionValidateMaxInterval)
}
func (factory *Factory) GetTraceDecoders() []channel2.TraceMessageDecoder {
func (factory *Factory) GetTraceDecoders() []channel.TraceMessageDecoder {
return nil
}
func (factory *Factory) Run(ctrl channel2.Channel, _ boltz.Db, closeNotify chan struct{}) error {
func (factory *Factory) Run(ctrl channel.Channel, _ boltz.Db, closeNotify chan struct{}) error {
factory.ctrl = ctrl
factory.stateManager.StartHeartbeat(ctrl, factory.edgeRouterConfig.HeartbeatIntervalSeconds, closeNotify)
factory.certChecker = NewCertExpirationChecker(factory.routerConfig.Id, factory.edgeRouterConfig, ctrl, closeNotify)
router/xgress_edge/listener.go
+8 -7
View File
@@ -21,6 +21,8 @@ import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/internal/cert"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/xgress_common"
@@ -28,7 +30,6 @@ import (
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
"github.com/openziti/foundation/util/protobufs"
"github.com/openziti/sdk-golang/ziti/edge"
"github.com/pkg/errors"
"time"
@@ -154,7 +155,7 @@ func (self *edgeClientConn) processConnect(req *channel2.Message, ch channel2.Ch
response := &edge_ctrl_pb.CreateCircuitResponse{}
timeout := self.listener.options.Options.GetCircuitTimeout
responseMsg, err := self.listener.factory.Channel().SendForReply(request, timeout)
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(self.listener.factory.Channel())
if err = getResultOrFailure(responseMsg, err, response); err != nil {
log.WithError(err).Warn("failed to dial fabric")
@@ -242,7 +243,7 @@ func (self *edgeClientConn) processBind(req *channel2.Message, ch channel2.Chann
}
timeout := self.listener.factory.DefaultRequestTimeout()
responseMsg, err := self.listener.factory.Channel().SendForReply(request, timeout)
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(self.listener.factory.Channel())
if err = xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_CreateTerminatorResponseType); err != nil {
log.WithError(err).Warn("error creating terminator")
messageSink.close(false, "") // don't notify here, as we're notifying next line with a response
@@ -294,7 +295,7 @@ func (self *edgeClientConn) removeTerminator(terminator *edgeTerminator) error {
}
timeout := self.listener.factory.DefaultRequestTimeout()
responseMsg, err := self.listener.factory.Channel().SendForReply(request, timeout)
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(self.listener.factory.Channel())
return xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_RemoveTerminatorResponseType)
}
@@ -340,7 +341,7 @@ func (self *edgeClientConn) processUpdateBind(req *channel2.Message, ch channel2
log.Debug("updating terminator")
timeout := self.listener.factory.DefaultRequestTimeout()
responseMsg, err := self.listener.factory.Channel().SendForReply(request, timeout)
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(self.listener.factory.Channel())
if err := xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_UpdateTerminatorResponseType); err != nil {
log.WithError(err).Error("terminator update failed")
} else {
@@ -372,7 +373,7 @@ func (self *edgeClientConn) processHealthEvent(req *channel2.Message, ch channel
WithField("checkPassed", checkPassed)
log.Debug("sending health event")
if err := protobufs.Send(self.listener.factory.Channel(), request); err != nil {
if err := protobufs.MarshalTyped(request).Send(self.listener.factory.Channel()); err != nil {
log.WithError(err).Error("send failed")
}
}
@@ -449,7 +450,7 @@ func (self *edgeClientConn) sendStateClosedReply(message string, req *channel2.M
}
}
func getResultOrFailure(msg *channel2.Message, err error, result channel2.TypedMessage) error {
func getResultOrFailure(msg *channel.Message, err error, result protobufs.TypedMessage) error {
if err != nil {
return err
}
router/xgress_edge_tunnel/fabric.go
+10 -9
View File
@@ -22,12 +22,13 @@ import (
"github.com/google/uuid"
"github.com/michaelquigley/pfxlog"
"github.com/netfoundry/secretstream/kx"
"github.com/openziti/channel"
"github.com/openziti/channel/protobufs"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/edge/router/xgress_common"
"github.com/openziti/edge/tunnel"
"github.com/openziti/fabric/build"
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/util/concurrenz"
"github.com/openziti/sdk-golang/ziti/edge"
@@ -156,7 +157,7 @@ func (self *fabricProvider) authenticate() error {
},
}
respMsg, err := self.factory.Channel().SendForReply(request, 30*time.Second)
respMsg, err := protobufs.MarshalTyped(request).WithTimeout(30 * time.Second).SendForReply(self.factory.Channel())
resp := &edge_ctrl_pb.CreateApiSessionResponse{}
if err = xgress_common.GetResultOrFailure(respMsg, err, resp); err != nil {
@@ -196,7 +197,7 @@ func (self *fabricProvider) TunnelService(service tunnel.Service, terminatorIden
PeerData: peerData,
}
responseMsg, err := self.factory.Channel().SendForReply(request, service.GetDialTimeout())
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(service.GetDialTimeout()).SendForReply(self.factory.Channel())
response := &edge_ctrl_pb.CreateCircuitForServiceResponse{}
if err = xgress_common.GetResultOrFailure(responseMsg, err, response); err != nil {
@@ -311,7 +312,7 @@ func (self *fabricProvider) establishTerminator(terminator *tunnelTerminator) er
request.GetContentType()
response := &edge_ctrl_pb.CreateTunnelTerminatorResponse{}
responseMsg, err := self.factory.Channel().SendForReply(request, self.factory.DefaultRequestTimeout())
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(self.factory.DefaultRequestTimeout()).SendForReply(self.factory.Channel())
if err = xgress_common.GetResultOrFailure(responseMsg, err, response); err != nil {
log.WithError(err).Error("error creating terminator")
return err
@@ -341,8 +342,8 @@ func (self *fabricProvider) establishTerminator(terminator *tunnelTerminator) er
}
func (self *fabricProvider) removeTerminator(terminator *tunnelTerminator) error {
msg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorRequestType), []byte(terminator.terminatorId))
responseMsg, err := self.factory.Channel().SendAndWaitWithTimeout(msg, self.factory.DefaultRequestTimeout())
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorRequestType), []byte(terminator.terminatorId))
responseMsg, err := msg.WithTimeout(self.factory.DefaultRequestTimeout()).SendForReply(self.factory.Channel())
return xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_RemoveTunnelTerminatorResponseType)
}
@@ -374,7 +375,7 @@ func (self *fabricProvider) updateTerminator(terminatorId string, cost *uint16,
log.Debug("updating terminator")
responseMsg, err := self.factory.Channel().SendForReply(request, self.factory.DefaultRequestTimeout())
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(self.factory.DefaultRequestTimeout()).SendForReply(self.factory.Channel())
if err := xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_UpdateTunnelTerminatorResponseType); err != nil {
log.WithError(err).Error("terminator update failed")
return err
@@ -385,7 +386,7 @@ func (self *fabricProvider) updateTerminator(terminatorId string, cost *uint16,
}
func (self *fabricProvider) sendHealthEvent(terminatorId string, checkPassed bool) error {
msg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_TunnelHealthEventType), nil)
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_TunnelHealthEventType), nil)
msg.Headers[int32(edge_ctrl_pb.Header_TerminatorId)] = []byte(terminatorId)
msg.PutBoolHeader(int32(edge_ctrl_pb.Header_CheckPassed), checkPassed)
@@ -403,7 +404,7 @@ func (self *fabricProvider) sendHealthEvent(terminatorId string, checkPassed boo
}
func (self *fabricProvider) requestServiceList(lastUpdateToken []byte) {
msg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ListServicesRequestType), lastUpdateToken)
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ListServicesRequestType), lastUpdateToken)
if err := self.factory.Channel().Send(msg); err != nil {
logrus.WithError(err).Error("failed to send service list request to controller")
}
router/xgress_edge_tunnel/factory.go
+8 -8
View File
@@ -17,11 +17,11 @@
package xgress_edge_tunnel
import (
"github.com/openziti/channel"
"github.com/openziti/edge/router/fabric"
"github.com/openziti/edge/router/handler_edge_ctrl"
"github.com/openziti/fabric/router"
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/storage/boltz"
"github.com/openziti/foundation/util/stringz"
@@ -38,7 +38,7 @@ const (
type Factory struct {
id identity.Identity
ctrl channel2.Channel
ctrl channel.Channel
routerConfig *router.Config
stateManager fabric.StateManager
serviceListHandler *handler_edge_ctrl.ServiceListHandler
@@ -48,11 +48,11 @@ type Factory struct {
func (self *Factory) NotifyOfReconnect() {
}
func (self *Factory) GetTraceDecoders() []channel2.TraceMessageDecoder {
func (self *Factory) GetTraceDecoders() []channel.TraceMessageDecoder {
return nil
}
func (self *Factory) Channel() channel2.Channel {
func (self *Factory) Channel() channel.Channel {
return self.ctrl
}
@@ -60,14 +60,14 @@ func (self *Factory) Enabled() bool {
return true
}
func (self *Factory) BindChannel(ch channel2.Channel) error {
self.ctrl = ch
func (self *Factory) BindChannel(binding channel.Binding) error {
self.ctrl = binding.GetChannel()
self.serviceListHandler = handler_edge_ctrl.NewServiceListHandler(self.tunneler.servicePoller.handleServiceListUpdate)
ch.AddReceiveHandler(self.serviceListHandler)
binding.AddTypedReceiveHandler(self.serviceListHandler)
return nil
}
func (self *Factory) Run(ctrl channel2.Channel, _ boltz.Db, notifyClose chan struct{}) error {
func (self *Factory) Run(ctrl channel.Channel, _ boltz.Db, notifyClose chan struct{}) error {
self.ctrl = ctrl
if self.tunneler.listenOptions != nil {
return self.tunneler.Start(notifyClose)