From e2d29d3f07cdc940c27e54639ff0ae883826df97 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 18 Apr 2022 16:45:51 -0400 Subject: [PATCH 1/2] Rework model for circuit inspections. Add inspect support back to mgmt channel --- controller/api_impl/inspections_api_model.go | 21 +++++- controller/handler_mgmt/bind.go | 19 +++--- controller/handler_mgmt/inspect.go | 72 ++++++++++++++++++++ controller/handler_mgmt/stream_circuits.go | 9 +-- controller/network/inspect.go | 26 ++----- controller/network/network.go | 4 +- health/handler.go | 16 +++++ inspect/circuit_detail.go | 55 +++++++++++++++ inspect/links_inspect_result.go | 29 ++++++++ pb/mgmt_pb/impl.go | 8 +++ router/forwarder/forwarder.go | 25 +++---- router/handler_ctrl/inspect.go | 17 +---- router/xgress/link_receive_buffer.go | 11 +-- router/xgress/link_send_buffer.go | 46 ++++++------- router/xgress/xgress.go | 20 +++--- router/xlink/xlink.go | 4 +- router/xlink_transport/xlink.go | 19 ++++-- router/xlink_transport/xlink_split.go | 19 ++++-- 18 files changed, 294 insertions(+), 126 deletions(-) create mode 100644 controller/handler_mgmt/inspect.go create mode 100644 inspect/circuit_detail.go create mode 100644 inspect/links_inspect_result.go diff --git a/controller/api_impl/inspections_api_model.go b/controller/api_impl/inspections_api_model.go index c6de03621..96642c579 100644 --- a/controller/api_impl/inspections_api_model.go +++ b/controller/api_impl/inspections_api_model.go @@ -17,8 +17,10 @@ package api_impl import ( + "encoding/json" "github.com/openziti/fabric/controller/network" "github.com/openziti/fabric/rest_model" + "strings" ) const EntityNameInspect = "inspections" @@ -29,10 +31,27 @@ func MapInspectResultToRestModel(inspectResult *network.InspectResult) *rest_mod Success: &inspectResult.Success, } for _, val := range inspectResult.Results { + var emitVal interface{} + if strings.HasPrefix(val.Value, "{") { + mapVal := map[string]interface{}{} + if err := json.Unmarshal([]byte(val.Value), &mapVal); err == nil { + emitVal = mapVal + } + } else if strings.HasPrefix(val.Value, "[") { + var arrayVal []interface{} + if err := json.Unmarshal([]byte(val.Value), &arrayVal); err == nil { + emitVal = arrayVal + } + } + + if emitVal == nil { + emitVal = val.Value + } + resp.Values = append(resp.Values, &rest_model.InspectResponseValue{ AppID: &val.AppId, Name: &val.Name, - Value: &val.Value, + Value: emitVal, }) } return resp diff --git a/controller/handler_mgmt/bind.go b/controller/handler_mgmt/bind.go index 077ab7411..d96b950a7 100644 --- a/controller/handler_mgmt/bind.go +++ b/controller/handler_mgmt/bind.go @@ -33,27 +33,26 @@ func NewBindHandler(network *network.Network, xmgmts []xmgmt.Xmgmt) channel.Bind } func (bindHandler *BindHandler) BindChannel(binding channel.Binding) error { - network := bindHandler.network + binding.AddTypedReceiveHandler(newInspectHandler(bindHandler.network)) + binding.AddTypedReceiveHandler(newCreateRouterHandler(bindHandler.network)) + binding.AddTypedReceiveHandler(newListServicesHandler(bindHandler.network)) - binding.AddTypedReceiveHandler(newCreateRouterHandler(network)) - binding.AddTypedReceiveHandler(newListServicesHandler(network)) - - streamMetricHandler := newStreamMetricsHandler(network) + streamMetricHandler := newStreamMetricsHandler(bindHandler.network) binding.AddTypedReceiveHandler(streamMetricHandler) binding.AddCloseHandler(streamMetricHandler) - streamCircuitsHandler := newStreamCircuitsHandler(network) + streamCircuitsHandler := newStreamCircuitsHandler(bindHandler.network) binding.AddTypedReceiveHandler(streamCircuitsHandler) binding.AddCloseHandler(streamCircuitsHandler) - streamTracesHandler := newStreamTracesHandler(network) + streamTracesHandler := newStreamTracesHandler(bindHandler.network) binding.AddTypedReceiveHandler(streamTracesHandler) binding.AddCloseHandler(streamTracesHandler) - binding.AddTypedReceiveHandler(newTogglePipeTracesHandler(network)) + binding.AddTypedReceiveHandler(newTogglePipeTracesHandler(bindHandler.network)) - traceDispatchWrapper := trace.NewDispatchWrapper(network.GetEventDispatcher().Dispatch) - binding.AddPeekHandler(trace.NewChannelPeekHandler(network.GetAppId(), binding.GetChannel(), network.GetTraceController(), traceDispatchWrapper)) + traceDispatchWrapper := trace.NewDispatchWrapper(bindHandler.network.GetEventDispatcher().Dispatch) + binding.AddPeekHandler(trace.NewChannelPeekHandler(bindHandler.network.GetAppId(), binding.GetChannel(), bindHandler.network.GetTraceController(), traceDispatchWrapper)) xmgmtDone := make(chan struct{}) for _, x := range bindHandler.xmgmts { diff --git a/controller/handler_mgmt/inspect.go b/controller/handler_mgmt/inspect.go new file mode 100644 index 000000000..dc5b5eb62 --- /dev/null +++ b/controller/handler_mgmt/inspect.go @@ -0,0 +1,72 @@ +/* + Copyright NetFoundry, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package handler_mgmt + +import ( + "fmt" + "github.com/golang/protobuf/proto" + "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" + "github.com/openziti/fabric/controller/network" + "github.com/openziti/fabric/pb/mgmt_pb" +) + +type inspectHandler struct { + network *network.Network +} + +func newInspectHandler(network *network.Network) *inspectHandler { + return &inspectHandler{network: network} +} + +func (*inspectHandler) ContentType() int32 { + return int32(mgmt_pb.ContentType_InspectRequestType) +} + +func (handler *inspectHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { + go func() { + response := &mgmt_pb.InspectResponse{} + request := &mgmt_pb.InspectRequest{} + if err := proto.Unmarshal(msg.Body, request); err != nil { + response.Success = false + response.Errors = append(response.Errors, fmt.Sprintf("%v: %v", handler.network.GetAppId(), err)) + } else { + result := handler.network.Controllers.Inspections.Inspect(request.AppRegex, request.RequestedValues) + response.Success = result.Success + response.Errors = result.Errors + for _, val := range result.Results { + response.Values = append(response.Values, &mgmt_pb.InspectResponse_InspectValue{ + AppId: val.AppId, + Name: val.Name, + Value: val.Value, + }) + } + } + + body, err := proto.Marshal(response) + if err != nil { + pfxlog.Logger().Errorf("unexpected error serializing InspectResponse (%s)", err) + return + } + + responseMsg := channel.NewMessage(int32(mgmt_pb.ContentType_InspectResponseType), body) + responseMsg.ReplyTo(msg) + if err := ch.Send(responseMsg); err != nil { + pfxlog.Logger().Errorf("unexpected error sending InspectResponse (%s)", err) + } + }() +} diff --git a/controller/handler_mgmt/stream_circuits.go b/controller/handler_mgmt/stream_circuits.go index 574bc390e..831a44c00 100644 --- a/controller/handler_mgmt/stream_circuits.go +++ b/controller/handler_mgmt/stream_circuits.go @@ -20,7 +20,6 @@ import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" - "github.com/openziti/fabric/controller/handler_common" "github.com/openziti/fabric/controller/network" "github.com/openziti/fabric/events" "github.com/openziti/fabric/pb/mgmt_pb" @@ -39,13 +38,7 @@ func (*streamCircuitsHandler) ContentType() int32 { return int32(mgmt_pb.ContentType_StreamCircuitsRequestType) } -func (handler *streamCircuitsHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { - request := &mgmt_pb.StreamCircuitsRequest{} - if err := proto.Unmarshal(msg.Body, request); err != nil { - handler_common.SendFailure(msg, ch, err.Error()) - return - } - +func (handler *streamCircuitsHandler) HandleReceive(_ *channel.Message, ch channel.Channel) { circuitsStreamHandler := &CircuitsStreamHandler{ch: ch} handler.streamHandlers = append(handler.streamHandlers, circuitsStreamHandler) events.AddCircuitEventHandler(circuitsStreamHandler) diff --git a/controller/network/inspect.go b/controller/network/inspect.go index a2ed93f3a..2e3c143c9 100644 --- a/controller/network/inspect.go +++ b/controller/network/inspect.go @@ -17,14 +17,12 @@ package network import ( - "encoding/json" "fmt" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel/protobufs" "github.com/openziti/fabric/pb/ctrl_pb" "github.com/openziti/foundation/util/concurrenz" "regexp" - "strings" "sync" "time" ) @@ -32,7 +30,7 @@ import ( type InspectResultValue struct { AppId string Name string - Value interface{} + Value string } type InspectResult struct { @@ -115,7 +113,7 @@ func (ctx *inspectRequestContext) inspectLocal() { for _, requested := range ctx.requestedValues { result := ctx.network.Inspect(requested) if result != nil { - ctx.appendValue(ctx.network.GetAppId(), requested, result) + ctx.appendValue(ctx.network.GetAppId(), requested, *result) } } close(notifier) @@ -160,27 +158,11 @@ func (ctx *inspectRequestContext) handleRouterMessaging(router *Router, notifier } for _, val := range resp.Values { - handled := false - if strings.HasPrefix(val.Value, "{") { - mapVal := map[string]interface{}{} - if err := json.Unmarshal([]byte(val.Value), &mapVal); err == nil { - ctx.appendValue(router.Id, val.Name, mapVal) - handled = true - } - } else if strings.HasPrefix(val.Value, "[") { - var arrayVal []interface{} - if err := json.Unmarshal([]byte(val.Value), &arrayVal); err == nil { - ctx.appendValue(router.Id, val.Name, arrayVal) - handled = true - } - } - if !handled { - ctx.appendValue(router.Id, val.Name, val.Value) - } + ctx.appendValue(router.Id, val.Name, val.Value) } } -func (ctx *inspectRequestContext) appendValue(appId string, name string, value interface{}) { +func (ctx *inspectRequestContext) appendValue(appId string, name string, value string) { ctx.lock.Lock() defer ctx.lock.Unlock() if !ctx.complete { diff --git a/controller/network/network.go b/controller/network/network.go index b8c0b85f5..5d8af96eb 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -938,10 +938,10 @@ func (network *Network) showOptions() { } } -func (network *Network) Inspect(name string) interface{} { +func (network *Network) Inspect(name string) *string { if strings.ToLower(name) == "stackdump" { result := debugz.GenerateStack() - return result + return &result } return nil } diff --git a/health/handler.go b/health/handler.go index 9229fb987..0c6eafea4 100644 --- a/health/handler.go +++ b/health/handler.go @@ -1,3 +1,19 @@ +/* + 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 health import ( diff --git a/inspect/circuit_detail.go b/inspect/circuit_detail.go new file mode 100644 index 000000000..e683accce --- /dev/null +++ b/inspect/circuit_detail.go @@ -0,0 +1,55 @@ +/* + 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 inspect + +type CircuitInspectDetail struct { + CircuitId string `json:"circuitId"` + Forwards map[string]string `json:"forwards"` + XgressDetails map[string]*XgressDetail `json:"xgressDetails"` + LinkDetails map[string]*LinkInspectDetail `json:"linkDetails"` +} + +type XgressDetail struct { + Address string `json:"address"` + Originator string `json:"originator"` + TimeSinceLastLinkRx string `json:"timeSinceLastLinkRx"` + SendBufferDetail XgressSendBufferDetail `json:"sendBufferDetail"` + RecvBufferDetail XgressRecvBufferDetail `json:"recvBufferDetail"` +} + +type XgressSendBufferDetail struct { + WindowSize uint32 `json:"windowSize"` + LinkSendBufferSize uint32 `json:"linkSendBufferSize"` + LinkRecvBufferSize uint32 `json:"linkRecvBufferSize"` + Accumulator uint32 `json:"accumulator"` + SuccessfulAcks uint32 `json:"successfulAcks"` + DuplicateAcks uint32 `json:"duplicateAcks"` + Retransmits uint32 `json:"retransmits"` + Closed bool `json:"closed"` + BlockedByLocalWindow bool `json:"blockedByLocalWindow"` + BlockedByRemoteWindow bool `json:"blockedByRemoteWindow"` + RetxScale float64 `json:"retxScale"` + RetxThreshold uint32 `json:"retxThreshold"` + TimeSinceLastRetx string `json:"timeSinceLastRetx"` + CloseWhenEmpty bool `json:"closeWhenEmpty"` + AcquiredSafely bool `json:"acquiredSafely"` +} + +type XgressRecvBufferDetail struct { + Size uint32 `json:"size"` + LastSizeSent uint32 `json:"lastSizeSent"` +} diff --git a/inspect/links_inspect_result.go b/inspect/links_inspect_result.go new file mode 100644 index 000000000..8f738bb26 --- /dev/null +++ b/inspect/links_inspect_result.go @@ -0,0 +1,29 @@ +/* + 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 inspect + +type LinksInspectResult struct { + Links []*LinkInspectDetail `json:"links"` +} + +type LinkInspectDetail struct { + Id string `json:"id"` + Split bool `json:"split"` + Protocol string `json:"protocol"` + Dest string `json:"dest"` + DestVersion string `json:"destVersion"` +} diff --git a/pb/mgmt_pb/impl.go b/pb/mgmt_pb/impl.go index da60c5ba9..aa90af56d 100644 --- a/pb/mgmt_pb/impl.go +++ b/pb/mgmt_pb/impl.go @@ -7,3 +7,11 @@ func (request *ListServicesRequest) GetContentType() int32 { func (request *CreateRouterRequest) GetContentType() int32 { return int32(ContentType_CreateRouterRequestType) } + +func (request *InspectRequest) GetContentType() int32 { + return int32(ContentType_InspectRequestType) +} + +func (request *InspectResponse) GetContentType() int32 { + return int32(ContentType_InspectResponseType) +} diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index 5afc1da0b..9a0a48f97 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -18,6 +18,7 @@ package forwarder import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/pb/ctrl_pb" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" @@ -44,7 +45,7 @@ type Destination interface { SendPayload(payload *xgress.Payload) error SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error SendControl(control *xgress.Control) error - Inspect() map[string]interface{} + InspectCircuit(detail *inspect.CircuitInspectDetail) } type XgressDestination interface { @@ -282,11 +283,13 @@ func (forwarder *Forwarder) getXgressForCircuit(circuitId string) XgressDestinat return nil } -func (forwarder *Forwarder) InspectCircuit(circuitId string) *CircuitReport { +func (forwarder *Forwarder) InspectCircuit(circuitId string) *inspect.CircuitInspectDetail { if val, found := forwarder.circuits.circuits.Get(circuitId); found { - result := &CircuitReport{ - Forwards: map[string]string{}, - Destinations: map[string]map[string]interface{}{}, + result := &inspect.CircuitInspectDetail{ + CircuitId: circuitId, + Forwards: map[string]string{}, + XgressDetails: map[string]*inspect.XgressDetail{}, + LinkDetails: map[string]*inspect.LinkInspectDetail{}, } ft := val.(*forwardTable) @@ -296,20 +299,10 @@ func (forwarder *Forwarder) InspectCircuit(circuitId string) *CircuitReport { for _, addr := range result.Forwards { if dest, _ := forwarder.destinations.getDestination(xgress.Address(addr)); dest != nil { - result.Destinations[addr] = dest.Inspect() - } else { - result.Destinations[addr] = map[string]interface{}{ - "type": "missing", - "addr": string(addr), - } + dest.InspectCircuit(result) } } return result } return nil } - -type CircuitReport struct { - Forwards map[string]string - Destinations map[string]map[string]interface{} -} diff --git a/router/handler_ctrl/inspect.go b/router/handler_ctrl/inspect.go index 06eadcdcb..5690523d6 100644 --- a/router/handler_ctrl/inspect.go +++ b/router/handler_ctrl/inspect.go @@ -21,6 +21,7 @@ import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/pb/ctrl_pb" "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xlink" @@ -85,14 +86,9 @@ func (context *inspectRequestContext) processLocal() { if strings.ToLower(lc) == "stackdump" { context.appendValue(requested, debugz.GenerateStack()) } else if strings.ToLower(lc) == "links" { - var result []*linkInspectResult + result := &inspect.LinksInspectResult{} for link := range context.handler.xlinkRegistry.Iter() { - result = append(result, &linkInspectResult{ - Id: link.Id().Token, - Protocol: link.LinkProtocol(), - Dest: link.DestinationId(), - DestVersion: link.DestVersion(), - }) + result.Links = append(result.Links, link.InspectLink()) } js, err := json.Marshal(result) if err != nil { @@ -140,10 +136,3 @@ func (context *inspectRequestContext) appendError(err string) { context.response.Success = false context.response.Errors = append(context.response.Errors, err) } - -type linkInspectResult struct { - Id string `json:"id"` - Protocol string `json:"protocol"` - Dest string `json:"dest"` - DestVersion string `json:"destVersion"` -} diff --git a/router/xgress/link_receive_buffer.go b/router/xgress/link_receive_buffer.go index 285598d35..b5e56dfba 100644 --- a/router/xgress/link_receive_buffer.go +++ b/router/xgress/link_receive_buffer.go @@ -20,6 +20,7 @@ import ( "github.com/emirpasic/gods/trees/btree" "github.com/emirpasic/gods/utils" "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/inspect" "sync/atomic" ) @@ -84,9 +85,9 @@ func (buffer *LinkReceiveBuffer) getLastBufferSizeSent() uint32 { return atomic.LoadUint32(&buffer.lastBufferSizeSent) } -func (buffer *LinkReceiveBuffer) Inspect() map[string]interface{} { - result := map[string]interface{}{} - result["size"] = buffer.Size() - result["lastSizeSent"] = buffer.getLastBufferSizeSent() - return result +func (buffer *LinkReceiveBuffer) Inspect() inspect.XgressRecvBufferDetail { + return inspect.XgressRecvBufferDetail{ + Size: buffer.Size(), + LastSizeSent: buffer.getLastBufferSizeSent(), + } } diff --git a/router/xgress/link_send_buffer.go b/router/xgress/link_send_buffer.go index c4cdd72c7..953fb8643 100644 --- a/router/xgress/link_send_buffer.go +++ b/router/xgress/link_send_buffer.go @@ -18,6 +18,7 @@ package xgress import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/inspect" "github.com/openziti/foundation/util/concurrenz" "github.com/openziti/foundation/util/info" "github.com/pkg/errors" @@ -335,39 +336,38 @@ func (buffer *LinkSendBuffer) scale(factor float64) { } } -func (buffer *LinkSendBuffer) inspect() map[string]interface{} { - result := map[string]interface{}{} - result["windowsSize"] = buffer.windowsSize - result["linkSendBufferSize"] = buffer.linkSendBufferSize - result["linkRecvBufferSize"] = buffer.linkRecvBufferSize - result["accumulator"] = buffer.accumulator - result["successfulAcks"] = buffer.successfulAcks - result["duplicateAcks"] = buffer.duplicateAcks - result["retransmits"] = buffer.retransmits - result["closed"] = buffer.closed.Get() - result["blockedByLocalWindow"] = buffer.blockedByLocalWindow - result["blockedByRemoteWindow"] = buffer.blockedByRemoteWindow - result["retxScale"] = buffer.retxScale - result["retxThreshold"] = buffer.retxThreshold - +func (buffer *LinkSendBuffer) inspect() *inspect.XgressSendBufferDetail { timeSinceLastRetransmit := time.Duration(info.NowInMilliseconds()-buffer.lastRetransmitTime) * time.Millisecond - result["timeSinceLastRetx"] = timeSinceLastRetransmit.String() - result["closeWhenEmpty"] = buffer.closeWhenEmpty.Get() - + result := &inspect.XgressSendBufferDetail{ + WindowSize: buffer.windowsSize, + LinkSendBufferSize: buffer.linkSendBufferSize, + LinkRecvBufferSize: buffer.linkRecvBufferSize, + Accumulator: buffer.accumulator, + SuccessfulAcks: buffer.successfulAcks, + DuplicateAcks: buffer.duplicateAcks, + Retransmits: buffer.retransmits, + Closed: buffer.closed.Get(), + BlockedByLocalWindow: buffer.blockedByLocalWindow, + BlockedByRemoteWindow: buffer.blockedByRemoteWindow, + RetxScale: buffer.retxScale, + RetxThreshold: buffer.retxThreshold, + TimeSinceLastRetx: timeSinceLastRetransmit.String(), + CloseWhenEmpty: buffer.closeWhenEmpty.Get(), + } return result } -func (buffer *LinkSendBuffer) Inspect() map[string]interface{} { +func (buffer *LinkSendBuffer) Inspect() *inspect.XgressSendBufferDetail { timeout := time.After(25 * time.Millisecond) inspectEvent := &InspectEvent{ - notifyComplete: make(chan map[string]interface{}, 1), + notifyComplete: make(chan *inspect.XgressSendBufferDetail, 1), } select { case buffer.inspectRequests <- inspectEvent: select { case result := <-inspectEvent.notifyComplete: - result["acquiredSafely"] = true + result.AcquiredSafely = true return result case <-timeout: } @@ -375,12 +375,12 @@ func (buffer *LinkSendBuffer) Inspect() map[string]interface{} { } result := buffer.inspect() - result["acquiredSafely"] = false + result.AcquiredSafely = false return result } type InspectEvent struct { - notifyComplete chan map[string]interface{} + notifyComplete chan *inspect.XgressSendBufferDetail } func (self *InspectEvent) handle(buffer *LinkSendBuffer) { diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 5d9bf03aa..b051be0fd 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -21,6 +21,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/channel" "github.com/openziti/fabric/controller/xt" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/logcontext" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/concurrenz" @@ -580,16 +581,15 @@ func (self *Xgress) SendEmptyAck() { acker.ack(ack, self.address) } -func (self *Xgress) Inspect() map[string]interface{} { - result := map[string]interface{}{} - result["type"] = "xgress" - result["addr"] = string(self.address) - result["originator"] = self.originator.String() - +func (self *Xgress) InspectCircuit(detail *inspect.CircuitInspectDetail) { timeSinceLastRxFromLink := time.Duration(info.NowInMilliseconds()-atomic.LoadInt64(&self.timeOfLastRxFromLink)) * time.Millisecond - result["timeSinceLastLinkRx"] = timeSinceLastRxFromLink.String() - result["sendBuffer"] = self.payloadBuffer.Inspect() - result["recvBuffer"] = self.linkRxBuffer.Inspect() + xgressDetail := &inspect.XgressDetail{ + Address: string(self.address), + Originator: self.originator.String(), + TimeSinceLastLinkRx: timeSinceLastRxFromLink.String(), + SendBufferDetail: *self.payloadBuffer.Inspect(), + RecvBufferDetail: self.linkRxBuffer.Inspect(), + } - return result + detail.XgressDetails[string(self.address)] = xgressDetail } diff --git a/router/xlink/xlink.go b/router/xlink/xlink.go index 641fe1019..b0f11de3d 100644 --- a/router/xlink/xlink.go +++ b/router/xlink/xlink.go @@ -18,6 +18,7 @@ package xlink import ( "github.com/openziti/fabric/controller/xctrl" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/identity/identity" "github.com/openziti/transport" @@ -89,7 +90,8 @@ type Xlink interface { DestVersion() string LinkProtocol() string HandleCloseNotification(f func()) - Inspect() map[string]interface{} + InspectCircuit(circuitDetail *inspect.CircuitInspectDetail) + InspectLink() *inspect.LinkInspectDetail } type Forwarder interface { diff --git a/router/xlink_transport/xlink.go b/router/xlink_transport/xlink.go index 3d2e49934..1b4826e3b 100644 --- a/router/xlink_transport/xlink.go +++ b/router/xlink_transport/xlink.go @@ -18,6 +18,7 @@ package xlink_transport import ( "github.com/openziti/channel" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/concurrenz" @@ -75,12 +76,16 @@ func (self *impl) HandleCloseNotification(f func()) { } } -func (self *impl) Inspect() map[string]interface{} { - return map[string]interface{}{ - "type": "link", - "split": false, - "id": self.Id().Token, - "dest": self.DestinationId(), - "protocol": self.LinkProtocol(), +func (self *impl) InspectCircuit(detail *inspect.CircuitInspectDetail) { + detail.LinkDetails[self.id.Token] = self.InspectLink() +} + +func (self *impl) InspectLink() *inspect.LinkInspectDetail { + return &inspect.LinkInspectDetail{ + Id: self.Id().Token, + Split: false, + Protocol: self.LinkProtocol(), + Dest: self.DestinationId(), + DestVersion: self.DestVersion(), } } diff --git a/router/xlink_transport/xlink_split.go b/router/xlink_transport/xlink_split.go index 783345c2d..c0113e97f 100644 --- a/router/xlink_transport/xlink_split.go +++ b/router/xlink_transport/xlink_split.go @@ -18,6 +18,7 @@ package xlink_transport import ( "github.com/openziti/channel" + "github.com/openziti/fabric/inspect" "github.com/openziti/fabric/router/xgress" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/util/concurrenz" @@ -91,12 +92,16 @@ func (self *splitImpl) HandleCloseNotification(f func()) { } } -func (self *splitImpl) Inspect() map[string]interface{} { - return map[string]interface{}{ - "type": "link", - "split": true, - "id": self.Id().Token, - "dest": self.DestinationId(), - "protocol": self.LinkProtocol(), +func (self *splitImpl) InspectCircuit(detail *inspect.CircuitInspectDetail) { + detail.LinkDetails[self.id.Token] = self.InspectLink() +} + +func (self *splitImpl) InspectLink() *inspect.LinkInspectDetail { + return &inspect.LinkInspectDetail{ + Id: self.Id().Token, + Split: true, + Protocol: self.LinkProtocol(), + Dest: self.DestinationId(), + DestVersion: self.DestVersion(), } } From b48b1d6fc52e122adf7d51ee4a7315f987ff97d3 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 19 Apr 2022 21:50:21 -0400 Subject: [PATCH 2/2] Add another path test. Address review comments --- controller/network/path_test.go | 66 ++++++++++++++++++++++++++++ inspect/circuit_detail.go | 10 ++--- router/xgress/link_receive_buffer.go | 4 +- router/xgress/xgress.go | 2 +- 4 files changed, 74 insertions(+), 8 deletions(-) diff --git a/controller/network/path_test.go b/controller/network/path_test.go index c7954cddd..6bed5aa87 100644 --- a/controller/network/path_test.go +++ b/controller/network/path_test.go @@ -17,6 +17,7 @@ package network import ( + "github.com/stretchr/testify/require" "runtime" "testing" "time" @@ -571,3 +572,68 @@ func TestShortestPathWithUntraversableEdgeRoutersAndUntraversableMiddle(t *testi req.Equal(int64(0), cost) } + +func TestRouterCost(t *testing.T) { + ctx := db.NewTestContext(t) + defer ctx.Cleanup() + + req := require.New(t) + + closeNotify := make(chan struct{}) + defer close(closeNotify) + + network, err := NewNetwork("test", nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify) + req.NoError(err) + + addr := "tcp:0.0.0.0:0" + transportAddr, err := tcp.AddressParser{}.Parse(addr) + req.NoError(err) + + r0 := newRouterForTest("r0", "", transportAddr, nil, 10, true) + network.Routers.markConnected(r0) + + r1 := newRouterForTest("r1", "", transportAddr, nil, 100, false) + network.Routers.markConnected(r1) + + r2 := newRouterForTest("r2", "", transportAddr, nil, 200, false) + network.Routers.markConnected(r2) + + r3 := newRouterForTest("r3", "", transportAddr, nil, 20, true) + network.Routers.markConnected(r3) + + newTestLink(network, "l0", r0, r1) + newTestLink(network, "l1", r0, r2) + newTestLink(network, "l2", r1, r3) + newTestLink(network, "l3", r2, r3) + + path, cost, err := network.shortestPath(r0, r3) + req.NoError(err) + req.NotNil(t, path) + req.Len(path, 3) + req.Equal("r0", path[0].Id) + req.Equal("r1", path[1].Id) + req.Equal("r3", path[2].Id) + + req.Equal(int64(122), cost) + + r1.Cost = 300 + + path, cost, err = network.shortestPath(r0, r3) + req.NoError(err) + req.NotNil(t, path) + req.Len(path, 3) + req.Equal("r0", path[0].Id) + req.Equal("r2", path[1].Id) + req.Equal("r3", path[2].Id) + + req.Equal(int64(222), cost) +} + +func newTestLink(network *Network, id string, srcR, destR *Router) *Link { + l := newLink(id, "tls") + l.Src = srcR + l.Dst = destR + l.addState(newLinkState(Connected)) + network.linkController.add(l) + return l +} diff --git a/inspect/circuit_detail.go b/inspect/circuit_detail.go index e683accce..591db441e 100644 --- a/inspect/circuit_detail.go +++ b/inspect/circuit_detail.go @@ -24,11 +24,11 @@ type CircuitInspectDetail struct { } type XgressDetail struct { - Address string `json:"address"` - Originator string `json:"originator"` - TimeSinceLastLinkRx string `json:"timeSinceLastLinkRx"` - SendBufferDetail XgressSendBufferDetail `json:"sendBufferDetail"` - RecvBufferDetail XgressRecvBufferDetail `json:"recvBufferDetail"` + Address string `json:"address"` + Originator string `json:"originator"` + TimeSinceLastLinkRx string `json:"timeSinceLastLinkRx"` + SendBufferDetail *XgressSendBufferDetail `json:"sendBufferDetail"` + RecvBufferDetail *XgressRecvBufferDetail `json:"recvBufferDetail"` } type XgressSendBufferDetail struct { diff --git a/router/xgress/link_receive_buffer.go b/router/xgress/link_receive_buffer.go index b5e56dfba..2c2bab2c4 100644 --- a/router/xgress/link_receive_buffer.go +++ b/router/xgress/link_receive_buffer.go @@ -85,8 +85,8 @@ func (buffer *LinkReceiveBuffer) getLastBufferSizeSent() uint32 { return atomic.LoadUint32(&buffer.lastBufferSizeSent) } -func (buffer *LinkReceiveBuffer) Inspect() inspect.XgressRecvBufferDetail { - return inspect.XgressRecvBufferDetail{ +func (buffer *LinkReceiveBuffer) Inspect() *inspect.XgressRecvBufferDetail { + return &inspect.XgressRecvBufferDetail{ Size: buffer.Size(), LastSizeSent: buffer.getLastBufferSizeSent(), } diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index b051be0fd..d24bbcb54 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -587,7 +587,7 @@ func (self *Xgress) InspectCircuit(detail *inspect.CircuitInspectDetail) { Address: string(self.address), Originator: self.originator.String(), TimeSinceLastLinkRx: timeSinceLastRxFromLink.String(), - SendBufferDetail: *self.payloadBuffer.Inspect(), + SendBufferDetail: self.payloadBuffer.Inspect(), RecvBufferDetail: self.linkRxBuffer.Inspect(), }