mirror of
https://github.com/openziti/ziti.git
synced 2026-10-12 23:32:02 +00:00
Merge pull request #453 from openziti/rename-session-to-circuit
Update for session -> circuit changes
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -60,7 +60,7 @@ mgmt:
|
||||
# - type: metrics
|
||||
# sourceFilter: .*
|
||||
# metricFilter: .*egress.*m1_rate*
|
||||
# - type: fabric.sessions
|
||||
# - type: fabric.circuits
|
||||
# include:
|
||||
# - created
|
||||
# - type: edge.sessions
|
||||
|
||||
+16
-10
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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 <sessionId>",
|
||||
Short: "Remove a session from the fabric",
|
||||
var removeCircuitCmd = &cobra.Command{
|
||||
Use: "circuit <circuitId>",
|
||||
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)
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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],
|
||||
|
||||
Reference in New Issue
Block a user