Merge pull request openziti/fabric#377 from openziti/add-xgress-inspect-goroutines

Add inspect variant which includes goroutines related to xgress

inspect/circuit_detail.go

@@ -17,18 +17,30 @@
 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"`
+	CircuitId         string                        `json:"circuitId"`
+	Forwards          map[string]string             `json:"forwards"`
+	XgressDetails     map[string]*XgressDetail      `json:"xgressDetails"`
+	LinkDetails       map[string]*LinkInspectDetail `json:"linkDetails"`
+	includeGoroutines bool
+}
+
+func (self *CircuitInspectDetail) SetIncludeGoroutines(includeGoroutines bool) {
+	self.includeGoroutines = includeGoroutines
+}
+
+func (self *CircuitInspectDetail) IncludeGoroutines() bool {
+	return self.includeGoroutines
 }
 
 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"`
+	XgressPointer         string                  `json:"xgressPointer"`
+	LinkSendBufferPointer string                  `json:"linkSendBufferPointer"`
+	Goroutines            []string                `json:"goroutines"`
 }
 
 type XgressSendBufferDetail struct {

router/forwarder/forwarder.go

@@ -306,7 +306,7 @@ func (forwarder *Forwarder) getXgressForCircuit(circuitId string) XgressDestinat
 	return nil
 }
 
-func (forwarder *Forwarder) InspectCircuit(circuitId string) *inspect.CircuitInspectDetail {
+func (forwarder *Forwarder) InspectCircuit(circuitId string, getRelatedGoroutines bool) *inspect.CircuitInspectDetail {
 	if val, found := forwarder.circuits.circuits.Get(circuitId); found {
 		result := &inspect.CircuitInspectDetail{
 			CircuitId:     circuitId,
@@ -314,6 +314,7 @@ func (forwarder *Forwarder) InspectCircuit(circuitId string) *inspect.CircuitIns
 			XgressDetails: map[string]*inspect.XgressDetail{},
 			LinkDetails:   map[string]*inspect.LinkInspectDetail{},
 		}
+		result.SetIncludeGoroutines(getRelatedGoroutines)
 
 		ft := val.(*forwardTable)
 		ft.destinations.IterCb(func(key string, v interface{}) {

router/handler_ctrl/inspect.go

@@ -98,7 +98,18 @@ func (context *inspectRequestContext) processLocal() {
 			}
 		} else if strings.HasPrefix(lc, "circuit:") {
 			circuitId := requested[len("circuit:"):]
-			result := context.handler.fwd.InspectCircuit(circuitId)
+			result := context.handler.fwd.InspectCircuit(circuitId, false)
+			if result != nil {
+				js, err := json.Marshal(result)
+				if err != nil {
+					context.appendError(errors.Wrap(err, "failed to marshal circuit report to json").Error())
+				} else {
+					context.appendValue(requested, string(js))
+				}
+			}
+		} else if strings.HasPrefix(lc, "circuitandstacks:") {
+			circuitId := requested[len("circuitAndStacks:"):]
+			result := context.handler.fwd.InspectCircuit(circuitId, true)
 			if result != nil {
 				js, err := json.Marshal(result)
 				if err != nil {

router/xgress/xgress.go

@@ -17,6 +17,8 @@
 package xgress
 
 import (
+	"bufio"
+	"bytes"
 	"fmt"
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel"
@@ -25,10 +27,12 @@ import (
 	"github.com/openziti/fabric/logcontext"
 	"github.com/openziti/foundation/identity/identity"
 	"github.com/openziti/foundation/util/concurrenz"
+	"github.com/openziti/foundation/util/debugz"
 	"github.com/openziti/foundation/util/info"
 	"github.com/sirupsen/logrus"
 	"io"
 	"math/rand"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -584,12 +588,68 @@ func (self *Xgress) SendEmptyAck() {
 func (self *Xgress) InspectCircuit(detail *inspect.CircuitInspectDetail) {
 	timeSinceLastRxFromLink := time.Duration(info.NowInMilliseconds()-atomic.LoadInt64(&self.timeOfLastRxFromLink)) * time.Millisecond
 	xgressDetail := &inspect.XgressDetail{
-		Address:             string(self.address),
-		Originator:          self.originator.String(),
-		TimeSinceLastLinkRx: timeSinceLastRxFromLink.String(),
-		SendBufferDetail:    self.payloadBuffer.Inspect(),
-		RecvBufferDetail:    self.linkRxBuffer.Inspect(),
+		Address:               string(self.address),
+		Originator:            self.originator.String(),
+		TimeSinceLastLinkRx:   timeSinceLastRxFromLink.String(),
+		SendBufferDetail:      self.payloadBuffer.Inspect(),
+		RecvBufferDetail:      self.linkRxBuffer.Inspect(),
+		XgressPointer:         fmt.Sprintf("%p", self),
+		LinkSendBufferPointer: fmt.Sprintf("%p", self.payloadBuffer),
 	}
 
 	detail.XgressDetails[string(self.address)] = xgressDetail
+
+	if detail.IncludeGoroutines() {
+		xgressDetail.Goroutines = self.getRelatedGoroutines(xgressDetail.XgressPointer, xgressDetail.LinkSendBufferPointer)
+	}
+}
+
+func (self *Xgress) getRelatedGoroutines(contains ...string) []string {
+	reader := bytes.NewBufferString(debugz.GenerateStack())
+	scanner := bufio.NewScanner(reader)
+	var result []string
+	var buf *bytes.Buffer
+	xgressRelated := false
+	for scanner.Scan() {
+		line := scanner.Text()
+		if strings.HasPrefix(line, "goroutine") && strings.HasSuffix(line, ":") {
+			result = self.addGoroutineIfRelated(buf, xgressRelated, result, contains...)
+			buf = &bytes.Buffer{}
+			xgressRelated = false
+		}
+
+		if buf != nil {
+			if strings.Contains(line, "xgress") {
+				xgressRelated = true
+			}
+			buf.WriteString(line)
+			buf.WriteByte('\n')
+		}
+	}
+	result = self.addGoroutineIfRelated(buf, xgressRelated, result, contains...)
+	if err := scanner.Err(); err != nil {
+		result = append(result, "goroutine parsing error: %v", err.Error())
+	}
+	return result
+}
+
+func (self *Xgress) addGoroutineIfRelated(buf *bytes.Buffer, xgressRelated bool, result []string, contains ...string) []string {
+	if !xgressRelated {
+		return result
+	}
+	if buf != nil {
+		gr := buf.String()
+		// ignore the current goroutine
+		if strings.Contains(gr, "GenerateStack") {
+			return result
+		}
+
+		for _, s := range contains {
+			if strings.Contains(gr, s) {
+				result = append(result, gr)
+				break
+			}
+		}
+	}
+	return result
 }