diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index cd714689d..e13380fd6 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -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 diff --git a/.github/workflows/update-dependency.yml b/.github/workflows/update-dependency.yml index 152407de0..2ccdfc026 100644 --- a/.github/workflows/update-dependency.yml +++ b/.github/workflows/update-dependency.yml @@ -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 diff --git a/controller/controller.go b/controller/controller.go index 0b1087a71..7c475e60c 100644 --- a/controller/controller.go +++ b/controller/controller.go @@ -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 } diff --git a/controller/network/circuit_test.go b/controller/network/circuit_test.go index 06ab68b8e..115610b5b 100644 --- a/controller/network/circuit_test.go +++ b/controller/network/circuit_test.go @@ -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" diff --git a/controller/network/network.go b/controller/network/network.go index a06c9d531..99e0b8989 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -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() diff --git a/controller/network/path_test.go b/controller/network/path_test.go index 09165e48f..5811b8375 100644 --- a/controller/network/path_test.go +++ b/controller/network/path_test.go @@ -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" diff --git a/controller/xctrl/xctrl.go b/controller/xctrl/xctrl.go index 6541b3fc6..0a7396857 100644 --- a/controller/xctrl/xctrl.go +++ b/controller/xctrl/xctrl.go @@ -28,4 +28,5 @@ type Xctrl interface { Enabled() bool Run(ctrl channel2.Channel, db boltz.Db, done chan struct{}) error NotifyOfReconnect() + GetTraceDecoders() []channel2.TraceMessageDecoder } diff --git a/controller/xctrl_example/example.go b/controller/xctrl_example/example.go index 10c873703..f00304f5d 100644 --- a/controller/xctrl_example/example.go +++ b/controller/xctrl_example/example.go @@ -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 diff --git a/ctrl_msg/messages.go b/ctrl_msg/messages.go index f483703a2..668d614ec 100644 --- a/ctrl_msg/messages.go +++ b/ctrl_msg/messages.go @@ -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 } diff --git a/pb/ctrl_pb/ctrl.pb.go b/pb/ctrl_pb/ctrl.pb.go index ecfd823b3..3947e26a9 100644 --- a/pb/ctrl_pb/ctrl.pb.go +++ b/pb/ctrl_pb/ctrl.pb.go @@ -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 ( diff --git a/pb/ctrl_pb/ctrl.proto b/pb/ctrl_pb/ctrl.proto index 42fae65ee..55337c682 100644 --- a/pb/ctrl_pb/ctrl.proto +++ b/pb/ctrl_pb/ctrl.proto @@ -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 peerData = 3; + string ingressId = 1; + string serviceId = 2; + map 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 peerData = 5; - uint32 cost = 6; - TerminatorPrecedence precedence = 7; - string identity = 8; - bytes identitySecret = 9; + string serviceId = 2; + string binding = 3; + string address = 4; + map 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 peerData = 4; - } - Egress egress = 3; + message Egress { + string binding = 1; + string address = 2; + string destination = 3; + map 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; + } } \ No newline at end of file diff --git a/pb/ctrl_pb/decoder.go b/pb/ctrl_pb/decoder.go index bc77d0b57..14dfbaf9b 100644 --- a/pb/ctrl_pb/decoder.go +++ b/pb/ctrl_pb/decoder.go @@ -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 diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index cda3c125c..c9e65554d 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -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 } } diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index ce07f78f9..473e196f1 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -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())) diff --git a/router/handler_ctrl/dial.go b/router/handler_ctrl/dial.go index 89f45272b..972e659e9 100644 --- a/router/handler_ctrl/dial.go +++ b/router/handler_ctrl/dial.go @@ -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) }) } diff --git a/router/handler_ctrl/pool.go b/router/handler_ctrl/pool.go index f9f05206b..0a230b1fe 100644 --- a/router/handler_ctrl/pool.go +++ b/router/handler_ctrl/pool.go @@ -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: + } } diff --git a/router/handler_ctrl/route.go b/router/handler_ctrl/route.go index c5cec5808..1c75a96a9 100644 --- a/router/handler_ctrl/route.go +++ b/router/handler_ctrl/route.go @@ -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) } } diff --git a/router/monitor_darwin.go b/router/monitor_darwin.go index 31528b8c5..eb30ca374 100644 --- a/router/monitor_darwin.go +++ b/router/monitor_darwin.go @@ -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 + } } } diff --git a/router/monitor_linux.go b/router/monitor_linux.go index 31528b8c5..eb30ca374 100644 --- a/router/monitor_linux.go +++ b/router/monitor_linux.go @@ -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 + } } } diff --git a/router/monitor_windows.go b/router/monitor_windows.go index 46707fdde..23a562c65 100644 --- a/router/monitor_windows.go +++ b/router/monitor_windows.go @@ -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) - } } diff --git a/router/router.go b/router/router.go index fdb68a306..a80328377 100644 --- a/router/router.go +++ b/router/router.go @@ -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) diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index 781e3a489..1f02df7ac 100644 --- a/router/xgress/ordering_test.go +++ b/router/xgress/ordering_test.go @@ -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) diff --git a/trace/controller.go b/trace/controller.go index 831e65a00..c32d34978 100644 --- a/trace/controller.go +++ b/trace/controller.go @@ -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 } } }