mirror of
https://github.com/openziti/ziti.git
synced 2026-10-13 23:59:53 +00:00
Merge pull request openziti/fabric#358 from openziti/inspect-updates
Rework model for circuit inspections. Add inspect support back to mgm…
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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{}
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+10
-10
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user