gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-12 23:32:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request openziti/fabric#192 from openziti/goroutines-cleanup

Browse Source 
Goroutines cleanup
This commit is contained in:
Paul Lorenz
2021-02-17 20:27:05 -05:00
committed by GitHub
parent 1823ec01fc 2b283dcdce
commit dac6f35ac9
23 changed files with 321 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/main.yml
+1 -1
View File
@@ -17,7 +17,7 @@ jobs:
- name: Install Go
uses: actions/setup-go@v2
with:
go-version: '^1.15.0'
go-version: '^1.16.0'
- name: Install Ziti CI
uses: netfoundry/ziti-ci@v1
.github/workflows/update-dependency.yml
+1 -1
View File
@@ -19,7 +19,7 @@ jobs:
- name: Install Go
uses: actions/setup-go@v2
with:
go-version: '^1.15.0'
go-version: '^1.16.0'
- name: Install Ziti CI
uses: netfoundry/ziti-ci@v1
controller/controller.go
+10 -3
View File
@@ -69,7 +69,7 @@ func NewController(cfg *Config, versionProvider common.VersionProvider) (*Contro
c.loadEventHandlers()
if n, err := network.NewNetwork(cfg.Id, cfg.Network, cfg.Db, cfg.Metrics, versionProvider); err == nil {
if n, err := network.NewNetwork(cfg.Id, cfg.Network, cfg.Db, cfg.Metrics, versionProvider, c.shutdownC); err == nil {
c.network = n
} else {
return nil, err
@@ -134,6 +134,10 @@ func (c *Controller) Run() error {
return nil
}
func (c *Controller) GetCloseNotifyChannel() <-chan struct{} {
return c.shutdownC
}
func (c *Controller) Shutdown() {
if c.isShutdown.CompareAndSwap(false, true) {
close(c.shutdownC)
@@ -150,8 +154,6 @@ func (c *Controller) Shutdown() {
}
}
c.network.Shutdown()
if c.config.Db != nil {
if err := c.config.Db.Close(); err != nil {
pfxlog.Logger().WithError(err).Error("failed to close db")
@@ -249,6 +251,11 @@ func (c *Controller) RegisterXctrl(x xctrl.Xctrl) error {
}
if x.Enabled() {
c.xctrls = append(c.xctrls, x)
if c.config.Trace.Handler != nil {
for _, decoder := range x.GetTraceDecoders() {
c.config.Trace.Handler.AddDecoder(decoder)
}
}
}
return nil
}
controller/network/circuit_test.go
+8 -2
View File
@@ -34,8 +34,11 @@ func TestSimpleCircuit2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest())
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
@@ -96,8 +99,11 @@ func TestTransitCircuit2(t *testing.T) {
ctx := db.NewTestContext(t)
defer ctx.Cleanup()
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest())
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
assert.Nil(t, err)
addr := "tcp:0.0.0.0:0"
controller/network/network.go
+13 -19
View File
@@ -34,7 +34,6 @@ import (
"github.com/openziti/foundation/metrics"
"github.com/openziti/foundation/metrics/metrics_pb"
"github.com/openziti/foundation/storage/boltz"
"github.com/openziti/foundation/util/concurrenz"
"github.com/openziti/foundation/util/sequence"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
@@ -62,8 +61,7 @@ type Network struct {
traceController trace.Controller
routerPresenceHandlers []RouterPresenceHandler
capabilities []string
shutdownChan chan struct{}
isShutdown concurrenz.AtomicBoolean
closeNotify <-chan struct{}
lock sync.Mutex
strategyRegistry xt.Registry
lastSnapshot time.Time
@@ -71,14 +69,13 @@ type Network struct {
VersionProvider common.VersionProvider
}
func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, metricsCfg *metrics.Config, versionProvider common.VersionProvider) (*Network, error) {
func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, metricsCfg *metrics.Config, versionProvider common.VersionProvider, closeNotify <-chan struct{}) (*Network, error) {
stores, err := db.InitStores(database)
if err != nil {
return nil, err
}
controllers := NewControllers(database, stores)
eventDispatcher := event.NewDispatcher()
network := &Network{
Controllers: controllers,
@@ -91,9 +88,9 @@ func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, m
sessionController: newSessionController(),
routeSenderController: newRouteSenderController(),
sequence: sequence.NewSequence(),
eventDispatcher: eventDispatcher,
traceController: trace.NewController(),
shutdownChan: make(chan struct{}),
eventDispatcher: event.NewDispatcher(closeNotify),
traceController: trace.NewController(closeNotify),
closeNotify: closeNotify,
strategyRegistry: xt.GlobalRegistry(),
lastSnapshot: time.Now().Add(-time.Hour),
metricsRegistry: metrics.NewRegistry(nodeId.Token, nil),
@@ -114,6 +111,8 @@ func (network *Network) relayControllerMetrics(cfg *metrics.Config) {
}
go func() {
timer := time.NewTicker(reportInterval)
defer timer.Stop()
dispatcher := metrics.NewDispatchWrapper(network.eventDispatcher.Dispatch)
for {
select {
@@ -121,6 +120,8 @@ func (network *Network) relayControllerMetrics(cfg *metrics.Config) {
if msg := network.metricsRegistry.Poll(); msg != nil {
dispatcher.AcceptMetrics(msg)
}
case <-network.closeNotify:
return
}
}
}()
@@ -638,21 +639,14 @@ func (network *Network) Run() {
network.clean()
network.smart()
case _, ok := <-network.shutdownChan:
if !ok {
return
}
case <-network.closeNotify:
events.RemoveMetricsEventHandler(network)
network.metricsRegistry.DisposeAll()
return
}
}
}
func (network *Network) Shutdown() {
if network.isShutdown.CompareAndSwap(false, true) {
close(network.shutdownChan)
events.RemoveMetricsEventHandler(network)
}
}
func (network *Network) AddCapability(capability string) {
network.lock.Lock()
defer network.lock.Unlock()
controller/network/path_test.go
+4 -1
View File
@@ -30,8 +30,11 @@ func TestShortestPath(t *testing.T) {
req := assert.New(t)
closeNotify := make(chan struct{})
defer close(closeNotify)
nodeId := &identity.TokenId{Token: "test"}
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest())
network, err := NewNetwork(nodeId, nil, ctx.GetDb(), nil, NewVersionProviderTest(), closeNotify)
req.NoError(err)
addr := "tcp:0.0.0.0:0"
controller/xctrl/xctrl.go
+1
View File
@@ -28,4 +28,5 @@ type Xctrl interface {
Enabled() bool
Run(ctrl channel2.Channel, db boltz.Db, done chan struct{}) error
NotifyOfReconnect()
GetTraceDecoders() []channel2.TraceMessageDecoder
}
controller/xctrl_example/example.go
+8 -4
View File
@@ -34,10 +34,6 @@ type example struct {
enabled bool
}
func (example *example) NotifyOfReconnect() {
logrus.Info("control channel reconnected")
}
func NewExample() xctrl.Xctrl {
return &example{delay: 1, enabled: false}
}
@@ -92,4 +88,12 @@ func (example *example) Run(ctrl channel2.Channel, _ boltz.Db, done chan struct{
return nil
}
func (example *example) NotifyOfReconnect() {
logrus.Info("control channel reconnected")
}
func (example *example) GetTraceDecoders() []channel2.TraceMessageDecoder {
return nil
}
const contentType = 99999
ctrl_msg/messages.go
+8 -18
View File
@@ -17,10 +17,7 @@
package ctrl_msg
import (
"bytes"
"encoding/binary"
"github.com/openziti/foundation/channel2"
"github.com/pkg/errors"
)
const (
@@ -45,24 +42,17 @@ func NewSessionFailedMsg(message string) *channel2.Message {
return channel2.NewMessage(SessionFailedType, []byte(message))
}
func NewRouteResultSuccessMsg(sessionId string, attempt int) (*channel2.Message, error) {
func NewRouteResultSuccessMsg(sessionId string, attempt int) *channel2.Message {
msg := channel2.NewMessage(RouteResultType, []byte(sessionId))
buf := new(bytes.Buffer)
if err := binary.Write(buf, binary.LittleEndian, uint32(attempt)); err != nil {
return nil, errors.Wrap(err, "error encoding route result attempt")
}
msg.Headers[RouteResultAttemptHeader] = buf.Bytes()
msg.Headers[RouteResultSuccessHeader] = []byte{1}
return msg, nil
msg.PutUint32Header(RouteResultAttemptHeader, uint32(attempt))
msg.PutUint32Header(RouteResultAttemptHeader, uint32(attempt))
msg.PutBoolHeader(RouteResultSuccessHeader, true)
return msg
}
func NewRouteResultFailedMessage(sessionId string, attempt int, rerr string) (*channel2.Message, error) {
func NewRouteResultFailedMessage(sessionId string, attempt int, rerr string) *channel2.Message {
msg := channel2.NewMessage(RouteResultType, []byte(sessionId))
buf := new(bytes.Buffer)
if err := binary.Write(buf, binary.LittleEndian, uint32(attempt)); err != nil {
return nil, errors.Wrap(err, "error encoding route result attempt")
}
msg.Headers[RouteResultAttemptHeader] = buf.Bytes()
msg.PutUint32Header(RouteResultAttemptHeader, uint32(attempt))
msg.Headers[RouteResultErrorHeader] = []byte(rerr)
return msg, nil
return msg
}
pb/ctrl_pb/ctrl.pb.go
+52 -62
View File
@@ -1,13 +1,12 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.23.0
// protoc-gen-go v1.25.0-devel
// protoc v3.14.0
// source: ctrl.proto
package ctrl_pb
import (
proto "github.com/golang/protobuf/proto"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
@@ -21,29 +20,27 @@ const (
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// This is a compile-time assertion that a sufficiently up-to-date version
// of the legacy proto package is being used.
const _ = proto.ProtoPackageIsVersion4
type ContentType int32
const (
ContentType_Zero ContentType = 0
ContentType_SessionRequestType ContentType = 1000
ContentType_SessionSuccessType ContentType = 1001
ContentType_DialType ContentType = 1002
ContentType_LinkType ContentType = 1003
ContentType_FaultType ContentType = 1004
ContentType_RouteType ContentType = 1005
ContentType_UnrouteType ContentType = 1006
ContentType_MetricsType ContentType = 1007
ContentType_TogglePipeTracesRequestType ContentType = 1008
ContentType_TraceEventType ContentType = 1010
ContentType_CreateTerminatorRequestType ContentType = 1011
ContentType_RemoveTerminatorRequestType ContentType = 1012
ContentType_InspectRequestType ContentType = 1013
ContentType_InspectResponseType ContentType = 1014
ContentType_SessionFailedType ContentType = 1016
ContentType_Zero ContentType = 0
ContentType_SessionRequestType ContentType = 1000
// defined in ctrl_msg/messages.go now
// SessionSuccessType = 1001;
ContentType_DialType ContentType = 1002
ContentType_LinkType ContentType = 1003
ContentType_FaultType ContentType = 1004
ContentType_RouteType ContentType = 1005
ContentType_UnrouteType ContentType = 1006
ContentType_MetricsType ContentType = 1007
ContentType_TogglePipeTracesRequestType ContentType = 1008
ContentType_TraceEventType ContentType = 1010
ContentType_CreateTerminatorRequestType ContentType = 1011
ContentType_RemoveTerminatorRequestType ContentType = 1012
ContentType_InspectRequestType ContentType = 1013
ContentType_InspectResponseType ContentType = 1014
// defined in ctrl_msg/messages.go now
// SessionFailedType = 1016;
ContentType_ValidateTerminatorsRequestType ContentType = 1017
ContentType_UpdateTerminatorRequestType ContentType = 1018
)
@@ -53,7 +50,6 @@ var (
ContentType_name = map[int32]string{
0: "Zero",
1000: "SessionRequestType",
1001: "SessionSuccessType",
1002: "DialType",
1003: "LinkType",
1004: "FaultType",
@@ -66,14 +62,12 @@ var (
1012: "RemoveTerminatorRequestType",
1013: "InspectRequestType",
1014: "InspectResponseType",
1016: "SessionFailedType",
1017: "ValidateTerminatorsRequestType",
1018: "UpdateTerminatorRequestType",
}
ContentType_value = map[string]int32{
"Zero": 0,
"SessionRequestType": 1000,
"SessionSuccessType": 1001,
"DialType": 1002,
"LinkType": 1003,
"FaultType": 1004,
@@ -86,7 +80,6 @@ var (
"RemoveTerminatorRequestType": 1012,
"InspectRequestType": 1013,
"InspectResponseType": 1014,
"SessionFailedType": 1016,
"ValidateTerminatorsRequestType": 1017,
"UpdateTerminatorRequestType": 1018,
}
@@ -1324,44 +1317,41 @@ var file_ctrl_proto_rawDesc = []byte{
0x74, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61,
0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
0x2a, 0xb8, 0x03, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65,
0x2a, 0x87, 0x03, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65,
0x12, 0x08, 0x0a, 0x04, 0x5a, 0x65, 0x72, 0x6f, 0x10, 0x00, 0x12, 0x17, 0x0a, 0x12, 0x53, 0x65,
0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65,
0x10, 0xe8, 0x07, 0x12, 0x17, 0x0a, 0x12, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x53, 0x75,
0x63, 0x63, 0x65, 0x73, 0x73, 0x54, 0x79, 0x70, 0x65, 0x10, 0xe9, 0x07, 0x12, 0x0d, 0x0a, 0x08,
0x44, 0x69, 0x61, 0x6c, 0x54, 0x79, 0x70, 0x65, 0x10, 0xea, 0x07, 0x12, 0x0d, 0x0a, 0x08, 0x4c,
0x69, 0x6e, 0x6b, 0x54, 0x79, 0x70, 0x65, 0x10, 0xeb, 0x07, 0x12, 0x0e, 0x0a, 0x09, 0x46, 0x61,
0x75, 0x6c, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xec, 0x07, 0x12, 0x0e, 0x0a, 0x09, 0x52, 0x6f,
0x75, 0x74, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xed, 0x07, 0x12, 0x10, 0x0a, 0x0b, 0x55, 0x6e,
0x72, 0x6f, 0x75, 0x74, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xee, 0x07, 0x12, 0x10, 0x0a, 0x0b,
0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x54, 0x79, 0x70, 0x65, 0x10, 0xef, 0x07, 0x12, 0x20,
0x0a, 0x1b, 0x54, 0x6f, 0x67, 0x67, 0x6c, 0x65, 0x50, 0x69, 0x70, 0x65, 0x54, 0x72, 0x61, 0x63,
0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf0, 0x07,
0x12, 0x13, 0x0a, 0x0e, 0x54, 0x72, 0x61, 0x63, 0x65, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79,
0x70, 0x65, 0x10, 0xf2, 0x07, 0x12, 0x20, 0x0a, 0x1b, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54,
0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
0x54, 0x79, 0x70, 0x65, 0x10, 0xf3, 0x07, 0x12, 0x20, 0x0a, 0x1b, 0x52, 0x65, 0x6d, 0x6f, 0x76,
0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf4, 0x07, 0x12, 0x17, 0x0a, 0x12, 0x49, 0x6e, 0x73,
0x70, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10,
0xf5, 0x07, 0x12, 0x18, 0x0a, 0x13, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73,
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf6, 0x07, 0x12, 0x16, 0x0a, 0x11,
0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x54, 0x79, 0x70,
0x65, 0x10, 0xf8, 0x07, 0x12, 0x23, 0x0a, 0x1e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65,
0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf9, 0x07, 0x12, 0x20, 0x0a, 0x1b, 0x55, 0x70, 0x64,
0x61, 0x74, 0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71,
0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xfa, 0x07, 0x2a, 0x3d, 0x0a, 0x14, 0x54,
0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x50, 0x72, 0x65, 0x63, 0x65, 0x64, 0x65,
0x6e, 0x63, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x44, 0x65, 0x66, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00,
0x12, 0x0c, 0x0a, 0x08, 0x52, 0x65, 0x71, 0x75, 0x69, 0x72, 0x65, 0x64, 0x10, 0x01, 0x12, 0x0a,
0x0a, 0x06, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x10, 0x02, 0x2a, 0x52, 0x0a, 0x0c, 0x46, 0x61,
0x75, 0x6c, 0x74, 0x53, 0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x10, 0x0a, 0x0c, 0x49, 0x6e,
0x67, 0x72, 0x65, 0x73, 0x73, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b,
0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x01, 0x12, 0x0d, 0x0a,
0x09, 0x4c, 0x69, 0x6e, 0x6b, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x02, 0x12, 0x10, 0x0a, 0x0c,
0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x03, 0x62, 0x06,
0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x10, 0xe8, 0x07, 0x12, 0x0d, 0x0a, 0x08, 0x44, 0x69, 0x61, 0x6c, 0x54, 0x79, 0x70, 0x65, 0x10,
0xea, 0x07, 0x12, 0x0d, 0x0a, 0x08, 0x4c, 0x69, 0x6e, 0x6b, 0x54, 0x79, 0x70, 0x65, 0x10, 0xeb,
0x07, 0x12, 0x0e, 0x0a, 0x09, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xec,
0x07, 0x12, 0x0e, 0x0a, 0x09, 0x52, 0x6f, 0x75, 0x74, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xed,
0x07, 0x12, 0x10, 0x0a, 0x0b, 0x55, 0x6e, 0x72, 0x6f, 0x75, 0x74, 0x65, 0x54, 0x79, 0x70, 0x65,
0x10, 0xee, 0x07, 0x12, 0x10, 0x0a, 0x0b, 0x4d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x54, 0x79,
0x70, 0x65, 0x10, 0xef, 0x07, 0x12, 0x20, 0x0a, 0x1b, 0x54, 0x6f, 0x67, 0x67, 0x6c, 0x65, 0x50,
0x69, 0x70, 0x65, 0x54, 0x72, 0x61, 0x63, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
0x54, 0x79, 0x70, 0x65, 0x10, 0xf0, 0x07, 0x12, 0x13, 0x0a, 0x0e, 0x54, 0x72, 0x61, 0x63, 0x65,
0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf2, 0x07, 0x12, 0x20, 0x0a, 0x1b,
0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf3, 0x07, 0x12, 0x20,
0x0a, 0x1b, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74,
0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf4, 0x07,
0x12, 0x17, 0x0a, 0x12, 0x49, 0x6e, 0x73, 0x70, 0x65, 0x63, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf5, 0x07, 0x12, 0x18, 0x0a, 0x13, 0x49, 0x6e, 0x73,
0x70, 0x65, 0x63, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65,
0x10, 0xf6, 0x07, 0x12, 0x23, 0x0a, 0x1e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x54,
0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf9, 0x07, 0x12, 0x20, 0x0a, 0x1b, 0x55, 0x70, 0x64, 0x61,
0x74, 0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xfa, 0x07, 0x2a, 0x3d, 0x0a, 0x14, 0x54, 0x65,
0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x50, 0x72, 0x65, 0x63, 0x65, 0x64, 0x65, 0x6e,
0x63, 0x65, 0x12, 0x0b, 0x0a, 0x07, 0x44, 0x65, 0x66, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00, 0x12,
0x0c, 0x0a, 0x08, 0x52, 0x65, 0x71, 0x75, 0x69, 0x72, 0x65, 0x64, 0x10, 0x01, 0x12, 0x0a, 0x0a,
0x06, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x10, 0x02, 0x2a, 0x52, 0x0a, 0x0c, 0x46, 0x61, 0x75,
0x6c, 0x74, 0x53, 0x75, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x12, 0x10, 0x0a, 0x0c, 0x49, 0x6e, 0x67,
0x72, 0x65, 0x73, 0x73, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x45,
0x67, 0x72, 0x65, 0x73, 0x73, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x01, 0x12, 0x0d, 0x0a, 0x09,
0x4c, 0x69, 0x6e, 0x6b, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x02, 0x12, 0x10, 0x0a, 0x0c, 0x46,
0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x46, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x03, 0x62, 0x06, 0x70,
0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
pb/ctrl_pb/ctrl.proto
+77 -75
View File
@@ -3,128 +3,130 @@ syntax = "proto3";
package ctrl.pb;
enum ContentType {
Zero = 0;
Zero = 0;
SessionRequestType = 1000;
SessionSuccessType = 1001;
DialType = 1002;
LinkType = 1003;
FaultType = 1004;
RouteType = 1005;
UnrouteType = 1006;
MetricsType = 1007;
TogglePipeTracesRequestType = 1008;
TraceEventType = 1010;
CreateTerminatorRequestType = 1011;
RemoveTerminatorRequestType = 1012;
InspectRequestType = 1013;
InspectResponseType = 1014;
SessionFailedType = 1016;
ValidateTerminatorsRequestType = 1017;
UpdateTerminatorRequestType = 1018;
SessionRequestType = 1000;
// defined in ctrl_msg/messages.go now
// SessionSuccessType = 1001;
DialType = 1002;
LinkType = 1003;
FaultType = 1004;
RouteType = 1005;
UnrouteType = 1006;
MetricsType = 1007;
TogglePipeTracesRequestType = 1008;
TraceEventType = 1010;
CreateTerminatorRequestType = 1011;
RemoveTerminatorRequestType = 1012;
InspectRequestType = 1013;
InspectResponseType = 1014;
// defined in ctrl_msg/messages.go now
// SessionFailedType = 1016;
ValidateTerminatorsRequestType = 1017;
UpdateTerminatorRequestType = 1018;
}
message SessionRequest {
string ingressId = 1;
string serviceId = 2;
map<uint32, bytes> peerData = 3;
string ingressId = 1;
string serviceId = 2;
map<uint32, bytes> peerData = 3;
}
enum TerminatorPrecedence {
Default = 0;
Required = 1;
Failed = 2;
Default = 0;
Required = 1;
Failed = 2;
}
message CreateTerminatorRequest {
string serviceId = 2;
string binding = 3;
string address = 4;
map<uint32, bytes> peerData = 5;
uint32 cost = 6;
TerminatorPrecedence precedence = 7;
string identity = 8;
bytes identitySecret = 9;
string serviceId = 2;
string binding = 3;
string address = 4;
map<uint32, bytes> peerData = 5;
uint32 cost = 6;
TerminatorPrecedence precedence = 7;
string identity = 8;
bytes identitySecret = 9;
}
message RemoveTerminatorRequest {
string terminatorId = 1;
string terminatorId = 1;
}
message Terminator {
string id = 1;
string binding = 2;
string address = 3;
string id = 1;
string binding = 2;
string address = 3;
}
message ValidateTerminatorsRequest {
repeated Terminator terminators = 1;
repeated Terminator terminators = 1;
}
message UpdateTerminatorRequest {
string terminatorId = 1;
bool updatePrecedence = 2;
bool updateCost = 3;
TerminatorPrecedence precedence = 4;
uint32 cost = 5;
string terminatorId = 1;
bool updatePrecedence = 2;
bool updateCost = 3;
TerminatorPrecedence precedence = 4;
uint32 cost = 5;
}
message Dial {
string id = 1;
string address = 2;
string id = 1;
string address = 2;
}
message Link {
string id = 1;
string id = 1;
}
enum FaultSubject {
IngressFault = 0;
EgressFault = 1;
LinkFault = 2;
ForwardFault = 3;
IngressFault = 0;
EgressFault = 1;
LinkFault = 2;
ForwardFault = 3;
}
message Fault {
FaultSubject subject = 1;
string id = 2;
FaultSubject subject = 1;
string id = 2;
}
message Route {
string sessionId = 1;
uint32 attempt = 2;
string sessionId = 1;
uint32 attempt = 2;
message Egress {
string binding = 1;
string address = 2;
string destination = 3;
map<uint32, bytes> peerData = 4;
}
Egress egress = 3;
message Egress {
string binding = 1;
string address = 2;
string destination = 3;
map<uint32, bytes> peerData = 4;
}
Egress egress = 3;
message Forward {
string srcAddress = 1;
string dstAddress = 2;
}
repeated Forward forwards = 4;
message Forward {
string srcAddress = 1;
string dstAddress = 2;
}
repeated Forward forwards = 4;
}
message Unroute {
string sessionId = 1;
bool now = 2;
string sessionId = 1;
bool now = 2;
}
message InspectRequest {
repeated string requestedValues = 1;
repeated string requestedValues = 1;
}
message InspectResponse {
bool success = 1;
repeated string errors = 2;
repeated InspectValue values = 3;
bool success = 1;
repeated string errors = 2;
repeated InspectValue values = 3;
message InspectValue {
string name = 1;
string value = 2;
}
message InspectValue {
string name = 1;
string value = 2;
}
}
pb/ctrl_pb/decoder.go
+18 -2
View File
@@ -113,7 +113,7 @@ func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) {
return nil, true
}
case int32(ContentType_SessionSuccessType):
case int32(ctrl_msg.SessionSuccessType):
meta := channel2.NewTraceMessageDecode(DECODER, "Session Success Response")
meta["sessionId"] = string(msg.Body)
meta["address"] = string(msg.Headers[ctrl_msg.SessionSuccessAddressHeader])
@@ -125,7 +125,7 @@ func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) {
return data, true
case int32(ContentType_SessionFailedType):
case int32(ctrl_msg.SessionFailedType):
meta := channel2.NewTraceMessageDecode(DECODER, "Session Failed Response")
message := string(msg.Body)
if message != "" {
@@ -278,6 +278,22 @@ func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) {
pfxlog.Logger().Errorf("unexpected error (%s)", err)
}
case int32(ctrl_msg.RouteResultType):
meta := channel2.NewTraceMessageDecode(DECODER, "Route Result")
meta["sessionId"] = 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
router/forwarder/forwarder.go
+20 -12
View File
@@ -36,6 +36,7 @@ type Forwarder struct {
metricsRegistry metrics.UsageRegistry
traceController trace.Controller
Options *Options
CloseNotify <-chan struct{}
}
type Destination interface {
@@ -52,14 +53,15 @@ type XgressDestination interface {
GetTimeOfLastRxFromLink() int64
}
func NewForwarder(metricsRegistry metrics.UsageRegistry, faulter *Faulter, options *Options) *Forwarder {
func NewForwarder(metricsRegistry metrics.UsageRegistry, faulter *Faulter, options *Options, closeNotify <-chan struct{}) *Forwarder {
forwarder := &Forwarder{
sessions: newSessionTable(),
destinations: newDestinationTable(),
faulter: faulter,
metricsRegistry: metricsRegistry,
traceController: trace.NewController(),
traceController: trace.NewController(closeNotify),
Options: options,
CloseNotify: closeNotify,
}
return forwarder
@@ -187,8 +189,8 @@ func (forwarder *Forwarder) ForwardAcknowledgement(srcAddr xgress.Address, ackno
func (forwarder *Forwarder) ReportForwardingFault(sessionId string) {
if forwarder.faulter != nil {
forwarder.faulter.report(sessionId)
} else {
logrus.Errorf("nil faulter, cannot accept forwarding fault report")
} else {
logrus.Errorf("nil faulter, cannot accept forwarding fault report")
}
}
@@ -205,19 +207,25 @@ func (forwarder *Forwarder) unrouteTimeout(sessionId string, interval time.Durat
log.Debug("scheduled")
defer log.Debug("timeout")
for {
time.Sleep(interval)
ticker := time.NewTicker(interval)
defer ticker.Stop()
if dest := forwarder.getXgressForSession(sessionId); dest != nil {
elapsedDelta := info.NowInMilliseconds() - dest.GetTimeOfLastRxFromLink()
if (time.Duration(elapsedDelta) * time.Millisecond) >= interval {
for {
select {
case <-ticker.C:
if dest := forwarder.getXgressForSession(sessionId); dest != nil {
elapsedDelta := info.NowInMilliseconds() - dest.GetTimeOfLastRxFromLink()
if (time.Duration(elapsedDelta) * time.Millisecond) >= interval {
forwarder.sessions.removeForwardTable(sessionId)
forwarder.EndSession(sessionId)
return
}
} else {
forwarder.sessions.removeForwardTable(sessionId)
forwarder.EndSession(sessionId)
return
}
} else {
forwarder.sessions.removeForwardTable(sessionId)
forwarder.EndSession(sessionId)
case <-forwarder.CloseNotify:
return
}
}
router/handler_ctrl/bind.go
+6 -3
View File
@@ -34,6 +34,7 @@ type bindHandler struct {
ctrl xgress.CtrlChannel
forwarder *forwarder.Forwarder
xctrls []xctrl.Xctrl
closeNotify chan struct{}
}
func NewBindHandler(id *identity.TokenId,
@@ -41,7 +42,8 @@ func NewBindHandler(id *identity.TokenId,
xlinkDialers []xlink.Dialer,
ctrl xgress.CtrlChannel,
forwarder *forwarder.Forwarder,
xctrls []xctrl.Xctrl) channel2.BindHandler {
xctrls []xctrl.Xctrl,
closeNotify chan struct{}) channel2.BindHandler {
return &bindHandler{
id: id,
dialerCfg: dialerCfg,
@@ -49,12 +51,13 @@ func NewBindHandler(id *identity.TokenId,
ctrl: ctrl,
forwarder: forwarder,
xctrls: xctrls,
closeNotify: closeNotify,
}
}
func (self *bindHandler) BindChannel(ch channel2.Channel) error {
ch.AddReceiveHandler(newDialHandler(self.id, self.ctrl, self.xlinkDialers, self.forwarder))
ch.AddReceiveHandler(newRouteHandler(self.id, self.ctrl, self.dialerCfg, self.forwarder))
ch.AddReceiveHandler(newDialHandler(self.id, self.ctrl, self.xlinkDialers, self.forwarder, self.closeNotify))
ch.AddReceiveHandler(newRouteHandler(self.id, self.ctrl, self.dialerCfg, self.forwarder, self.closeNotify))
ch.AddReceiveHandler(newValidateTerminatorsHandler(self.ctrl, self.dialerCfg))
ch.AddReceiveHandler(newUnrouteHandler(self.forwarder))
ch.AddReceiveHandler(newTraceHandler(self.id, self.forwarder.TraceController()))
router/handler_ctrl/dial.go
+6 -3
View File
@@ -35,12 +35,15 @@ type dialHandler struct {
pool handlerPool
}
func newDialHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialers []xlink.Dialer, forwarder *forwarder.Forwarder) *dialHandler {
func newDialHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialers []xlink.Dialer, forwarder *forwarder.Forwarder, closeNotify chan struct{}) *dialHandler {
handler := &dialHandler{
id: id,
ctrl: ctrl,
dialers: dialers,
pool: handlerPool{options: forwarder.Options.LinkDial},
pool: handlerPool{
options: forwarder.Options.LinkDial,
closeNotify: closeNotify,
},
}
handler.pool.Start()
@@ -52,7 +55,7 @@ func (self *dialHandler) ContentType() int32 {
}
func (self *dialHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
self.pool.Queue(func(){
self.pool.Queue(func() {
self.handle(msg, ch)
})
}
router/handler_ctrl/pool.go
+32 -19
View File
@@ -23,33 +23,46 @@ import (
)
type handlerPool struct {
options forwarder.WorkerPoolOptions
startOnce sync.Once
queue chan func()
options forwarder.WorkerPoolOptions
startOnce sync.Once
queue chan func()
closeNotify chan struct{}
}
func (pool *handlerPool) Start() {
pool.startOnce.Do(func() {
pool.queue = make(chan func(), pool.options.QueueLength)
for i := uint16(0); i < pool.options.WorkerCount; i++ {
go func() {
for handler := range pool.queue {
if handler != nil {
func() {
defer func() {
if err := recover(); err != nil {
pfxlog.Logger().Errorf("worker error: %v", err)
}
}()
handler()
}()
}
}
}()
go pool.worker()
}
})
}
func (pool *handlerPool) Queue(handler func()) {
pool.queue <- handler
func (pool *handlerPool) worker() {
for {
select {
case work := <-pool.queue:
if work != nil {
pool.doWork(work)
}
case <-pool.closeNotify:
return
}
}
}
func (pool *handlerPool) doWork(work func()) {
defer func() {
if err := recover(); err != nil {
pfxlog.Logger().Errorf("worker error: %v", err)
}
}()
work()
}
func (pool *handlerPool) Queue(handler func()) {
select {
case pool.queue <- handler:
case <-pool.closeNotify:
}
}
router/handler_ctrl/route.go
+17 -20
View File
@@ -39,13 +39,16 @@ type routeHandler struct {
pool handlerPool
}
func newRouteHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialerCfg map[string]xgress.OptionsData, forwarder *forwarder.Forwarder) *routeHandler {
func newRouteHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialerCfg map[string]xgress.OptionsData, forwarder *forwarder.Forwarder, closeNotify chan struct{}) *routeHandler {
handler := &routeHandler{
id: id,
ctrl: ctrl,
dialerCfg: dialerCfg,
forwarder: forwarder,
pool: handlerPool{options: forwarder.Options.XgressDial},
pool: handlerPool{
options: forwarder.Options.XgressDial,
closeNotify: closeNotify,
},
}
handler.pool.Start()
@@ -85,20 +88,17 @@ func (rh *routeHandler) success(msg *channel2.Message, attempt int, ch channel2.
rh.forwarder.Route(route)
log := pfxlog.ContextLogger(ch.Label())
if response, err := ctrl_msg.NewRouteResultSuccessMsg(route.SessionId, attempt); err == nil {
for k, v := range peerData {
response.Headers[int32(k)] = v
}
response := ctrl_msg.NewRouteResultSuccessMsg(route.SessionId, attempt)
for k, v := range peerData {
response.Headers[int32(k)] = v
}
response.ReplyTo(msg)
response.ReplyTo(msg)
if err := rh.ctrl.Channel().Send(response); err == nil {
log.Debugf("handled route for [s/%s]", route.SessionId)
} else {
log.Errorf("send response failed for [s/%s] (%s)", route.SessionId, err)
}
if err := rh.ctrl.Channel().Send(response); err == nil {
log.Debugf("handled route for [s/%s]", route.SessionId)
} else {
log.Errorf("error creating success response for [s/%s] (%v)", route.SessionId, err)
log.Errorf("send response failed for [s/%s] (%s)", route.SessionId, err)
}
}
@@ -106,13 +106,10 @@ func (rh *routeHandler) fail(msg *channel2.Message, attempt int, ch channel2.Cha
log := pfxlog.ContextLogger(ch.Label())
log.WithError(err).Errorf("failed to connect egress for [s/%s]", route.SessionId)
if response, err := ctrl_msg.NewRouteResultFailedMessage(route.SessionId, attempt, err.Error()); err == nil {
response.ReplyTo(msg)
if err := rh.ctrl.Channel().Send(response); err != nil {
log.Errorf("send failure response failed for [s/%s] (%s)", route.SessionId, err)
}
} else {
log.Errorf("error creating failure response for [s/%s] (%v)", route.SessionId, err)
response := ctrl_msg.NewRouteResultFailedMessage(route.SessionId, attempt, err.Error())
response.ReplyTo(msg)
if err := rh.ctrl.Channel().Send(response); err != nil {
log.Errorf("send failure response failed for [s/%s] (%s)", route.SessionId, err)
}
}
router/monitor_darwin.go
+10 -5
View File
@@ -25,18 +25,23 @@ import (
)
type routerMonitor struct {
forwarder *forwarder.Forwarder
forwarder *forwarder.Forwarder
closeNotify <-chan struct{}
}
func newRouterMonitor(forwarder *forwarder.Forwarder) *routerMonitor {
return &routerMonitor{forwarder: forwarder}
func newRouterMonitor(forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *routerMonitor {
return &routerMonitor{forwarder: forwarder, closeNotify: closeNotify}
}
func (routerMonitor *routerMonitor) Monitor() {
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, syscall.SIGUSR1)
for {
<-signalChan
pfxlog.Logger().Info("\n" + routerMonitor.forwarder.Debug())
select {
case <-signalChan:
pfxlog.Logger().Info("\n" + routerMonitor.forwarder.Debug())
case <-routerMonitor.closeNotify:
return
}
}
}
router/monitor_linux.go
+10 -5
View File
@@ -25,18 +25,23 @@ import (
)
type routerMonitor struct {
forwarder *forwarder.Forwarder
forwarder *forwarder.Forwarder
closeNotify <-chan struct{}
}
func newRouterMonitor(forwarder *forwarder.Forwarder) *routerMonitor {
return &routerMonitor{forwarder: forwarder}
func newRouterMonitor(forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *routerMonitor {
return &routerMonitor{forwarder: forwarder, closeNotify: closeNotify}
}
func (routerMonitor *routerMonitor) Monitor() {
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, syscall.SIGUSR1)
for {
<-signalChan
pfxlog.Logger().Info("\n" + routerMonitor.forwarder.Debug())
select {
case <-signalChan:
pfxlog.Logger().Info("\n" + routerMonitor.forwarder.Debug())
case <-routerMonitor.closeNotify:
return
}
}
}
router/monitor_windows.go
+1 -5
View File
@@ -18,17 +18,13 @@ package router
import (
"github.com/openziti/fabric/router/forwarder"
"time"
)
type routerMonitor struct{}
func newRouterMonitor(forwarder *forwarder.Forwarder) *routerMonitor {
func newRouterMonitor(forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *routerMonitor {
return &routerMonitor{}
}
func (routerMonitor *routerMonitor) Monitor() {
for {
time.Sleep(24 * time.Hour)
}
}
router/router.go
+7 -6
View File
@@ -82,18 +82,19 @@ func (self *Router) DefaultRequestTimeout() time.Duration {
}
func Create(config *Config, versionProvider common.VersionProvider) *Router {
eventDispatcher := event.NewDispatcher()
closeNotify := make(chan struct{})
eventDispatcher := event.NewDispatcher(closeNotify)
metricsConfig := &metrics.Config{
Source: config.Id.Token,
ReportInterval: time.Second * 15,
EventSink: metrics.NewDispatchWrapper(eventDispatcher.Dispatch),
}
metricsRegistry := metrics.NewUsageRegistryFromConfig(metricsConfig)
metricsRegistry := metrics.NewUsageRegistryFromConfig(metricsConfig, closeNotify)
xgress.InitMetrics(metricsRegistry)
closeNotify := make(chan struct{})
faulter := forwarder.NewFaulter(config.Forwarder.FaultTxInterval, closeNotify)
fwd := forwarder.NewForwarder(metricsRegistry, faulter, config.Forwarder)
fwd := forwarder.NewForwarder(metricsRegistry, faulter, config.Forwarder, closeNotify)
xgress.InitPayloadIngester(closeNotify)
xgress.InitAcker(fwd, metricsRegistry, closeNotify)
@@ -144,7 +145,6 @@ func (self *Router) Start() error {
func (self *Router) Shutdown() error {
var errors []error
if self.isShutdown.CompareAndSwap(false, true) {
self.eventDispatcher.Stop()
if self.metricsReporter != nil {
events.RemoveMetricsEventHandler(self.metricsReporter)
}
@@ -204,7 +204,7 @@ func (self *Router) startProfiling() {
logrus.Errorf("unexpected error launching cpu profiling (%v)", err)
}
}
go newRouterMonitor(self.forwarder).Monitor()
go newRouterMonitor(self.forwarder, self.shutdownC).Monitor()
}
func (self *Router) registerComponents() error {
@@ -313,6 +313,7 @@ func (self *Router) startControlPlane() error {
self,
self.forwarder,
self.xctrls,
self.shutdownC,
)
self.config.Ctrl.Options.BindHandlers = append(self.config.Ctrl.Options.BindHandlers, bindHandler)
router/xgress/ordering_test.go
+1 -1
View File
@@ -56,7 +56,7 @@ func (n noopReceiveHandler) HandleXgressReceive(payload *Payload, x *Xgress) {
}
func Test_Ordering(t *testing.T) {
metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, time.Minute, noopMetricsHandler{})
metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, time.Minute, noopMetricsHandler{}, nil)
closeNotify := make(chan struct{})
InitPayloadIngester(closeNotify)
InitMetrics(metricsRegistry)
trace/controller.go
+10 -16
View File
@@ -23,8 +23,6 @@ import (
"strings"
)
type controllerShutdown struct{}
type SourceType int
const (
@@ -150,13 +148,13 @@ type Controller interface {
DisableTracing(sourceType SourceType, matcher SourceMatcher, resultChan chan<- ToggleApplyResult)
AddSource(source Source)
RemoveSource(source Source)
Shutdown()
}
func NewController() Controller {
func NewController(closeNotify <-chan struct{}) Controller {
controller := &controllerImpl{
events: make(chan interface{}, 1),
sources: make(map[Source]Source),
events: make(chan interface{}, 1),
sources: make(map[Source]Source),
closeNotify: closeNotify,
}
go controller.run()
return controller
@@ -183,8 +181,9 @@ type sourceRemovedEvent struct {
}
type controllerImpl struct {
events chan interface{}
sources map[Source]Source
events chan interface{}
sources map[Source]Source
closeNotify <-chan struct{}
}
func (controller *controllerImpl) EnableTracing(sourceType SourceType, matcher SourceMatcher, resultChan chan<- ToggleApplyResult) {
@@ -203,14 +202,11 @@ func (controller *controllerImpl) RemoveSource(source Source) {
controller.events <- &sourceRemovedEvent{source}
}
func (controller *controllerImpl) Shutdown() {
controller.events <- &controllerShutdown{}
}
func (controller *controllerImpl) run() {
running := true
for running {
for {
select {
case <-controller.closeNotify:
return
case i := <-controller.events:
switch event := i.(type) {
case *enableSourcesEvent:
@@ -227,8 +223,6 @@ func (controller *controllerImpl) run() {
controller.sources[event.source] = event.source
case *sourceRemovedEvent:
delete(controller.sources, event.source)
case *controllerShutdown:
running = false
}
}
}