From 088ee9f8e25bcab556ab6b6881ec8649e2ec0a87 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 27 Jul 2021 00:00:29 -0400 Subject: [PATCH] Update for session -> circuit changes --- CHANGELOG.md | 34 +++++++ etc/ctrl.with.edge.yml | 2 +- etc/ctrl.yml | 26 +++-- etc/edge.router.yml | 2 - go.mod | 6 +- go.sum | 14 ++- ziti-fabric-gw/subcmd/list_sessions.go | 26 ++--- ziti-fabric-gw/subcmd/messages.go | 12 +-- ziti-fabric-gw/subcmd/run.go | 6 +- .../{list_sessions.go => list_circuits.go} | 44 ++++----- .../{remove_session.go => remove_circuit.go} | 35 +++---- ziti-fabric/subcmd/stream_circuits.go | 95 +++++++++++++++++++ ziti-fabric/subcmd/stream_sessions.go | 95 ------------------- ziti/cmd/ziti/cmd/ps_router_add_route.go | 2 +- 14 files changed, 218 insertions(+), 181 deletions(-) rename ziti-fabric/subcmd/{list_sessions.go => list_circuits.go} (61%) rename ziti-fabric/subcmd/{remove_session.go => remove_circuit.go} (70%) create mode 100644 ziti-fabric/subcmd/stream_circuits.go delete mode 100644 ziti-fabric/subcmd/stream_sessions.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e8565387..b6d897df3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,37 @@ +# Release 0.22.0 + +## Breaking Changes +Fabric sessions renamed to circuits. See below for details. + +# What's New + +* Refactor: Fabric Sessions renamed to Circuits +* Feature: Links will now wait for a timeout for retrying +* Bug fix: Sessions created on the controller when circuit creation fails are now cleaned up + +## Fabric Sessions renamed to Circuits +Previously we had three separate entities named session: fabric sessions, edge sessions and edge API sessions. In order to reduce confusion, fabric sessions +have been renamed to circuits. This has the following impacts: + +* ziti-fabric CLI + * `list sessions` renamed to `list circuits` + * `remove session` renamed to `remove circuit` + * `stream sessions` renamed to `stream circuits` +* Config properties + * In the controller config, under `networks`, `createSessionRetries` is now `createCircuitRetries` + * In the router config, under xgress dialer/listener options, `getSessionTimeout` is now `getCircuitTimeout` + * In the router config, under xgress dialer/listener options, `sessionStartTimeout` is now `circuitStartTimeout` + * In the router, under `forwarder`, `idleSessionTimeout` is now `idleCircuitTimeout` + +In the context of the fabric there was an existing construct call `Circuit` which has now been renamed to `Path`. This may be visible in a few `ziti-fabric` CLI outputs + +# Pending Link Timeout +Previously whenever a router connected we'd look for new links possiblities and create new links between routers where any were missing. +If lots of routers connected at the same time, we might create duplicate links because the links hadn't been reported as established yet. +Now we'll checking for links in Pending state, and if they haven't hit a configurable timeout, we won't create another link. + +The new config property is `pendingLinkTimeoutSeconds` in the controller config file under `network`, and defaults to 10 seconds. + # Release 0.21.0 ## Semantic now Required for policies (BREAKING CHANGE) diff --git a/etc/ctrl.with.edge.yml b/etc/ctrl.with.edge.yml index b78e4dd3f..97acb4205 100644 --- a/etc/ctrl.with.edge.yml +++ b/etc/ctrl.with.edge.yml @@ -60,7 +60,7 @@ mgmt: # - type: metrics # sourceFilter: .* # metricFilter: .*egress.*m1_rate* -# - type: fabric.sessions +# - type: fabric.circuits # include: # - created # - type: edge.sessions diff --git a/etc/ctrl.yml b/etc/ctrl.yml index ea46153d0..092f193ae 100644 --- a/etc/ctrl.yml +++ b/etc/ctrl.yml @@ -29,28 +29,34 @@ identity: # #routeTimeoutSeconds: 10 # - # createSessionRetries controls the number of retries that will be attempted to create a circuit (and terminate it) - # for new sessions. + # createCircuitRetries controls the number of retries that will be attempted to create a path (and terminate it) + # for new circuits. # - #createSessionRetries: 3 + #createCircuitRetries: 3 + # + # pendingLinkTimeoutSeconds controls how long we'll wait before creating a new link between routers where + # there isn't an established link, but a link request has been sent # - # Defines the period that the controller re-evaluates the performance of all of the sessions + #pendingLinkTimeoutSeconds: 10 + # + # + # Defines the period that the controller re-evaluates the performance of all of the circuits # running on the network. # cycleSeconds: 15 # #smart: # - # Defines the fractional upper limit of underperforming sessions that are candidates to be re-routed. If - # smart routing detects 100 sessions that are underperforming, and `smart.rerouteFraction` is set to `0.02`, - # then the upper limit of sessions that will be re-routed in this `cycleSeconds` period will be limited to + # Defines the fractional upper limit of underperforming circuits that are candidates to be re-routed. If + # smart routing detects 100 circuits that are underperforming, and `smart.rerouteFraction` is set to `0.02`, + # then the upper limit of circuits that will be re-routed in this `cycleSeconds` period will be limited to # 2 (2% of 100). # #rerouteFraction: 0.02 # - # Defines the hard upper limit of underperforming sessions that are candidates to be re-routed. If smart - # routing detects 100 sessions that are underperforming, and `smart.rerouteCap` is set to `1`, and - # `smart.rerouteFraction` is set to `0.02`, then the upper limit of sessions that will be re-routed in this + # Defines the hard upper limit of underperforming circuits that are candidates to be re-routed. If smart + # routing detects 100 circuits that are underperforming, and `smart.rerouteCap` is set to `1`, and + # `smart.rerouteFraction` is set to `0.02`, then the upper limit of circuits that will be re-routed in this # `cycleSeconds` period will be limited to 1. # #rerouteCap: 4 diff --git a/etc/edge.router.yml b/etc/edge.router.yml index 6f559dab4..c358e637a 100644 --- a/etc/edge.router.yml +++ b/etc/edge.router.yml @@ -106,8 +106,6 @@ listeners: #connectTimeoutMs: 3000 # how long to wait before timing out when looking up api-sessions after client connect. Default 5 seconds. #lookupApiSessionTimeout: 5s - # how long to wait before timing out when looking up sessions after client dial/bind. Default 5 seconds. - #lookupSessionTimeout: 5s - binding: transport address: tls:0.0.0.0:7099 diff --git a/go.mod b/go.mod index f3222ebd6..de47bf6d0 100644 --- a/go.mod +++ b/go.mod @@ -29,10 +29,10 @@ require ( github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d github.com/keybase/go-ps v0.0.0-20190827175125-91aafc93ba19 github.com/michaelquigley/pfxlog v0.6.0 - github.com/openziti/edge v0.20.2 - github.com/openziti/fabric v0.16.82 + github.com/openziti/edge v0.20.8 + github.com/openziti/fabric v0.16.84 github.com/openziti/foundation v0.15.65 - github.com/openziti/sdk-golang v0.15.72 + github.com/openziti/sdk-golang v0.15.73 github.com/pborman/uuid v1.2.0 github.com/pkg/errors v0.9.1 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 diff --git a/go.sum b/go.sum index d5f89357a..b500c0d2f 100644 --- a/go.sum +++ b/go.sum @@ -677,16 +677,14 @@ github.com/openziti/bbolt v1.3.6-0.20210317142109-547da822475e h1:ST9+54UtCr96S5 github.com/openziti/bbolt v1.3.6-0.20210317142109-547da822475e/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ= github.com/openziti/dilithium v0.3.3 h1:PLgQ6PMNLSTzCFbX/h98cmudgz/cU6TmjdSv5NAPD8k= github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo= -github.com/openziti/edge v0.20.2 h1:ZjkLOV12uzj+SE/0chzeRAIJtDbjqn0HeD4X5EJCly8= -github.com/openziti/edge v0.20.2/go.mod h1:GSg6dhOBVfToZcn8RJbGyVdi9UJaxEiza5bLqVFxkjo= -github.com/openziti/fabric v0.16.81/go.mod h1:2Y6bOG67R9n9XTvnIFz8A+oc+9pX+h6pDmbDZKot1Hk= -github.com/openziti/fabric v0.16.82 h1:G0w+kcBICCaPCS0VcDKpoZpqB4i8xHdqAQ7/nCdf/6M= -github.com/openziti/fabric v0.16.82/go.mod h1:2Y6bOG67R9n9XTvnIFz8A+oc+9pX+h6pDmbDZKot1Hk= +github.com/openziti/edge v0.20.8 h1:5MiqedRzx2AH+YDazV85v0q+78Vc9eqB44pfXxnOUuo= +github.com/openziti/edge v0.20.8/go.mod h1:J1ZVM0nOamuLZ/J1jClwVPaJzBrZdhAynJhpCcg4HLU= +github.com/openziti/fabric v0.16.84 h1:4LthxZk7wrHZ8nO6GesI/d89aLlmfAoZWmNlC+4Fxsg= +github.com/openziti/fabric v0.16.84/go.mod h1:2Y6bOG67R9n9XTvnIFz8A+oc+9pX+h6pDmbDZKot1Hk= github.com/openziti/foundation v0.15.65 h1:s8xOYQ271uN+G+PHWXhZ9jOFoqrkeWyB3fY1uxE/OYA= github.com/openziti/foundation v0.15.65/go.mod h1:ksnMJvct0z7WtQk2fdz00vDUfN4SuDEbxgOvIiHysI8= -github.com/openziti/sdk-golang v0.15.71/go.mod h1:qdGTdtVUHQG08gBk9u9YySM76IXXHrOfAFlIPmkul6M= -github.com/openziti/sdk-golang v0.15.72 h1:PLTeCvfslbQ5aWX8t6H2Wt37TfK8ucGZdv0lGjdam2I= -github.com/openziti/sdk-golang v0.15.72/go.mod h1:qdGTdtVUHQG08gBk9u9YySM76IXXHrOfAFlIPmkul6M= +github.com/openziti/sdk-golang v0.15.73 h1:BouaCcjt8DNibq8wB5fkoSceTk9QqtTyw0h2Snp8oYI= +github.com/openziti/sdk-golang v0.15.73/go.mod h1:qdGTdtVUHQG08gBk9u9YySM76IXXHrOfAFlIPmkul6M= github.com/oracle/oci-go-sdk v24.2.0+incompatible/go.mod h1:VQb79nF8Z2cwLkLS35ukwStZIg5F66tcBccjip/j888= github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI= github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 h1:fa50YL1pzKW+1SsBnJDOHppJN9stOEwS+CRWyUtyYGU= diff --git a/ziti-fabric-gw/subcmd/list_sessions.go b/ziti-fabric-gw/subcmd/list_sessions.go index 0d2cd21a6..ccc1c4bb9 100644 --- a/ziti-fabric-gw/subcmd/list_sessions.go +++ b/ziti-fabric-gw/subcmd/list_sessions.go @@ -17,17 +17,17 @@ package subcmd import ( - "github.com/openziti/foundation/channel2" - "github.com/openziti/fabric/pb/mgmt_pb" "encoding/json" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/pb/mgmt_pb" + "github.com/openziti/foundation/channel2" "net/http" "time" ) -func handleListSessions(w http.ResponseWriter, _ *http.Request) { - request := &mgmt_pb.ListSessionsRequest{} +func handleListCircuits(w http.ResponseWriter, _ *http.Request) { + request := &mgmt_pb.ListCircuitsRequest{} body, err := proto.Marshal(request) if err != nil { w.WriteHeader(http.StatusInternalServerError) @@ -35,16 +35,16 @@ func handleListSessions(w http.ResponseWriter, _ *http.Request) { return } - requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_ListSessionsRequestType), body) + requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_ListCircuitsRequestType), body) waitCh, err := mgmtCh.SendAndWait(requestMsg) if err == nil { select { case responseMsg := <-waitCh: - if responseMsg.ContentType == int32(mgmt_pb.ContentType_ListSessionsResponseType) { - response := &mgmt_pb.ListSessionsResponse{} + if responseMsg.ContentType == int32(mgmt_pb.ContentType_ListCircuitsResponseType) { + response := &mgmt_pb.ListCircuitsResponse{} err := proto.Unmarshal(responseMsg.Body, response) if err == nil { - if err := json.NewEncoder(w).Encode(convertSessions(response.Sessions)); err != nil { + if err := json.NewEncoder(w).Encode(convertCircuits(response.Circuits)); err != nil { pfxlog.Logger().Errorf("error encoding json (%s)", err) } @@ -75,17 +75,17 @@ func handleListSessions(w http.ResponseWriter, _ *http.Request) { } } -func convertSessions(in []*mgmt_pb.Session) interface{} { +func convertCircuits(in []*mgmt_pb.Circuit) interface{} { d := &data{} for _, s := range in { - c := &circuit{} - for _, p := range s.Circuit.Path { + c := &path{} + for _, p := range s.Path.Nodes { c.Path = append(c.Path, p) } - for _, l := range s.Circuit.Links { + for _, l := range s.Path.Links { c.Links = append(c.Links, l) } - d.Data = append(d.Data, &session{ + d.Data = append(d.Data, &circuit{ Id: s.Id, ClientId: s.ClientId, ServiceId: s.ServiceId, diff --git a/ziti-fabric-gw/subcmd/messages.go b/ziti-fabric-gw/subcmd/messages.go index e1dde1a40..ccb650a34 100644 --- a/ziti-fabric-gw/subcmd/messages.go +++ b/ziti-fabric-gw/subcmd/messages.go @@ -16,7 +16,7 @@ package subcmd -type circuit struct { +type path struct { Path []string `json:"path"` Links []string `json:"links"` } @@ -50,9 +50,9 @@ type service struct { Egress string `json:"egress"` } -type session struct { - Id string `json:"id"` - ClientId string `json:"clientId"` - ServiceId string `json:"serviceId"` - Circuit *circuit `json:"circuit"` +type circuit struct { + Id string `json:"id"` + ClientId string `json:"clientId"` + ServiceId string `json:"serviceId"` + Circuit *path `json:"circuit"` } diff --git a/ziti-fabric-gw/subcmd/run.go b/ziti-fabric-gw/subcmd/run.go index 24b7770e8..d72153a74 100644 --- a/ziti-fabric-gw/subcmd/run.go +++ b/ziti-fabric-gw/subcmd/run.go @@ -17,12 +17,12 @@ package subcmd import ( + "github.com/gorilla/mux" + "github.com/michaelquigley/pfxlog" "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/common/constants" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" - "github.com/gorilla/mux" - "github.com/michaelquigley/pfxlog" "github.com/rs/cors" "github.com/spf13/cobra" "github.com/urfave/negroni" @@ -112,7 +112,7 @@ func NewRouter() http.Handler { router.HandleFunc("/ctrl/services", handleCreateService).Methods("POST") router.HandleFunc("/ctrl/services/{id}", handleGetService).Methods("GET") router.HandleFunc("/ctrl/services/{id}", handleRemoveService).Methods("DELETE") - router.HandleFunc("/ctrl/sessions", handleListSessions).Methods("GET") + router.HandleFunc("/ctrl/sessions", handleListCircuits).Methods("GET") return router } diff --git a/ziti-fabric/subcmd/list_sessions.go b/ziti-fabric/subcmd/list_circuits.go similarity index 61% rename from ziti-fabric/subcmd/list_sessions.go rename to ziti-fabric/subcmd/list_circuits.go index ee1ab1656..4413cda3e 100644 --- a/ziti-fabric/subcmd/list_sessions.go +++ b/ziti-fabric/subcmd/list_circuits.go @@ -17,56 +17,56 @@ package subcmd import ( - "github.com/openziti/foundation/channel2" - "github.com/openziti/fabric/pb/mgmt_pb" "errors" "fmt" "github.com/golang/protobuf/proto" + "github.com/openziti/fabric/pb/mgmt_pb" + "github.com/openziti/foundation/channel2" "github.com/spf13/cobra" "sort" "time" ) func init() { - listSessionsClient = NewMgmtClient(listSessionsCmd) - listCmd.AddCommand(listSessionsCmd) + listCircuitsClient = NewMgmtClient(listCircuitsCmd) + listCmd.AddCommand(listCircuitsCmd) } -var listSessionsCmd = &cobra.Command{ - Use: "sessions", - Short: "List sessions", - Run: listSessions, +var listCircuitsCmd = &cobra.Command{ + Use: "circuits", + Short: "List circuits", + Run: listCircuits, } -var listSessionsClient *mgmtClient +var listCircuitsClient *mgmtClient -func listSessions(cmd *cobra.Command, args []string) { - if ch, err := listSessionsClient.Connect(); err == nil { - request := &mgmt_pb.ListSessionsRequest{} +func listCircuits(cmd *cobra.Command, args []string) { + if ch, err := listCircuitsClient.Connect(); err == nil { + request := &mgmt_pb.ListCircuitsRequest{} body, err := proto.Marshal(request) if err != nil { panic(err) } - requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_ListSessionsRequestType), body) + requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_ListCircuitsRequestType), body) waitCh, err := ch.SendAndWait(requestMsg) if err != nil { panic(err) } select { case responseMsg := <-waitCh: - if responseMsg.ContentType == int32(mgmt_pb.ContentType_ListSessionsResponseType) { - response := &mgmt_pb.ListSessionsResponse{} + if responseMsg.ContentType == int32(mgmt_pb.ContentType_ListCircuitsResponseType) { + response := &mgmt_pb.ListCircuitsResponse{} err := proto.Unmarshal(responseMsg.Body, response) if err == nil { - out := fmt.Sprintf("\nSessions: (%d)\n\n", len(response.Sessions)) + out := fmt.Sprintf("\nCircuits: (%d)\n\n", len(response.Circuits)) out += fmt.Sprintf("%-12s | %-12s | %-12s | %s\n", "Id", "Client", "Service", "Path") - sort.Slice(response.Sessions, func(i, j int) bool { - if response.Sessions[i].ServiceId == response.Sessions[j].ServiceId { - return response.Sessions[i].Id < response.Sessions[j].Id + sort.Slice(response.Circuits, func(i, j int) bool { + if response.Circuits[i].ServiceId == response.Circuits[j].ServiceId { + return response.Circuits[i].Id < response.Circuits[j].Id } - return response.Sessions[i].ServiceId < response.Sessions[j].ServiceId + return response.Circuits[i].ServiceId < response.Circuits[j].ServiceId }) - for _, s := range response.Sessions { - out += fmt.Sprintf("%-12s | %-12s | %-12s | %s", s.Id, s.ClientId, s.ServiceId, s.Circuit.CalculateDisplayPath()) + for _, s := range response.Circuits { + out += fmt.Sprintf("%-12s | %-12s | %-12s | %s", s.Id, s.ClientId, s.ServiceId, s.Path.CalculateDisplayPath()) } out += "\n" fmt.Print(out) diff --git a/ziti-fabric/subcmd/remove_session.go b/ziti-fabric/subcmd/remove_circuit.go similarity index 70% rename from ziti-fabric/subcmd/remove_session.go rename to ziti-fabric/subcmd/remove_circuit.go index e0ae5c9ee..17e4eecaf 100644 --- a/ziti-fabric/subcmd/remove_session.go +++ b/ziti-fabric/subcmd/remove_circuit.go @@ -17,41 +17,42 @@ package subcmd import ( - "github.com/openziti/foundation/channel2" - "github.com/openziti/fabric/pb/mgmt_pb" "errors" "fmt" "github.com/golang/protobuf/proto" + "github.com/openziti/fabric/pb/mgmt_pb" + "github.com/openziti/foundation/channel2" "github.com/spf13/cobra" "time" ) func init() { - removeSessionClient = NewMgmtClient(removeSessionCmd) - removeSessionCmd.Flags().BoolVar(&removeSessionNow, "now", false, "Remove session now") - removeCmd.AddCommand(removeSessionCmd) + removeCircuitClient = NewMgmtClient(removeCircuitCmd) + removeCircuitCmd.Flags().BoolVar(&removeCircuitNow, "now", false, "Remove circuit now") + removeCmd.AddCommand(removeCircuitCmd) } -var removeSessionCmd = &cobra.Command{ - Use: "session ", - Short: "Remove a session from the fabric", +var removeCircuitCmd = &cobra.Command{ + Use: "circuit ", + Short: "Remove a circuit from the fabric", Args: cobra.ExactArgs(1), - Run: removeSession, + Run: removeCircuit, } -var removeSessionNow bool -var removeSessionClient *mgmtClient -func removeSession(cmd *cobra.Command, args []string) { - if ch, err := removeSessionClient.Connect(); err == nil { - request := &mgmt_pb.RemoveSessionRequest{ - SessionId: args[0], - Now: removeSessionNow, +var removeCircuitNow bool +var removeCircuitClient *mgmtClient + +func removeCircuit(cmd *cobra.Command, args []string) { + if ch, err := removeCircuitClient.Connect(); err == nil { + request := &mgmt_pb.RemoveCircuitRequest{ + CircuitId: args[0], + Now: removeCircuitNow, } body, err := proto.Marshal(request) if err != nil { panic(err) } - requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_RemoveSessionRequestType), body) + requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_RemoveCircuitRequestType), body) waitCh, err := ch.SendAndWait(requestMsg) if err != nil { panic(err) diff --git a/ziti-fabric/subcmd/stream_circuits.go b/ziti-fabric/subcmd/stream_circuits.go new file mode 100644 index 000000000..8fb3f8b33 --- /dev/null +++ b/ziti-fabric/subcmd/stream_circuits.go @@ -0,0 +1,95 @@ +/* + 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 subcmd + +import ( + "errors" + "fmt" + "github.com/golang/protobuf/proto" + "github.com/openziti/fabric/pb/mgmt_pb" + "github.com/openziti/foundation/channel2" + "github.com/spf13/cobra" + "time" +) + +func init() { + streamCircuitsClient = NewMgmtClient(streamCircuitsCmd) + streamCmd.AddCommand(streamCircuitsCmd) +} + +var streamCircuitsCmd = &cobra.Command{ + Use: "circuits", + Short: "Stream circuit data as circuits are created/destroyed", + Run: streamCircuits, +} +var streamCircuitsClient *mgmtClient + +func streamCircuits(*cobra.Command, []string) { + ch, err := streamCircuitsClient.Connect() + if err != nil { + panic(err) + } + + request := &mgmt_pb.StreamCircuitsRequest{} + body, err := proto.Marshal(request) + if err != nil { + panic(err) + } + + ch.AddReceiveHandler(&circuitsHandler{}) + requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_StreamCircuitsRequestType), body) + + waitCh, err := ch.SendAndSync(requestMsg) + if err != nil { + panic(err) + } + select { + case err := <-waitCh: + if err != nil { + panic(err) + } + case <-time.After(5 * time.Second): + panic(errors.New("timeout")) + } + + waitForChannelClose(ch) +} + +type circuitsHandler struct{} + +func (*circuitsHandler) ContentType() int32 { + return int32(mgmt_pb.ContentType_StreamCircuitsEventType) +} + +func (*circuitsHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { + event := &mgmt_pb.StreamCircuitsEvent{} + err := proto.Unmarshal(msg.Body, event) + if err != nil { + panic(err) + } + + eventType := mgmt_pb.StreamCircuitEventType_name[int32(event.EventType)] + if event.EventType == mgmt_pb.StreamCircuitEventType_CircuitDeleted { + fmt.Printf("%v: circuitId: %v\n", eventType, event.CircuitId) + } else if event.EventType == mgmt_pb.StreamCircuitEventType_CircuitCreated { + fmt.Printf("%v: circuitId: %v, clientId: %v, serviceId: %v, path: %v\n", + eventType, event.CircuitId, event.ClientId, event.ServiceId, event.Path.CalculateDisplayPath()) + } else if event.EventType == mgmt_pb.StreamCircuitEventType_PathUpdated { + fmt.Printf("%v: circuitId: %v, path: %v\n", + eventType, event.CircuitId, event.Path.CalculateDisplayPath()) + } +} diff --git a/ziti-fabric/subcmd/stream_sessions.go b/ziti-fabric/subcmd/stream_sessions.go deleted file mode 100644 index bf6eab031..000000000 --- a/ziti-fabric/subcmd/stream_sessions.go +++ /dev/null @@ -1,95 +0,0 @@ -/* - 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 subcmd - -import ( - "github.com/openziti/foundation/channel2" - "github.com/openziti/fabric/pb/mgmt_pb" - "errors" - "fmt" - "github.com/golang/protobuf/proto" - "github.com/spf13/cobra" - "time" -) - -func init() { - streamSessionsClient = NewMgmtClient(streamSessionsCmd) - streamCmd.AddCommand(streamSessionsCmd) -} - -var streamSessionsCmd = &cobra.Command{ - Use: "sessions", - Short: "Stream session data as sessions are created/destroyed", - Run: streamSessions, -} -var streamSessionsClient *mgmtClient - -func streamSessions(cmd *cobra.Command, args []string) { - ch, err := streamSessionsClient.Connect() - if err != nil { - panic(err) - } - - request := &mgmt_pb.StreamSessionsRequest{} - body, err := proto.Marshal(request) - if err != nil { - panic(err) - } - - ch.AddReceiveHandler(&sessionsHandler{}) - requestMsg := channel2.NewMessage(int32(mgmt_pb.ContentType_StreamSessionsRequestType), body) - - waitCh, err := ch.SendAndSync(requestMsg) - if err != nil { - panic(err) - } - select { - case err := <-waitCh: - if err != nil { - panic(err) - } - case <-time.After(5 * time.Second): - panic(errors.New("timeout")) - } - - waitForChannelClose(ch) -} - -type sessionsHandler struct{} - -func (*sessionsHandler) ContentType() int32 { - return int32(mgmt_pb.ContentType_StreamSessionsEventType) -} - -func (*sessionsHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - event := &mgmt_pb.StreamSessionsEvent{} - err := proto.Unmarshal(msg.Body, event) - if err != nil { - panic(err) - } - - eventType := mgmt_pb.StreamSessionEventType_name[int32(event.EventType)] - if event.EventType == mgmt_pb.StreamSessionEventType_SessionDeleted { - fmt.Printf("%v: sessionId: %v\n", eventType, event.SessionId) - } else if event.EventType == mgmt_pb.StreamSessionEventType_SessionCreated { - fmt.Printf("%v: sessionId: %v, clientId: %v, serviceId: %v, path: %v\n", - eventType, event.SessionId, event.ClientId, event.ServiceId, event.Circuit.CalculateDisplayPath()) - } else if event.EventType == mgmt_pb.StreamSessionEventType_CircuitUpdated { - fmt.Printf("%v: sessionId: %v, path: %v\n", - eventType, event.SessionId, event.Circuit.CalculateDisplayPath()) - } -} diff --git a/ziti/cmd/ziti/cmd/ps_router_add_route.go b/ziti/cmd/ziti/cmd/ps_router_add_route.go index d41b93e7a..ef4b677da 100644 --- a/ziti/cmd/ziti/cmd/ps_router_add_route.go +++ b/ziti/cmd/ziti/cmd/ps_router_add_route.go @@ -78,7 +78,7 @@ func (o *PsRouteOptions) Run() error { } route := &ctrl_pb.Route{ - SessionId: o.Args[offset], + CircuitId: o.Args[offset], Forwards: []*ctrl_pb.Route_Forward{ { SrcAddress: o.Args[offset+1],