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 #867 from openziti/update-ctrl-channel

Browse Source 
Update for control channel API changes
This commit is contained in:
Paul Lorenz
2022-01-19 09:17:08 -05:00
committed by GitHub
parent 072ed173cb 3f69bfdd43
commit 3577ae4135
25 changed files with 101 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/handler_edge_ctrl/common.go
+8 -8
View File
@@ -12,7 +12,7 @@ import (
"github.com/openziti/fabric/controller/network"
"github.com/openziti/fabric/controller/xt"
"github.com/openziti/fabric/logcontext"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/storage/boltz"
"github.com/openziti/foundation/util/stringz"
@@ -28,7 +28,7 @@ import (
type requestHandler interface {
getAppEnv() *env.AppEnv
getNetwork() *network.Network
getChannel() channel2.Channel
getChannel() channel.Channel
ContentType() int32
Label() string
}
@@ -41,7 +41,7 @@ type UpdateTerminatorRequest interface {
}
type baseRequestHandler struct {
ch channel2.Channel
ch channel.Channel
appEnv *env.AppEnv
}
@@ -53,13 +53,13 @@ func (self *baseRequestHandler) getAppEnv() *env.AppEnv {
return self.appEnv
}
func (self *baseRequestHandler) getChannel() channel2.Channel {
func (self *baseRequestHandler) getChannel() channel.Channel {
return self.ch
}
func (self *baseRequestHandler) returnError(ctx requestContext, err controllerError) {
ctx.CleanupOnError()
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error()))
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error()))
responseMsg.PutUint32Header(edge.ErrorCodeHeader, err.ErrorCode())
responseMsg.ReplyTo(ctx.GetMessage())
logger := pfxlog.
@@ -97,7 +97,7 @@ func (self *baseRequestHandler) logResult(ctx requestContext, err error) {
type requestContext interface {
GetHandler() requestHandler
GetMessage() *channel2.Message
GetMessage() *channel.Message
CleanupOnError()
}
@@ -108,7 +108,7 @@ type sessionRequestContext interface {
type baseSessionRequestContext struct {
handler requestHandler
msg *channel2.Message
msg *channel.Message
err controllerError
sourceRouter *network.Router
session *model.Session
@@ -129,7 +129,7 @@ func (self *baseSessionRequestContext) CleanupOnError() {
}
}
func (self *baseSessionRequestContext) GetMessage() *channel2.Message {
func (self *baseSessionRequestContext) GetMessage() *channel.Message {
return self.msg
}
controller/handler_edge_ctrl/create_circuit.go
+4 -4
View File
@@ -22,14 +22,14 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
)
type createCircuitHandler struct {
baseRequestHandler
}
func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &createCircuitHandler{
baseRequestHandler: baseRequestHandler{
ch: ch,
@@ -55,14 +55,14 @@ func (self *createCircuitHandler) sendResponse(ctx *CreateCircuitRequestContext,
return
}
responseMsg := channel2.NewMessage(response.GetContentType(), body)
responseMsg := channel.NewMessage(response.GetContentType(), body)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to send create circuit response")
}
}
func (self *createCircuitHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *createCircuitHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.CreateCircuitRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateCircuitRequest")
controller/handler_edge_ctrl/create_terminator.go
+4 -4
View File
@@ -25,7 +25,7 @@ import (
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/storage/boltz"
"go.etcd.io/bbolt"
"math"
@@ -35,7 +35,7 @@ type createTerminatorHandler struct {
baseRequestHandler
}
func NewCreateTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewCreateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &createTerminatorHandler{
baseRequestHandler{
ch: ch,
@@ -52,7 +52,7 @@ func (self *createTerminatorHandler) Label() string {
return "create.terminator"
}
func (self *createTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *createTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.CreateTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateTerminatorRequest")
@@ -123,7 +123,7 @@ func (self *createTerminatorHandler) CreateTerminator(ctx *CreateTerminatorReque
log.WithField("terminator", id).Info("created terminator")
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_CreateTerminatorResponseType), []byte(id))
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateTerminatorResponseType), []byte(id))
responseMsg.ReplyTo(ctx.msg)
if err := self.ch.Send(responseMsg); err != nil {
log.WithError(err).Error("failed to send create terminator response")
controller/handler_edge_ctrl/create_tunnel_api_session.go
+4 -4
View File
@@ -4,7 +4,7 @@ import (
"github.com/golang/protobuf/proto"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
)
@@ -13,7 +13,7 @@ type createApiSessionHandler struct {
*TunnelState
}
func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler {
func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler {
return &createApiSessionHandler{
baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv},
TunnelState: tunnelState,
@@ -32,7 +32,7 @@ func (self *createApiSessionHandler) Label() string {
return "tunnel.create.api_session"
}
func (self *createApiSessionHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
func (self *createApiSessionHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
req := &edge_ctrl_pb.CreateApiSessionRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
logrus.WithField("router", self.ch.Id().Token).WithError(err).Error("could not unmarshal CreateApiSessionRequest")
@@ -77,7 +77,7 @@ func (self *createApiSessionHandler) createApiSession(ctx *createApiSessionReque
return
}
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_CreateApiSessionResponseType), body)
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateApiSessionResponseType), body)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
logrus.WithError(err).Error("failed to send response")
controller/handler_edge_ctrl/create_tunnel_circuit.go
+4 -4
View File
@@ -22,7 +22,7 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
)
type createCircuitForServiceHandler struct {
@@ -30,7 +30,7 @@ type createCircuitForServiceHandler struct {
*TunnelState
}
func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler {
func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler {
return &createCircuitForServiceHandler{
baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv},
TunnelState: tunnelState,
@@ -58,14 +58,14 @@ func (self *createCircuitForServiceHandler) sendResponse(ctx *CreateCircuitForSe
return
}
responseMsg := channel2.NewMessage(response.GetContentType(), body)
responseMsg := channel.NewMessage(response.GetContentType(), body)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
log.WithError(err).WithField("service", ctx.req.ServiceName).Error("failed to send CreateCircuitForServiceResponse")
}
}
func (self *createCircuitForServiceHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *createCircuitForServiceHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.CreateCircuitForServiceRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateCircuitForServiceRequest")
controller/handler_edge_ctrl/create_tunnel_terminator.go
+4 -4
View File
@@ -25,7 +25,7 @@ import (
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
"math"
)
@@ -35,7 +35,7 @@ type createTunnelTerminatorHandler struct {
*TunnelState
}
func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler {
func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler {
return &createTunnelTerminatorHandler{
baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv},
TunnelState: tunnelState,
@@ -54,7 +54,7 @@ func (self *createTunnelTerminatorHandler) Label() string {
return "tunnel.create.terminator"
}
func (self *createTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *createTunnelTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.CreateTunnelTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateTerminatorRequest")
@@ -139,7 +139,7 @@ func (self *createTunnelTerminatorHandler) CreateTerminator(ctx *CreateTunnelTer
return
}
responseMsg := channel2.NewMessage(response.GetContentType(), body)
responseMsg := channel.NewMessage(response.GetContentType(), body)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
logger.WithError(err).Error("failed to send CreateTunnelTerminatorResponse")
controller/handler_edge_ctrl/extend_enrollment.go
+3 -3
View File
@@ -23,7 +23,7 @@ import (
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/internal/cert"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
)
type extendEnrollmentHandler struct {
@@ -40,7 +40,7 @@ func (h *extendEnrollmentHandler) ContentType() int32 {
return env.EnrollmentExtendRouterRequestType
}
func (h *extendEnrollmentHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *extendEnrollmentHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
go func() {
req := &edge_ctrl_pb.EnrollmentExtendRouterRequest{}
certs := ch.Underlay().Certificates()
@@ -126,7 +126,7 @@ func (h *extendEnrollmentHandler) HandleReceive(msg *channel2.Message, ch channe
return
}
msg := channel2.NewMessage(env.EnrollmentCertsResponseType, body)
msg := channel.NewMessage(env.EnrollmentCertsResponseType, body)
if err := ch.Send(msg); err != nil {
log.WithError(err).Errorf("request to extend router enrollment failed to send enrollment certificate response")
controller/handler_edge_ctrl/extend_enrollment_verify.go
+4 -4
View File
@@ -22,7 +22,7 @@ import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
)
type extendEnrollmentVerifyHandler struct {
@@ -39,16 +39,16 @@ func (h *extendEnrollmentVerifyHandler) ContentType() int32 {
return env.EnrollmentExtendRouterVerifyRequestType
}
func (h *extendEnrollmentVerifyHandler) respond(respErr *edge_ctrl_pb.Error, msg *channel2.Message, ch channel2.Channel) {
func (h *extendEnrollmentVerifyHandler) respond(respErr *edge_ctrl_pb.Error, msg *channel.Message, ch channel.Channel) {
respErrBody, _ := proto.Marshal(respErr)
respMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), respErrBody)
respMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), respErrBody)
respMsg.ReplyTo(msg)
if err := ch.Send(respMsg); err != nil {
pfxlog.Logger().Errorf("could not send enrollment verification response, channel error: %v", err)
}
}
func (h *extendEnrollmentVerifyHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *extendEnrollmentVerifyHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
go func() {
verifyMsg := &edge_ctrl_pb.EnrollmentExtendRouterVerifyRequest{}
controller/handler_edge_ctrl/health_event.go
+3 -3
View File
@@ -22,7 +22,7 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/metrics"
"time"
)
@@ -33,7 +33,7 @@ type healthEventHandler struct {
serviceHealthCheckFailedCounter metrics.IntervalCounter
}
func NewHealthEventHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry()
return &healthEventHandler{
baseRequestHandler: baseRequestHandler{
@@ -53,7 +53,7 @@ func (self *healthEventHandler) Label() string {
return "health.event"
}
func (self *healthEventHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *healthEventHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.HealthEventRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal health event")
controller/handler_edge_ctrl/heartbeat.go
+3 -3
View File
@@ -21,7 +21,7 @@ import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
)
type sessionHeartbeatHandler struct {
@@ -38,7 +38,7 @@ func (h *sessionHeartbeatHandler) ContentType() int32 {
return env.ApiSessionHeartbeatType
}
func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *sessionHeartbeatHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
go func() {
req := &edge_ctrl_pb.ApiSessionHeartbeat{}
routerId := ch.Id().Token
@@ -63,7 +63,7 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channe
}
msgBytes, _ := proto.Marshal(msgStruct)
msg := channel2.NewMessage(env.ApiSessionRemovedType, msgBytes)
msg := channel.NewMessage(env.ApiSessionRemovedType, msgBytes)
_ = ch.Send(msg)
}
controller/handler_edge_ctrl/hello.go
+2 -2
View File
@@ -22,7 +22,7 @@ import (
"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"
"github.com/openziti/foundation/channel"
)
type helloHandler struct {
@@ -41,7 +41,7 @@ func (h *helloHandler) ContentType() int32 {
return env.ClientHelloType
}
func (h *helloHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *helloHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
respHello := &edge_ctrl_pb.ClientHello{}
if err := proto.Unmarshal(msg.Body, respHello); err != nil {
pfxlog.Logger().WithError(err).Error("could not unmarshal clientHello after serverHello")
controller/handler_edge_ctrl/list_tunnel_services.go
+4 -4
View File
@@ -5,7 +5,7 @@ import (
"github.com/golang/protobuf/proto"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/storage/ast"
"github.com/sirupsen/logrus"
"time"
@@ -16,7 +16,7 @@ type listTunnelServicesHandler struct {
*TunnelState
}
func NewListTunnelServicesHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler {
func NewListTunnelServicesHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler {
return &listTunnelServicesHandler{
baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv},
TunnelState: tunnelState,
@@ -35,7 +35,7 @@ func (self *listTunnelServicesHandler) Label() string {
return "tunnel.list.services"
}
func (self *listTunnelServicesHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
func (self *listTunnelServicesHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
logger := logrus.WithField("router", self.ch.Id().Token)
ctx := &listTunnelServicesRequestContext{
@@ -134,7 +134,7 @@ func (self *listTunnelServicesHandler) listServices(ctx *listTunnelServicesReque
return
}
serviceListMsg := channel2.NewMessage(serviceList.GetContentType(), body)
serviceListMsg := channel.NewMessage(serviceList.GetContentType(), body)
if err = self.ch.Send(serviceListMsg); err != nil {
logger.Error("failed to send services list")
}
controller/handler_edge_ctrl/remove_terminator.go
+4 -4
View File
@@ -23,7 +23,7 @@ import (
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
)
@@ -31,7 +31,7 @@ type removeTerminatorHandler struct {
baseRequestHandler
}
func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &removeTerminatorHandler{
baseRequestHandler{
ch: ch,
@@ -51,14 +51,14 @@ func (self *removeTerminatorHandler) Label() string {
func (self *removeTerminatorHandler) sendResponse(ctx *RemoveTerminatorRequestContext) {
log := pfxlog.ContextLogger(self.ch.Label())
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTerminatorResponseType), nil)
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTerminatorResponseType), nil)
responseMsg.ReplyTo(ctx.msg)
if err := self.ch.Send(responseMsg); err != nil {
log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to send remove terminator response")
}
}
func (self *removeTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *removeTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.RemoveTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal RemoveTerminator")
controller/handler_edge_ctrl/remove_tunnel_terminator.go
+4 -4
View File
@@ -20,7 +20,7 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
)
@@ -28,7 +28,7 @@ type removeTunnelTerminatorHandler struct {
baseRequestHandler
}
func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &removeTunnelTerminatorHandler{
baseRequestHandler{
ch: ch,
@@ -45,7 +45,7 @@ func (self *removeTunnelTerminatorHandler) Label() string {
return "tunnel.remove.terminator"
}
func (self *removeTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
func (self *removeTunnelTerminatorHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
ctx := &RemoveTunnelTerminatorRequestContext{
baseSessionRequestContext: baseSessionRequestContext{handler: self, msg: msg},
terminatorId: string(msg.Body),
@@ -77,7 +77,7 @@ func (self *removeTunnelTerminatorHandler) RemoveTerminator(ctx *RemoveTunnelTer
logger.Info("removed terminator")
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorResponseType), nil)
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorResponseType), nil)
responseMsg.ReplyTo(ctx.msg)
if err := self.ch.Send(responseMsg); err != nil {
logger.WithError(err).Error("failed to send remove tunnel terminator response")
controller/handler_edge_ctrl/resync.go
+2 -2
View File
@@ -22,7 +22,7 @@ import (
"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"
"github.com/openziti/foundation/channel"
)
type resyncHandler struct {
@@ -41,7 +41,7 @@ func (h *resyncHandler) ContentType() int32 {
return env.RequestClientReSyncType
}
func (h *resyncHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (h *resyncHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
resyncReq := &edge_ctrl_pb.RequestClientReSync{}
if err := proto.Unmarshal(msg.Body, resyncReq); err != nil {
pfxlog.Logger().WithError(err).Error("could not unmarshal RequestClientReSync")
controller/handler_edge_ctrl/tunnel_health_event.go
+3 -3
View File
@@ -20,7 +20,7 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/metrics"
"time"
)
@@ -31,7 +31,7 @@ type tunnelHealthEventHandler struct {
serviceHealthCheckFailedCounter metrics.IntervalCounter
}
func NewTunnelHealthEventHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewTunnelHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry()
return &tunnelHealthEventHandler{
baseRequestHandler: baseRequestHandler{
@@ -51,7 +51,7 @@ func (self *tunnelHealthEventHandler) Label() string {
return "tunnel.health.event"
}
func (self *tunnelHealthEventHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
func (self *tunnelHealthEventHandler) HandleReceive(msg *channel.Message, _ channel.Channel) {
terminatorId, _ := msg.GetStringHeader(int32(edge_ctrl_pb.Header_TerminatorId))
checkPassed, _ := msg.GetBoolHeader(int32(edge_ctrl_pb.Header_CheckPassed))
controller/handler_edge_ctrl/update_terminator.go
+4 -4
View File
@@ -23,7 +23,7 @@ import (
"github.com/openziti/edge/controller/persistence"
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
)
@@ -31,7 +31,7 @@ type updateTerminatorHandler struct {
baseRequestHandler
}
func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &updateTerminatorHandler{
baseRequestHandler{
ch: ch,
@@ -48,7 +48,7 @@ func (self *updateTerminatorHandler) Label() string {
return "update.terminator"
}
func (self *updateTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *updateTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.UpdateTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal UpdateTerminator")
@@ -93,7 +93,7 @@ func (self *updateTerminatorHandler) UpdateTerminator(ctx *UpdateTerminatorReque
logger = logger.WithField("serviceId", terminator.Service)
logger.Info("updated terminator")
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTerminatorResponseType), nil)
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTerminatorResponseType), nil)
responseMsg.ReplyTo(ctx.msg)
if err := self.ch.Send(responseMsg); err != nil {
logger.WithError(err).Error("failed to send update terminator response")
controller/handler_edge_ctrl/update_tunnel_terminator.go
+4 -4
View File
@@ -22,7 +22,7 @@ import (
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/edge_common"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/sirupsen/logrus"
)
@@ -30,7 +30,7 @@ type updateTunnelTerminatorHandler struct {
baseRequestHandler
}
func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &updateTunnelTerminatorHandler{
baseRequestHandler{
ch: ch,
@@ -47,7 +47,7 @@ func (self *updateTunnelTerminatorHandler) Label() string {
return "tunnel.update.terminator"
}
func (self *updateTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *updateTunnelTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.UpdateTunnelTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal UpdateTerminator")
@@ -86,7 +86,7 @@ func (self *updateTunnelTerminatorHandler) UpdateTerminator(ctx *UpdateTunnelTer
logger = logger.WithField("serviceId", terminator.Service)
logger.Info("updated terminator")
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTunnelTerminatorResponseType), nil)
responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTunnelTerminatorResponseType), nil)
responseMsg.ReplyTo(ctx.msg)
if err := self.ch.Send(responseMsg); err != nil {
logger.WithError(err).Error("failed to send update tunnel terminator response")
controller/handler_edge_ctrl/validate_sessions.go
+4 -4
View File
@@ -21,7 +21,7 @@ import (
"github.com/michaelquigley/pfxlog"
"github.com/openziti/edge/controller/env"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"go.etcd.io/bbolt"
)
@@ -29,7 +29,7 @@ type validateSessionsHandler struct {
baseRequestHandler
}
func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler {
func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler {
return &validateSessionsHandler{
baseRequestHandler{
ch: ch,
@@ -46,7 +46,7 @@ func (self *validateSessionsHandler) Label() string {
return "validate.sessions"
}
func (self *validateSessionsHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
func (self *validateSessionsHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.ValidateSessionsRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal ValidateSessionsRequest")
@@ -86,7 +86,7 @@ func (self *validateSessionsHandler) validateSessions(req *edge_ctrl_pb.Validate
return
}
msg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_SessionRemovedType), body)
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_SessionRemovedType), body)
if err := self.ch.Send(msg); err != nil {
pfxlog.ContextLogger(self.ch.Label()).WithError(err).Error("failed to send validate sessions request")
return
controller/server/controller.go
+9 -8
View File
@@ -22,6 +22,7 @@ import (
sync2 "github.com/openziti/edge/controller/sync_strats"
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/controller/api_impl"
"github.com/openziti/foundation/channel"
"io/ioutil"
"sync"
"time"
@@ -141,10 +142,10 @@ func (c *Controller) SetHostController(h env.HostController) {
}
}
func (c *Controller) GetCtrlHandlers(ch channel2.Channel) []channel2.ReceiveHandler {
func (c *Controller) GetCtrlHandlers(ch channel.Channel) []channel.ReceiveHandler {
tunnelState := handler_edge_ctrl.NewTunnelState()
return []channel2.ReceiveHandler{
return []channel.ReceiveHandler{
handler_edge_ctrl.NewSessionHeartbeatHandler(c.AppEnv),
handler_edge_ctrl.NewCreateCircuitHandler(c.AppEnv, ch),
handler_edge_ctrl.NewCreateTerminatorHandler(c.AppEnv, ch),
@@ -321,12 +322,12 @@ func (c *Controller) Shutdown() {
type subctrl struct {
parent *Controller
channel channel2.Channel
channel channel.Channel
}
func (c *subctrl) GetTraceDecoders() []channel2.TraceMessageDecoder {
return []channel2.TraceMessageDecoder{
&edge_ctrl_pb.Decoder{},
func (c *subctrl) GetTraceDecoders() []channel.TraceMessageDecoder {
return []channel.TraceMessageDecoder{
edge_ctrl_pb.Decoder{},
}
}
@@ -341,7 +342,7 @@ func (c *subctrl) Enabled() bool {
return c.parent.Enabled()
}
func (c *subctrl) BindChannel(ch channel2.Channel) error {
func (c *subctrl) BindChannel(ch channel.Channel) error {
for _, h := range c.parent.GetCtrlHandlers(ch) {
ch.AddReceiveHandler(h)
}
@@ -349,7 +350,7 @@ func (c *subctrl) BindChannel(ch channel2.Channel) error {
return nil
}
func (c *subctrl) Run(channel2.Channel, boltz.Db, chan struct{}) error {
func (c *subctrl) Run(channel.Channel, boltz.Db, chan struct{}) error {
return nil
}
controller/sync_strats/rtx.go
+4 -3
View File
@@ -22,6 +22,7 @@ import (
"github.com/openziti/edge/controller/model"
"github.com/openziti/edge/eid"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/util/concurrenz"
"github.com/pkg/errors"
@@ -36,7 +37,7 @@ type RouterSender struct {
Id string
EdgeRouter *model.EdgeRouter
Router *network.Router
send chan *channel2.Message
send chan *channel.Message
closeNotify chan struct{}
running concurrenz.AtomicBoolean
@@ -48,7 +49,7 @@ func newRouterSender(edgeRouter *model.EdgeRouter, router *network.Router, sendB
Id: eid.New(),
EdgeRouter: edgeRouter,
Router: router,
send: make(chan *channel2.Message, sendBufferSize),
send: make(chan *channel.Message, sendBufferSize),
closeNotify: make(chan struct{}, 0),
running: concurrenz.AtomicBoolean(1),
RouterState: env.NewLockingRouterStatus(),
@@ -93,7 +94,7 @@ func (rtx *RouterSender) logger() *logrus.Entry {
WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed())
}
func (rtx *RouterSender) Send(msg *channel2.Message) error {
func (rtx *RouterSender) Send(msg *channel.Message) error {
if !rtx.running.Get() {
rtx.logger().Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
return errors.Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
controller/sync_strats/sync_instant.go
+8 -8
View File
@@ -29,7 +29,7 @@ import (
"github.com/openziti/edge/pb/edge_ctrl_pb"
"github.com/openziti/fabric/build"
"github.com/openziti/fabric/controller/network"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/storage/ast"
"github.com/openziti/foundation/util/debugz"
"go.etcd.io/bbolt"
@@ -80,8 +80,8 @@ type InstantStrategy struct {
rtxMap *routerTxMap
helloHandler channel2.ReceiveHandler
resyncHandler channel2.ReceiveHandler
helloHandler channel.ReceiveHandler
resyncHandler channel.ReceiveHandler
ae *env.AppEnv
routerConnectedQueue chan *RouterSender
@@ -222,7 +222,7 @@ func (strategy *InstantStrategy) ApiSessionUpdated(apiSession *persistence.ApiSe
strategy.rtxMap.Range(func(rtx *RouterSender) {
content, _ := proto.Marshal(apiSessionAdded)
msg := channel2.NewMessage(env.ApiSessionUpdatedType, content)
msg := channel.NewMessage(env.ApiSessionUpdatedType, content)
msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
msg.Headers[env.SyncStrategyStateHeader] = nil
_ = rtx.Send(msg)
@@ -236,7 +236,7 @@ func (strategy *InstantStrategy) ApiSessionDeleted(apiSession *persistence.ApiSe
strategy.rtxMap.Range(func(rtx *RouterSender) {
content, _ := proto.Marshal(sessionRemoved)
msg := channel2.NewMessage(env.ApiSessionRemovedType, content)
msg := channel.NewMessage(env.ApiSessionRemovedType, content)
_ = rtx.Send(msg)
})
}
@@ -248,7 +248,7 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) {
strategy.rtxMap.Range(func(rtx *RouterSender) {
content, _ := proto.Marshal(sessionRemoved)
msg := channel2.NewMessage(env.SessionRemovedType, content)
msg := channel.NewMessage(env.SessionRemovedType, content)
_ = rtx.Send(msg)
})
}
@@ -324,7 +324,7 @@ func (strategy *InstantStrategy) sendHello(rtx *RouterSender) {
return
}
if err = rtx.Router.Control.SendWithTimeout(channel2.NewMessage(env.ServerHelloType, buf), strategy.HelloSendTimeout); err != nil {
if err = rtx.Router.Control.Send(channel.NewMessage(env.ServerHelloType, buf).WithTimeout(strategy.HelloSendTimeout)); err != nil {
if rtx.Router.Control.IsClosed() {
rtx.SetSyncStatus(env.RouterSyncDisconnected)
rtx.logger().WithError(err).Error("timed out sending serverHello message for edge router, connection closed, giving up")
@@ -482,7 +482,7 @@ func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullSt
msgContentBytes, _ := proto.Marshal(msgContent)
msg := channel2.NewMessage(env.ApiSessionAddedType, msgContentBytes)
msg := channel.NewMessage(env.ApiSessionAddedType, msgContentBytes)
msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type())
msg.Headers[env.SyncStrategyStateHeader] = stateBytes
go.mod
+2 -2
View File
@@ -42,8 +42,8 @@ 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.17
github.com/openziti/foundation v0.16.13
github.com/openziti/fabric v0.17.18
github.com/openziti/foundation v0.16.14
github.com/openziti/sdk-golang v0.15.130
github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231
github.com/pkg/errors v0.9.1
go.sum
+4 -3
View File
@@ -705,10 +705,11 @@ github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+
github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/fabric v0.17.17 h1:sjAQqmYoC3cF1Sh6RsSkKtNBDkO/NiLqqS1n8Is7Jbw=
github.com/openziti/fabric v0.17.17/go.mod h1:n1ysT2vSj1CFFXCsLB+LpzaVwPnuHUelSC8Q4n5knws=
github.com/openziti/foundation v0.16.13 h1:LvZ6pzTE9NayzqWBfW2J/bGv4X/67iJ3H5cqsB3DYI4=
github.com/openziti/fabric v0.17.18 h1:0nmCwTG3Tmwhjm4f3khcKtIS9IPBcNajNQ3OAYIf7xM=
github.com/openziti/fabric v0.17.18/go.mod h1:HIquY6ph5+yC9vL38EQh2n5qWSeu5y81Pgc1YPhtPVQ=
github.com/openziti/foundation v0.16.13/go.mod h1:SHG4x9kLP+M7sn0BJpoc/RbSIh1Hm1BxwBPHFmCf7qE=
github.com/openziti/foundation v0.16.14 h1:fwiQK7bdhs+z9idFVpEfWHzk7p0DGIvbrFbYJ6UcU8c=
github.com/openziti/foundation v0.16.14/go.mod h1:SHG4x9kLP+M7sn0BJpoc/RbSIh1Hm1BxwBPHFmCf7qE=
github.com/openziti/sdk-golang v0.15.130 h1:XnkWlUZEVJb1qusSyNFUiQc8Bc/IAJbv+W/va0O6MLg=
github.com/openziti/sdk-golang v0.15.130/go.mod h1:gx60OgzPgn/7pbeOu5OerSA98jWL0s+jnITiefEJFNo=
github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
pb/edge_ctrl_pb/decoder.go
+2 -1
View File
@@ -20,6 +20,7 @@ import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/foundation/channel"
"github.com/openziti/foundation/channel2"
"strings"
)
@@ -28,7 +29,7 @@ type Decoder struct{}
const DECODER = "edge_ctrl"
func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) {
func (d Decoder) Decode(msg *channel.Message) ([]byte, bool) {
switch msg.ContentType {
case int32(ContentType_ServerHelloType):
request := &ServerHello{}