From 3cfea85bf73b431f08b42e082f59ce72d1502906 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 31 Jan 2022 13:50:01 -0500 Subject: [PATCH] Update links to use channel library --- controller/controller.go | 4 +- controller/handler_ctrl/accept.go | 4 +- controller/handler_mgmt/accept.go | 16 +- pb/ctrl_pb/ch2_decoder.go | 299 --------------- pb/mgmt_pb/ch2_decoder.go | 425 ---------------------- router/accepter.go | 2 +- router/handler_link/accepter.go | 91 ----- router/handler_link/ack.go | 6 +- router/handler_link/bind.go | 127 +++++++ router/handler_link/close.go | 4 +- router/handler_link/control.go | 29 +- router/handler_link/error.go | 4 +- router/handler_link/payload.go | 6 +- router/handler_link/queue_ack.go | 6 +- router/metrics/peekhandler.go | 16 +- router/router.go | 23 +- router/xgress/ch2_decoder.go | 61 ---- router/xgress/decoder.go | 5 +- router/xgress/messages.go | 66 +--- router/xgress/ordering_test.go | 4 +- router/xgress/request.go | 3 +- router/xgress/xgress.go | 8 +- router/xgress_proxy/connection.go | 4 +- router/xgress_transport/connection.go | 4 +- router/xgress_transport_udp/connection.go | 4 +- router/xgress_udp/session.go | 4 +- router/xlink/xlink.go | 2 +- router/xlink_transport/accepter.go | 13 - router/xlink_transport/bind.go | 13 + router/xlink_transport/config.go | 20 +- router/xlink_transport/connect.go | 4 +- router/xlink_transport/dialer.go | 114 +++--- router/xlink_transport/factory.go | 49 ++- router/xlink_transport/listener.go | 286 ++++++++------- router/xlink_transport/xlink.go | 4 +- router/xlink_transport/xlink_split.go | 16 +- tests/context.go | 2 +- tests/link_test.go | 32 +- tests/mgmt_test.go | 2 +- trace/channel2_peekhandler.go | 155 -------- trace/xgress_peekhandler.go | 8 +- 41 files changed, 507 insertions(+), 1438 deletions(-) delete mode 100644 pb/ctrl_pb/ch2_decoder.go delete mode 100644 pb/mgmt_pb/ch2_decoder.go delete mode 100644 router/handler_link/accepter.go create mode 100644 router/handler_link/bind.go delete mode 100644 router/xgress/ch2_decoder.go delete mode 100644 router/xlink_transport/accepter.go create mode 100644 router/xlink_transport/bind.go delete mode 100644 trace/channel2_peekhandler.go diff --git a/controller/controller.go b/controller/controller.go index 124851dea..a377ff110 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -149,7 +149,7 @@ func (c *Controller) Run() error { if err := c.mgmtListener.Listen(c.mgmtConnectHandler); err != nil { panic(err) } - mgmtAccepter := handler_mgmt.NewMgmtAccepter(c.mgmtListener, c.config.Mgmt.Options) + mgmtAccepter := handler_mgmt.NewMgmtAccepter(c.network, c.xmgmts, c.mgmtListener, c.config.Mgmt.Options) go mgmtAccepter.Run() /* @@ -253,8 +253,6 @@ func (c *Controller) registerComponents() error { c.ctrlConnectHandler = handler_ctrl.NewConnectHandler(c.config.Id, c.network, c.xctrls) c.mgmtConnectHandler = handler_mgmt.NewConnectHandler(c.config.Id, c.network) - c.config.Mgmt.Options.BindHandler = handler_mgmt.NewBindHandler(c.network, c.xmgmts) - //add default REST XWeb if err := c.RegisterXweb(xweb.NewXwebImpl(c.xwebFactoryRegistry, c.config.Id)); err != nil { return err diff --git a/controller/handler_ctrl/accept.go b/controller/handler_ctrl/accept.go index eead60e28..c46f09a4f 100644 --- a/controller/handler_ctrl/accept.go +++ b/controller/handler_ctrl/accept.go @@ -51,10 +51,8 @@ func (self *CtrlAccepter) Run() { log.Info("started") defer log.Warn("exited") - self.options.SetBindHandlerF(self.Bind) - for { - _, err := channel.NewChannel("ctrl", self.listener, self.options) + _, err := channel.NewChannel("ctrl", self.listener, channel.BindHandlerF(self.Bind), self.options) if err != nil { log.WithError(err).Error("error accepting control channel connection") if err.Error() == "closed" { diff --git a/controller/handler_mgmt/accept.go b/controller/handler_mgmt/accept.go index 8ceee9b57..66c078f1c 100644 --- a/controller/handler_mgmt/accept.go +++ b/controller/handler_mgmt/accept.go @@ -19,28 +19,38 @@ package handler_mgmt import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" + "github.com/openziti/fabric/controller/network" + "github.com/openziti/fabric/controller/xmgmt" ) type MgmtAccepter struct { listener channel.UnderlayListener options *channel.Options + network *network.Network + xmgmts []xmgmt.Xmgmt } -func NewMgmtAccepter(listener channel.UnderlayListener, +func NewMgmtAccepter(network *network.Network, + xmgmts []xmgmt.Xmgmt, + listener channel.UnderlayListener, options *channel.Options) *MgmtAccepter { return &MgmtAccepter{ + network: network, + xmgmts: xmgmts, listener: listener, options: options, } } -func (mgmtAccepter *MgmtAccepter) Run() { +func (self *MgmtAccepter) Run() { log := pfxlog.Logger() log.Info("started") defer log.Warn("exited") + bindHandler := NewBindHandler(self.network, self.xmgmts) + for { - ch, err := channel.NewChannel("mgmt", mgmtAccepter.listener, mgmtAccepter.options) + ch, err := channel.NewChannel("mgmt", self.listener, bindHandler, self.options) if err == nil { log.Debugf("accepted mgmt connection [%s]", ch.Id().Token) diff --git a/pb/ctrl_pb/ch2_decoder.go b/pb/ctrl_pb/ch2_decoder.go deleted file mode 100644 index 8c92a16e7..000000000 --- a/pb/ctrl_pb/ch2_decoder.go +++ /dev/null @@ -1,299 +0,0 @@ -/* - 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 ctrl_pb - -import ( - "fmt" - "github.com/golang/protobuf/proto" - "github.com/michaelquigley/pfxlog" - "github.com/openziti/fabric/ctrl_msg" - "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/metrics/metrics_pb" - "strconv" - "strings" -) - -type Channel2Decoder struct{} - -func (d Channel2Decoder) Decode(msg *channel2.Message) ([]byte, bool) { - switch msg.ContentType { - case int32(ContentType_CircuitRequestType): - circuitRequest := &CircuitRequest{} - if err := proto.Unmarshal(msg.Body, circuitRequest); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Circuit Request") - meta["ingressId"] = circuitRequest.IngressId - meta["serviceId"] = circuitRequest.ServiceId - headers := make([]string, 0) - for k := range circuitRequest.PeerData { - headers = append(headers, strconv.Itoa(int(k))) - } - meta["peerData"] = strings.Join(headers, ",") - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_CreateTerminatorRequestType): - createTerminator := &CreateTerminatorRequest{} - if err := proto.Unmarshal(msg.Body, createTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Create Terminator Request") - meta["terminator"] = terminatorToString(createTerminator) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_RemoveTerminatorRequestType): - removeTerminator := &RemoveTerminatorRequest{} - if err := proto.Unmarshal(msg.Body, removeTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Remove Terminator Request") - meta["terminatorId"] = removeTerminator.TerminatorId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_ValidateTerminatorsRequestType): - request := &ValidateTerminatorsRequest{} - if err := proto.Unmarshal(msg.Body, request); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Validate Terminators") - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ctrl_msg.CircuitSuccessType): - meta := channel2.NewTraceMessageDecode(DECODER, "Circuit Success Response") - meta["circuitId"] = string(msg.Body) - meta["address"] = string(msg.Headers[ctrl_msg.CircuitSuccessAddressHeader]) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - case int32(ctrl_msg.CircuitFailedType): - meta := channel2.NewTraceMessageDecode(DECODER, "Circuit Failed Response") - message := string(msg.Body) - if message != "" { - meta["message"] = message - } - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - case int32(ContentType_DialType): - connect := &Dial{} - if err := proto.Unmarshal(msg.Body, connect); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Dial") - meta["linkId"] = connect.LinkId - meta["address"] = connect.Address - meta["routerId"] = connect.RouterId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_LinkType): - link := &Link{} - if err := proto.Unmarshal(msg.Body, link); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Link") - meta["id"] = link.Id - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_FaultType): - fault := &Fault{} - if err := proto.Unmarshal(msg.Body, fault); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Fault") - meta["subject"] = fault.Subject.String() - meta["id"] = fault.Id - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_RouteType): - route := &Route{} - if err := proto.Unmarshal(msg.Body, route); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Route") - meta["circuitId"] = route.CircuitId - if route.Egress != nil { - meta["egress.address"] = route.Egress.Address - meta["egress.destination"] = route.Egress.Destination - } - for i, forward := range route.Forwards { - meta[fmt.Sprintf("forward[%d].srcAddress", i)] = forward.SrcAddress - meta[fmt.Sprintf("forward[%d].dstAddress", i)] = forward.DstAddress - } - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_UnrouteType): - unroute := &Unroute{} - if err := proto.Unmarshal(msg.Body, unroute); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Unroute") - meta["circuitId"] = unroute.CircuitId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - } - - case int32(ContentType_MetricsType): - metricsMsg := &metrics_pb.MetricsMessage{} - if err := proto.Unmarshal(msg.Body, metricsMsg); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Metrics") - - for name, metric := range metricsMsg.Histograms { - meta[name+".min"] = metric.Min - meta[name+".mean"] = metric.Mean - meta[name+".max"] = metric.Max - meta[name+".p95"] = metric.P95 - meta[name+".p99"] = metric.P99 - } - - for name, metric := range metricsMsg.Meters { - meta[name+".count"] = metric.Count - meta[name+".meanRate"] = metric.MeanRate - meta[name+".m1Rate"] = metric.M1Rate - meta[name+".m5Rate"] = metric.M5Rate - meta[name+".m15Rate"] = metric.M15Rate - } - - for name, counter := range metricsMsg.IntervalCounters { - for _, bucket := range counter.Buckets { - for key, val := range bucket.Values { - meta[name+"."+key+"["+strconv.FormatInt(bucket.IntervalStartUTC, 10)+"]"] = val - } - } - } - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - } - - case int32(ctrl_msg.RouteResultType): - meta := channel2.NewTraceMessageDecode(DECODER, "Route Result") - meta["circuitId"] = string(msg.Body) - meta["attempt"], _ = msg.GetUint32Header(ctrl_msg.RouteResultAttemptHeader) - success, _ := msg.GetBoolHeader(ctrl_msg.RouteResultSuccessHeader) - meta["success"] = success - if !success { - meta["errormsg"], _ = msg.GetStringHeader(ctrl_msg.RouteResultErrorHeader) - } - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - } - - return nil, false -} diff --git a/pb/mgmt_pb/ch2_decoder.go b/pb/mgmt_pb/ch2_decoder.go deleted file mode 100644 index 64b077c23..000000000 --- a/pb/mgmt_pb/ch2_decoder.go +++ /dev/null @@ -1,425 +0,0 @@ -/* - 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 mgmt_pb - -import ( - "github.com/golang/protobuf/proto" - "github.com/michaelquigley/pfxlog" - "github.com/openziti/foundation/channel2" -) - -type Channel2Decoder struct{} - -func (d Channel2Decoder) Decode(msg *channel2.Message) ([]byte, bool) { - switch msg.ContentType { - case int32(ContentType_ListServicesRequestType): - data, err := channel2.NewTraceMessageDecode(DECODER, "List Services Request").MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - case int32(ContentType_ListServicesResponseType): - listServices := &ListServicesResponse{} - if err := proto.Unmarshal(msg.Body, listServices); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "List Services Response") - meta["services"] = len(listServices.Services) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_CreateServiceRequestType): - createService := &CreateServiceRequest{} - if err := proto.Unmarshal(msg.Body, createService); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Create Service Request") - meta["service"] = serviceToString(createService.Service) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_RemoveServiceRequestType): - removeService := &RemoveServiceRequest{} - if err := proto.Unmarshal(msg.Body, removeService); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Remove Service Request") - meta["serviceId"] = removeService.ServiceId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_GetServiceRequestType): - getService := &GetServiceRequest{} - if err := proto.Unmarshal(msg.Body, getService); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Get Service Request") - meta["serviceId"] = getService.ServiceId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_GetServiceResponseType): - getService := &GetServiceResponse{} - if err := proto.Unmarshal(msg.Body, getService); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Get Service Response") - meta["service"] = serviceToString(getService.Service) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - // terminator messages - case int32(ContentType_ListTerminatorsRequestType): - data, err := channel2.NewTraceMessageDecode(DECODER, "List Terminators Request").MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - case int32(ContentType_ListTerminatorsResponseType): - listTerminators := &ListTerminatorsResponse{} - if err := proto.Unmarshal(msg.Body, listTerminators); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "List Terminators Response") - meta["terminators"] = len(listTerminators.Terminators) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_CreateTerminatorRequestType): - createTerminator := &CreateTerminatorRequest{} - if err := proto.Unmarshal(msg.Body, createTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Create Terminator Request") - meta["terminator"] = terminatorToString(createTerminator.Terminator) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_RemoveTerminatorRequestType): - removeTerminator := &RemoveTerminatorRequest{} - if err := proto.Unmarshal(msg.Body, removeTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Remove Terminator Request") - meta["terminatorId"] = removeTerminator.TerminatorId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_GetTerminatorRequestType): - getTerminator := &GetTerminatorRequest{} - if err := proto.Unmarshal(msg.Body, getTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Get Terminator Request") - meta["terminatorId"] = getTerminator.TerminatorId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_GetTerminatorResponseType): - getTerminator := &GetTerminatorResponse{} - if err := proto.Unmarshal(msg.Body, getTerminator); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Get Terminator Response") - meta["terminator"] = terminatorToString(getTerminator.Terminator) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_SetTerminatorCostRequestType): - request := &SetTerminatorCostRequest{} - if err := proto.Unmarshal(msg.Body, request); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Set Terminator Weight Request") - meta["terminatorId"] = request.TerminatorId - meta["staticCost"] = request.StaticCost - meta["precedence"] = request.Precedence.String() - meta["dynamicCost"] = request.DynamicCost - meta["changeMask"] = request.UpdateMask - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - // router messages - case int32(ContentType_ListRoutersRequestType): - data, err := channel2.NewTraceMessageDecode(DECODER, "List Routers Request").MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - case int32(ContentType_ListRoutersResponseType): - listRouters := &ListRoutersResponse{} - if err := proto.Unmarshal(msg.Body, listRouters); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "List Routers Response") - meta["routers"] = len(listRouters.Routers) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_CreateRouterRequestType): - createRouter := &CreateRouterRequest{} - if err := proto.Unmarshal(msg.Body, createRouter); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Create Router Request") - meta["router"] = routerToString(createRouter.Router) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_RemoveRouterRequestType): - removeRouter := &RemoveRouterRequest{} - if err := proto.Unmarshal(msg.Body, removeRouter); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Remove Router Request") - meta["routerId"] = removeRouter.RouterId - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_ListLinksRequestType): - data, err := channel2.NewTraceMessageDecode(DECODER, "List Links Request").MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - case int32(ContentType_ListLinksResponseType): - listLinks := &ListLinksResponse{} - if err := proto.Unmarshal(msg.Body, listLinks); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "List Links Response") - meta["links"] = len(listLinks.Links) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_SetLinkCostRequestType): - setLinkCost := &SetLinkCostRequest{} - if err := proto.Unmarshal(msg.Body, setLinkCost); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Set Link Cost Request") - meta["linkId"] = setLinkCost.LinkId - meta["cost"] = setLinkCost.Cost - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_SetLinkDownRequestType): - setLinkDown := &SetLinkDownRequest{} - if err := proto.Unmarshal(msg.Body, setLinkDown); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Set Link Down Request") - meta["linkId"] = setLinkDown.LinkId - meta["down"] = setLinkDown.Down - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - case int32(ContentType_ListCircuitsRequestType): - data, err := channel2.NewTraceMessageDecode(DECODER, "List Circuits Request").MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - case int32(ContentType_ListCircuitsResponseType): - listCircuits := &ListCircuitsResponse{} - if err := proto.Unmarshal(msg.Body, listCircuits); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "List Circuits Response") - meta["circuits"] = len(listCircuits.Circuits) - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - return nil, true - } - } - - return nil, false -} diff --git a/router/accepter.go b/router/accepter.go index 09f6e3f3b..12dd9d969 100644 --- a/router/accepter.go +++ b/router/accepter.go @@ -6,7 +6,7 @@ import ( "github.com/sirupsen/logrus" ) -func newXlinkAccepter(f *forwarder2.Forwarder) xlink.Accepter { +func newXlinkAccepter(f *forwarder2.Forwarder) xlink.Acceptor { return &xlinkAccepter{forwarder: f} } diff --git a/router/handler_link/accepter.go b/router/handler_link/accepter.go deleted file mode 100644 index 04a5f9f19..000000000 --- a/router/handler_link/accepter.go +++ /dev/null @@ -1,91 +0,0 @@ -package handler_link - -import ( - "crypto/sha1" - "fmt" - "github.com/openziti/channel" - "github.com/openziti/channel/protobufs" - "github.com/openziti/fabric/pb/ctrl_pb" - "github.com/openziti/fabric/router/forwarder" - metrics2 "github.com/openziti/fabric/router/metrics" - "github.com/openziti/fabric/router/xgress" - "github.com/openziti/fabric/router/xlink" - "github.com/openziti/fabric/router/xlink_transport" - "github.com/openziti/fabric/trace" - "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/metrics" - "github.com/pkg/errors" - "github.com/sirupsen/logrus" - "time" -) - -func NewChannelAccepter(c xgress.CtrlChannel, f *forwarder.Forwarder, fo *forwarder.Options, mr metrics.Registry) xlink_transport.ChannelAccepter { - return &channelAccepter{ - ctrl: c, - forwarder: f, - forwarderOptions: fo, - metricsRegistry: mr, - } -} - -func (self *channelAccepter) AcceptChannel(xlink xlink.Xlink, ch channel2.Channel, trackLatency bool, listenerSide bool) error { - if listenerSide { - if err := self.verifyLink(xlink, ch); err != nil { - return err - } - } - - closeNotify := make(chan struct{}) - - ch.SetLogicalName("l/" + xlink.Id().Token) - ch.SetUserData(xlink.Id().Token) - ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder, closeNotify)) - ch.AddErrorHandler(newErrorHandler(xlink, self.ctrl)) - ch.AddReceiveHandler(newPayloadHandler(xlink, self.forwarder)) - ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.forwarder, closeNotify)) - ch.AddReceiveHandler(&channel2.LatencyHandler{}) - ch.AddReceiveHandler(newControlHandler(xlink, ch, self.forwarder, closeNotify)) - ch.AddPeekHandler(metrics2.NewChannelPeekHandler(xlink.Id().Token, self.forwarder.MetricsRegistry())) - ch.AddPeekHandler(trace.NewChannel2PeekHandler(xlink.Id().Token, ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel()))) - - if trackLatency { - go metrics.ProbeLatency( - ch, - self.metricsRegistry.Histogram("link."+xlink.Id().Token+".latency"), - self.forwarderOptions.LatencyProbeInterval, - self.forwarderOptions.LatencyProbeTimeout, - ) - } - - return nil -} - -func (self *channelAccepter) verifyLink(l xlink.Xlink, ch channel2.Channel) error { - verifyLink := &ctrl_pb.VerifyLink{ - LinkId: l.Id().Token, - } - for _, cert := range ch.Certificates() { - fingerprint := fmt.Sprintf("%x", sha1.Sum(cert.Raw)) - verifyLink.Fingerprints = append(verifyLink.Fingerprints, fingerprint) - } - reply, err := protobufs.MarshalTyped(verifyLink).WithTimeout(10 * time.Second).SendForReply(self.ctrl.Channel()) - if err != nil { - return errors.Wrapf(err, "unable to verify link %v", l.Id().Token) - } - if reply.ContentType != channel.ContentTypeResultType { - return errors.Errorf("unexpected response type to verify link: %v", reply.ContentType) - } - result := channel.UnmarshalResult(reply) - if result.Success { - logrus.WithField("linkId", l.Id().Token).Info("successfully verified link") - return nil - } - return errors.Errorf("unable to verify link [%v]", result.Message) -} - -type channelAccepter struct { - ctrl xgress.CtrlChannel - forwarder *forwarder.Forwarder - forwarderOptions *forwarder.Options - metricsRegistry metrics.Registry -} diff --git a/router/handler_link/ack.go b/router/handler_link/ack.go index fcfa5566d..797d94511 100644 --- a/router/handler_link/ack.go +++ b/router/handler_link/ack.go @@ -18,10 +18,10 @@ package handler_link import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" ) type ackHandler struct { @@ -40,12 +40,12 @@ func (self *ackHandler) ContentType() int32 { return xgress.ContentTypeAcknowledgementType } -func (self *ackHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *ackHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { log := pfxlog.ContextLogger(ch.Label()). WithField("linkId", self.link.Id().Token). WithField("routerId", self.link.DestinationId()) - if ack, err := xgress.UnmarshallChannel2Acknowledgement(msg); err == nil { + if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil { if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.link.Id().Token), ack); err != nil { log.WithError(err).Debug("unable to forward acknowledgement") } diff --git a/router/handler_link/bind.go b/router/handler_link/bind.go new file mode 100644 index 000000000..5f2eb1bbb --- /dev/null +++ b/router/handler_link/bind.go @@ -0,0 +1,127 @@ +package handler_link + +import ( + "crypto/sha1" + "fmt" + "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" + "github.com/openziti/channel/latency" + "github.com/openziti/channel/protobufs" + "github.com/openziti/fabric/pb/ctrl_pb" + "github.com/openziti/fabric/router/forwarder" + metrics2 "github.com/openziti/fabric/router/metrics" + "github.com/openziti/fabric/router/xgress" + "github.com/openziti/fabric/router/xlink" + "github.com/openziti/fabric/trace" + "github.com/openziti/foundation/metrics" + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + "time" +) + +func NewBindHandlerFactory(c xgress.CtrlChannel, f *forwarder.Forwarder, fo *forwarder.Options, mr metrics.Registry) *bindHandlerFactory { + return &bindHandlerFactory{ + ctrl: c, + forwarder: f, + forwarderOptions: fo, + metricsRegistry: mr, + } +} + +type bindHandlerFactory struct { + ctrl xgress.CtrlChannel + forwarder *forwarder.Forwarder + forwarderOptions *forwarder.Options + metricsRegistry metrics.Registry +} + +func (self *bindHandlerFactory) NewBindHandler(link xlink.Xlink, latency bool, listenerSide bool) channel.BindHandler { + return &bindHandler{ + bindHandlerFactory: self, + xlink: link, + trackLatency: latency, + listenerSide: listenerSide, + } +} + +type bindHandler struct { + *bindHandlerFactory + xlink xlink.Xlink + trackLatency bool + listenerSide bool +} + +func (self *bindHandler) BindChannel(binding channel.Binding) error { + ch := binding.GetChannel() + if self.listenerSide { + if err := self.verifyLink(self.xlink, ch); err != nil { + return err + } + } + + closeNotify := make(chan struct{}) + + binding.GetChannel().SetLogicalName("l/" + self.xlink.Id().Token) + binding.SetUserData(self.xlink.Id().Token) + binding.AddCloseHandler(newCloseHandler(self.xlink, self.ctrl, self.forwarder, closeNotify)) + binding.AddErrorHandler(newErrorHandler(self.xlink, self.ctrl)) + binding.AddTypedReceiveHandler(newPayloadHandler(self.xlink, self.forwarder)) + binding.AddTypedReceiveHandler(newQueuingAckHandler(self.xlink, self.forwarder, closeNotify)) + binding.AddTypedReceiveHandler(&latency.LatencyHandler{}) + binding.AddTypedReceiveHandler(newControlHandler(self.xlink, self.forwarder, closeNotify)) + binding.AddPeekHandler(metrics2.NewChannelPeekHandler(self.xlink.Id().Token, self.forwarder.MetricsRegistry())) + binding.AddPeekHandler(trace.NewChannelPeekHandler(self.xlink.Id().Token, ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel()))) + + if self.trackLatency { + latencyMetric := self.metricsRegistry.Histogram("link." + self.xlink.Id().Token + ".latency") + latencyTimeout := self.forwarderOptions.LatencyProbeTimeout + config := &latency.ProbeConfig{ + Channel: ch, + Interval: self.forwarderOptions.LatencyProbeInterval, + Timeout: latencyTimeout, + ResultHandler: func(resultNanos int64) { + latencyMetric.Update(resultNanos) + }, + TimeoutHandler: func() { + pfxlog.ContextLogger(ch.Label()).Errorf("latency timeout after [%s] on channel [%s]", latencyTimeout, ch.Label()) + }, + ExitHandler: func() { + latencyMetric.Dispose() + }, + } + + go latency.ProbeLatencyConfigurable(config) + } + + return nil +} + +func (self *bindHandler) verifyLink(l xlink.Xlink, ch channel.Channel) error { + var fingerprints []string + for _, cert := range ch.Certificates() { + fingerprint := fmt.Sprintf("%x", sha1.Sum(cert.Raw)) + fingerprints = append(fingerprints, fingerprint) + } + + verifyLink := &ctrl_pb.VerifyLink{ + LinkId: l.Id().Token, + Fingerprints: fingerprints, + } + + reply, err := protobufs.MarshalTyped(verifyLink).WithTimeout(10 * time.Second).SendForReply(self.ctrl.Channel()) + if err != nil { + return errors.Wrapf(err, "unable to verify link %v", l.Id().Token) + } + + if reply.ContentType != channel.ContentTypeResultType { + return errors.Errorf("unexpected response type to verify link: %v", reply.ContentType) + } + + result := channel.UnmarshalResult(reply) + if result.Success { + logrus.WithField("linkId", l.Id().Token).Info("successfully verified link") + return nil + } + + return errors.Errorf("unable to verify link [%v]", result.Message) +} diff --git a/router/handler_link/close.go b/router/handler_link/close.go index 78bdd61bc..c71be4652 100644 --- a/router/handler_link/close.go +++ b/router/handler_link/close.go @@ -18,12 +18,12 @@ package handler_link import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/channel/protobufs" "github.com/openziti/fabric/pb/ctrl_pb" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/concurrenz" ) @@ -44,7 +44,7 @@ func newCloseHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwa } } -func (self *closeHandler) HandleClose(ch channel2.Channel) { +func (self *closeHandler) HandleClose(ch channel.Channel) { if self.closed.CompareAndSwap(false, true) { log := pfxlog.ContextLogger(ch.Label()). WithField("linkId", self.link.Id().Token). diff --git a/router/handler_link/control.go b/router/handler_link/control.go index 9c1a4bffd..5683f38b9 100644 --- a/router/handler_link/control.go +++ b/router/handler_link/control.go @@ -18,25 +18,22 @@ package handler_link import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" ) type controlHandler struct { link xlink.Xlink forwarder *forwarder.Forwarder - ctrls chan *xgress.Control } -func newControlHandler(link xlink.Xlink, ch channel2.Channel, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *controlHandler { +func newControlHandler(link xlink.Xlink, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *controlHandler { result := &controlHandler{ link: link, forwarder: forwarder, - ctrls: make(chan *xgress.Control, 4), } - go result.run(ch.Label(), closeNotify) return result } @@ -44,7 +41,7 @@ func (self *controlHandler) ContentType() int32 { return xgress.ContentTypeControlType } -func (self *controlHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *controlHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { log := pfxlog.ContextLogger(ch.Label()) control, err := xgress.UnmarshallControl(msg) @@ -56,23 +53,3 @@ func (self *controlHandler) HandleReceive(msg *channel2.Message, ch channel2.Cha log.Errorf("unexpected error (%v)", err) } } - -func (self *controlHandler) run(label string, closeNotify <-chan struct{}) { - log := pfxlog.ContextLogger(label). - WithField("linkId", self.link.Id().Token). - WithField("routerId", self.link.DestinationId()) - - log.Info("starting") - defer log.Info("exiting") - - for { - select { - case control := <-self.ctrls: - if err := self.forwarder.ForwardControl(xgress.Address(self.link.Id().Token), control); err != nil { - log.WithError(err).Debug("unable to forward") - } - case <-closeNotify: - return - } - } -} diff --git a/router/handler_link/error.go b/router/handler_link/error.go index 00490b1a9..7bb650a69 100644 --- a/router/handler_link/error.go +++ b/router/handler_link/error.go @@ -18,9 +18,9 @@ package handler_link import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" ) type errorHandler struct { @@ -32,7 +32,7 @@ func newErrorHandler(link xlink.Xlink, ctrl xgress.CtrlChannel) *errorHandler { return &errorHandler{link: link, ctrl: ctrl} } -func (self *errorHandler) HandleError(err error, ch channel2.Channel) { +func (self *errorHandler) HandleError(err error, ch channel.Channel) { log := pfxlog.ContextLogger(ch.Label()). WithField("linkId", self.link.Id().Token). WithField("routerId", self.link.DestinationId()) diff --git a/router/handler_link/payload.go b/router/handler_link/payload.go index 36e2f55e1..ae3f0f600 100644 --- a/router/handler_link/payload.go +++ b/router/handler_link/payload.go @@ -18,10 +18,10 @@ package handler_link import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" ) type payloadHandler struct { @@ -40,12 +40,12 @@ func (self *payloadHandler) ContentType() int32 { return xgress.ContentTypePayloadType } -func (self *payloadHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *payloadHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { log := pfxlog.ContextLogger(ch.Label()). WithField("linkId", self.link.Id().Token). WithField("routerId", self.link.DestinationId()) - payload, err := xgress.UnmarshallChannel2Payload(msg) + payload, err := xgress.UnmarshallPayload(msg) if err == nil { if err := self.forwarder.ForwardPayload(xgress.Address(self.link.Id().Token), payload); err != nil { log.WithError(err).Debug("unable to forward") diff --git a/router/handler_link/queue_ack.go b/router/handler_link/queue_ack.go index 11c5c00a4..ed89fcb2d 100644 --- a/router/handler_link/queue_ack.go +++ b/router/handler_link/queue_ack.go @@ -19,10 +19,10 @@ package handler_link import ( "github.com/ef-ds/deque" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" "sync/atomic" ) @@ -60,8 +60,8 @@ func (self *queuingAckHandler) ContentType() int32 { return xgress.ContentTypeAcknowledgementType } -func (self *queuingAckHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - if ack, err := xgress.UnmarshallChannel2Acknowledgement(msg); err == nil { +func (self *queuingAckHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { + if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil { select { case self.ackIngest <- ack: case <-self.closeNotify: diff --git a/router/metrics/peekhandler.go b/router/metrics/peekhandler.go index eb0c977e0..dc7510a3c 100644 --- a/router/metrics/peekhandler.go +++ b/router/metrics/peekhandler.go @@ -18,14 +18,14 @@ package metrics import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/metrics" "time" ) // NewChannelPeekHandler creates a channel PeekHandler which tracks latency, message rate and message size distribution -func NewChannelPeekHandler(linkId string, registry metrics.UsageRegistry) channel2.PeekHandler { +func NewChannelPeekHandler(linkId string, registry metrics.UsageRegistry) channel.PeekHandler { appTxBytesMeter := registry.Meter("fabric.tx.bytesrate") appTxMsgMeter := registry.Meter("fabric.tx.msgrate") appTxMsgSizeHistogram := registry.Histogram("fabric.tx.msgsize") @@ -95,10 +95,10 @@ type channelPeekHandler struct { closeHook func() } -func (h *channelPeekHandler) Connect(channel2.Channel, string) { +func (h *channelPeekHandler) Connect(channel.Channel, string) { } -func (h *channelPeekHandler) Rx(msg *channel2.Message, _ channel2.Channel) { +func (h *channelPeekHandler) Rx(msg *channel.Message, _ channel.Channel) { msgSize := int64(len(msg.Body)) h.linkRxBytesMeter.Mark(msgSize) h.linkRxMsgMeter.Mark(1) @@ -108,7 +108,7 @@ func (h *channelPeekHandler) Rx(msg *channel2.Message, _ channel2.Channel) { h.appRxMsgSizeHistogram.Update(msgSize) if msg.ContentType == int32(xgress.ContentTypePayloadType) { - if payload, err := xgress.UnmarshallChannel2Payload(msg); err != nil { + if payload, err := xgress.UnmarshallPayload(msg); err != nil { pfxlog.Logger().Errorf("Failed to unmarshal payload. Error: %v", err) } else { h.usageRxCounter.Update(payload.CircuitId, time.Now(), uint64(len(payload.Data))) @@ -116,7 +116,7 @@ func (h *channelPeekHandler) Rx(msg *channel2.Message, _ channel2.Channel) { } } -func (h *channelPeekHandler) Tx(msg *channel2.Message, _ channel2.Channel) { +func (h *channelPeekHandler) Tx(msg *channel.Message, _ channel.Channel) { msgSize := int64(len(msg.Body)) h.linkTxBytesMeter.Mark(msgSize) h.linkTxMsgMeter.Mark(1) @@ -126,7 +126,7 @@ func (h *channelPeekHandler) Tx(msg *channel2.Message, _ channel2.Channel) { h.appTxMsgSizeHistogram.Update(msgSize) if msg.ContentType == int32(xgress.ContentTypePayloadType) { - if payload, err := xgress.UnmarshallChannel2Payload(msg); err != nil { + if payload, err := xgress.UnmarshallPayload(msg); err != nil { pfxlog.Logger().Errorf("Failed to unmarshal payload. Error: %v", err) } else { h.usageTxCounter.Update(payload.CircuitId, time.Now(), uint64(len(payload.Data))) @@ -134,7 +134,7 @@ func (h *channelPeekHandler) Tx(msg *channel2.Message, _ channel2.Channel) { } } -func (h *channelPeekHandler) Close(channel2.Channel) { +func (h *channelPeekHandler) Close(channel.Channel) { if h.closeHook != nil { h.closeHook() } diff --git a/router/router.go b/router/router.go index 343be2d86..1d81f8120 100644 --- a/router/router.go +++ b/router/router.go @@ -28,7 +28,6 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" "github.com/openziti/fabric/controller/xctrl" - "github.com/openziti/fabric/event" "github.com/openziti/fabric/health" fabricMetrics "github.com/openziti/fabric/metrics" "github.com/openziti/fabric/pb/ctrl_pb" @@ -44,7 +43,6 @@ import ( "github.com/openziti/fabric/router/xlink" "github.com/openziti/fabric/router/xlink_transport" "github.com/openziti/fabric/xweb" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/common" "github.com/openziti/foundation/metrics" "github.com/openziti/foundation/profiler" @@ -62,9 +60,6 @@ import ( type Router struct { config *Config ctrl channel.Channel - ctrlOptions *channel.Options - linkOptions *channel2.Options - linkListener channel2.UnderlayListener faulter *forwarder.Faulter scanner *forwarder.Scanner forwarder *forwarder.Forwarder @@ -78,7 +73,6 @@ type Router struct { shutdownC chan struct{} shutdownDoneC chan struct{} isShutdown concurrenz.AtomicBoolean - eventDispatcher event.Dispatcher metricsReporter metrics.Handler versionProvider common.VersionProvider debugOperations map[byte]func(c *bufio.ReadWriter) error @@ -108,7 +102,6 @@ func (self *Router) DefaultRequestTimeout() time.Duration { func Create(config *Config, versionProvider common.VersionProvider) *Router { closeNotify := make(chan struct{}) - eventDispatcher := event.NewDispatcher(closeNotify) metricsRegistry := metrics.NewUsageRegistry(config.Id.Token, map[string]string{}, closeNotify) xgress.InitMetrics(metricsRegistry) @@ -128,7 +121,6 @@ func Create(config *Config, versionProvider common.VersionProvider) *Router { metricsRegistry: metricsRegistry, shutdownC: closeNotify, shutdownDoneC: make(chan struct{}), - eventDispatcher: eventDispatcher, versionProvider: versionProvider, debugOperations: map[byte]func(c *bufio.ReadWriter) error{}, xwebFactoryRegistry: xweb.NewWebHandlerFactoryRegistryImpl(), @@ -274,7 +266,7 @@ func (self *Router) startProfiling() { func (self *Router) registerComponents() error { self.xlinkFactories = make(map[string]xlink.Factory) xlinkAccepter := newXlinkAccepter(self.forwarder) - xlinkChAccepter := handler_link.NewChannelAccepter(self, + xlinkChAccepter := handler_link.NewBindHandlerFactory(self, self.forwarder, self.config.Forwarder, self.metricsRegistry, @@ -385,10 +377,10 @@ func (self *Router) startControlPlane() error { return fmt.Errorf("error with version header information value: %v", err) } - attributes[channel2.HelloVersionHeader] = version + attributes[channel.HelloVersionHeader] = version if len(self.xlinkListeners) == 1 { - attributes[channel2.HelloRouterAdvertisementsHeader] = []byte(self.xlinkListeners[0].GetAdvertisement()) + attributes[channel.HelloRouterAdvertisementsHeader] = []byte(self.xlinkListeners[0].GetAdvertisement()) } reconnectHandler := func() { @@ -411,9 +403,7 @@ func (self *Router) startControlPlane() error { self.shutdownC, ) - self.config.Ctrl.Options.BindHandler = bindHandler - - ch, err := channel.NewChannel("ctrl", dialer, self.config.Ctrl.Options) + ch, err := channel.NewChannel("ctrl", dialer, bindHandler, self.config.Ctrl.Options) if err != nil { return fmt.Errorf("error connecting ctrl (%v)", err) } @@ -594,11 +584,12 @@ func (self *controllerPinger) PingContext(ctx context.Context) error { return errors.Errorf("control channel not yet established") } - msg := channel.NewMessage(channel2.ContentTypePingType, nil) + msg := channel.NewMessage(channel.ContentTypePingType, nil) deadline, ok := ctx.Deadline() if !ok { deadline = time.Now().Add(30 * time.Second) } timeout := deadline.Sub(time.Now()) - return msg.WithTimeout(timeout).SendAndWaitForWire(self.router.ctrl) + _, err := msg.WithTimeout(timeout).SendForReply(self.router.ctrl) + return err } diff --git a/router/xgress/ch2_decoder.go b/router/xgress/ch2_decoder.go deleted file mode 100644 index 50c0d689d..000000000 --- a/router/xgress/ch2_decoder.go +++ /dev/null @@ -1,61 +0,0 @@ -/* - 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 xgress - -import ( - "fmt" - "github.com/michaelquigley/pfxlog" - "github.com/openziti/foundation/channel2" -) - -type Channel2Decoder struct{} - -func (d Channel2Decoder) Decode(msg *channel2.Message) ([]byte, bool) { - switch msg.ContentType { - case int32(ContentTypePayloadType): - if payload, err := UnmarshallChannel2Payload(msg); err == nil { - return DecodePayload(payload) - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - } - - case int32(ContentTypeAcknowledgementType): - if ack, err := UnmarshallChannel2Acknowledgement(msg); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Acknowledgement") - meta["circuitId"] = ack.CircuitId - meta["sequence"] = fmt.Sprintf("len(%d)", len(ack.Sequence)) - switch ack.GetOriginator() { - case Initiator: - meta["originator"] = "i" - case Terminator: - meta["originator"] = "e" - } - - data, err := meta.MarshalTraceMessageDecode() - if err != nil { - return nil, true - } - - return data, true - - } else { - pfxlog.Logger().Errorf("unexpected error (%s)", err) - } - } - - return nil, false -} diff --git a/router/xgress/decoder.go b/router/xgress/decoder.go index d371fd986..5505cd501 100644 --- a/router/xgress/decoder.go +++ b/router/xgress/decoder.go @@ -20,7 +20,6 @@ import ( "fmt" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" - "github.com/openziti/foundation/channel2" ) type Decoder struct{} @@ -38,7 +37,7 @@ func (d Decoder) Decode(msg *channel.Message) ([]byte, bool) { case int32(ContentTypeAcknowledgementType): if ack, err := UnmarshallAcknowledgement(msg); err == nil { - meta := channel2.NewTraceMessageDecode(DECODER, "Acknowledgement") + meta := channel.NewTraceMessageDecode(DECODER, "Acknowledgement") meta["circuitId"] = ack.CircuitId meta["sequence"] = fmt.Sprintf("len(%d)", len(ack.Sequence)) switch ack.GetOriginator() { @@ -64,7 +63,7 @@ func (d Decoder) Decode(msg *channel.Message) ([]byte, bool) { } func DecodePayload(payload *Payload) ([]byte, bool) { - meta := channel2.NewTraceMessageDecode(DECODER, "Payload") + meta := channel.NewTraceMessageDecode(DECODER, "Payload") meta["circuitId"] = payload.CircuitId meta["sequence"] = payload.Sequence switch payload.GetOriginator() { diff --git a/router/xgress/messages.go b/router/xgress/messages.go index 7abed07df..83a522455 100644 --- a/router/xgress/messages.go +++ b/router/xgress/messages.go @@ -20,7 +20,6 @@ import ( "encoding/binary" "fmt" "github.com/openziti/channel" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/info" "github.com/openziti/foundation/util/uuidz" "github.com/pkg/errors" @@ -93,7 +92,7 @@ func (header *Header) GetOriginator() Originator { return Initiator } -func (header *Header) unmarshallHeader(msg *channel2.Message) error { +func (header *Header) unmarshallHeader(msg *channel.Message) error { circuitId, ok := msg.Headers[HeaderKeyCircuitId] if !ok { return fmt.Errorf("no circuitId found in xgress payload message") @@ -133,7 +132,7 @@ func (header *Header) unmarshallChannelHeader(msg *channel.Message) error { return nil } -func (header *Header) marshallHeader(msg *channel2.Message) { +func (header *Header) marshallHeader(msg *channel.Message) { msg.Headers[HeaderKeyCircuitId] = []byte(header.CircuitId) if header.Flags != 0 { msg.PutUint32Header(HeaderKeyFlags, header.Flags) @@ -195,25 +194,12 @@ func (ack *Acknowledgement) unmarshallSequence(data []byte) error { return nil } -func (ack *Acknowledgement) Marshall() *channel2.Message { - msg := channel2.NewMessage(ContentTypeAcknowledgementType, ack.marshallSequence()) +func (ack *Acknowledgement) Marshall() *channel.Message { + msg := channel.NewMessage(ContentTypeAcknowledgementType, ack.marshallSequence()) ack.marshallHeader(msg) return msg } -func UnmarshallChannel2Acknowledgement(msg *channel2.Message) (*Acknowledgement, error) { - ack := &Acknowledgement{} - - if err := ack.unmarshallHeader(msg); err != nil { - return nil, err - } - if err := ack.unmarshallSequence(msg.Body); err != nil { - return nil, err - } - - return ack, nil -} - func UnmarshallAcknowledgement(msg *channel.Message) (*Acknowledgement, error) { ack := &Acknowledgement{} @@ -247,9 +233,9 @@ func (payload *Payload) GetSequence() int32 { return payload.Sequence } -func (payload *Payload) Marshall() *channel2.Message { +func (payload *Payload) Marshall() *channel.Message { payload.RTT = uint16(info.NowInMilliseconds()) - msg := channel2.NewMessage(ContentTypePayloadType, payload.Data) + msg := channel.NewMessage(ContentTypePayloadType, payload.Data) for key, value := range payload.Headers { msgHeaderKey := MinHeaderKey + int32(key) msg.Headers[msgHeaderKey] = value @@ -260,36 +246,6 @@ func (payload *Payload) Marshall() *channel2.Message { return msg } -func UnmarshallChannel2Payload(msg *channel2.Message) (*Payload, error) { - var headers map[uint8][]byte - for key, val := range msg.Headers { - if key >= MinHeaderKey && key <= MaxHeaderKey { - if headers == nil { - headers = make(map[uint8][]byte) - } - xgressHeaderKey := uint8(key - MinHeaderKey) - headers[xgressHeaderKey] = val - } - } - - payload := &Payload{ - Headers: headers, - Data: msg.Body, - } - - if err := payload.unmarshallHeader(msg); err != nil { - return nil, err - } - - sequence, ok := msg.GetUint64Header(HeaderKeySequence) - if !ok { - return nil, fmt.Errorf("no sequence found in xgress payload message") - } - payload.Sequence = int32(sequence) - - return payload, nil -} - func UnmarshallPayload(msg *channel.Message) (*Payload, error) { var headers map[uint8][]byte for key, val := range msg.Headers { @@ -367,16 +323,16 @@ const ( type Control struct { Type ControlType CircuitId string - Headers channel2.Headers + Headers channel.Headers } -func (self *Control) Marshall() *channel2.Message { - msg := channel2.NewMessage(ContentTypeControlType, append([]byte{byte(self.Type)}, self.CircuitId...)) +func (self *Control) Marshall() *channel.Message { + msg := channel.NewMessage(ContentTypeControlType, append([]byte{byte(self.Type)}, self.CircuitId...)) msg.Headers = self.Headers return msg } -func UnmarshallControl(msg *channel2.Message) (*Control, error) { +func UnmarshallControl(msg *channel.Message) (*Control, error) { if len(msg.Body) < 2 { return nil, errors.New("control message body too short") } @@ -424,7 +380,7 @@ func (self *Control) GetLoggerFields() logrus.Fields { } } -func RespondToTraceRequest(headers channel2.Headers, hopType, hopId string, response ControlReceiver) { +func RespondToTraceRequest(headers channel.Headers, hopType, hopId string, response ControlReceiver) { resp := &Control{Headers: headers} resp.DecrementAndGetHop() resp.Headers.PutStringHeader(ControlHopType, hopType) diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index f76919fd9..4b7c7adae 100644 --- a/router/xgress/ordering_test.go +++ b/router/xgress/ordering_test.go @@ -2,7 +2,7 @@ package xgress import ( "encoding/binary" - "github.com/openziti/foundation/channel2" + "github.com/openziti/channel" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/metrics" "github.com/openziti/foundation/metrics/metrics_pb" @@ -41,7 +41,7 @@ func (conn *testConn) WritePayload(bytes []byte, _ map[uint8][]byte) (int, error return len(bytes), nil } -func (conn *testConn) HandleControlMsg(ControlType, channel2.Headers, ControlReceiver) error { +func (conn *testConn) HandleControlMsg(ControlType, channel.Headers, ControlReceiver) error { return nil } diff --git a/router/xgress/request.go b/router/xgress/request.go index 87820a18f..197607416 100644 --- a/router/xgress/request.go +++ b/router/xgress/request.go @@ -23,7 +23,6 @@ import ( "github.com/openziti/channel/protobufs" "github.com/openziti/fabric/ctrl_msg" "github.com/openziti/fabric/pb/ctrl_pb" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" "github.com/pkg/errors" @@ -210,7 +209,7 @@ func RemoveTerminator(ctrl CtrlChannel, terminatorId string) error { return err } - if responseMsg != nil && responseMsg.ContentType == channel2.ContentTypeResultType { + if responseMsg != nil && responseMsg.ContentType == channel.ContentTypeResultType { result := channel.UnmarshalResult(responseMsg) if result.Success { log.Debugf("successfully removed service terminator [s/%s]", terminatorId) diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 9562605ed..0dda5c5de 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -21,7 +21,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/fabric/controller/xt" "github.com/openziti/fabric/logcontext" - "github.com/openziti/foundation/channel2" + "github.com/openziti/channel" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/foundation/util/info" @@ -69,7 +69,7 @@ type BindHandler interface { } type ControlReceiver interface { - HandleControlReceive(controlType ControlType, headers channel2.Headers) + HandleControlReceive(controlType ControlType, headers channel.Headers) } // ReceiveHandler is invoked by an xgress whenever data is received from the connected peer. Generally a ReceiveHandler @@ -109,7 +109,7 @@ type Connection interface { LogContext() string ReadPayload() ([]byte, map[uint8][]byte, error) WritePayload([]byte, map[uint8][]byte) (int, error) - HandleControlMsg(controlType ControlType, headers channel2.Headers, responder ControlReceiver) error + HandleControlMsg(controlType ControlType, headers channel.Headers, responder ControlReceiver) error } type Xgress struct { @@ -338,7 +338,7 @@ func (self *Xgress) SendControl(control *Control) error { return self.peer.HandleControlMsg(control.Type, control.Headers, self) } -func (self *Xgress) HandleControlReceive(controlType ControlType, headers channel2.Headers) { +func (self *Xgress) HandleControlReceive(controlType ControlType, headers channel.Headers) { control := &Control{ Type: controlType, CircuitId: self.circuitId, diff --git a/router/xgress_proxy/connection.go b/router/xgress_proxy/connection.go index e1b1479f5..1689939e3 100644 --- a/router/xgress_proxy/connection.go +++ b/router/xgress_proxy/connection.go @@ -17,8 +17,8 @@ package xgress_proxy import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/transport" "github.com/pkg/errors" ) @@ -45,7 +45,7 @@ func (c *proxyXgressConnection) WritePayload(p []byte, headers map[uint8][]byte) return c.Writer().Write(p) } -func (self *proxyXgressConnection) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { +func (self *proxyXgressConnection) HandleControlMsg(controlType xgress.ControlType, headers channel.Headers, responder xgress.ControlReceiver) error { if controlType == xgress.ControlTypeTraceRoute { xgress.RespondToTraceRequest(headers, "xgress/proxy", "", responder) return nil diff --git a/router/xgress_transport/connection.go b/router/xgress_transport/connection.go index 61411e1cb..e88588d92 100644 --- a/router/xgress_transport/connection.go +++ b/router/xgress_transport/connection.go @@ -17,8 +17,8 @@ package xgress_transport import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/transport" "github.com/pkg/errors" ) @@ -48,7 +48,7 @@ func (c *transportXgressConn) WritePayload(p []byte, _ map[uint8][]byte) (n int, return c.Writer().Write(p) } -func (self *transportXgressConn) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { +func (self *transportXgressConn) HandleControlMsg(controlType xgress.ControlType, headers channel.Headers, responder xgress.ControlReceiver) error { if controlType == xgress.ControlTypeTraceRoute { xgress.RespondToTraceRequest(headers, "xgress/transport", "", responder) return nil diff --git a/router/xgress_transport_udp/connection.go b/router/xgress_transport_udp/connection.go index c26bafbef..a14b83836 100644 --- a/router/xgress_transport_udp/connection.go +++ b/router/xgress_transport_udp/connection.go @@ -17,8 +17,8 @@ package xgress_transport_udp import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/info" "github.com/pkg/errors" "net" @@ -41,7 +41,7 @@ func (p *packetConn) WritePayload(data []byte, headers map[uint8][]byte) (n int, return p.Write(data) } -func (self *packetConn) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { +func (self *packetConn) HandleControlMsg(controlType xgress.ControlType, headers channel.Headers, responder xgress.ControlReceiver) error { if controlType == xgress.ControlTypeTraceRoute { xgress.RespondToTraceRequest(headers, "xgress/transport_udp", "", responder) return nil diff --git a/router/xgress_udp/session.go b/router/xgress_udp/session.go index 231b640ff..134b180c7 100644 --- a/router/xgress_udp/session.go +++ b/router/xgress_udp/session.go @@ -17,8 +17,8 @@ package xgress_udp import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/pkg/errors" "io" "net" @@ -66,7 +66,7 @@ func (s *PacketSession) WritePayload(p []byte, _ map[uint8][]byte) (n int, err e return s.Write(p) } -func (s *PacketSession) HandleControlMsg(controlType xgress.ControlType, headers channel2.Headers, responder xgress.ControlReceiver) error { +func (s *PacketSession) HandleControlMsg(controlType xgress.ControlType, headers channel.Headers, responder xgress.ControlReceiver) error { if controlType == xgress.ControlTypeTraceRoute { xgress.RespondToTraceRequest(headers, "xgress/udp", "", responder) return nil diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index aef907422..6a3266747 100644 --- a/router/xlink/xlink.go +++ b/router/xlink/xlink.go @@ -33,7 +33,7 @@ type Listener interface { Close() error } -type Accepter interface { +type Acceptor interface { Accept(xlink Xlink) error } diff --git a/router/xlink_transport/accepter.go b/router/xlink_transport/accepter.go deleted file mode 100644 index 6d5cf8104..000000000 --- a/router/xlink_transport/accepter.go +++ /dev/null @@ -1,13 +0,0 @@ -package xlink_transport - -import ( - "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" -) - -// ChannelAccepter can be implemented and provided to the factory to perform channel2 binding and other channel setup -// tasks at accept time. -// -type ChannelAccepter interface { - AcceptChannel(xlink xlink.Xlink, payloadCh channel2.Channel, latency bool, listenerSide bool) error -} diff --git a/router/xlink_transport/bind.go b/router/xlink_transport/bind.go new file mode 100644 index 000000000..6db82fb83 --- /dev/null +++ b/router/xlink_transport/bind.go @@ -0,0 +1,13 @@ +package xlink_transport + +import ( + "github.com/openziti/channel" + "github.com/openziti/fabric/router/xlink" +) + +// BindHandlerFactory can be implemented and provided to the factory to perform channel binding and other channel setup +// tasks at accept time. +// +type BindHandlerFactory interface { + NewBindHandler(xlink xlink.Xlink, latency bool, listenerSide bool) channel.BindHandler +} diff --git a/router/xlink_transport/config.go b/router/xlink_transport/config.go index b6efaa8e6..5f078e530 100644 --- a/router/xlink_transport/config.go +++ b/router/xlink_transport/config.go @@ -18,7 +18,7 @@ package xlink_transport import ( "fmt" - "github.com/openziti/foundation/channel2" + "github.com/openziti/channel" "github.com/openziti/foundation/transport" "github.com/pkg/errors" "reflect" @@ -56,12 +56,16 @@ func loadListenerConfig(data map[interface{}]interface{}) (*listenerConfig, erro if value, found := data["options"]; found { if submap, ok := value.(map[interface{}]interface{}); ok { - config.options = channel2.LoadOptions(submap) + options, err := channel.LoadOptions(submap) + if err != nil { + return nil, errors.Wrap(err, "failed to parse link listener options") + } + config.options = options } else { return nil, fmt.Errorf("invalid 'options' in listener config (%s)", reflect.TypeOf(value)) } } else { - config.options = channel2.DefaultOptions() + config.options = channel.DefaultOptions() } return config, nil @@ -70,7 +74,7 @@ func loadListenerConfig(data map[interface{}]interface{}) (*listenerConfig, erro type listenerConfig struct { bind transport.Address advertise transport.Address - options *channel2.Options + options *channel.Options } func loadDialerConfig(data map[interface{}]interface{}) (*dialerConfig, error) { @@ -86,7 +90,11 @@ func loadDialerConfig(data map[interface{}]interface{}) (*dialerConfig, error) { if value, found := data["options"]; found { if submap, ok := value.(map[interface{}]interface{}); ok { - config.options = channel2.LoadOptions(submap) + options, err := channel.LoadOptions(submap) + if err != nil { + return nil, errors.Wrap(err, "failed to parse link dialer options") + } + config.options = options } else { return nil, fmt.Errorf("invalid 'options' in dialer config (%s)", reflect.TypeOf(value)) } @@ -97,5 +105,5 @@ func loadDialerConfig(data map[interface{}]interface{}) (*dialerConfig, error) { type dialerConfig struct { split bool - options *channel2.Options + options *channel.Options } diff --git a/router/xlink_transport/connect.go b/router/xlink_transport/connect.go index 6cff3e523..69c9050e9 100644 --- a/router/xlink_transport/connect.go +++ b/router/xlink_transport/connect.go @@ -2,7 +2,7 @@ package xlink_transport import ( "crypto/x509" - "github.com/openziti/foundation/channel2" + "github.com/openziti/channel" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/errorz" "github.com/pkg/errors" @@ -12,7 +12,7 @@ type ConnectionHandler struct { routerId identity.Identity } -func (self *ConnectionHandler) HandleConnection(_ *channel2.Hello, certificates []*x509.Certificate) error { +func (self *ConnectionHandler) HandleConnection(_ *channel.Hello, certificates []*x509.Certificate) error { if len(certificates) == 0 { return errors.New("no certificates provided, unable to verify dialer") } diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go index d1a2ada9a..ab2671a84 100644 --- a/router/xlink_transport/dialer.go +++ b/router/xlink_transport/dialer.go @@ -18,14 +18,22 @@ package xlink_transport import ( "github.com/google/uuid" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" "github.com/pkg/errors" "github.com/sirupsen/logrus" ) +type dialer struct { + id *identity.TokenId + config *dialerConfig + bindHandlerFactory BindHandlerFactory + acceptor xlink.Acceptor + transportConfig transport.Configuration +} + func (self *dialer) Dial(addressString string, linkId *identity.TokenId, routerId string) error { address, err := transport.ParseAddress(addressString) if err != nil { @@ -44,7 +52,7 @@ func (self *dialer) Dial(addressString string, linkId *identity.TokenId, routerI return errors.Wrapf(err, "error dialing outgoing link [l/%s]", linkId.Token) } - if err := self.accepter.Accept(xli); err != nil { + if err := self.acceptor.Accept(xli); err != nil { return errors.Wrapf(err, "error accepting link [l/%s]", linkId.Token) } @@ -55,81 +63,91 @@ func (self *dialer) Dial(addressString string, linkId *identity.TokenId, routerI func (self *dialer) dialSplit(linkId *identity.TokenId, address transport.Address, routerId, connId string) (xlink.Xlink, error) { logrus.Debugf("dialing link with split payload/ack channels [l/%s]", linkId.Token) - payloadDialer := channel2.NewClassicDialer(linkId, address, map[int32][]byte{ + payloadDialer := channel.NewClassicDialer(linkId, address, map[int32][]byte{ LinkHeaderRouterId: []byte(routerId), LinkHeaderConnId: []byte(connId), - LinkHeaderType: {PayloadChannel}, + LinkHeaderType: {byte(PayloadChannel)}, }) logrus.Debugf("dialing payload channel for [l/%s]", linkId.Token) - payloadCh, err := channel2.NewChannelWithTransportConfiguration("l/"+linkId.Token, payloadDialer, self.config.options, self.tcfg) + bindHandler := &splitDialBindHandler{ + dialer: self, + link: &splitImpl{id: linkId, routerId: routerId}, + } + + payloadCh, err := channel.NewChannelWithTransportConfiguration("l/"+linkId.Token, payloadDialer, channel.BindHandlerF(bindHandler.bindPayloadChannel), self.config.options, self.transportConfig) if err != nil { return nil, errors.Wrapf(err, "error dialing payload channel for [l/%s]", linkId.Token) } - ackDialer := channel2.NewClassicDialer(linkId, address, map[int32][]byte{ - LinkHeaderConnId: []byte(connId), - LinkHeaderType: {AckChannel}, - }) - logrus.Debugf("dialing ack channel for [l/%s]", linkId.Token) - ackCh, err := channel2.NewChannelWithTransportConfiguration("l/"+linkId.Token, ackDialer, self.config.options, self.tcfg) + ackDialer := channel.NewClassicDialer(linkId, address, map[int32][]byte{ + LinkHeaderConnId: []byte(connId), + LinkHeaderType: {byte(AckChannel)}, + }) + + _, err = channel.NewChannelWithTransportConfiguration("l/"+linkId.Token, ackDialer, channel.BindHandlerF(bindHandler.bindAckChannel), self.config.options, self.transportConfig) if err != nil { _ = payloadCh.Close() return nil, errors.Wrapf(err, "error dialing ack channel for [l/%s]", linkId.Token) } - xli := &splitImpl{id: linkId, payloadCh: payloadCh, ackCh: ackCh} - - if self.chAccepter != nil { - if err := self.chAccepter.AcceptChannel(xli, payloadCh, true, false); err != nil { - _ = xli.Close() - return nil, errors.Wrapf(err, "error accepting outgoing payload channel for [l/%s]", linkId.Token) - } - if err := self.chAccepter.AcceptChannel(xli, ackCh, false, false); err != nil { - _ = xli.Close() - return nil, errors.Wrapf(err, "error accepting outgoing ack channel for [l/%s]", linkId.Token) - } - } - - return xli, nil + return bindHandler.link, nil } func (self *dialer) dialSingle(linkId *identity.TokenId, address transport.Address, routerId, connId string) (xlink.Xlink, error) { logrus.Debugf("dialing link with single channel [l/%s]", linkId.Token) - payloadDialer := channel2.NewClassicDialer(linkId, address, map[int32][]byte{ + payloadDialer := channel.NewClassicDialer(linkId, address, map[int32][]byte{ LinkHeaderRouterId: []byte(routerId), LinkHeaderConnId: []byte(connId), }) - payloadCh, err := channel2.NewChannelWithTransportConfiguration("l/"+linkId.Token, payloadDialer, self.config.options, self.tcfg) + bindHandler := &dialBindHandler{ + dialer: self, + link: &impl{id: linkId, routerId: routerId}, + } + + _, err := channel.NewChannelWithTransportConfiguration("l/"+linkId.Token, payloadDialer, bindHandler, self.config.options, self.transportConfig) if err != nil { - return nil, errors.Wrapf(err, "dialing link [l/%s] for payload", linkId.Token) + return nil, errors.Wrapf(err, "error dialing link [l/%s]", linkId.Token) } - xli := &impl{ - id: linkId, - ch: payloadCh, - routerId: routerId, - } - - if self.chAccepter != nil { - if err := self.chAccepter.AcceptChannel(xli, payloadCh, true, false); err != nil { - _ = xli.Close() - return nil, errors.Wrapf(err, "error accepting outgoing channel for [l/%s]", linkId.Token) - } - } - - return xli, nil + return bindHandler.link, nil } -type dialer struct { - id *identity.TokenId - config *dialerConfig - accepter xlink.Accepter - chAccepter ChannelAccepter - tcfg transport.Configuration +type dialBindHandler struct { + dialer *dialer + link *impl +} + +func (self *dialBindHandler) BindChannel(binding channel.Binding) error { + self.link.ch = binding.GetChannel() + bindHandler := self.dialer.bindHandlerFactory.NewBindHandler(self.link, true, false) + return bindHandler.BindChannel(binding) +} + +type splitDialBindHandler struct { + link *splitImpl + dialer *dialer +} + +func (self *splitDialBindHandler) bindPayloadChannel(binding channel.Binding) error { + self.link.payloadCh = binding.GetChannel() + bindHandler := self.dialer.bindHandlerFactory.NewBindHandler(self.link, true, false) + if err := bindHandler.BindChannel(binding); err != nil { + return errors.Wrapf(err, "error accepting outgoing payload channel for [l/%s]", self.link.id.Token) + } + return nil +} + +func (self *splitDialBindHandler) bindAckChannel(binding channel.Binding) error { + self.link.ackCh = binding.GetChannel() + bindHandler := self.dialer.bindHandlerFactory.NewBindHandler(self.link, false, false) + if err := bindHandler.BindChannel(binding); err != nil { + return errors.Wrapf(err, "error accepting outgoing ack channel for [l/%s]", self.link.id.Token) + } + return nil } diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go index 416790be7..1b03ea6e9 100644 --- a/router/xlink_transport/factory.go +++ b/router/xlink_transport/factory.go @@ -23,17 +23,29 @@ import ( "github.com/openziti/foundation/transport" ) +type channelType byte + const ( LinkHeaderConnId = 0 LinkHeaderType = 1 LinkHeaderRouterId = 2 - PayloadChannel = 1 - AckChannel = 2 + PayloadChannel channelType = 1 + AckChannel channelType = 2 ) -func NewFactory(accepter xlink.Accepter, chAccepter ChannelAccepter, c transport.Configuration) xlink.Factory { - return &factory{accepter: accepter, chAccepter: chAccepter, tcfg: c} +func (self channelType) String() string { + if self == PayloadChannel { + return "payload" + } + if self == AckChannel { + return "ack" + } + return "invalid" +} + +func NewFactory(accepter xlink.Acceptor, bindHandlerFactory BindHandlerFactory, c transport.Configuration) xlink.Factory { + return &factory{acceptor: accepter, bindHandlerFactory: bindHandlerFactory, transportConfig: c} } func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, configData transport.Configuration) (xlink.Listener, error) { @@ -42,13 +54,12 @@ func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, con return nil, fmt.Errorf("error loading listener configuration (%w)", err) } return &listener{ - id: id, - config: config, - accepter: self.accepter, - chAccepter: self.chAccepter, - tcfg: self.tcfg, - eventC: make(chan linkEvent, 2), - pendingChannels: map[string]*newChannelEvent{}, + id: id, + config: config, + accepter: self.acceptor, + bindHandlerFactory: self.bindHandlerFactory, + tcfg: self.transportConfig, + pendingLinks: map[string]*pendingLink{}, }, nil } @@ -58,16 +69,16 @@ func (self *factory) CreateDialer(id *identity.TokenId, _ xlink.Forwarder, confi return nil, fmt.Errorf("error loading dialer configuration (%w)", err) } return &dialer{ - id: id, - config: config, - accepter: self.accepter, - chAccepter: self.chAccepter, - tcfg: self.tcfg, + id: id, + config: config, + acceptor: self.acceptor, + bindHandlerFactory: self.bindHandlerFactory, + transportConfig: self.transportConfig, }, nil } type factory struct { - accepter xlink.Accepter - chAccepter ChannelAccepter - tcfg transport.Configuration + acceptor xlink.Acceptor + bindHandlerFactory BindHandlerFactory + transportConfig transport.Configuration } diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index 3beabf8b3..2c9ae598a 100644 --- a/router/xlink_transport/listener.go +++ b/router/xlink_transport/listener.go @@ -17,19 +17,31 @@ package xlink_transport import ( - "errors" "fmt" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xlink" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" + "github.com/pkg/errors" "github.com/sirupsen/logrus" + "sync" "time" ) +type listener struct { + id *identity.TokenId + config *listenerConfig + listener channel.UnderlayListener + accepter xlink.Acceptor + bindHandlerFactory BindHandlerFactory + tcfg transport.Configuration + pendingLinks map[string]*pendingLink + lock sync.Mutex +} + func (self *listener) Listen() error { - listener := channel2.NewClassicListenerWithTransportConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.tcfg, nil) + listener := channel.NewClassicListenerWithTransportConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.tcfg, nil) self.listener = listener connectionHandler := &ConnectionHandler{self.id} @@ -37,7 +49,7 @@ func (self *listener) Listen() error { return fmt.Errorf("error listening (%w)", err) } go self.acceptLoop() - go self.handleEvents() + go self.cleanupExpiredPartialLinks() return nil } @@ -51,177 +63,163 @@ func (self *listener) Close() error { func (self *listener) acceptLoop() { for { - ch, err := channel2.NewChannelWithTransportConfiguration("link", self.listener, self.config.options, self.tcfg) - if err != nil && errors.Is(err, channel2.ListenerClosedError) { + _, err := channel.NewChannelWithTransportConfiguration("link", self.listener, self, self.config.options, self.tcfg) + if err != nil && errors.Is(err, channel.ListenerClosedError) { logrus.Errorf("link underlay acceptor closed") return } else if err != nil { logrus.Errorf("error creating link underlay (%v)", err) continue } + } +} - log := pfxlog.ChannelLogger("link", "linkListener").WithField("linkId", ch.Id().Token) +func (self *listener) BindChannel(binding channel.Binding) error { + log := pfxlog.ChannelLogger("link", "linkListener"). + WithField("linkId", binding.GetChannel().Id().Token) - headers := ch.Underlay().Headers() - channelType := byte(0) - routerId := "" - if headers != nil { - if v, ok := headers[LinkHeaderRouterId]; ok { - routerId = string(v) - log = log.WithField("routerId", routerId) - log.Info("accepting link") - } - if val, ok := headers[LinkHeaderType]; ok { - channelType = val[0] - } + headers := binding.GetChannel().Underlay().Headers() + var chanType channelType + routerId := "" + + if headers != nil { + if v, ok := headers[LinkHeaderRouterId]; ok { + routerId = string(v) + log = log.WithField("routerId", routerId) + log.Info("accepting link") } - if channelType != 0 { - id, ok := headers[LinkHeaderConnId] - if !ok { - log.Error("split conn received but missing connection id. closing") - _ = ch.Close() - continue - } - - log.Infof("accepted %v part of split conn", channelType) - - event := &newChannelEvent{ - ch: ch, - channelType: channelType, - id: string(id), - eventTime: time.Now(), - routerId: routerId, - } - self.eventC <- event - continue + if val, ok := headers[LinkHeaderType]; ok { + chanType = channelType(val[0]) } + } - xli := &impl{ - id: ch.Id(), - ch: ch, - routerId: routerId, - } + if chanType != 0 { + log = log.WithField("channelType", chanType) + return self.bindSplitChannel(binding, chanType, routerId, log) + } - log.Info("accepting link") + return self.bindNonSplitChannel(binding, routerId, log) +} - if self.chAccepter != nil { - if err := self.chAccepter.AcceptChannel(xli, ch, true, true); err != nil { - log.WithError(err).Error("error accepting incoming channel") - if err := xli.Close(); err != nil { - log.WithError(err).Debugf("error closing link") - } - } - } +func (self *listener) bindSplitChannel(binding channel.Binding, chanType channelType, routerId string, log *logrus.Entry) error { + headers := binding.GetChannel().Underlay().Headers() + id, ok := headers[LinkHeaderConnId] + if !ok { + return errors.New("split conn received but missing connection id. closing") + } + log.Info("accepted part of split conn") + + xli, err := self.getOrCreateSplitLink(string(id), routerId, binding, chanType) + if err != nil { + log.WithError(err).Error("erroring binding link channel") + } + + latencyPing := chanType == PayloadChannel + if err := self.bindHandlerFactory.NewBindHandler(xli, latencyPing, true).BindChannel(binding); err != nil { + return err + } + + if xli.payloadCh != nil && xli.ackCh != nil { if err := self.accepter.Accept(xli); err != nil { log.WithError(err).Error("error accepting incoming Xlink") - if err := xli.Close(); err != nil { - log.WithError(err).Debugf("error closing link") - } } log.Info("accepted link") } + + return nil } -func (self *listener) handleEvents() { - ticker := time.NewTicker(time.Minute) - for { - select { - case event := <-self.eventC: - event.handle(self) - case <-ticker.C: - now := time.Now() - for k, v := range self.pendingChannels { - if now.Sub(v.eventTime) > (30 * time.Second) { - _ = v.ch.Close() - delete(self.pendingChannels, k) - } - } +func (self *listener) getOrCreateSplitLink(id, routerId string, binding channel.Binding, chanType channelType) (*splitImpl, error) { + self.lock.Lock() + defer self.lock.Unlock() + + var link *splitImpl + + if pending, found := self.pendingLinks[id]; found { + delete(self.pendingLinks, id) + link = pending.link + } else { + pending = &pendingLink{ + link: &splitImpl{ + id: binding.GetChannel().Id(), + routerId: routerId, + }, + eventTime: time.Now(), } - } -} - -type listener struct { - id *identity.TokenId - config *listenerConfig - listener channel2.UnderlayListener - accepter xlink.Accepter - chAccepter ChannelAccepter - tcfg transport.Configuration - eventC chan linkEvent - pendingChannels map[string]*newChannelEvent -} - -type linkEvent interface { - handle(l *listener) -} - -type newChannelEvent struct { - ch channel2.Channel - channelType byte - id string - eventTime time.Time - routerId string -} - -func (event *newChannelEvent) handle(l *listener) { - log := pfxlog.ChannelLogger("link", "linkListener"). - WithField("linkId", event.ch.Id().Token). - WithField("routerId", event.routerId) - - partner, ok := l.pendingChannels[event.id] - if !ok { - l.pendingChannels[event.id] = event - return - } - delete(l.pendingChannels, event.id) - - var payloadCh channel2.Channel - - if partner.channelType == PayloadChannel { - payloadCh = partner.ch - } else if event.channelType == PayloadChannel { - payloadCh = event.ch + self.pendingLinks[id] = pending + link = pending.link } - var ackCh channel2.Channel - - if partner.channelType == AckChannel { - ackCh = partner.ch - } else if event.channelType == AckChannel { - ackCh = event.ch - } - - if payloadCh == nil || ackCh == nil { - log.Errorf("got two link channels, but types aren't correct. %v %v", partner.channelType, event.channelType) - return - } - - xli := &splitImpl{ - id: event.ch.Id(), - payloadCh: payloadCh, - ackCh: ackCh, - routerId: event.routerId, - } - - log.Info("accepting split link") - - if l.chAccepter != nil { - if err := l.chAccepter.AcceptChannel(xli, xli.payloadCh, true, true); err != nil { - log.WithError(err).Error("error accepting incoming channel") - _ = xli.Close() + if chanType == PayloadChannel { + if link.payloadCh == nil { + link.payloadCh = binding.GetChannel() + } else { + return nil, errors.Errorf("got two payload channels for link %v", binding.GetChannel().Id()) } - - if err := l.chAccepter.AcceptChannel(xli, xli.ackCh, true, true); err != nil { - log.WithError(err).Error("error accepting incoming channel") - _ = xli.Close() + } else if chanType == AckChannel { + if link.ackCh == nil { + link.ackCh = binding.GetChannel() + } else { + return nil, errors.Errorf("got two ack channels for link %v", binding.GetChannel().Id()) } + } else { + return nil, errors.Errorf("invalid channel type %v", chanType) } - if err := l.accepter.Accept(xli); err != nil { + return link, nil +} + +func (self *listener) bindNonSplitChannel(binding channel.Binding, routerId string, log *logrus.Entry) error { + xli := &impl{ + id: binding.GetChannel().Id(), + ch: binding.GetChannel(), + routerId: routerId, + } + + bindHandler := self.bindHandlerFactory.NewBindHandler(xli, true, true) + if err := bindHandler.BindChannel(binding); err != nil { + return errors.Wrapf(err, "error binding channel for link [l/%v]", binding.GetChannel().Id()) + } + + log.Info("accepting link") + + if err := self.accepter.Accept(xli); err != nil { log.WithError(err).Error("error accepting incoming Xlink") + if err := xli.Close(); err != nil { + log.WithError(err).Debugf("error closing link") + } + return err } log.Info("accepted link") + return nil +} + +func (self *listener) cleanupExpiredPartialLinks() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + func() { + self.lock.Lock() + defer self.lock.Unlock() + now := time.Now() + for k, v := range self.pendingLinks { + if now.Sub(v.eventTime) > (30 * time.Second) { + _ = v.link.Close() + delete(self.pendingLinks, k) + } + } + }() + } + } +} + +type pendingLink struct { + link *splitImpl + eventTime time.Time } diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index d8ec56df8..fbaa8d927 100644 --- a/router/xlink_transport/xlink.go +++ b/router/xlink_transport/xlink.go @@ -17,8 +17,8 @@ package xlink_transport import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" ) @@ -48,6 +48,6 @@ func (self *impl) DestinationId() string { type impl struct { id *identity.TokenId - ch channel2.Channel + ch channel.Channel routerId string } diff --git a/router/xlink_transport/xlink_split.go b/router/xlink_transport/xlink_split.go index 98eb10496..4da6fe888 100644 --- a/router/xlink_transport/xlink_split.go +++ b/router/xlink_transport/xlink_split.go @@ -17,8 +17,8 @@ package xlink_transport import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/pkg/errors" ) @@ -40,8 +40,14 @@ func (self *splitImpl) SendAcknowledgement(acknowledgement *xgress.Acknowledgeme } func (self *splitImpl) Close() error { - err := self.payloadCh.Close() - err2 := self.ackCh.Close() + var err, err2 error + if self.payloadCh != nil { + err = self.payloadCh.Close() + } + + if self.ackCh != nil { + err2 = self.ackCh.Close() + } if err == nil { return err2 } @@ -57,7 +63,7 @@ func (self *splitImpl) DestinationId() string { type splitImpl struct { id *identity.TokenId - payloadCh channel2.Channel - ackCh channel2.Channel + payloadCh channel.Channel + ackCh channel.Channel routerId string } diff --git a/tests/context.go b/tests/context.go index 73ee83ec8..e7d667898 100644 --- a/tests/context.go +++ b/tests/context.go @@ -303,7 +303,7 @@ func (ctx *TestContext) createMgmtClient() *MgmtClient { mgmtAddress, err := transport.ParseAddress("tls:localhost:10001") ctx.Req.NoError(err) dialer := channel.NewClassicDialer(id, mgmtAddress, nil) - ch, err := channel.NewChannel("mgmt", dialer, nil) + ch, err := channel.NewChannel("mgmt", dialer, nil, nil) ctx.Req.NoError(err) return &MgmtClient{ diff --git a/tests/link_test.go b/tests/link_test.go index 89c77a1b9..187148d26 100644 --- a/tests/link_test.go +++ b/tests/link_test.go @@ -3,10 +3,10 @@ package tests import ( + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/fabric/router/xlink_transport" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" "github.com/sirupsen/logrus" @@ -30,12 +30,16 @@ func (self *testXlinkAcceptor) getLink() xlink.Xlink { return self.link } -type testChannelAcceptor struct{} +type testBindHandlerFactory struct{} -func (t testChannelAcceptor) AcceptChannel(xlink xlink.Xlink, payloadCh channel2.Channel, latency bool, listenerSide bool) error { +func (t testBindHandlerFactory) BindChannel(binding channel.Binding) error { return nil } +func (t testBindHandlerFactory) NewBindHandler(xlink xlink.Xlink, latency bool, listenerSide bool) channel.BindHandler { + return t +} + func Test_LinkWithValidCertFromUnknownChain(t *testing.T) { ctx := NewTestContext(t) defer ctx.Teardown() @@ -55,7 +59,7 @@ func Test_LinkWithValidCertFromUnknownChain(t *testing.T) { tcfg := transport.Configuration{ "split": false, } - factory := xlink_transport.NewFactory(xla, testChannelAcceptor{}, tcfg) + factory := xlink_transport.NewFactory(xla, testBindHandlerFactory{}, tcfg) dialer, err := factory.CreateDialer(badId, nil, tcfg) ctx.Req.NoError(err) linkId := badId.ShallowCloneWithNewToken("testLinkId") @@ -84,7 +88,7 @@ func Test_UnrequestedLinkFromValidRouter(t *testing.T) { tcfg := transport.Configuration{ "split": false, } - factory := xlink_transport.NewFactory(xla, testChannelAcceptor{}, tcfg) + factory := xlink_transport.NewFactory(xla, testBindHandlerFactory{}, tcfg) dialer, err := factory.CreateDialer(router2Id, nil, tcfg) ctx.Req.NoError(err) linkId := router2Id.ShallowCloneWithNewToken("testLinkId") @@ -92,15 +96,17 @@ func Test_UnrequestedLinkFromValidRouter(t *testing.T) { if err != nil { ctx.Req.ErrorIs(err, io.EOF, "unexpected error: %v", err) } else { - payload := &xgress.Payload{ - Header: xgress.Header{ - CircuitId: "hello", - }, - Sequence: 0, - Headers: nil, - Data: []byte{1, 2, 3, 4}, + for i := int32(0); i < 100 && err == nil; i++ { + payload := &xgress.Payload{ + Header: xgress.Header{ + CircuitId: "hello", + }, + Sequence: i, + Headers: nil, + Data: []byte{1, 2, 3, 4}, + } + err = xla.getLink().SendPayload(payload) } - err = xla.getLink().SendPayload(payload) ctx.Req.Error(err) ctx.Req.EqualError(err, "channel closed", "unexpected error: %v", err) } diff --git a/tests/mgmt_test.go b/tests/mgmt_test.go index 18c27b927..90b14db1d 100644 --- a/tests/mgmt_test.go +++ b/tests/mgmt_test.go @@ -23,7 +23,7 @@ func Test_MgmtChannelInvalidClient(t *testing.T) { mgmtAddress, err := transport.ParseAddress("tls:localhost:10001") ctx.Req.NoError(err) dialer := channel.NewClassicDialer(badId, mgmtAddress, nil) - _, err = channel.NewChannel("mgmt", dialer, nil) + _, err = channel.NewChannel("mgmt", dialer, nil, nil) ctx.Req.Error(err) } diff --git a/trace/channel2_peekhandler.go b/trace/channel2_peekhandler.go deleted file mode 100644 index 6ee4c3e69..000000000 --- a/trace/channel2_peekhandler.go +++ /dev/null @@ -1,155 +0,0 @@ -/* - 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 trace - -import ( - "fmt" - "github.com/golang/protobuf/proto" - "github.com/michaelquigley/pfxlog" - "github.com/openziti/fabric/pb/ctrl_pb" - "github.com/openziti/fabric/pb/mgmt_pb" - "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" - trace_pb "github.com/openziti/foundation/trace/pb" - "github.com/openziti/foundation/util/concurrenz" - "time" -) - -var channel2decoders = []channel2.TraceMessageDecoder{&channel2.Decoder{}, ctrl_pb.Channel2Decoder{}, xgress.Channel2Decoder{}, mgmt_pb.Channel2Decoder{}} - -type Channel2PeekHandler struct { - appId string - ch channel2.Channel - enabled concurrenz.AtomicBoolean - controller Controller - decoders []channel2.TraceMessageDecoder - eventSink EventHandler -} - -func (handler *Channel2PeekHandler) EnableTracing(sourceType SourceType, matcher SourceMatcher, resultChan chan<- ToggleApplyResult) { - handler.ToggleTracing(sourceType, matcher, true, resultChan) -} - -func (handler *Channel2PeekHandler) DisableTracing(sourceType SourceType, matcher SourceMatcher, resultChan chan<- ToggleApplyResult) { - handler.ToggleTracing(sourceType, matcher, false, resultChan) -} - -func (handler *Channel2PeekHandler) ToggleTracing(sourceType SourceType, matcher SourceMatcher, enable bool, resultChan chan<- ToggleApplyResult) { - name := handler.ch.LogicalName() - matched := sourceType == SourceTypePipe && matcher.Matches(name) - prevState := handler.IsEnabled() - nextState := prevState - if matched { - handler.enable(enable) - nextState = enable - } - resultChan <- &ToggleApplyResultImpl{matched, - fmt.Sprintf("Link %v.%v matched? %v. Old trace state: %v, New trace state: %v", - handler.appId, name, matched, prevState, nextState)} -} - -func NewChannel2PeekHandler(appId string, ch channel2.Channel, controller Controller, eventSink EventHandler) *Channel2PeekHandler { - handler := &Channel2PeekHandler{ - appId: appId, - ch: ch, - controller: controller, - decoders: channel2decoders, - eventSink: eventSink, - } - controller.AddSource(handler) - return handler -} - -func (handler *Channel2PeekHandler) enable(enabled bool) { - handler.enabled.Set(true) -} - -func (handler *Channel2PeekHandler) IsEnabled() bool { - return handler.enabled.Get() -} - -func (*Channel2PeekHandler) Connect(ch channel2.Channel, remoteAddress string) { -} - -func (handler *Channel2PeekHandler) Rx(msg *channel2.Message, ch channel2.Channel) { - handler.trace(msg, ch, false) -} - -func (handler *Channel2PeekHandler) Tx(msg *channel2.Message, ch channel2.Channel) { - handler.trace(msg, ch, true) -} - -func (handler *Channel2PeekHandler) Close(ch channel2.Channel) { - handler.controller.RemoveSource(handler) -} - -func (handler *Channel2PeekHandler) trace(msg *channel2.Message, ch channel2.Channel, rx bool) { - if !handler.IsEnabled() || msg.ContentType == int32(ctrl_pb.ContentType_TraceEventType) || - msg.ContentType == int32(mgmt_pb.ContentType_StreamTracesEventType) { - return - } - - var decode []byte - for _, decoder := range handler.decoders { - if str, ok := decoder.Decode(msg); ok { - decode = str - break - } - } - - traceMsg := &trace_pb.ChannelMessage{ - Timestamp: time.Now().UnixNano(), - Identity: handler.appId, - Channel: ch.LogicalName(), - IsRx: rx, - ContentType: msg.ContentType, - Sequence: msg.Sequence(), - ReplyFor: msg.ReplyFor(), - Length: int32(len(msg.Body)), - Decode: decode, - } - - // This can result in a message send. Doing a send from inside a peekhandler can cause deadlocks, so it's best avoided - go handler.eventSink.Accept(traceMsg) -} - -func NewChannel2Sink(ch channel2.Channel) EventHandler { - return &channel2Sink{ch} -} - -type channel2Sink struct { - ch channel2.Channel -} - -func (sink *channel2Sink) Accept(event *trace_pb.ChannelMessage) { - log := pfxlog.Logger() - - bytes, err := proto.Marshal(event) - if err != nil { - log.Errorf("Failed to encode metrics message: %v", err) - return - } - - chMsg := channel2.NewMessage(int32(ctrl_pb.ContentType_TraceEventType), bytes) - - err = sink.ch.Send(chMsg) - if err != nil { - log.Errorf("Failed to send trace message: %v", err) - } else { - log.Tracef("Reported trace to fabric controller") - } -} diff --git a/trace/xgress_peekhandler.go b/trace/xgress_peekhandler.go index 3e2aa0c3d..d8c38762c 100644 --- a/trace/xgress_peekhandler.go +++ b/trace/xgress_peekhandler.go @@ -18,8 +18,8 @@ package trace import ( "fmt" + "github.com/openziti/channel" "github.com/openziti/fabric/router/xgress" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" trace_pb "github.com/openziti/foundation/trace/pb" "github.com/openziti/foundation/util/concurrenz" @@ -30,7 +30,7 @@ type XgressPeekHandler struct { appId *identity.TokenId enabled concurrenz.AtomicBoolean controller Controller - decoders []channel2.TraceMessageDecoder + decoders []channel.TraceMessageDecoder eventSink EventHandler } @@ -69,11 +69,10 @@ func (handler *XgressPeekHandler) Close(x *xgress.Xgress) { } func NewXgressPeekHandler(appId *identity.TokenId, controller Controller, eventSink EventHandler) *XgressPeekHandler { - handler := &XgressPeekHandler{ appId: appId, controller: controller, - decoders: channel2decoders, + decoders: decoders, eventSink: eventSink, } controller.AddSource(handler) @@ -89,7 +88,6 @@ func (handler *XgressPeekHandler) IsEnabled() bool { } func (handler *XgressPeekHandler) trace(x *xgress.Xgress, payload *xgress.Payload, rx bool) { - decode, _ := xgress.DecodePayload(payload) traceMsg := &trace_pb.ChannelMessage{